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

Simplify differential test helpers: extract shared measurement/SQL-building code 

- CursorCompactionAllocationGateTest: extract withMeasurementEnv (the repeated
  preemptive-open/cursor-enabled save-set-restore boilerplate in every @Test), measureBest
  (the warmup/measured min-allocation loop), captureLastInputBytes, and dumpAllocationProfile
  (the JFR recording/dump block shared by the three diagnostic tests). No behavior change.
- DifferentialCompactionTester: move the allJson helper (duplicated identically in
  PartialSetDifferentialCompactionTest and PurgeBoundaryDifferentialCompactionTest) to the
  shared base class and delete both copies.
- RandomDifferentialCompactionTest: extract appendNames/appendEqPredicates/appendPlaceholders
  for the column-name-list and "name = ?" predicate-list building repeated across
  insertStmt/pkOnlyInsertStmt/updateStmt/cellDeleteStmt/rangeDeleteStmt/deleteStmt.

Verified by rerunning the full differential suite (all 15 test classes).
This commit is contained in:
Jon Haddad 2026-07-27 16:09:43 -07:00
parent ab01848741
commit 8f88969e61
5 changed files with 158 additions and 242 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

286
test/unit/org/apache/cassandra/db/compaction/differential/CursorCompactionAllocationGateTest.java
View File

