diff --git a/test/unit/org/apache/cassandra/db/compaction/differential/DifferentialCompactionTester.java b/test/unit/org/apache/cassandra/db/compaction/differential/DifferentialCompactionTester.java index 4a451a5658..369efe0918 100644 --- a/test/unit/org/apache/cassandra/db/compaction/differential/DifferentialCompactionTester.java +++ b/test/unit/org/apache/cassandra/db/compaction/differential/DifferentialCompactionTester.java @@ -143,7 +143,22 @@ public abstract class DifferentialCompactionTester extends CQLTester Set byteDiffAllowlist, TaskFactory taskFactory) throws Exception { - long gcBefore = cfs.getDefaultGcBefore(FBUtilities.nowInSeconds()); + return assertCursorMatchesIterator(cfs, inputs, byteDiffAllowlist, taskFactory, + cfs.getDefaultGcBefore(FBUtilities.nowInSeconds())); + } + + /** + * Variant with an explicit gcBefore: lets scenarios place purge decisions EXACTLY at the + * boundary (purge requires localDeletionTime < gcBefore) without controlling the wall + * clock — read the actual deletion time from the flushed sstable's stats, then run with + * gcBefore == ldt (retained) and gcBefore == ldt + 1 (purged). + */ + protected CapturedOutput assertCursorMatchesIterator(ColumnFamilyStore cfs, + Set inputs, + Set byteDiffAllowlist, + TaskFactory taskFactory, + long gcBefore) throws Exception + { Path scratch = Files.createTempDirectory("differential-compaction"); // Early open must be off for keepOriginals to actually keep originals: diff --git a/test/unit/org/apache/cassandra/db/compaction/differential/HarryDifferentialCompactionTest.java b/test/unit/org/apache/cassandra/db/compaction/differential/HarryDifferentialCompactionTest.java new file mode 100644 index 0000000000..972d0e2957 --- /dev/null +++ b/test/unit/org/apache/cassandra/db/compaction/differential/HarryDifferentialCompactionTest.java @@ -0,0 +1,138 @@ +/* + * 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.compaction.differential; + +import java.util.Arrays; +import java.util.Set; +import java.util.concurrent.atomic.AtomicInteger; + +import org.junit.Test; + +import org.apache.cassandra.db.ColumnFamilyStore; +import org.apache.cassandra.db.Keyspace; +import org.apache.cassandra.harry.ColumnSpec; +import org.apache.cassandra.harry.SchemaSpec; +import org.apache.cassandra.harry.dsl.HistoryBuilder; +import org.apache.cassandra.harry.dsl.HistoryBuilderHelper; +import org.apache.cassandra.harry.execution.CQLTesterVisitExecutor; +import org.apache.cassandra.harry.execution.CQLVisitExecutor; +import org.apache.cassandra.harry.execution.DataTracker; +import org.apache.cassandra.harry.model.QuiescentChecker; +import org.apache.cassandra.harry.op.Visit; + +import static org.apache.cassandra.harry.checker.TestHelper.withRandom; + +/** + * Harry-driven differential soak: deep probabilistic tombstone/overwrite histories (the + * workload shapes Harry's DSL is strongest at — interleaved row/column/range/partition + * deletes against overlapping inserts) executed through plain CQL in this JVM, flushed into + * several sstables, then both compaction paths compared byte-for-byte. The differential + * harness is the oracle, so no Harry read-validation visits are issued. + * + * Schema uses simple types only (the currently supported cursor surface; Harry cannot + * generate multi-cell columns yet — ColumnSpec TODOs). Reversed clustering included. + */ +public class HarryDifferentialCompactionTest extends DifferentialCompactionTester +{ + private static final Set ALLOWLIST = Set.of(); + private static final AtomicInteger idGen = new AtomicInteger(0); + + private static final int PARTITIONS = 30; + private static final int ROWS = 20; + private static final int REG_COLS = 5; + private static final int ROUNDS = 3; + private static final int OPS_PER_ROUND = 120; + + @Test + public void harryTombstoneHistories() throws Throwable + { + long seed = System.currentTimeMillis(); + logger.info("harryTombstoneHistories seed={}", seed); + withRandom(seed, rng -> { + String ks = "harry_diff_ks" + idGen.incrementAndGet(); + String table = "tbl" + idGen.incrementAndGet(); + SchemaSpec schema = new SchemaSpec(rng.next(), + 1000, + ks, + table, + Arrays.asList(ColumnSpec.pk("pk1", ColumnSpec.asciiType), + ColumnSpec.pk("pk2", ColumnSpec.int64Type)), + Arrays.asList(ColumnSpec.ck("ck1", ColumnSpec.asciiType, false), + ColumnSpec.ck("ck2", ColumnSpec.int64Type, true)), + Arrays.asList(ColumnSpec.regularColumn("r1", ColumnSpec.asciiType), + ColumnSpec.regularColumn("r2", ColumnSpec.int64Type), + ColumnSpec.regularColumn("r3", ColumnSpec.doubleType), + ColumnSpec.regularColumn("r4", ColumnSpec.int32Type), + ColumnSpec.regularColumn("r5", ColumnSpec.textType)), + Arrays.asList(ColumnSpec.staticColumn("s1", ColumnSpec.asciiType), + ColumnSpec.staticColumn("s2", ColumnSpec.int64Type))); + + schemaChange(String.format("CREATE KEYSPACE IF NOT EXISTS %s WITH replication = " + + "{'class': 'SimpleStrategy', 'replication_factor': '1'}", ks)); + createTable(schema.compile()); + ColumnFamilyStore cfs = Keyspace.open(ks).getColumnFamilyStore(table); + cfs.disableAutoCompaction(); + + HistoryBuilder history = new HistoryBuilder(schema.valueGenerators); + + for (int round = 0; round < ROUNDS; round++) + { + for (int op = 0; op < OPS_PER_ROUND; op++) + { + int pd = rng.nextInt(0, PARTITIONS); + int row = rng.nextInt(0, ROWS); + int kind = rng.nextInt(0, 100); + if (kind < 70) + history.insert(pd, row); + else if (kind < 80) + history.deleteRow(pd, row); + else if (kind < 88) + { + int lower = rng.nextInt(0, ROWS); + int upper = rng.nextInt(lower, 2 * ROWS); + history.deleteRowRange(pd, lower, upper, + rng.nextInt(schema.clusteringKeys.size()), + rng.nextBoolean(), + rng.nextBoolean()); + } + else if (kind < 96) + HistoryBuilderHelper.deleteRandomColumns(schema, pd, row, rng, history); + else + history.deletePartition(pd); + } + history.customThrowing(() -> flush(ks, table), "flush round"); + } + + replay(schema, history); + + assertCursorMatchesIterator(cfs, ALLOWLIST); + }); + } + + private void replay(SchemaSpec schema, HistoryBuilder history) + { + DataTracker tracker = new DataTracker.SequentialDataTracker(); + CQLVisitExecutor executor = + new CQLTesterVisitExecutor(schema, tracker, + new QuiescentChecker(schema.valueGenerators, tracker, history), + statement -> execute(statement.cql(), statement.bindings())); + for (Visit visit : history) + executor.execute(visit); + } +} diff --git a/test/unit/org/apache/cassandra/db/compaction/differential/PurgeBoundaryDifferentialCompactionTest.java b/test/unit/org/apache/cassandra/db/compaction/differential/PurgeBoundaryDifferentialCompactionTest.java new file mode 100644 index 0000000000..248aa0a721 --- /dev/null +++ b/test/unit/org/apache/cassandra/db/compaction/differential/PurgeBoundaryDifferentialCompactionTest.java @@ -0,0 +1,145 @@ +/* + * 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.compaction.differential; + +import java.util.Set; + +import org.junit.Test; + +import org.apache.cassandra.config.Config.DiskAccessMode; +import org.apache.cassandra.config.DatabaseDescriptor; +import org.apache.cassandra.db.ColumnFamilyStore; +import org.apache.cassandra.io.sstable.format.SSTableReader; + +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +/** + * Exact purge-boundary scenarios plus configuration-axis variants (compression algorithms, + * compaction disk access mode). Purge requires localDeletionTime < gcBefore; instead of + * freezing the clock, scenarios read the REAL deletion time from the flushed sstable's stats + * and pass an explicit gcBefore placed exactly at, and one past, that boundary. + */ +public class PurgeBoundaryDifferentialCompactionTest extends DifferentialCompactionTester +{ + private static final Set ALLOWLIST = Set.of(); + + private static String allJson(CapturedOutput out) + { + StringBuilder sb = new StringBuilder(); + for (CapturedSSTable s : out.sstables) + sb.append(s.json); + return sb.toString(); + } + + /** gcBefore exactly AT the tombstone's deletion time: NOT purgeable (strict less-than). */ + @Test + public void boundaryExactlyAtDeletionTime() throws Exception + { + createTable("CREATE TABLE %s (pk bigint, ck bigint, v text, PRIMARY KEY (pk, ck)) " + + "WITH gc_grace_seconds = 0"); + ColumnFamilyStore cfs = getCurrentColumnFamilyStore(); + cfs.disableAutoCompaction(); + + for (long pk = 0; pk < 4; pk++) + for (long ck = 0; ck < 10; ck++) + execute("INSERT INTO %s (pk, ck, v) VALUES (?, ?, ?) USING TIMESTAMP 100", pk, ck, "v" + ck); + flush(); + execute("DELETE FROM %s USING TIMESTAMP 200 WHERE pk = 1"); + for (long ck = 0; ck < 5; ck++) + execute("DELETE FROM %s USING TIMESTAMP 200 WHERE pk = 2 AND ck = ?", ck); + flush(); + + // sstables with no tombstones report the NO_DELETION sentinel (Long.MAX_VALUE) as their + // max local deletion time — use the min across sstables that DO have deletions, which + // for this scenario is the single deletion second of the tombstone-only sstable + long maxLdt = Long.MIN_VALUE; + for (SSTableReader sstable : cfs.getLiveSSTables()) + { + long ldt = sstable.getSSTableMetadata().maxLocalDeletionTime; + if (ldt != Long.MAX_VALUE) + maxLdt = Math.max(maxLdt, ldt); + } + assertTrue("scenario produced no tombstone deletion times", maxLdt > 0 && maxLdt < Long.MAX_VALUE); + + // boundary: ldt == gcBefore -> NOT purgeable (purge requires ldt < gcBefore) + CapturedOutput at = assertCursorMatchesIterator(cfs, cfs.getLiveSSTables(), ALLOWLIST, DEFAULT_TASK, maxLdt); + assertTrue("tombstones at exactly gcBefore must be retained", + allJson(at).contains("\"marked_deleted\":\"200\"")); + + // one past: ldt < gcBefore -> purgeable, shadowed data and tombstones both gone + CapturedOutput past = assertCursorMatchesIterator(cfs, cfs.getLiveSSTables(), ALLOWLIST, DEFAULT_TASK, maxLdt + 1); + assertFalse("tombstones strictly before gcBefore must purge", + allJson(past).contains("\"marked_deleted\":\"200\"")); + } + + /** Same differential flow under direct I/O for the compaction read path. */ + @Test + public void directDiskAccessMode() throws Exception + { + createTable("CREATE TABLE %s (pk bigint, ck bigint, v text, PRIMARY KEY (pk, ck)) " + + "WITH gc_grace_seconds = 864000"); + ColumnFamilyStore cfs = getCurrentColumnFamilyStore(); + cfs.disableAutoCompaction(); + + for (int round = 0; round < 3; round++) + { + for (long pk = 0; pk < 10; pk++) + for (long ck = 0; ck < 15; ck++) + execute("INSERT INTO %s (pk, ck, v) VALUES (?, ?, ?)", pk, ck, "r" + round + "v" + ck); + execute("DELETE FROM %s WHERE pk = ? AND ck >= 3 AND ck < 9", (long) round); + flush(); + } + + DiskAccessMode original = DatabaseDescriptor.getCompactionReadDiskAccessMode(); + DatabaseDescriptor.setCompactionReadDiskAccessMode(DiskAccessMode.direct); + try + { + assertCursorMatchesIterator(cfs, ALLOWLIST); + } + finally + { + DatabaseDescriptor.setCompactionReadDiskAccessMode(original); + } + } + + /** Output equivalence must hold for every compressor, not just the default LZ4. */ + @Test + public void compressionVariants() throws Exception + { + for (String compression : new String[]{ "ZstdCompressor", "DeflateCompressor", "SnappyCompressor" }) + { + createTable("CREATE TABLE %s (pk bigint, ck bigint, v text, PRIMARY KEY (pk, ck)) " + + "WITH gc_grace_seconds = 864000 AND compression = {'class': '" + compression + "'}"); + ColumnFamilyStore cfs = getCurrentColumnFamilyStore(); + cfs.disableAutoCompaction(); + + for (int round = 0; round < 2; round++) + { + for (long pk = 0; pk < 8; pk++) + for (long ck = 0; ck < 10; ck++) + execute("INSERT INTO %s (pk, ck, v) VALUES (?, ?, ?)", pk, ck, compression + round + ck); + execute("DELETE FROM %s WHERE pk = ? AND ck >= 2 AND ck < 6", (long) round); + flush(); + } + + assertCursorMatchesIterator(cfs, ALLOWLIST); + } + } +}