diff --git a/CHANGES.txt b/CHANGES.txt index aeef6ebc6c..a524728bc4 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -298,6 +298,7 @@ Merged from 4.1: * Enforce CQL message size limit on multiframe messages (CASSANDRA-20052) * Fix race condition in DecayingEstimatedHistogramReservoir during rescale (CASSANDRA-19365) Merged from 4.0: + * Make secondary index implementations notified about rows in fully expired SSTables in compaction (CASSANDRA-20829) * Ensure prepared_statement INSERT timestamp precedes eviction DELETE (CASSANDRA-19703) * Gossip doesn't converge due to race condition when updating EndpointStates multiple fields (CASSANDRA-20659) * Handle sstable metadata stats file getting a new mtime after compaction has finished (CASSANDRA-18119) diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionTask.java b/src/java/org/apache/cassandra/db/compaction/CompactionTask.java index eb331dda8f..1f9e8ea403 100644 --- a/src/java/org/apache/cassandra/db/compaction/CompactionTask.java +++ b/src/java/org/apache/cassandra/db/compaction/CompactionTask.java @@ -17,6 +17,7 @@ */ package org.apache.cassandra.db.compaction; +import java.util.ArrayList; import java.util.Collection; import java.util.HashMap; import java.util.HashSet; @@ -32,6 +33,8 @@ import com.google.common.collect.Iterables; import com.google.common.util.concurrent.RateLimiter; import org.apache.cassandra.db.compaction.unified.UnifiedCompactionTask; +import org.apache.cassandra.db.rows.Unfiltered; +import org.apache.cassandra.db.rows.UnfilteredRowIterator; import org.apache.cassandra.dht.Range; import org.apache.cassandra.dht.Token; import org.apache.commons.lang3.StringUtils; @@ -42,9 +45,14 @@ import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.db.ColumnFamilyStore; import org.apache.cassandra.db.Directories; import org.apache.cassandra.db.SystemKeyspace; +import org.apache.cassandra.db.WriteContext; import org.apache.cassandra.db.compaction.writers.CompactionAwareWriter; import org.apache.cassandra.db.compaction.writers.DefaultCompactionWriter; import org.apache.cassandra.db.lifecycle.ILifecycleTransaction; +import org.apache.cassandra.db.rows.Row; +import org.apache.cassandra.index.Index; +import org.apache.cassandra.index.transactions.IndexTransaction; +import org.apache.cassandra.io.sstable.ISSTableScanner; import org.apache.cassandra.io.sstable.format.SSTableReader; import org.apache.cassandra.io.sstable.metadata.MetadataCollector; import org.apache.cassandra.io.util.File; @@ -172,6 +180,9 @@ public class CompactionTask extends AbstractCompactionTask Set actuallyCompact = new HashSet<>(inputSSTables()); final Set fullyExpiredSSTables = controller.getFullyExpiredSSTables(); + + maybeNotifyIndexersAboutRowsInFullyExpiredSSTables(fullyExpiredSSTables); + if (!fullyExpiredSSTables.isEmpty()) { logger.debug("Compaction {} dropping expired sstables: {}", transaction.opIdString(), fullyExpiredSSTables); @@ -516,4 +527,67 @@ public class CompactionTask extends AbstractCompactionTask } return max; } + + private void maybeNotifyIndexersAboutRowsInFullyExpiredSSTables(Set fullyExpiredSSTables) + { + if (fullyExpiredSSTables.isEmpty()) + return; + + List indexes = new ArrayList<>(); + for (Index index : cfs.indexManager.listIndexes()) + { + if (index.notifyIndexerAboutRowsInFullyExpiredSSTables()) + indexes.add(index); + } + + if (indexes.isEmpty()) + return; + + for (SSTableReader expiredSSTable : fullyExpiredSSTables) + { + try (ISSTableScanner scanner = expiredSSTable.getScanner()) + { + while (scanner.hasNext()) + { + UnfilteredRowIterator partition = scanner.next(); + + try (WriteContext ctx = cfs.keyspace.getWriteHandler().createContextForIndexing()) + { + List indexers = new ArrayList<>(); + for (int i = 0; i < indexes.size(); i++) + { + Index.Indexer indexer = indexes.get(i).indexerFor(partition.partitionKey(), + partition.columns(), + FBUtilities.nowInSeconds(), + ctx, + IndexTransaction.Type.COMPACTION, + null); + + if (indexer != null) + indexers.add(indexer); + } + + if (!indexers.isEmpty()) + { + for (Index.Indexer indexer : indexers) + indexer.begin(); + + while (partition.hasNext()) + { + Unfiltered unfiltered = partition.next(); + if (unfiltered instanceof Row) + { + for (Index.Indexer indexer : indexers) + indexer.removeRow((Row) unfiltered); + } + } + + for (Index.Indexer indexer : indexers) + indexer.finish(); + } + } + } + } + } + } } diff --git a/src/java/org/apache/cassandra/index/Index.java b/src/java/org/apache/cassandra/index/Index.java index 5225b82dee..b830d75c98 100644 --- a/src/java/org/apache/cassandra/index/Index.java +++ b/src/java/org/apache/cassandra/index/Index.java @@ -520,6 +520,28 @@ public interface Index return Collections.emptySet(); } + + /** + * When a secondary index is created on a column for a table with e.g. TWCS strategy, + * when this table contains SSTables which are evaluated as fully expired upon compaction, + * they are by default filtered out as they can be dropped in their entirety. However, once dropped like that, + * the index implementation is not notified about this fact via IndexGCTransaction as compaction on + * non-fully expired tables would do. This in turn means that custom index will never know that some data have + * been removed hence data custom index implementation is responsible for will grow beyond any limit. + * + * Override this method and return false in index implementation only in case you do not want to be notified about + * dropped fully-expired data. This will eventually mean that {@link Indexer#removeRow(Row)} will not be called + * for rows contained in fully expired table. Return true if you do want to be notified about that fact. + * + * This method returns true by default. + * + * @return true when fully expired tables should be included in compaction process, false otherwise. + */ + public default boolean notifyIndexerAboutRowsInFullyExpiredSSTables() + { + return true; + } + /* * Update processing */ diff --git a/src/java/org/apache/cassandra/index/sai/StorageAttachedIndex.java b/src/java/org/apache/cassandra/index/sai/StorageAttachedIndex.java index c93923562c..9f13496b52 100644 --- a/src/java/org/apache/cassandra/index/sai/StorageAttachedIndex.java +++ b/src/java/org/apache/cassandra/index/sai/StorageAttachedIndex.java @@ -557,6 +557,12 @@ public class StorageAttachedIndex implements Index .collect(Collectors.toSet()); } + @Override + public boolean notifyIndexerAboutRowsInFullyExpiredSSTables() + { + return false; + } + @Override public Indexer indexerFor(DecoratedKey key, RegularAndStaticColumns columns, diff --git a/src/java/org/apache/cassandra/index/sasi/SASIIndex.java b/src/java/org/apache/cassandra/index/sasi/SASIIndex.java index b2989cb8d8..9b069827d3 100644 --- a/src/java/org/apache/cassandra/index/sasi/SASIIndex.java +++ b/src/java/org/apache/cassandra/index/sasi/SASIIndex.java @@ -271,6 +271,12 @@ public class SASIIndex implements Index, INotificationConsumer public void validate(PartitionUpdate update, ClientState state) throws InvalidRequestException {} + @Override + public boolean notifyIndexerAboutRowsInFullyExpiredSSTables() + { + return false; + } + @Override public Indexer indexerFor(DecoratedKey key, RegularAndStaticColumns columns, long nowInSec, WriteContext context, IndexTransaction.Type transactionType, Memtable memtable) { diff --git a/test/unit/org/apache/cassandra/index/CustomIndexTest.java b/test/unit/org/apache/cassandra/index/CustomIndexTest.java index 3c0d75b5fc..3cd756b9e3 100644 --- a/test/unit/org/apache/cassandra/index/CustomIndexTest.java +++ b/test/unit/org/apache/cassandra/index/CustomIndexTest.java @@ -44,6 +44,8 @@ import com.google.common.collect.Maps; import com.google.common.collect.Sets; import org.junit.Assume; import org.junit.BeforeClass; +import com.google.common.util.concurrent.Uninterruptibles; +import org.junit.Assert; import org.junit.Test; import com.datastax.driver.core.exceptions.QueryValidationException; @@ -683,6 +685,64 @@ public class CustomIndexTest extends CQLTester assertEquals(0, deletedClustering.intValue()); } + + // two stub indexes just to track number of row deletions + // when we have fully-expired tables and indexes on different columns + public static class StubIndex1 extends StubIndex + { + public StubIndex1(ColumnFamilyStore baseCfs, IndexMetadata metadata) + { + super(baseCfs, metadata); + } + } + + public static class StubIndex2 extends StubIndex + { + + public StubIndex2(ColumnFamilyStore baseCfs, IndexMetadata metadata) + { + super(baseCfs, metadata); + } + } + + @Test + public void notifyIndexesOfFullyExpiredSSTablesDuringCompaction() + { + createTable("CREATE TABLE %s (id int primary key, col1 int, col2 int) " + + "WITH compaction = {'class': 'TimeWindowCompactionStrategy', " + + " 'compaction_window_size': 1," + + " 'compaction_window_unit': 'MINUTES'," + + " 'expired_sstable_check_frequency_seconds': 10} " + + "AND gc_grace_seconds = 0"); + + createIndex(String.format("CREATE CUSTOM INDEX row_ttl_test_index_1 ON %%s(col1) USING '%s'", StubIndex1.class.getName())); + createIndex(String.format("CREATE CUSTOM INDEX row_ttl_test_index_2 ON %%s(col2) USING '%s'", StubIndex2.class.getName())); + + ColumnFamilyStore cfs = getCurrentColumnFamilyStore(); + StubIndex index1 = (StubIndex1)cfs.indexManager.getIndexByName("row_ttl_test_index_1"); + StubIndex index2 = (StubIndex2)cfs.indexManager.getIndexByName("row_ttl_test_index_2"); + + execute("INSERT INTO %s (id, col1) VALUES (?, ?) USING TTL 20", 0, 0); + execute("INSERT INTO %s (id, col1) VALUES (?, ?) USING TTL 20", 1, 1); + execute("INSERT INTO %s (id, col1) VALUES (?, ?) USING TTL 20", 2, 2); + + execute("INSERT INTO %s (id, col2) VALUES (?, ?) USING TTL 20", 0, 0); + execute("INSERT INTO %s (id, col2) VALUES (?, ?) USING TTL 20", 1, 1); + execute("INSERT INTO %s (id, col2) VALUES (?, ?) USING TTL 20", 2, 2); + + flush(); + + Uninterruptibles.sleepUninterruptibly(60, TimeUnit.SECONDS); + + compact(); + + Assert.assertFalse(index1.rowsDeleted.isEmpty()); + Assert.assertEquals(3, index1.rowsDeleted.size()); + + Assert.assertFalse(index2.rowsDeleted.isEmpty()); + Assert.assertEquals(3, index2.rowsDeleted.size()); + } + @Test public void validateOptions() {