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.compaction;
  
  import java.util.Collection;
  import java.util.concurrent.ExecutionException;
  
  import com.google.common.collect.Iterables;
  import org.junit.BeforeClass;
  import org.junit.Test;
  
  import org.apache.cassandra.SchemaLoader;
  import org.apache.cassandra.Util;
  import org.apache.cassandra.cql3.QueryProcessor;
  import org.apache.cassandra.cql3.UntypedResultSet;
  import org.apache.cassandra.cql3.statements.schema.CreateTableStatement;
  import org.apache.cassandra.db.*;
  import org.apache.cassandra.db.partitions.ImmutableBTreePartition;
  import org.apache.cassandra.db.rows.Row;
  import org.apache.cassandra.db.partitions.PartitionUpdate;
  import org.apache.cassandra.exceptions.ConfigurationException;
  import org.apache.cassandra.io.sstable.format.SSTableReader;
  import org.apache.cassandra.schema.CachingParams;
  import org.apache.cassandra.schema.KeyspaceParams;
  import org.apache.cassandra.utils.ByteBufferUtil;
  import org.apache.cassandra.utils.FBUtilities;
  
  import static org.junit.Assert.*;
  import static org.apache.cassandra.Util.dk;
  
  public class CompactionsPurgeTest
  {
      private static final String KEYSPACE1 = "CompactionsPurgeTest1";
      private static final String CF_STANDARD1 = "Standard1";
      private static final String CF_STANDARD2 = "Standard2";
      private static final String KEYSPACE2 = "CompactionsPurgeTest2";
      private static final String KEYSPACE_CACHED = "CompactionsPurgeTestCached";
      private static final String CF_CACHED = "CachedCF";
      private static final String KEYSPACE_CQL = "cql_keyspace";
      private static final String CF_CQL = "table1";
  
      @BeforeClass
      public static void defineSchema() throws ConfigurationException
      {
          SchemaLoader.prepareServer();
  
          SchemaLoader.createKeyspace(KEYSPACE1,
                                      KeyspaceParams.simple(1),
                                      SchemaLoader.standardCFMD(KEYSPACE1, CF_STANDARD1),
                                      SchemaLoader.standardCFMD(KEYSPACE1, CF_STANDARD2));
  
          SchemaLoader.createKeyspace(KEYSPACE2,
                                      KeyspaceParams.simple(1),
                                      SchemaLoader.standardCFMD(KEYSPACE2, CF_STANDARD1));
  
          SchemaLoader.createKeyspace(KEYSPACE_CACHED,
                                      KeyspaceParams.simple(1),
                                      SchemaLoader.standardCFMD(KEYSPACE_CACHED, CF_CACHED).caching(CachingParams.CACHE_EVERYTHING));
  
          SchemaLoader.createKeyspace(KEYSPACE_CQL,
                                      KeyspaceParams.simple(1),
                                      CreateTableStatement.parse("CREATE TABLE " + CF_CQL + " ("
                                                                 + "k int PRIMARY KEY,"
                                                                 + "v1 text,"
                                                                 + "v2 int"
                                                                 + ")", KEYSPACE_CQL));
      }
  
      @Test
      public void testMajorCompactionPurge()
      {
          CompactionManager.instance.disableAutoCompaction();
  
          Keyspace keyspace = Keyspace.open(KEYSPACE1);
          String cfName = "Standard1";
          ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(cfName);
  
          String key = "key1";
  
          // inserts
          for (int i = 0; i < 10; i++)
          {
              RowUpdateBuilder builder = new RowUpdateBuilder(cfs.metadata(), 0, key);
              builder.clustering(String.valueOf(i))
                     .add("val", ByteBufferUtil.EMPTY_BYTE_BUFFER)
                     .build().applyUnsafe();
          }
  
          Util.flush(cfs);
  
          // deletes
          for (int i = 0; i < 10; i++)
          {
              RowUpdateBuilder.deleteRow(cfs.metadata(), 1, key, String.valueOf(i)).applyUnsafe();
          }
          Util.flush(cfs);
  
          // resurrect one column
          RowUpdateBuilder builder = new RowUpdateBuilder(cfs.metadata(), 2, key);
          builder.clustering(String.valueOf(5))
                 .add("val", ByteBufferUtil.EMPTY_BYTE_BUFFER)
                 .build().applyUnsafe();
  
          Util.flush(cfs);
  
          // major compact and test that all columns but the resurrected one is completely gone
          FBUtilities.waitOnFutures(CompactionManager.instance.submitMaximal(cfs, Long.MAX_VALUE, false, 0));
          cfs.invalidateCachedPartition(dk(key));
  
          ImmutableBTreePartition partition = Util.getOnlyPartitionUnfiltered(Util.cmd(cfs, key).build());
          assertEquals(1, partition.rowCount());
      }
  
      @Test
      public void testMajorCompactionPurgeTombstonesWithMaxTimestamp()
      {
          CompactionManager.instance.disableAutoCompaction();
  
          Keyspace keyspace = Keyspace.open(KEYSPACE1);
          String cfName = "Standard1";
          ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(cfName);
  
          String key = "key1";
  
          // inserts
          for (int i = 0; i < 10; i++)
          {
              RowUpdateBuilder builder = new RowUpdateBuilder(cfs.metadata(), 0, key);
              builder.clustering(String.valueOf(i))
                     .add("val", ByteBufferUtil.EMPTY_BYTE_BUFFER)
                     .build().applyUnsafe();
          }
          Util.flush(cfs);
  
          // deletes
          for (int i = 0; i < 10; i++)
          {
              RowUpdateBuilder.deleteRow(cfs.metadata(), Long.MAX_VALUE, key, String.valueOf(i)).applyUnsafe();
          }
          Util.flush(cfs);
  
          // major compact - tombstones should be purged
          FBUtilities.waitOnFutures(CompactionManager.instance.submitMaximal(cfs, Long.MAX_VALUE, false, 0));
  
          // resurrect one column
          RowUpdateBuilder builder = new RowUpdateBuilder(cfs.metadata(), 2, key);
          builder.clustering(String.valueOf(5))
                 .add("val", ByteBufferUtil.EMPTY_BYTE_BUFFER)
                 .build().applyUnsafe();
  
          Util.flush(cfs);
  
          cfs.invalidateCachedPartition(dk(key));
  
          ImmutableBTreePartition partition = Util.getOnlyPartitionUnfiltered(Util.cmd(cfs, key).build());
          assertEquals(1, partition.rowCount());
      }
  
      @Test
      public void testMajorCompactionPurgeTopLevelTombstoneWithMaxTimestamp()
      {
          CompactionManager.instance.disableAutoCompaction();
  
          Keyspace keyspace = Keyspace.open(KEYSPACE1);
          String cfName = "Standard1";
          ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(cfName);
  
          String key = "key1";
  
          // inserts
          for (int i = 0; i < 10; i++)
          {
              RowUpdateBuilder builder = new RowUpdateBuilder(cfs.metadata(), 0, key);
              builder.clustering(String.valueOf(i))
                     .add("val", ByteBufferUtil.EMPTY_BYTE_BUFFER)
                     .build().applyUnsafe();
          }
          Util.flush(cfs);
  
          new Mutation.PartitionUpdateCollector(KEYSPACE1, dk(key))
              .add(PartitionUpdate.fullPartitionDelete(cfs.metadata(), dk(key), Long.MAX_VALUE, FBUtilities.nowInSeconds()))
              .build()
              .applyUnsafe();
          Util.flush(cfs);
  
          // major compact - tombstones should be purged
          FBUtilities.waitOnFutures(CompactionManager.instance.submitMaximal(cfs, Long.MAX_VALUE, false, 0));
  
          // resurrect one column
          RowUpdateBuilder builder = new RowUpdateBuilder(cfs.metadata(), 2, key);
          builder.clustering(String.valueOf(5))
                 .add("val", ByteBufferUtil.EMPTY_BYTE_BUFFER)
                 .build().applyUnsafe();
  
          Util.flush(cfs);
  
          cfs.invalidateCachedPartition(dk(key));
  
          ImmutableBTreePartition partition = Util.getOnlyPartitionUnfiltered(Util.cmd(cfs, key).build());
          assertEquals(1, partition.rowCount());
      }
  
      @Test
      public void testMajorCompactionPurgeRangeTombstoneWithMaxTimestamp()
      {
          CompactionManager.instance.disableAutoCompaction();
  
          Keyspace keyspace = Keyspace.open(KEYSPACE1);
          String cfName = "Standard1";
          ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(cfName);
  
          String key = "key1";
  
          // inserts
          for (int i = 0; i < 10; i++)
          {
              RowUpdateBuilder builder = new RowUpdateBuilder(cfs.metadata(), 0, key);
              builder.clustering(String.valueOf(i))
                     .add("val", ByteBufferUtil.EMPTY_BYTE_BUFFER)
                     .build().applyUnsafe();
          }
          Util.flush(cfs);
  
          new RowUpdateBuilder(cfs.metadata(), Long.MAX_VALUE, dk(key))
              .addRangeTombstone(String.valueOf(0), String.valueOf(9)).build().applyUnsafe();
          Util.flush(cfs);
  
          // major compact - tombstones should be purged
          FBUtilities.waitOnFutures(CompactionManager.instance.submitMaximal(cfs, Long.MAX_VALUE, false, 0));
  
          // resurrect one column
          RowUpdateBuilder builder = new RowUpdateBuilder(cfs.metadata(), 2, key);
          builder.clustering(String.valueOf(5))
                 .add("val", ByteBufferUtil.EMPTY_BYTE_BUFFER)
                 .build().applyUnsafe();
  
          Util.flush(cfs);
  
          cfs.invalidateCachedPartition(dk(key));
  
          ImmutableBTreePartition partition = Util.getOnlyPartitionUnfiltered(Util.cmd(cfs, key).build());
          assertEquals(1, partition.rowCount());
      }
  
      @Test
      public void testMinorCompactionPurge()
      {
          CompactionManager.instance.disableAutoCompaction();
  
          Keyspace keyspace = Keyspace.open(KEYSPACE2);
          String cfName = "Standard1";
          ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(cfName);
  
          for (int k = 1; k <= 2; ++k) {
              String key = "key" + k;
  
              // inserts
              for (int i = 0; i < 10; i++)
              {
                  RowUpdateBuilder builder = new RowUpdateBuilder(cfs.metadata(), 0, key);
                  builder.clustering(String.valueOf(i))
                          .add("val", ByteBufferUtil.EMPTY_BYTE_BUFFER)
                          .build().applyUnsafe();
              }
              Util.flush(cfs);
  
              // deletes
              for (int i = 0; i < 10; i++)
              {
                  RowUpdateBuilder.deleteRow(cfs.metadata(), 1, key, String.valueOf(i)).applyUnsafe();
              }
  
              Util.flush(cfs);
          }
  
          DecoratedKey key1 = Util.dk("key1");
          DecoratedKey key2 = Util.dk("key2");
  
          // flush, remember the current sstable and then resurrect one column
          // for first key. Then submit minor compaction on remembered sstables.
          Util.flush(cfs);
          Collection<SSTableReader> sstablesIncomplete = cfs.getLiveSSTables();
  
          RowUpdateBuilder builder = new RowUpdateBuilder(cfs.metadata(), 2, "key1");
          builder.clustering(String.valueOf(5))
                  .add("val", ByteBufferUtil.EMPTY_BYTE_BUFFER)
                  .build().applyUnsafe();
  
          Util.flush(cfs);
          try (CompactionTasks tasks = cfs.getCompactionStrategyManager().getUserDefinedTasks(sstablesIncomplete, Integer.MAX_VALUE))
          {
              Iterables.getOnlyElement(tasks).execute(ActiveCompactionsTracker.NOOP);
          }
  
          // verify that minor compaction does GC when key is provably not
          // present in a non-compacted sstable
          Util.assertEmpty(Util.cmd(cfs, key2).build());
  
          // verify that minor compaction still GC when key is present
          // in a non-compacted sstable but the timestamp ensures we won't miss anything
          ImmutableBTreePartition partition = Util.getOnlyPartitionUnfiltered(Util.cmd(cfs, key1).build());
          assertEquals(1, partition.rowCount());
      }
  
      /**
       * verify that we don't drop tombstones during a minor compaction that might still be relevant
       */
      @Test
      public void testMinTimestampPurge()
      {
          CompactionManager.instance.disableAutoCompaction();
  
          Keyspace keyspace = Keyspace.open(KEYSPACE2);
          String cfName = "Standard1";
          ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(cfName);
          final boolean enforceStrictLiveness = cfs.metadata().enforceStrictLiveness();
          String key3 = "key3";
  
          // inserts
          new RowUpdateBuilder(cfs.metadata(), 8, key3)
              .clustering("c1")
              .add("val", ByteBufferUtil.EMPTY_BYTE_BUFFER)
              .build().applyUnsafe();
  
          new RowUpdateBuilder(cfs.metadata(), 8, key3)
          .clustering("c2")
          .add("val", ByteBufferUtil.EMPTY_BYTE_BUFFER)
          .build().applyUnsafe();
  
          Util.flush(cfs);
          // delete c1
          RowUpdateBuilder.deleteRow(cfs.metadata(), 10, key3, "c1").applyUnsafe();
  
          Util.flush(cfs);
          Collection<SSTableReader> sstablesIncomplete = cfs.getLiveSSTables();
  
          // delete c2 so we have new delete in a diffrent SSTable
          RowUpdateBuilder.deleteRow(cfs.metadata(), 9, key3, "c2").applyUnsafe();
          Util.flush(cfs);
  
          // compact the sstables with the c1/c2 data and the c1 tombstone
          try (CompactionTasks tasks = cfs.getCompactionStrategyManager().getUserDefinedTasks(sstablesIncomplete, Integer.MAX_VALUE))
          {
              Iterables.getOnlyElement(tasks).execute(ActiveCompactionsTracker.NOOP);
          }
  
          // We should have both the c1 and c2 tombstones still. Since the min timestamp in the c2 tombstone
          // sstable is older than the c1 tombstone, it is invalid to throw out the c1 tombstone.
          ImmutableBTreePartition partition = Util.getOnlyPartitionUnfiltered(Util.cmd(cfs, key3).build());
          assertEquals(2, partition.rowCount());
          for (Row row : partition)
              assertFalse(row.hasLiveData(FBUtilities.nowInSeconds(), enforceStrictLiveness));
      }
  
      @Test
      public void testCompactionPurgeOneFile() throws ExecutionException, InterruptedException
      {
          CompactionManager.instance.disableAutoCompaction();
  
          Keyspace keyspace = Keyspace.open(KEYSPACE1);
          String cfName = "Standard2";
          ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(cfName);
  
          String key = "key1";
  
          // inserts
          for (int i = 0; i < 5; i++)
          {
              RowUpdateBuilder builder = new RowUpdateBuilder(cfs.metadata(), 0, key);
              builder.clustering(String.valueOf(i))
                     .add("val", ByteBufferUtil.EMPTY_BYTE_BUFFER)
                     .build().applyUnsafe();
          }
  
          // deletes
          for (int i = 0; i < 5; i++)
          {
              RowUpdateBuilder.deleteRow(cfs.metadata(), 1, key, String.valueOf(i)).applyUnsafe();
          }
          Util.flush(cfs);
          assertEquals(String.valueOf(cfs.getLiveSSTables()), 1, cfs.getLiveSSTables().size()); // inserts & deletes were in the same memtable -> only deletes in sstable
  
          // compact and test that the row is completely gone
          Util.compactAll(cfs, Integer.MAX_VALUE).get();
          assertTrue(cfs.getLiveSSTables().isEmpty());
  
          Util.assertEmpty(Util.cmd(cfs, key).build());
      }
  
  
      @Test
      public void testCompactionPurgeCachedRow() throws ExecutionException, InterruptedException
      {
          CompactionManager.instance.disableAutoCompaction();
  
          String keyspaceName = KEYSPACE_CACHED;
          String cfName = CF_CACHED;
          Keyspace keyspace = Keyspace.open(keyspaceName);
          ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(cfName);
  
          String key = "key3";
  
          // inserts
          for (int i = 0; i < 10; i++)
          {
              RowUpdateBuilder builder = new RowUpdateBuilder(cfs.metadata(), 0, key);
              builder.clustering(String.valueOf(i))
                     .add("val", ByteBufferUtil.EMPTY_BYTE_BUFFER)
                     .build().applyUnsafe();
          }
  
          // deletes partition
          Mutation.PartitionUpdateCollector rm = new Mutation.PartitionUpdateCollector(KEYSPACE_CACHED, dk(key));
          rm.add(PartitionUpdate.fullPartitionDelete(cfs.metadata(), dk(key), 1, FBUtilities.nowInSeconds()));
          rm.build().applyUnsafe();
  
          // Adds another unrelated partition so that the sstable is not considered fully expired. We do not
          // invalidate the row cache in that latter case.
          new RowUpdateBuilder(cfs.metadata(), 0, "key4").clustering("c").add("val", ByteBufferUtil.EMPTY_BYTE_BUFFER).build().applyUnsafe();
  
          // move the key up in row cache (it should not be empty since we have the partition deletion info)
          assertFalse(Util.getOnlyPartitionUnfiltered(Util.cmd(cfs, key).build()).isEmpty());
  
          // flush and major compact
          Util.flush(cfs);
          Util.compactAll(cfs, Integer.MAX_VALUE).get();
  
          // Since we've force purging (by passing MAX_VALUE for gc_before), the row should have been invalidated and we should have no deletion info anymore
          Util.assertEmpty(Util.cmd(cfs, key).build());
      }
  
      @Test
      public void testCompactionPurgeTombstonedRow() throws ExecutionException, InterruptedException
      {
          CompactionManager.instance.disableAutoCompaction();
  
          String keyspaceName = KEYSPACE1;
          String cfName = "Standard1";
          Keyspace keyspace = Keyspace.open(keyspaceName);
          ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(cfName);
          String key = "key3";
  
          // inserts
          for (int i = 0; i < 10; i++)
          {
              RowUpdateBuilder builder = new RowUpdateBuilder(cfs.metadata(), i, key);
              builder.clustering(String.valueOf(i))
                     .add("val", ByteBufferUtil.EMPTY_BYTE_BUFFER)
                     .build().applyUnsafe();
          }
  
          // deletes partition with timestamp such that not all columns are deleted
          Mutation.PartitionUpdateCollector rm = new Mutation.PartitionUpdateCollector(KEYSPACE1, dk(key));
          rm.add(PartitionUpdate.fullPartitionDelete(cfs.metadata(), dk(key), 4, FBUtilities.nowInSeconds()));
          rm.build().applyUnsafe();
  
          ImmutableBTreePartition partition = Util.getOnlyPartitionUnfiltered(Util.cmd(cfs, key).build());
          assertFalse(partition.partitionLevelDeletion().isLive());
  
          // flush and major compact (with tombstone purging)
          Util.flush(cfs);
          Util.compactAll(cfs, Integer.MAX_VALUE).get();
          assertFalse(Util.getOnlyPartitionUnfiltered(Util.cmd(cfs, key).build()).isEmpty());
  
          // re-inserts with timestamp lower than delete
          for (int i = 0; i < 5; i++)
          {
              RowUpdateBuilder builder = new RowUpdateBuilder(cfs.metadata(), i, key);
              builder.clustering(String.valueOf(i))
                     .add("val", ByteBufferUtil.EMPTY_BYTE_BUFFER)
                     .build().applyUnsafe();
          }
  
          // Check that the second insert went in
          partition = Util.getOnlyPartitionUnfiltered(Util.cmd(cfs, key).build());
          assertEquals(10, partition.rowCount());
      }
  
  
      @Test
      public void testRowTombstoneObservedBeforePurging()
      {
          String keyspace = "cql_keyspace";
          String table = "table1";
          ColumnFamilyStore cfs = Keyspace.open(keyspace).getColumnFamilyStore(table);
          cfs.disableAutoCompaction();
  
          // write a row out to one sstable
          QueryProcessor.executeInternal(String.format("INSERT INTO %s.%s (k, v1, v2) VALUES (%d, '%s', %d)",
                                                       keyspace, table, 1, "foo", 1));
          Util.flush(cfs);
  
          UntypedResultSet result = QueryProcessor.executeInternal(String.format("SELECT * FROM %s.%s WHERE k = %d", keyspace, table, 1));
          assertEquals(1, result.size());
  
          // write a row tombstone out to a second sstable
          QueryProcessor.executeInternal(String.format("DELETE FROM %s.%s WHERE k = %d", keyspace, table, 1));
          Util.flush(cfs);
  
          // basic check that the row is considered deleted
          assertEquals(2, cfs.getLiveSSTables().size());
          result = QueryProcessor.executeInternal(String.format("SELECT * FROM %s.%s WHERE k = %d", keyspace, table, 1));
          assertEquals(0, result.size());
  
          // compact the two sstables with a gcBefore that does *not* allow the row tombstone to be purged
          FBUtilities.waitOnFutures(CompactionManager.instance.submitMaximal(cfs, (int) (System.currentTimeMillis() / 1000) - 10000, false, 0));
  
          // the data should be gone, but the tombstone should still exist
          assertEquals(1, cfs.getLiveSSTables().size());
          result = QueryProcessor.executeInternal(String.format("SELECT * FROM %s.%s WHERE k = %d", keyspace, table, 1));
          assertEquals(0, result.size());
  
          // write a row out to one sstable
          QueryProcessor.executeInternal(String.format("INSERT INTO %s.%s (k, v1, v2) VALUES (%d, '%s', %d)",
                                                       keyspace, table, 1, "foo", 1));
          Util.flush(cfs);
          assertEquals(2, cfs.getLiveSSTables().size());
          result = QueryProcessor.executeInternal(String.format("SELECT * FROM %s.%s WHERE k = %d", keyspace, table, 1));
          assertEquals(1, result.size());
  
          // write a row tombstone out to a different sstable
          QueryProcessor.executeInternal(String.format("DELETE FROM %s.%s WHERE k = %d", keyspace, table, 1));
          Util.flush(cfs);
  
          // compact the two sstables with a gcBefore that *does* allow the row tombstone to be purged
          FBUtilities.waitOnFutures(CompactionManager.instance.submitMaximal(cfs, (int) (System.currentTimeMillis() / 1000) + 10000, false, 0));
  
          // both the data and the tombstone should be gone this time
          assertEquals(0, cfs.getLiveSSTables().size());
          result = QueryProcessor.executeInternal(String.format("SELECT * FROM %s.%s WHERE k = %d", keyspace, table, 1));
          assertEquals(0, result.size());
      }
  }
  