From 547187a2295581e085c69b90ca7ef428295e9a51 Mon Sep 17 00:00:00 2001 From: Stefan Miklosovic Date: Tue, 3 Mar 2026 12:57:39 +0100 Subject: [PATCH] Obsolete expired SSTables before compaction starts patch by Stefan Miklosovic; reviewed by Branimir Lambov for CASSANDRA-19776 --- CHANGES.txt | 1 + .../db/compaction/CompactionTask.java | 6 + .../cassandra/io/sstable/SSTableRewriter.java | 11 +- .../db/compaction/CompactionTaskTest.java | 187 +++++++++++++++++- .../TestableLifecycleTransaction.java | 49 +++++ 5 files changed, 249 insertions(+), 5 deletions(-) create mode 100644 test/unit/org/apache/cassandra/db/lifecycle/TestableLifecycleTransaction.java diff --git a/CHANGES.txt b/CHANGES.txt index ae4ff28dd4..75b82edcd8 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -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) diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionTask.java b/src/java/org/apache/cassandra/db/compaction/CompactionTask.java index ce28e6b26f..d150610722 100644 --- a/src/java/org/apache/cassandra/db/compaction/CompactionTask.java +++ b/src/java/org/apache/cassandra/db/compaction/CompactionTask.java @@ -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 actuallyCompact = Sets.difference(transaction.originals(), fullyExpiredSSTables); Collection newSStables; diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableRewriter.java b/src/java/org/apache/cassandra/io/sstable/SSTableRewriter.java index a3d5ae9a2b..93e74dd6a3 100644 --- a/src/java/org/apache/cassandra/io/sstable/SSTableRewriter.java +++ b/src/java/org/apache/cassandra/io/sstable/SSTableRewriter.java @@ -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); + } } } diff --git a/test/unit/org/apache/cassandra/db/compaction/CompactionTaskTest.java b/test/unit/org/apache/cassandra/db/compaction/CompactionTaskTest.java index bdd2014230..1721ffa2ac 100644 --- a/test/unit/org/apache/cassandra/db/compaction/CompactionTaskTest.java +++ b/test/unit/org/apache/cassandra/db/compaction/CompactionTaskTest.java @@ -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 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 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 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 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 nonExpiredSSTables) + { + return new MaxSSTableSizeWriter(cfs, directories, transaction, nonExpiredSSTables, + maxSSTableSize, 0); + } + }; + + try (CompactionController compactionController = task.getCompactionController(txn.originals())) + { + Set 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); diff --git a/test/unit/org/apache/cassandra/db/lifecycle/TestableLifecycleTransaction.java b/test/unit/org/apache/cassandra/db/lifecycle/TestableLifecycleTransaction.java new file mode 100644 index 0000000000..ad3b4fddc6 --- /dev/null +++ b/test/unit/org/apache/cassandra/db/lifecycle/TestableLifecycleTransaction.java @@ -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. + *

+ * 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 readers) + { + super(markCompactingAndReturn(tracker, readers), operationType, readers); + } + + private static Tracker markCompactingAndReturn(Tracker tracker, Iterable readers) + { + tracker.apply(permitCompacting(readers), updateCompacting(emptySet(), readers)); + return tracker; + } +}