@ -73,6 +73,83 @@ public class CursorCompactionAllocationGateTest extends DifferentialCompactionTe
private static final int MEASURED_ITERATIONS = 3;
private static final long CEILING_BYTES = 512 * 1024;
private interface ThrowingRunnable
{
void run() throws Exception;
}
/**
* Disables preemptive-open (so the gate sees a deterministic, stable sstable set) for the
* duration of {@code body}, restoring it and cursorCompactionEnabled to their original
* values afterward. Callers that need cursor compaction on set it themselves inside
* {@code body} (some measure both cursor and iterator paths within the same call).
*/
private void withMeasurementEnv(ThrowingRunnable body) throws Exception
{
int originalPreemptiveOpen = DatabaseDescriptor.getSSTablePreemptiveOpenIntervalInMiB();
DatabaseDescriptor.setSSTablePreemptiveOpenIntervalInMiB(-1);
boolean originalCursorEnabled = DatabaseDescriptor.cursorCompactionEnabled();
try
{
body.run();
}
finally
{
DatabaseDescriptor.setSSTablePreemptiveOpenIntervalInMiB(originalPreemptiveOpen);
DatabaseDescriptor.setCursorCompactionEnabled(originalCursorEnabled);
}
}
/** Runs warmup + measured compactions, returning the minimum allocated over the measured tail. */
private long measureBest(com.sun.management.ThreadMXBean threadMXBean, ColumnFamilyStore cfs, long gcBefore,
int warmup, int measured) throws Exception
{
long best = Long.MAX_VALUE;
for (int i = 0; i < warmup + measured; i++)
{
long allocated = compactOnceMeasured(threadMXBean, cfs, gcBefore);
if (i >= warmup)
best = Math.min(best, allocated);
}
return best;
}
/** Records lastInputBytes as the total on-disk length of cfs's current live sstables. */
private void captureLastInputBytes(ColumnFamilyStore cfs)
{
lastInputBytes = 0;
for (SSTableReader sstable : cfs.getLiveSSTables())
lastInputBytes += sstable.onDiskLength();
}
private void dumpAllocationProfile(java.nio.file.Path dest, int iterations,
com.sun.management.ThreadMXBean threadMXBean,
ColumnFamilyStore cfs, long gcBefore) throws Exception
{
dumpAllocationProfile(dest, WARMUP_ITERATIONS, iterations, threadMXBean, cfs, gcBefore);
}
/** Warms up, then records a JFR allocation profile (with stacks) over `iterations` cursor
* compactions of cfs, dumped to dest for offline attribution. */
private void dumpAllocationProfile(java.nio.file.Path dest, int warmup, int iterations,
com.sun.management.ThreadMXBean threadMXBean,
ColumnFamilyStore cfs, long gcBefore) throws Exception
{
for (int i = 0; i < warmup; i++)
compactOnceMeasured(threadMXBean, cfs, gcBefore);
try (jdk.jfr.Recording recording = new jdk.jfr.Recording())
{
recording.enable("jdk.ObjectAllocationInNewTLAB").withStackTrace();
recording.enable("jdk.ObjectAllocationOutsideTLAB").withStackTrace();
recording.start();
for (int i = 0; i < iterations; i++)
compactOnceMeasured(threadMXBean, cfs, gcBefore);
recording.stop();
recording.dump(dest);
}
}
@Test
public void allocationDoesNotScaleWithRows() throws Exception
{
@ -80,12 +157,8 @@ public class CursorCompactionAllocationGateTest extends DifferentialCompactionTe
Assume.assumeTrue("thread allocation measurement unsupported on this JVM",
threadMXBean != null);
int originalPreemptiveOpen = DatabaseDescriptor.getSSTablePreemptiveOpenIntervalInMiB();
DatabaseDescriptor.setSSTablePreemptiveOpenIntervalInMiB(-1);
boolean originalCursorEnabled = DatabaseDescriptor.cursorCompactionEnabled();
DatabaseDescriptor.setCursorCompactionEnabled(true);
try
{
withMeasurementEnv(() -> {
DatabaseDescriptor.setCursorCompactionEnabled(true);
long smallAlloc = measureSteadyStateAllocation(threadMXBean, SMALL_PARTITIONS, true);
long bigAlloc = measureSteadyStateAllocation(threadMXBean, SMALL_PARTITIONS * SCALE, true);
long delta = bigAlloc - smallAlloc;
@ -104,12 +177,7 @@ public class CursorCompactionAllocationGateTest extends DifferentialCompactionTe
"A per-row/cell allocation has been introduced on the cursor hot path.",
smallAlloc, bigAlloc, delta, CEILING_BYTES),
delta <= CEILING_BYTES);
}
finally
{
DatabaseDescriptor.setSSTablePreemptiveOpenIntervalInMiB(originalPreemptiveOpen);
DatabaseDescriptor.setCursorCompactionEnabled(originalCursorEnabled);
}
});
}
private long measureSteadyStateAllocation(com.sun.management.ThreadMXBean threadMXBean, int partitions, boolean cursor) throws Exception
@ -140,18 +208,8 @@ public class CursorCompactionAllocationGateTest extends DifferentialCompactionTe
if (cursor)
assertCursorPathWillRun(cfs, cfs.getLiveSSTables(), gcBefore);
lastInputBytes = 0;
for (SSTableReader sstable : cfs.getLiveSSTables())
lastInputBytes += sstable.onDiskLength();
long best = Long.MAX_VALUE;
for (int i = 0; i < warmup + measured; i++)
{
long allocated = compactOnceMeasured(threadMXBean, cfs, gcBefore);
if (i >= warmup)
best = Math.min(best, allocated);
}
return best;
captureLastInputBytes(cfs);
return measureBest(threadMXBean, cfs, gcBefore, warmup, measured);
}
/** Total on-disk input bytes of the most recent measureSteadyStateAllocation call. */
@ -170,11 +228,7 @@ public class CursorCompactionAllocationGateTest extends DifferentialCompactionTe
Assume.assumeTrue(threadMXBean != null);
String padding = "v".repeat(500);
int originalPreemptiveOpen = DatabaseDescriptor.getSSTablePreemptiveOpenIntervalInMiB();
DatabaseDescriptor.setSSTablePreemptiveOpenIntervalInMiB(-1);
boolean originalCursorEnabled = DatabaseDescriptor.cursorCompactionEnabled();
try
{
withMeasurementEnv(() -> {
// 4 rounds = 4 input files; big: 192 partitions * 100 rows * ~520B = ~10MB/file
long smallAlloc = measureSteadyStateAllocation(threadMXBean, 19, true, 4, padding, 2, 2);
long smallBytes = lastInputBytes;
@ -200,12 +254,7 @@ public class CursorCompactionAllocationGateTest extends DifferentialCompactionTe
assertTrue(String.format("cursor allocation per input byte too high: %.3f B/B (delta %,dB over %,dB)",
perInputByte, delta, extraBytes),
perInputByte <= 0.5);
}
finally
{
DatabaseDescriptor.setSSTablePreemptiveOpenIntervalInMiB(originalPreemptiveOpen);
DatabaseDescriptor.setCursorCompactionEnabled(originalCursorEnabled);
}
});
}
/** Compacts all live sstables on the cursor path, measuring ONLY execute(); restores inputs. */
@ -247,12 +296,8 @@ public class CursorCompactionAllocationGateTest extends DifferentialCompactionTe
com.sun.management.ThreadMXBean threadMXBean = threadMXBean();
Assume.assumeTrue(threadMXBean != null);
int originalPreemptiveOpen = DatabaseDescriptor.getSSTablePreemptiveOpenIntervalInMiB();
DatabaseDescriptor.setSSTablePreemptiveOpenIntervalInMiB(-1);
boolean originalCursorEnabled = DatabaseDescriptor.cursorCompactionEnabled();
DatabaseDescriptor.setCursorCompactionEnabled(true);
try
{
withMeasurementEnv(() -> {
DatabaseDescriptor.setCursorCompactionEnabled(true);
long smallAlloc = measureSparse(threadMXBean, SMALL_PARTITIONS);
long bigAlloc = measureSparse(threadMXBean, SMALL_PARTITIONS * SCALE);
long delta = bigAlloc - smallAlloc;
@ -262,12 +307,7 @@ public class CursorCompactionAllocationGateTest extends DifferentialCompactionTe
"%,dB -> %,dB, delta %,dB exceeds ceiling %,dB",
smallAlloc, bigAlloc, delta, CEILING_BYTES),
delta <= CEILING_BYTES);
}
finally
{
DatabaseDescriptor.setSSTablePreemptiveOpenIntervalInMiB(originalPreemptiveOpen);
DatabaseDescriptor.setCursorCompactionEnabled(originalCursorEnabled);
}
});
}
private long measureSparse(com.sun.management.ThreadMXBean threadMXBean, int partitions) throws Exception
@ -291,14 +331,7 @@ public class CursorCompactionAllocationGateTest extends DifferentialCompactionTe
}
long gcBefore = cfs.getDefaultGcBefore(FBUtilities.nowInSeconds());
assertCursorPathWillRun(cfs, cfs.getLiveSSTables(), gcBefore);
long best = Long.MAX_VALUE;
for (int i = 0; i < WARMUP_ITERATIONS + MEASURED_ITERATIONS; i++)
{
long allocated = compactOnceMeasured(threadMXBean, cfs, gcBefore);
if (i >= WARMUP_ITERATIONS)
best = Math.min(best, allocated);
}
return best;
return measureBest(threadMXBean, cfs, gcBefore, WARMUP_ITERATIONS, MEASURED_ITERATIONS);
}
/**
@ -316,12 +349,8 @@ public class CursorCompactionAllocationGateTest extends DifferentialCompactionTe
com.sun.management.ThreadMXBean threadMXBean = threadMXBean();
Assume.assumeTrue(threadMXBean != null);
int originalPreemptiveOpen = DatabaseDescriptor.getSSTablePreemptiveOpenIntervalInMiB();
DatabaseDescriptor.setSSTablePreemptiveOpenIntervalInMiB(-1);
boolean originalCursorEnabled = DatabaseDescriptor.cursorCompactionEnabled();
DatabaseDescriptor.setCursorCompactionEnabled(true);
try
{
withMeasurementEnv(() -> {
DatabaseDescriptor.setCursorCompactionEnabled(true);
long smallAlloc = measureWideSparse(threadMXBean, SMALL_PARTITIONS);
long smallBytes = lastInputBytes;
long bigAlloc = measureWideSparse(threadMXBean, SMALL_PARTITIONS * SCALE);
@ -342,12 +371,7 @@ public class CursorCompactionAllocationGateTest extends DifferentialCompactionTe
"%.3f B/B (delta %,dB over %,dB extra input)",
perInputByte, delta, extraBytes),
perInputByte <= 0.6);
}
finally
{
DatabaseDescriptor.setSSTablePreemptiveOpenIntervalInMiB(originalPreemptiveOpen);
DatabaseDescriptor.setCursorCompactionEnabled(originalCursorEnabled);
}
});
}
private long measureWideSparse(com.sun.management.ThreadMXBean threadMXBean, int partitions) throws Exception
@ -398,17 +422,8 @@ public class CursorCompactionAllocationGateTest extends DifferentialCompactionTe
}
long gcBefore = cfs.getDefaultGcBefore(FBUtilities.nowInSeconds());
assertCursorPathWillRun(cfs, cfs.getLiveSSTables(), gcBefore);
lastInputBytes = 0;
for (SSTableReader sstable : cfs.getLiveSSTables())
lastInputBytes += sstable.onDiskLength();
long best = Long.MAX_VALUE;
for (int i = 0; i < WARMUP_ITERATIONS + MEASURED_ITERATIONS; i++)
{
long allocated = compactOnceMeasured(threadMXBean, cfs, gcBefore);
if (i >= WARMUP_ITERATIONS)
best = Math.min(best, allocated);
}
return best;
captureLastInputBytes(cfs);
return measureBest(threadMXBean, cfs, gcBefore, WARMUP_ITERATIONS, MEASURED_ITERATIONS);
}
/**
@ -431,12 +446,8 @@ public class CursorCompactionAllocationGateTest extends DifferentialCompactionTe
com.sun.management.ThreadMXBean threadMXBean = threadMXBean();
Assume.assumeTrue(threadMXBean != null);
int originalPreemptiveOpen = DatabaseDescriptor.getSSTablePreemptiveOpenIntervalInMiB();
DatabaseDescriptor.setSSTablePreemptiveOpenIntervalInMiB(-1);
boolean originalCursorEnabled = DatabaseDescriptor.cursorCompactionEnabled();
DatabaseDescriptor.setCursorCompactionEnabled(true);
try
{
withMeasurementEnv(() -> {
DatabaseDescriptor.setCursorCompactionEnabled(true);
long smallAlloc = measureRangeTombstones(threadMXBean, 12);
long smallBytes = lastInputBytes;
long bigAlloc = measureRangeTombstones(threadMXBean, 96);
@ -455,12 +466,7 @@ public class CursorCompactionAllocationGateTest extends DifferentialCompactionTe
smallAlloc, bigAlloc, delta, extraBytes, perInputByte,
rtPerInputByteCeiling()),
perInputByte <= rtPerInputByteCeiling());
}
finally
{
DatabaseDescriptor.setSSTablePreemptiveOpenIntervalInMiB(originalPreemptiveOpen);
DatabaseDescriptor.setCursorCompactionEnabled(originalCursorEnabled);
}
});
}
/** Calibrated from measured 0.684 B/B — all test-env residual by JFR attribution
@ -493,17 +499,8 @@ public class CursorCompactionAllocationGateTest extends DifferentialCompactionTe
}
long gcBefore = cfs.getDefaultGcBefore(FBUtilities.nowInSeconds());
assertCursorPathWillRun(cfs, cfs.getLiveSSTables(), gcBefore);
lastInputBytes = 0;
for (SSTableReader sstable : cfs.getLiveSSTables())
lastInputBytes += sstable.onDiskLength();
long best = Long.MAX_VALUE;
for (int i = 0; i < WARMUP_ITERATIONS + MEASURED_ITERATIONS; i++)
{
long allocated = compactOnceMeasured(threadMXBean, cfs, gcBefore);
if (i >= WARMUP_ITERATIONS)
best = Math.min(best, allocated);
}
return best;
captureLastInputBytes(cfs);
return measureBest(threadMXBean, cfs, gcBefore, WARMUP_ITERATIONS, MEASURED_ITERATIONS);
}
/**
@ -517,12 +514,8 @@ public class CursorCompactionAllocationGateTest extends DifferentialCompactionTe
com.sun.management.ThreadMXBean threadMXBean = threadMXBean();
Assume.assumeTrue(threadMXBean != null);
int originalPreemptiveOpen = DatabaseDescriptor.getSSTablePreemptiveOpenIntervalInMiB();
DatabaseDescriptor.setSSTablePreemptiveOpenIntervalInMiB(-1);
boolean originalCursorEnabled = DatabaseDescriptor.cursorCompactionEnabled();
DatabaseDescriptor.setCursorCompactionEnabled(true);
try
{
withMeasurementEnv(() -> {
DatabaseDescriptor.setCursorCompactionEnabled(true);
createTable("CREATE TABLE %s (pk bigint, ck bigint, v1 bigint, v2 text, PRIMARY KEY (pk, ck)) " +
"WITH compression = {'enabled': 'false'}");
ColumnFamilyStore cfs = getCurrentColumnFamilyStore();
@ -538,26 +531,9 @@ public class CursorCompactionAllocationGateTest extends DifferentialCompactionTe
long gcBefore = cfs.getDefaultGcBefore(FBUtilities.nowInSeconds());
assertCursorPathWillRun(cfs, cfs.getLiveSSTables(), gcBefore);
for (int i = 0; i < WARMUP_ITERATIONS; i++)
compactOnceMeasured(threadMXBean, cfs, gcBefore);
try (jdk.jfr.Recording recording = new jdk.jfr.Recording())
{
recording.enable("jdk.ObjectAllocationInNewTLAB").withStackTrace();
recording.enable("jdk.ObjectAllocationOutsideTLAB").withStackTrace();
recording.start();
for (int i = 0; i < 30; i++)
compactOnceMeasured(threadMXBean, cfs, gcBefore);
recording.stop();
recording.dump(java.nio.file.Path.of("/tmp/cursor-alloc.jfr"));
}
dumpAllocationProfile(java.nio.file.Path.of("/tmp/cursor-alloc.jfr"), 30, threadMXBean, cfs, gcBefore);
logger.info("allocation profile dumped to /tmp/cursor-alloc.jfr");
}
finally
{
DatabaseDescriptor.setSSTablePreemptiveOpenIntervalInMiB(originalPreemptiveOpen);
DatabaseDescriptor.setCursorCompactionEnabled(originalCursorEnabled);
}
});
}
/** Diagnostic, not a gate: JFR allocation profile over warmed cursor compactions of the
@ -568,12 +544,8 @@ public class CursorCompactionAllocationGateTest extends DifferentialCompactionTe
com.sun.management.ThreadMXBean threadMXBean = threadMXBean();
Assume.assumeTrue(threadMXBean != null);
int originalPreemptiveOpen = DatabaseDescriptor.getSSTablePreemptiveOpenIntervalInMiB();
DatabaseDescriptor.setSSTablePreemptiveOpenIntervalInMiB(-1);
boolean originalCursorEnabled = DatabaseDescriptor.cursorCompactionEnabled();
DatabaseDescriptor.setCursorCompactionEnabled(true);
try
{
withMeasurementEnv(() -> {
DatabaseDescriptor.setCursorCompactionEnabled(true);
createTable("CREATE TABLE %s (pk bigint, ck bigint, v text, PRIMARY KEY (pk, ck)) " +
"WITH compression = {'enabled': 'false'} AND gc_grace_seconds = 864000");
ColumnFamilyStore cfs = getCurrentColumnFamilyStore();
@ -594,26 +566,9 @@ public class CursorCompactionAllocationGateTest extends DifferentialCompactionTe
long gcBefore = cfs.getDefaultGcBefore(FBUtilities.nowInSeconds());
assertCursorPathWillRun(cfs, cfs.getLiveSSTables(), gcBefore);
for (int i = 0; i < WARMUP_ITERATIONS; i++)
compactOnceMeasured(threadMXBean, cfs, gcBefore);
try (jdk.jfr.Recording recording = new jdk.jfr.Recording())
{
recording.enable("jdk.ObjectAllocationInNewTLAB").withStackTrace();
recording.enable("jdk.ObjectAllocationOutsideTLAB").withStackTrace();
recording.start();
for (int i = 0; i < 30; i++)
compactOnceMeasured(threadMXBean, cfs, gcBefore);
recording.stop();
recording.dump(java.nio.file.Path.of("/tmp/cursor-alloc-rt.jfr"));
}
dumpAllocationProfile(java.nio.file.Path.of("/tmp/cursor-alloc-rt.jfr"), 30, threadMXBean, cfs, gcBefore);
logger.info("allocation profile dumped to /tmp/cursor-alloc-rt.jfr");
}
finally
{
DatabaseDescriptor.setSSTablePreemptiveOpenIntervalInMiB(originalPreemptiveOpen);
DatabaseDescriptor.setCursorCompactionEnabled(originalCursorEnabled);
}
});
}
/** Diagnostic: JFR profile at large file sizes; dumps /tmp/cursor-alloc-large.jfr. */
@ -623,12 +578,8 @@ public class CursorCompactionAllocationGateTest extends DifferentialCompactionTe
com.sun.management.ThreadMXBean threadMXBean = threadMXBean();
Assume.assumeTrue(threadMXBean != null);
String padding = "v".repeat(500);
int originalPreemptiveOpen = DatabaseDescriptor.getSSTablePreemptiveOpenIntervalInMiB();
DatabaseDescriptor.setSSTablePreemptiveOpenIntervalInMiB(-1);
boolean originalCursorEnabled = DatabaseDescriptor.cursorCompactionEnabled();
DatabaseDescriptor.setCursorCompactionEnabled(true);
try
{
withMeasurementEnv(() -> {
DatabaseDescriptor.setCursorCompactionEnabled(true);
createTable("CREATE TABLE %s (pk bigint, ck bigint, v1 bigint, v2 text, PRIMARY KEY (pk, ck)) " +
"WITH compression = {'enabled': 'false'}");
ColumnFamilyStore cfs = getCurrentColumnFamilyStore();
@ -642,24 +593,9 @@ public class CursorCompactionAllocationGateTest extends DifferentialCompactionTe
}
long gcBefore = cfs.getDefaultGcBefore(FBUtilities.nowInSeconds());
assertCursorPathWillRun(cfs, cfs.getLiveSSTables(), gcBefore);
for (int i = 0; i < 2; i++)
compactOnceMeasured(threadMXBean, cfs, gcBefore);
try (jdk.jfr.Recording recording = new jdk.jfr.Recording())
{
recording.enable("jdk.ObjectAllocationInNewTLAB").withStackTrace();
recording.enable("jdk.ObjectAllocationOutsideTLAB").withStackTrace();
recording.start();
for (int i = 0; i < 8; i++)
compactOnceMeasured(threadMXBean, cfs, gcBefore);
recording.stop();
recording.dump(java.nio.file.Path.of("/tmp/cursor-alloc-large.jfr"));
}
}
finally
{
DatabaseDescriptor.setSSTablePreemptiveOpenIntervalInMiB(originalPreemptiveOpen);
DatabaseDescriptor.setCursorCompactionEnabled(originalCursorEnabled);
}
dumpAllocationProfile(java.nio.file.Path.of("/tmp/cursor-alloc-large.jfr"), 2, 8, threadMXBean, cfs, gcBefore);
});
}
private static com.sun.management.ThreadMXBean threadMXBean()

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

