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

Obsolete expired SSTables before compaction starts 

patch by Stefan Miklosovic; reviewed by Branimir Lambov for CASSANDRA-19776
This commit is contained in:
Stefan Miklosovic 2026-03-03 12:57:39 +01:00
parent 1461e754e8
commit 547187a229
No known key found for this signature in database
GPG Key ID: 32F35CB2F546D93E
5 changed files with 249 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,4 +1,5 @@
4.0.20
* Obsolete expired SSTables before compaction starts (CASSANDRA-19776)
* No need to evict already prepared statements, as it creates a race condition between multiple threads (CASSANDRA-17401)
* Switch lz4-java to at.yawk.lz4 version due to CVE (CASSANDRA-20152)
* Restrict BytesType compatibility to scalar types only (CASSANDRA-20982)

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

@ -174,6 +174,12 @@ public class CompactionTask extends AbstractCompactionTask
maybeNotifyIndexersAboutRowsInFullyExpiredSSTables(fullyExpiredSSTables);
if (!fullyExpiredSSTables.isEmpty())
{
logger.debug("Compaction {} dropping expired sstables: {}", transaction.opId().toString(), fullyExpiredSSTables);
fullyExpiredSSTables.forEach(transaction::obsolete);
}
Set<SSTableReader> actuallyCompact = Sets.difference(transaction.originals(), fullyExpiredSSTables);
Collection<SSTableReader> newSStables;

11
src/java/org/apache/cassandra/io/sstable/SSTableRewriter.java
View File

@ -255,10 +255,13 @@ public class SSTableRewriter extends Transactional.AbstractTransactional impleme
continue;
}
DecoratedKey newStart = latest.firstKeyBeyond(lowerbound);
assert newStart != null;
SSTableReader replacement = latest.cloneWithNewStart(newStart, runOnClose);
transaction.update(replacement, true);
if (!transaction.isObsolete(latest))
{
DecoratedKey newStart = latest.firstKeyBeyond(lowerbound);
assert newStart != null;
SSTableReader replacement = latest.cloneWithNewStart(newStart, runOnClose);
transaction.update(replacement, true);
}
}
}

187
test/unit/org/apache/cassandra/db/compaction/CompactionTaskTest.java
View File

