natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'cassandra-5.0' into trunk

This commit is contained in:
Stefan Miklosovic 2025-08-20 10:54:42 +02:00
parent acec78abf4 74e0794d92
commit b7e2b3b9dc
No known key found for this signature in database
GPG Key ID: 32F35CB2F546D93E
6 changed files with 169 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -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)

74
src/java/org/apache/cassandra/db/compaction/CompactionTask.java
View File

@ -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<SSTableReader> actuallyCompact = new HashSet<>(inputSSTables());
final Set<SSTableReader> 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<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();
}
}
}
}
}
}
}

22
src/java/org/apache/cassandra/index/Index.java
View File

@ -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
*/

6
src/java/org/apache/cassandra/index/sai/StorageAttachedIndex.java
View File

@ -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,

6
src/java/org/apache/cassandra/index/sasi/SASIIndex.java
View File

@ -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)
{

60
test/unit/org/apache/cassandra/index/CustomIndexTest.java
View File

@ -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()
{