@ -130,6 +130,15 @@ public abstract class DifferentialCompactionTester extends CQLTester
final List<CapturedSSTable> sstables = new ArrayList<>();
}
/** Concatenates the canonical logical dump of every captured output sstable. */
protected static String allJson(CapturedOutput out)
{
StringBuilder sb = new StringBuilder();
for (CapturedSSTable s : out.sstables)
sb.append(s.json);
return sb.toString();
}
/** Creates the CompactionTask for one differential run. MUST honor keepOriginals=true. */
public interface TaskFactory
{

8
test/unit/org/apache/cassandra/db/compaction/differential/PartialSetDifferentialCompactionTest.java
View File

@ -41,14 +41,6 @@ import org.apache.cassandra.io.sstable.format.SSTableReader;
public class PartialSetDifferentialCompactionTest extends DifferentialCompactionTester
{
private static String allJson(CapturedOutput out)
{
StringBuilder sb = new StringBuilder();
for (CapturedSSTable s : out.sstables)
sb.append(s.json);
return sb.toString();
}
/** Flushes and returns the sstable that flush produced; tracks flush order explicitly. */
private SSTableReader flushAndTrack(ColumnFamilyStore cfs, List<SSTableReader> flushed) throws Throwable
{

8
test/unit/org/apache/cassandra/db/compaction/differential/PurgeBoundaryDifferentialCompactionTest.java
View File

@ -38,14 +38,6 @@ import static org.junit.Assert.assertTrue;
public class PurgeBoundaryDifferentialCompactionTest extends DifferentialCompactionTester
{
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

89
test/unit/org/apache/cassandra/db/compaction/differential/RandomDifferentialCompactionTest.java
View File

@ -270,21 +270,43 @@ public class RandomDifferentialCompactionTest extends DifferentialCompactionTest
assertCursorMatchesIterator(cfs);
}
private static String insertStmt(TableMetadata metadata)
/** Appends {@code columns[i].name}, joined by separator. */
private static void appendNames(StringBuilder sb, List<ColumnMetadata> columns, String separator)
{
StringBuilder sb = new StringBuilder("INSERT INTO ").append(metadata).append(" (");
Iterator<ColumnMetadata> cols = metadata.allColumnsInSelectOrder();
while (cols.hasNext())
for (int i = 0; i < columns.size(); i++)
{
sb.append(cols.next().name.toCQLString());
if (cols.hasNext()) sb.append(", ");
if (i > 0) sb.append(separator);
sb.append(columns.get(i).name.toCQLString());
}
sb.append(") VALUES (");
for (int i = 0; i < metadata.columns().size(); i++)
}
/** Appends "name = ?" equality predicates over columns, joined by separator. */
private static void appendEqPredicates(StringBuilder sb, List<ColumnMetadata> columns, String separator)
{
for (int i = 0; i < columns.size(); i++)
{
if (i > 0) sb.append(separator);
sb.append(columns.get(i).name.toCQLString()).append(" = ?");
}
}
/** Appends {@code count} "?" placeholders, joined by ", ". */
private static void appendPlaceholders(StringBuilder sb, int count)
{
for (int i = 0; i < count; i++)
{
if (i > 0) sb.append(", ");
sb.append('?');
}
}
private static String insertStmt(TableMetadata metadata)
{
List<ColumnMetadata> cols = ImmutableList.copyOf(metadata.allColumnsInSelectOrder());
StringBuilder sb = new StringBuilder("INSERT INTO ").append(metadata).append(" (");
appendNames(sb, cols, ", ");
sb.append(") VALUES (");
appendPlaceholders(sb, cols.size());
return sb.append(')').toString();
}
@ -293,17 +315,9 @@ public class RandomDifferentialCompactionTest extends DifferentialCompactionTest
{
List<ColumnMetadata> keys = primaryKeyColumns(metadata);
StringBuilder sb = new StringBuilder("INSERT INTO ").append(metadata).append(" (");
for (int i = 0; i < keys.size(); i++)
{
if (i > 0) sb.append(", ");
sb.append(keys.get(i).name.toCQLString());
}
appendNames(sb, keys, ", ");
sb.append(") VALUES (");
for (int i = 0; i < keys.size(); i++)
{
if (i > 0) sb.append(", ");
sb.append('?');
}
appendPlaceholders(sb, keys.size());
return sb.append(')').toString();
}
@ -311,18 +325,9 @@ public class RandomDifferentialCompactionTest extends DifferentialCompactionTest
private static String updateStmt(TableMetadata metadata, List<ColumnMetadata> regularColumns)
{
StringBuilder sb = new StringBuilder("UPDATE ").append(metadata).append(" SET ");
for (int i = 0; i < regularColumns.size(); i++)
{
if (i > 0) sb.append(", ");
sb.append(regularColumns.get(i).name.toCQLString()).append(" = ?");
}
appendEqPredicates(sb, regularColumns, ", ");
sb.append(" WHERE ");
List<ColumnMetadata> keys = primaryKeyColumns(metadata);
for (int i = 0; i < keys.size(); i++)
{
if (i > 0) sb.append(" AND ");
sb.append(keys.get(i).name.toCQLString()).append(" = ?");
}
appendEqPredicates(sb, primaryKeyColumns(metadata), " AND ");
return sb.toString();
}
@ -341,18 +346,9 @@ public class RandomDifferentialCompactionTest extends DifferentialCompactionTest
private static String cellDeleteStmt(TableMetadata metadata, List<ColumnMetadata> columns, int keyColumnCount)
{
StringBuilder sb = new StringBuilder("DELETE ");
for (int i = 0; i < columns.size(); i++)
{
if (i > 0) sb.append(", ");
sb.append(columns.get(i).name.toCQLString());
}
appendNames(sb, columns, ", ");
sb.append(" FROM ").append(metadata).append(" WHERE ");
List<ColumnMetadata> keys = primaryKeyColumns(metadata);
for (int i = 0; i < keyColumnCount; i++)
{
if (i > 0) sb.append(" AND ");
sb.append(keys.get(i).name.toCQLString()).append(" = ?");
}
appendEqPredicates(sb, primaryKeyColumns(metadata).subList(0, keyColumnCount), " AND ");
return sb.toString();
}
@ -366,11 +362,7 @@ public class RandomDifferentialCompactionTest extends DifferentialCompactionTest
List<ColumnMetadata> keys = primaryKeyColumns(metadata);
int partitionColumnCount = metadata.partitionKeyColumns().size();
int bound = partitionColumnCount + eqDepth;
for (int i = 0; i < bound; i++)
{
if (i > 0) sb.append(" AND ");
sb.append(keys.get(i).name.toCQLString()).append(" = ?");
}
appendEqPredicates(sb, keys.subList(0, bound), " AND ");
sb.append(" AND ").append(keys.get(bound).name.toCQLString()).append(' ').append(op).append(" ?");
return sb.toString();
}
@ -394,12 +386,7 @@ public class RandomDifferentialCompactionTest extends DifferentialCompactionTest
private static String deleteStmt(TableMetadata metadata, int keyColumnCount)
{
StringBuilder sb = new StringBuilder("DELETE FROM ").append(metadata).append(" WHERE ");
List<ColumnMetadata> keys = primaryKeyColumns(metadata);
for (int i = 0; i < keyColumnCount; i++)
{
if (i > 0) sb.append(" AND ");
sb.append(keys.get(i).name.toCQLString()).append(" = ?");
}
appendEqPredicates(sb, primaryKeyColumns(metadata).subList(0, keyColumnCount), " AND ");
return sb.toString();
}