content: /*
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
  * regarding copyright ownership.  The ASF licenses this file
  * to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
  *
  *    http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing,
  * software distributed under the License is distributed on an
  * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
  * KIND, either express or implied.  See the License for the
  * specific language governing permissions and limitations
  * under the License.
  */
  package org.apache.cassandra.db.lifecycle;
  
  import java.util.ArrayList;
  import java.util.Collections;
  import java.util.List;
  import java.util.Set;
  
  import com.google.common.base.Function;
  import com.google.common.base.Predicates;
  import com.google.common.collect.ImmutableList;
  import com.google.common.collect.ImmutableSet;
  import com.google.common.collect.Iterables;
  import org.junit.Assert;
  import org.junit.BeforeClass;
  import org.junit.Test;
  
  import org.apache.cassandra.ServerTestUtils;
  import org.apache.cassandra.db.ColumnFamilyStore;
  import org.apache.cassandra.db.PartitionPosition;
  import org.apache.cassandra.db.commitlog.CommitLog;
  import org.apache.cassandra.db.memtable.Memtable;
  import org.apache.cassandra.dht.AbstractBounds;
  import org.apache.cassandra.io.sstable.format.SSTableReader;
  import org.apache.cassandra.schema.MockSchema;
  
  import static com.google.common.collect.ImmutableSet.copyOf;
  import static com.google.common.collect.ImmutableSet.of;
  import static com.google.common.collect.Iterables.concat;
  import static java.util.Collections.singleton;
  import static org.apache.cassandra.db.lifecycle.Helpers.emptySet;
  import static org.apache.cassandra.db.lifecycle.SSTableIntervalTree.buildSSTableIntervalTree;
  
  public class ViewTest
  {
      @BeforeClass
      public static void setUp()
      {
          ServerTestUtils.prepareServerNoRegister();
          CommitLog.instance.start();
      }
  
      @Test
      public void testSSTablesInBounds()
      {
          ColumnFamilyStore cfs = MockSchema.newCFS();
          View initialView = fakeView(0, 5, cfs);
          for (int i = 0 ; i < 5 ; i++)
          {
              for (int j = i ; j < 5 ; j++)
              {
                  PartitionPosition min = MockSchema.readerBounds(i);
                  PartitionPosition max = MockSchema.readerBounds(j);
                  for (boolean minInc : new boolean[] { true })//, false} )
                  {
                      for (boolean maxInc : new boolean[] { true })//, false} )
                      {
                          if (i == j && !(minInc && maxInc))
                              continue;
  
                          AbstractBounds<PartitionPosition> bounds = AbstractBounds.bounds(min, minInc, max, maxInc);
                          List<SSTableReader> r = ImmutableList.copyOf(initialView.liveSSTablesInBounds(bounds.left, bounds.right));
                          Assert.assertEquals(String.format("%d(%s) %d(%s)", i, minInc, j, maxInc), j - i + (minInc ? 0 : -1) + (maxInc ? 1 : 0), r.size());
                      }
                  }
              }
          }
      }
  
      @Test
      public void testCompaction()
      {
          ColumnFamilyStore cfs = MockSchema.newCFS();
          View initialView = fakeView(0, 5, cfs, true);
          View cur = initialView;
          List<SSTableReader> readers = ImmutableList.copyOf(initialView.sstables);
          Assert.assertTrue(View.permitCompacting(readers).apply(cur));
          // check we permit compacting duplicates in the predicate, so we don't spin infinitely if there is a screw up
          Assert.assertTrue(View.permitCompacting(ImmutableList.copyOf(concat(readers, readers))).apply(cur));
          // check we fail in the application in the presence of duplicates
          testFailure(View.updateCompacting(emptySet(), concat(readers.subList(0, 1), readers.subList(0, 1))), cur);
  
          // do lots of trivial checks that the compacting set and related methods behave properly for a simple update
          cur = View.updateCompacting(emptySet(), readers.subList(0, 2)).apply(cur);
          Assert.assertTrue(View.permitCompacting(readers.subList(2, 5)).apply(cur));
          Assert.assertFalse(View.permitCompacting(readers.subList(0, 2)).apply(cur));
          Assert.assertFalse(View.permitCompacting(readers.subList(0, 1)).apply(cur));
          Assert.assertFalse(View.permitCompacting(readers.subList(1, 2)).apply(cur));
          Assert.assertTrue(readers.subList(2, 5).containsAll(copyOf(cur.getUncompacting(readers))));
          Assert.assertEquals(3, copyOf(cur.getUncompacting(readers)).size());
          Assert.assertTrue(ImmutableSet.copyOf(cur.select(SSTableSet.NONCOMPACTING)).containsAll(readers.subList(2, 5)));
          Assert.assertEquals(3, ImmutableSet.copyOf(cur.select(SSTableSet.NONCOMPACTING)).size());
  
          // check marking already compacting readers fails with an exception
          testFailure(View.updateCompacting(emptySet(), readers.subList(0, 1)), cur);
          testFailure(View.updateCompacting(emptySet(), readers.subList(1, 2)), cur);
          testFailure(View.updateCompacting(copyOf(readers.subList(0, 1)), readers.subList(1, 2)), cur);
  
          // make equivalents of readers.subList(0, 3) that are different instances
          SSTableReader r0 = MockSchema.sstable(0, cfs), r1 = MockSchema.sstable(1, cfs), r2 = MockSchema.sstable(2, cfs);
          // attempt to mark compacting a version not in the live set
          testFailure(View.updateCompacting(emptySet(), of(r2)), cur);
          // update one compacting, one non-compacting, of the liveset to another instance of the same readers;
          // confirm liveset changes but compacting does not
          cur = View.updateLiveSet(copyOf(readers.subList(1, 3)), of(r1, r2), cfs.metric.viewSSTableIntervalTree).apply(cur);
          Assert.assertSame(readers.get(0), cur.sstablesMap.get(r0));
          Assert.assertSame(r1, cur.sstablesMap.get(r1));
          Assert.assertSame(r2, cur.sstablesMap.get(r2));
          testFailure(View.updateCompacting(emptySet(), readers.subList(2, 3)), cur);
          Assert.assertSame(readers.get(1), Iterables.getFirst(Iterables.filter(cur.compacting, Predicates.equalTo(r1)), null));
  
          // unmark compacting, and check our methods are all correctly updated
          cur = View.updateCompacting(copyOf(readers.subList(0, 1)), emptySet()).apply(cur);
          Assert.assertTrue(View.permitCompacting(concat(readers.subList(0, 1), of(r2), readers.subList(3, 5))).apply(cur));
          Assert.assertFalse(View.permitCompacting(readers.subList(1, 2)).apply(cur));
          testFailure(View.updateCompacting(emptySet(), readers.subList(1, 2)), cur);
          testFailure(View.updateCompacting(copyOf(readers.subList(0, 2)), emptySet()), cur);
          Assert.assertTrue(copyOf(concat(readers.subList(0, 1), readers.subList(2, 5))).containsAll(copyOf(cur.getUncompacting(readers))));
          Assert.assertEquals(4, copyOf(cur.getUncompacting(readers)).size());
          Set<SSTableReader> nonCompacting = ImmutableSet.copyOf(cur.select(SSTableSet.NONCOMPACTING));
          Assert.assertTrue(nonCompacting.containsAll(readers.subList(2, 5)));
          Assert.assertTrue(nonCompacting.containsAll(readers.subList(0, 1)));
          Assert.assertEquals(4, nonCompacting.size());
  
          for (SSTableReader sstable : initialView.sstables)
              sstable.selfRef().release();
      }
  
      private static void testFailure(Function<View, ?> function, View view)
      {
          boolean failed = true;
          try
          {
              function.apply(view);
              failed = false;
          }
          catch (Throwable t)
          {
          }
          Assert.assertTrue(failed);
      }
  
      @Test
      public void testFlushing()
      {
          ColumnFamilyStore cfs = MockSchema.newCFS();
          View initialView = fakeView(1, 0, cfs);
          View cur = initialView;
          Memtable memtable1 = initialView.getCurrentMemtable();
          Memtable memtable2 = MockSchema.memtable(cfs);
  
          cur = View.switchMemtable(memtable2).apply(cur);
          Assert.assertEquals(2, cur.liveMemtables.size());
          Assert.assertEquals(memtable1, cur.liveMemtables.get(0));
          Assert.assertEquals(memtable2, cur.getCurrentMemtable());
  
          Memtable memtable3 = MockSchema.memtable(cfs);
          cur = View.switchMemtable(memtable3).apply(cur);
          Assert.assertEquals(3, cur.liveMemtables.size());
          Assert.assertEquals(0, cur.flushingMemtables.size());
          Assert.assertEquals(memtable1, cur.liveMemtables.get(0));
          Assert.assertEquals(memtable2, cur.liveMemtables.get(1));
          Assert.assertEquals(memtable3, cur.getCurrentMemtable());
  
          testFailure(View.replaceFlushed(memtable2, null, cfs.metric.viewSSTableIntervalTree), cur);
  
          cur = View.markFlushing(memtable2).apply(cur);
          Assert.assertTrue(cur.flushingMemtables.contains(memtable2));
          Assert.assertEquals(2, cur.liveMemtables.size());
          Assert.assertEquals(1, cur.flushingMemtables.size());
          Assert.assertEquals(memtable2, cur.flushingMemtables.get(0));
          Assert.assertEquals(memtable1, cur.liveMemtables.get(0));
          Assert.assertEquals(memtable3, cur.getCurrentMemtable());
  
          cur = View.markFlushing(memtable1).apply(cur);
          Assert.assertEquals(1, cur.liveMemtables.size());
          Assert.assertEquals(2, cur.flushingMemtables.size());
          Assert.assertEquals(memtable1, cur.flushingMemtables.get(0));
          Assert.assertEquals(memtable2, cur.flushingMemtables.get(1));
          Assert.assertEquals(memtable3, cur.getCurrentMemtable());
  
          cur = View.replaceFlushed(memtable2, null, cfs.metric.viewSSTableIntervalTree).apply(cur);
          Assert.assertEquals(1, cur.liveMemtables.size());
          Assert.assertEquals(1, cur.flushingMemtables.size());
          Assert.assertEquals(memtable1, cur.flushingMemtables.get(0));
          Assert.assertEquals(memtable3, cur.getCurrentMemtable());
  
          SSTableReader sstable = MockSchema.sstable(1, cfs);
          cur = View.replaceFlushed(memtable1, singleton(sstable), cfs.metric.viewSSTableIntervalTree).apply(cur);
          Assert.assertEquals(0, cur.flushingMemtables.size());
          Assert.assertEquals(1, cur.liveMemtables.size());
          Assert.assertEquals(memtable3, cur.getCurrentMemtable());
          Assert.assertEquals(1, cur.sstables.size());
          Assert.assertEquals(sstable, cur.sstablesMap.get(sstable));
      }
  
      static View fakeView(int memtableCount, int sstableCount, ColumnFamilyStore cfs)
      {
          return fakeView(memtableCount, sstableCount, cfs, false);
      }
  
      static View fakeView(int memtableCount, int sstableCount, ColumnFamilyStore cfs, boolean keepRef)
      {
          List<Memtable> memtables = new ArrayList<>();
          List<SSTableReader> sstables = new ArrayList<>();
          for (int i = 0 ; i < memtableCount ; i++)
              memtables.add(MockSchema.memtable(cfs));
          for (int i = 0 ; i < sstableCount ; i++)
              sstables.add(MockSchema.sstable(i, keepRef, cfs));
          return new View(ImmutableList.copyOf(memtables), Collections.<Memtable>emptyList(), Helpers.identityMap(sstables),
                          Collections.<SSTableReader, SSTableReader>emptyMap(), buildSSTableIntervalTree(sstables));
      }
  }
  