diff --git a/CHANGES.txt b/CHANGES.txt index 114520abfc..755372a335 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -12,6 +12,7 @@ Merged from 4.1: * Disk usage guardrail cannot be disabled when failure threshold is reached (CASSANDRA-21057) * ReadCommandController should close fast to avoid deadlock when building secondary index (CASSANDRA-19564) Merged from 4.0: + * Obsolete expired SSTables before compaction starts (CASSANDRA-19776) * Switch lz4-java to at.yawk.lz4 version due to CVE (CASSANDRA-20152) * Restrict BytesType compatibility to scalar types only (CASSANDRA-20982) * Backport fix to nodetool gcstats output for direct memory (CASSANDRA-21037) diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionTask.java b/src/java/org/apache/cassandra/db/compaction/CompactionTask.java index 996ca6e0cc..9566ef8430 100644 --- a/src/java/org/apache/cassandra/db/compaction/CompactionTask.java +++ b/src/java/org/apache/cassandra/db/compaction/CompactionTask.java @@ -184,6 +184,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/test/unit/org/apache/cassandra/db/compaction/CompactionTaskTest.java b/test/unit/org/apache/cassandra/db/compaction/CompactionTaskTest.java index 066aa92067..35763140e6 100644 --- a/test/unit/org/apache/cassandra/db/compaction/CompactionTaskTest.java +++ b/test/unit/org/apache/cassandra/db/compaction/CompactionTaskTest.java @@ -21,8 +21,10 @@ 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.concurrent.atomic.AtomicInteger; import org.junit.Assert; import org.junit.Before; @@ -31,12 +33,20 @@ import org.junit.Test; import org.apache.cassandra.SchemaLoader; import org.apache.cassandra.Util; +import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.cql3.QueryProcessor; import org.apache.cassandra.cql3.UntypedResultSet; 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.SystemKeyspace; +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; @@ -44,6 +54,7 @@ import org.apache.cassandra.schema.TableMetadata; import org.apache.cassandra.service.ActiveRepairService; import org.apache.cassandra.utils.FBUtilities; import org.apache.cassandra.utils.TimeUUID; +import org.apache.cassandra.utils.concurrent.Refs; import org.apache.cassandra.utils.concurrent.Transactional; import static java.lang.String.format; @@ -55,13 +66,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)", "ks").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 @@ -69,6 +85,9 @@ public class CompactionTaskTest { cfs.getCompactionStrategyManager().enable(); cfs.truncateBlocking(); + + gcGraceCfs.getCompactionStrategyManager().enable(); + gcGraceCfs.truncateBlocking(); } @Test @@ -136,6 +155,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); + } + Util.flush(gcGraceCfs); + + // 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); + } + Util.flush(gcGraceCfs); + + Set sstables = gcGraceCfs.getLiveSSTables(); + Assert.assertEquals(4, sstables.size()); + + // Enable preemptive opening so that SSTableRewriter.switchWriter() calls checkpoint(). + int originalPreemptiveOpenInterval = DatabaseDescriptor.getSSTablePreemptiveOpenIntervalInMiB(); + DatabaseDescriptor.setSSTablePreemptiveOpenIntervalInMiB(2); + + // collector of stacktraces to later check if checkpoint was called in switchWriter + final List stacks = new ArrayList<>(); + + try + { + AtomicInteger checkpointCount = new AtomicInteger(0); + + // 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.setSSTablePreemptiveOpenIntervalInMiB(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, TimeUUID pendingRepair, boolean isTransient) throws IOException { sstable.descriptor.getMetadataSerializer().mutateRepairMetadata(sstable.descriptor, repairedAt, pendingRepair, isTransient); @@ -228,11 +416,11 @@ public class CompactionTaskTest cfs.getTracker().removeUnsafe(sstables); } } - + @Test public void testMajorCompactTask() { - //major compact without range/pk specified + //major compact without range/pk specified CompactionTasks compactionTasks = cfs.getCompactionStrategyManager().getMaximalTasks(Integer.MAX_VALUE, false, OperationType.MAJOR_COMPACTION); Assert.assertTrue(compactionTasks.stream().allMatch(task -> task.compactionType.equals(OperationType.MAJOR_COMPACTION))); } 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; + } +}