@ -21,9 +21,11 @@ package org.apache.cassandra.db.compaction;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicInteger;
import org.junit.Assert;
import org.junit.Before;
@ -31,10 +33,18 @@ import org.junit.BeforeClass;
import org.junit.Test;
import org.apache.cassandra.SchemaLoader;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.cql3.QueryProcessor;
import org.apache.cassandra.cql3.statements.schema.CreateTableStatement;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.Directories;
import org.apache.cassandra.db.compaction.writers.CompactionAwareWriter;
import org.apache.cassandra.db.compaction.writers.MaxSSTableSizeWriter;
import org.apache.cassandra.db.lifecycle.LifecycleTransaction;
import org.apache.cassandra.db.lifecycle.SSTableSet;
import org.apache.cassandra.db.lifecycle.TestableLifecycleTransaction;
import org.apache.cassandra.db.lifecycle.View;
import org.apache.cassandra.io.sstable.SSTableRewriter;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.schema.KeyspaceParams;
import org.apache.cassandra.schema.Schema;
@ -42,6 +52,7 @@ import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.service.ActiveRepairService;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.UUIDGen;
import org.apache.cassandra.utils.concurrent.Refs;
import org.apache.cassandra.utils.concurrent.Transactional;
public class CompactionTaskTest
@ -49,13 +60,18 @@ public class CompactionTaskTest
private static TableMetadata cfm;
private static ColumnFamilyStore cfs;
private static TableMetadata gcGraceCfm;
private static ColumnFamilyStore gcGraceCfs;
@BeforeClass
public static void setUpClass() throws Exception
{
SchemaLoader.prepareServer();
cfm = CreateTableStatement.parse("CREATE TABLE tbl (k INT PRIMARY KEY, v INT)", "coordinatorsessiontest").build();
SchemaLoader.createKeyspace("ks", KeyspaceParams.simple(1), cfm);
gcGraceCfm = CreateTableStatement.parse("CREATE TABLE tbl2 (k INT PRIMARY KEY, col1 INT, col2 INT, col3 INT, data TEXT) WITH gc_grace_seconds = 0", "ks").build();
SchemaLoader.createKeyspace("ks", KeyspaceParams.simple(1), cfm, gcGraceCfm);
cfs = Schema.instance.getColumnFamilyStoreInstance(cfm.id);
gcGraceCfs = Schema.instance.getColumnFamilyStoreInstance(gcGraceCfm.id);
}
@Before
@ -96,6 +112,175 @@ public class CompactionTaskTest
Assert.assertEquals(Transactional.AbstractTransactional.State.ABORTED, txn.state());
}
/**
* Test that even some SSTables are fully expired, we can still select and reference them
* while they are part of compaction.
*
* @see <a href="https://issues.apache.org/jira/browse/CASSANDRA-19776>CASSANDRA-19776</a>
*/
@Test
public void testFullyExpiredSSTablesAreNotReleasedPrematurely()
{
Assert.assertEquals(0, gcGraceCfs.getLiveSSTables().size());
gcGraceCfs.getCompactionStrategyManager().disable();
// Use large SSTables (10+ MiB) so that switching to new output SSTables happens during
// compaction. Without large enough SSTables, the output fits in one SSTable and no switch happens
int numKeys = 5000;
String data = new String(new char[2048]).replace('\0', 'x'); // ~2KB padding per row
// Similar technique to get fully expired SSTables as in TTLExpiryTest#testAggressiveFullyExpired
// SSTable 1 (will be fully expired - superseded by SSTable 2)
for (int k = 0; k < numKeys; k++)
{
QueryProcessor.executeInternal("INSERT INTO ks.tbl2 (k, col1, data) VALUES (?, 1, ?) USING TIMESTAMP 1 AND TTL 1", k, data);
QueryProcessor.executeInternal("INSERT INTO ks.tbl2 (k, col2, data) VALUES (?, 1, ?) USING TIMESTAMP 3 AND TTL 1", k, data);
}
gcGraceCfs.forceBlockingFlush();
// SSTable 2 (will be fully expired - superseded by SSTables 3 and 4)
for (int k = 0; k < numKeys; k++)
{
QueryProcessor.executeInternal("INSERT INTO ks.tbl2 (k, col1, data) VALUES (?, 1, ?) USING TIMESTAMP 2 AND TTL 1", k, data);
QueryProcessor.executeInternal("INSERT INTO ks.tbl2 (k, col2, data) VALUES (?, 1, ?) USING TIMESTAMP 5 AND TTL 1", k, data);
}
gcGraceCfs.forceBlockingFlush();
Set<SSTableReader> toBeObsolete = new HashSet<>(gcGraceCfs.getLiveSSTables());
Assert.assertEquals(2, toBeObsolete.size());
// SSTable 3 (not fully expired)
for (int k = 0; k < numKeys; k++)
{
QueryProcessor.executeInternal("INSERT INTO ks.tbl2 (k, col1, data) VALUES (?, 1, ?) USING TIMESTAMP 4 AND TTL 1", k, data);
QueryProcessor.executeInternal("INSERT INTO ks.tbl2 (k, col3, data) VALUES (?, 1, ?) USING TIMESTAMP 7 AND TTL 1", k, data);
}
gcGraceCfs.forceBlockingFlush();
// SSTable 4 (not fully expired - col3 has longer TTL)
for (int k = 0; k < numKeys; k++)
{
QueryProcessor.executeInternal("INSERT INTO ks.tbl2 (k, col3, data) VALUES (?, 1, ?) USING TIMESTAMP 6 AND TTL 3", k, data);
QueryProcessor.executeInternal("INSERT INTO ks.tbl2 (k, col2, data) VALUES (?, 1, ?) USING TIMESTAMP 8 AND TTL 1", k, data);
}
gcGraceCfs.forceBlockingFlush();
Set<SSTableReader> sstables = gcGraceCfs.getLiveSSTables();
Assert.assertEquals(4, sstables.size());
// Enable preemptive opening so that SSTableRewriter.switchWriter() calls checkpoint().
int originalPreemptiveOpenInterval = DatabaseDescriptor.getSSTablePreemptiveOpenIntervalInMB();
DatabaseDescriptor.setSSTablePreemptiveOpenIntervalInMB(2);
// collector of stacktraces to later check if checkpoint was called in switchWriter
final List<StackTraceElement[]> stacks = new ArrayList<>();
AtomicInteger checkpointCount = new AtomicInteger(0);
try
{
// Hook into transaction's checkpoint and doCommit methods to verify that after
// checkpointing, all SSTables (including fully expired ones) remain referenceable.
// We use TestableLifecycleTransaction because in cassandra-5.0, AbstractCompactionTask.transaction
// is LifecycleTransaction (concrete), not ILifecycleTransaction (interface), so
// WrappedLifecycleTransaction cannot be used with CompactionTask directly.
LifecycleTransaction txn = new TestableLifecycleTransaction(gcGraceCfs.getTracker(), OperationType.COMPACTION, sstables)
{
@Override
public void checkpoint()
{
stacks.add(Thread.currentThread().getStackTrace());
for (SSTableReader r : toBeObsolete)
Assert.assertTrue(this.isObsolete(r));
assertAllSSTablesAreReferenceable();
super.checkpoint();
// This is the critical assertion: after checkpoint(), all SSTables in the
// CANONICAL view must still be referenceable. Before the fix, fully expired
// SSTables would lose their references here, causing selectAndReference() to
// spin loop (CASSANDRA-19776).
assertAllSSTablesAreReferenceable();
checkpointCount.incrementAndGet();
}
@Override
public Throwable doCommit(Throwable accumulate)
{
assertAllSSTablesAreReferenceable();
return super.doCommit(accumulate);
}
private void assertAllSSTablesAreReferenceable()
{
// This simulates what EstimatedPartitionCount metric and similar code paths do.
// It is crucial that tryRef does not return null; a null result means some SSTables
// are not referenceable, which would cause selectAndReference() to spin loop.
ColumnFamilyStore.ViewFragment view = gcGraceCfs.select(View.selectFunction(SSTableSet.CANONICAL));
Refs<SSTableReader> refs = Refs.tryRef(view.sstables);
Assert.assertNotNull("Some SSTables in CANONICAL view are not referenceable (CASSANDRA-19776)", refs);
refs.close();
}
};
// Use MaxSSTableSizeWriter with a small max size (2 MiB) to force output sstable switches during compaction.
long maxSSTableSize = 2L * 1024 * 1024; // 2 MiB
CompactionTask task = new CompactionTask(gcGraceCfs, txn, FBUtilities.nowInSeconds() + 2)
{
@Override
public CompactionAwareWriter getCompactionAwareWriter(ColumnFamilyStore cfs,
Directories directories,
LifecycleTransaction transaction,
Set<SSTableReader> nonExpiredSSTables)
{
return new MaxSSTableSizeWriter(cfs, directories, transaction, nonExpiredSSTables,
maxSSTableSize, 0);
}
};
try (CompactionController compactionController = task.getCompactionController(txn.originals()))
{
Set<SSTableReader> fullyExpiredSSTables = compactionController.getFullyExpiredSSTables();
Assert.assertEquals(2, fullyExpiredSSTables.size());
task.execute(null);
}
// Verify that checkpoint was called more than once, proving that output sstable switching
// happened during compaction. Without MaxSSTableSizeWriter and large enough SSTables,
// checkpoint would only be called at the end, which would not exercise the CASSANDRA-19776 fix.
Assert.assertTrue("Expected checkpoint() to be called more than once during compaction, but was called "
+ checkpointCount.get() + " time(s). Output sstable switching did not occur.",
checkpointCount.get() > 1);
}
finally
{
DatabaseDescriptor.setSSTablePreemptiveOpenIntervalInMB(originalPreemptiveOpenInterval);
}
Assert.assertFalse(stacks.isEmpty());
boolean checkpointCalledInSSTableRewriter = false;
for (int i = 0; i < stacks.size(); i++)
{
for (StackTraceElement element : stacks.get(i))
{
if (element.getClassName().equals(SSTableRewriter.class.getName())
&& (element.getMethodName().equals("switchWriter")
|| element.getMethodName().equals("maybeReopenEarly")))
{
checkpointCalledInSSTableRewriter = true;
break;
}
}
}
Assert.assertTrue(checkpointCalledInSSTableRewriter);
}
private static void mutateRepaired(SSTableReader sstable, long repairedAt, UUID pendingRepair, boolean isTransient) throws IOException
{
sstable.descriptor.getMetadataSerializer().mutateRepairMetadata(sstable.descriptor, repairedAt, pendingRepair, isTransient);

49
test/unit/org/apache/cassandra/db/lifecycle/TestableLifecycleTransaction.java Normal file
View File

@ -0,0 +1,49 @@
/*
* 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 org.apache.cassandra.db.compaction.OperationType;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import static java.util.Collections.emptySet;
import static org.apache.cassandra.db.lifecycle.View.permitCompacting;
import static org.apache.cassandra.db.lifecycle.View.updateCompacting;
/**
* A {@link LifecycleTransaction} subclass that exposes the package-private constructor
* for use in tests. This allows tests to create anonymous subclasses that override
* {@link #checkpoint()} and {@link #doCommit(Throwable)} for verification purposes.
* <p>
* The constructor marks the given SSTables as compacting in the tracker (equivalent to
* what {@link Tracker#tryModify(Iterable, OperationType)} does) before delegating to
* the parent constructor.
*/
public class TestableLifecycleTransaction extends LifecycleTransaction
{
public TestableLifecycleTransaction(Tracker tracker, OperationType operationType, Iterable<SSTableReader> readers)
{
super(markCompactingAndReturn(tracker, readers), operationType, readers);
}
private static Tracker markCompactingAndReturn(Tracker tracker, Iterable<SSTableReader> readers)
{
tracker.apply(permitCompacting(readers), updateCompacting(emptySet(), readers));
return tracker;
}
}