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

Add purge-boundary, Harry history, and config-axis differential coverage 

PurgeBoundaryDifferentialCompactionTest: exact purge-boundary scenarios
without clock control. The harness gains an explicit-gcBefore overload;
the scenario reads the real deletion second from the flushed sstable's
stats (excluding the Long.MAX_VALUE no-deletion sentinel) and runs the
differential at gcBefore == ldt (tombstones retained: purge requires
localDeletionTime strictly less than gcBefore) and gcBefore == ldt + 1
(purged), asserting the outcome in the captured dump both ways. Also
covers compaction reads with direct disk access mode and the Zstd,
Deflate and Snappy compressors.

HarryDifferentialCompactionTest: deep probabilistic tombstone histories
(interleaved inserts with row, column, range and partition deletes over
multiple flush rounds, reversed clustering, static columns) built with
Harry's HistoryBuilder and executed as plain CQL in this JVM via
CQLTesterVisitExecutor. Seeded and logged for reproduction. The
differential harness is the oracle, so no Harry validation reads are
issued.

The TTL-expiry-exactly-at-nowInSec boundary remains uncovered:
nowInSec is taken inside CompactionTask per run and a custom Clock can
only be installed at JVM startup, so deterministic coverage needs a
dedicated test-runner configuration.
This commit is contained in:
Jon Haddad 2026-06-09 21:04:04 -07:00
parent e2eba63a72
commit e57649a0b0
3 changed files with 299 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
test/unit/org/apache/cassandra/db/compaction/differential/DifferentialCompactionTester.java
View File

@ -143,7 +143,22 @@ public abstract class DifferentialCompactionTester extends CQLTester
Set<String> 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<SSTableReader> inputs,
Set<String> byteDiffAllowlist,
TaskFactory taskFactory,
long gcBefore) throws Exception
{
Path scratch = Files.createTempDirectory("differential-compaction");
// Early open must be off for keepOriginals to actually keep originals:

138
test/unit/org/apache/cassandra/db/compaction/differential/HarryDifferentialCompactionTest.java Normal file
View File

@ -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<String> 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);
}
}

145
test/unit/org/apache/cassandra/db/compaction/differential/PurgeBoundaryDifferentialCompactionTest.java Normal file
View File

@ -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<String> 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);
}
}
}