mirror of https://github.com/apache/cassandra
Merge branch 'cassandra-4.1' into cassandra-5.0
This commit is contained in:
Stefan Miklosovic
commit
74e0794d92
|
|
@ -1,5 +1,7 @@
|
|||
5.0.6
|
||||
* Sort SSTable TOC entries for determinism (CASSANDRA-20494)
|
||||
Merged from 4.0:
|
||||
* Make secondary index implementations notified about rows in fully expired SSTables in compaction (CASSANDRA-20829)
|
||||
|
||||
|
||||
5.0.5
|
||||
|
|
|
|||
|
|
@ -18,6 +18,7 @@
|
|||
package org.apache.cassandra.db.compaction;
|
||||
|
||||
import java.time.Instant;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Collection;
|
||||
import java.util.HashMap;
|
||||
import java.util.HashSet;
|
||||
|
|
@ -39,9 +40,16 @@ 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.LifecycleTransaction;
|
||||
import org.apache.cassandra.db.rows.Row;
|
||||
import org.apache.cassandra.db.rows.Unfiltered;
|
||||
import org.apache.cassandra.db.rows.UnfilteredRowIterator;
|
||||
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;
|
||||
|
|
@ -174,6 +182,8 @@ public class CompactionTask extends AbstractCompactionTask
|
|||
long inputSizeBytes;
|
||||
long timeSpentWritingKeys;
|
||||
|
||||
maybeNotifyIndexersAboutRowsInFullyExpiredSSTables(fullyExpiredSSTables);
|
||||
|
||||
Set<SSTableReader> actuallyCompact = Sets.difference(transaction.originals(), fullyExpiredSSTables);
|
||||
Collection<SSTableReader> newSStables;
|
||||
|
||||
|
|
@ -465,4 +475,67 @@ public class CompactionTask extends AbstractCompactionTask
|
|||
}
|
||||
return max;
|
||||
}
|
||||
|
||||
private void maybeNotifyIndexersAboutRowsInFullyExpiredSSTables(Set<SSTableReader> fullyExpiredSSTables)
|
||||
{
|
||||
if (fullyExpiredSSTables.isEmpty())
|
||||
return;
|
||||
|
||||
List<Index> 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<Index.Indexer> 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();
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -515,6 +515,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
|
||||
*/
|
||||
|
|
|
|||
|
|
@ -553,6 +553,12 @@ public class StorageAttachedIndex implements Index
|
|||
.collect(Collectors.toSet());
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean notifyIndexerAboutRowsInFullyExpiredSSTables()
|
||||
{
|
||||
return false;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Indexer indexerFor(DecoratedKey key,
|
||||
RegularAndStaticColumns columns,
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
{
|
||||
|
|
|
|||
|
|
@ -33,6 +33,8 @@ import com.google.common.collect.ImmutableSet;
|
|||
import com.google.common.collect.Maps;
|
||||
import com.google.common.collect.Sets;
|
||||
import org.junit.Assume;
|
||||
import com.google.common.util.concurrent.Uninterruptibles;
|
||||
import org.junit.Assert;
|
||||
import org.junit.Test;
|
||||
|
||||
import com.datastax.driver.core.exceptions.QueryValidationException;
|
||||
|
|
@ -649,6 +651,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()
|
||||
{
|
||||
|
|
|
|||
Loading…
Reference in New Issue