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

Add two-million-row volume scenario; stream the harness comparisons 

Writes 2,000,000 rows across twenty input sstables — 2,000 partitions x 50 rows per round
x 20 flush rounds, each round's clustering window overlapping the previous by half so
most output rows genuinely merge from two or three inputs — with a realistic mutation mix
(TTL'd rows, explicit-timestamp ties on overwritten keys, null-overwrite cell tombstones,
row/range/partition deletes).

To keep harness memory flat at this scale, capture gains a scale mode: the logical dump
streams through the expired-field normalization into a SHA-256 digest instead of being
retained as a String, and the byte-level comparison streams with 64KB buffers instead of
reading whole components into memory. Byte-identical on both paths.
This commit is contained in:
Jon Haddad 2026-07-27 11:10:02 -07:00
parent 277c4059d3
commit c5506c2372
2 changed files with 285 additions and 42 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

114
test/unit/org/apache/cassandra/db/compaction/differential/BigVolumeDifferentialCompactionTest.java Normal file
View File

@ -0,0 +1,114 @@
/*
* 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.db.ColumnFamilyStore;
/**
* Volume scenario: TWO MILLION written rows across TWENTY input sstables — 2,000 partitions
* x 50 rows per round x 20 flush rounds, with each round's clustering window overlapping the
* previous round's by half so most output rows genuinely merge from 2-3 inputs. The mutation
* mix runs at scale: ~14% TTL'd rows, ~8% explicit-timestamp ties on overwritten keys, ~3%
* null-overwrite cell tombstones, plus per-round row deletes, bounded and single-sided range
* deletes, and cycling partition deletes with resurrection.
*
* Runs in scale-capture mode (see DifferentialCompactionTester.scaleCapture): the logical
* dump streams into a SHA-256 digest and byte comparison streams, so harness memory stays
* flat. Two generations as everywhere.
*
* Runtime is minutes, dominated by the 2M inserts — run with a raised junit timeout:
* ant testsome -Dtest.name=...BigVolumeDifferentialCompactionTest -Dtest.timeout=2400000
*/
public class BigVolumeDifferentialCompactionTest extends DifferentialCompactionTester
{
private static final Set<String> ALLOWLIST = Set.of();
private static final int ROUNDS = 20;
private static final int PARTITIONS = 2000;
private static final int ROWS_PER_ROUND = 50;
private static final int CK_STRIDE = 23; // < ROWS_PER_ROUND: consecutive rounds overlap by 27 cks
@Override
protected boolean scaleCapture()
{
return true;
}
@Test
public void twoMillionRowsTwentySSTables() throws Throwable
{
createTable("CREATE TABLE %s (pk bigint, ck bigint, v1 bigint, v2 text, " +
"PRIMARY KEY (pk, ck)) WITH gc_grace_seconds = 864000");
ColumnFamilyStore cfs = getCurrentColumnFamilyStore();
cfs.disableAutoCompaction();
String insert = "INSERT INTO %s (pk, ck, v1, v2) VALUES (?, ?, ?, ?)";
String insertTtl = insert + " USING TTL 86400";
String insertTs = insert + " USING TIMESTAMP 5000";
for (int round = 0; round < ROUNDS; round++)
{
long ckBase = (long) round * CK_STRIDE;
for (long pk = 0; pk < PARTITIONS; pk++)
{
for (int j = 0; j < ROWS_PER_ROUND; j++)
{
long ck = ckBase + j;
long v1 = ck * 31 + round;
String v2 = j % 31 == 30 ? null : "v" + round + "_" + ck;
if (j % 7 == 3)
execute(insertTtl, pk, ck, v1, v2);
else if (j % 13 == 7)
execute(insertTs, pk, ck, v1, "tie" + round + "_" + ck);
else
execute(insert, pk, ck, v1, v2);
}
}
for (int i = 0; i < 100; i++)
execute("DELETE FROM %s WHERE pk = ? AND ck = ?",
(long) ((round * 97 + i * 13) % PARTITIONS), ckBase + (i % ROWS_PER_ROUND));
for (int i = 0; i < 30; i++)
{
long start = ckBase + (i % 40);
execute("DELETE FROM %s WHERE pk = ? AND ck >= ? AND ck < ?",
(long) ((round * 53 + i * 29) % PARTITIONS), start, start + 4);
}
for (int i = 0; i < 20; i++)
{
long pk = (round * 41 + i * 17) % PARTITIONS;
if (i % 2 == 0)
execute("DELETE FROM %s WHERE pk = ? AND ck >= ?", pk, ckBase + 25);
else
execute("DELETE FROM %s WHERE pk = ? AND ck <= ?", pk, ckBase + 5);
}
// cycling partition deletes over the same 5 partitions: tombstone + resurrection
execute("DELETE FROM %s WHERE pk = ?", (long) (1995 + round % 5));
flush();
logger.info("big-volume round {}/{} flushed", round + 1, ROUNDS);
}
assertCursorMatchesIteratorAcrossGenerations(cfs, ALLOWLIST);
}
}

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

@ -91,6 +91,18 @@ public abstract class DifferentialCompactionTester extends CQLTester
/** Fixed "now" used for JSON dumps so rendering cannot depend on wall clock. */
private static final long DUMP_NOW_SEC = 0;
/**
* Scale mode for very large scenarios (millions of rows): the logical dump is streamed
* into a SHA-256 digest instead of being retained as a String, so capture memory stays
* flat regardless of row count. Byte comparison always streams. On a digest mismatch
* the byte-level comparison (which still reports exact offsets) is the debugging tool;
* rerun a reduced scenario without scale mode for a row-level JSON diff.
*/
protected boolean scaleCapture()
{
return false;
}
public static final class CapturedSSTable
{
final Path dir; // copied component files, named by component (e.g. "Data.db")
@ -374,20 +386,42 @@ public abstract class DifferentialCompactionTester extends CQLTester
}
// 2. canonical logical dump
ByteArrayOutputStream baos = new ByteArrayOutputStream();
try (ISSTableScanner scanner = sstable.getScanner())
{
JsonTransformer.toJsonLines(scanner, Util.iterToStream(scanner), true, false,
sstable.metadata(), DUMP_NOW_SEC, baos);
}
// JsonTransformer's "expired" fields are computed from WALL CLOCK
// (currentTimeMillis), ignoring the fixed nowInSec passed above — so byte-identical
// (currentTimeMillis), ignoring the fixed nowInSec passed below — so byte-identical
// outputs can render differently when a localExpirationTime falls between the two
// paths' captures, which run seconds apart (materialized-view expired-liveness rows
// sit permanently on that boundary: their expiration IS the write second). The flag
// is derived from expires_at, which is still compared, so normalize it out.
String json = baos.toString(StandardCharsets.UTF_8)
.replaceAll("\"expired\"\\s*:\\s*(true|false)", "\"expired\":\"normalized\"");
String json;
if (scaleCapture())
{
// stream into a digest: capture memory stays flat at millions of rows
try (ISSTableScanner scanner = sstable.getScanner())
{
java.security.MessageDigest digest = java.security.MessageDigest.getInstance("SHA-256");
NormalizingDigestOutputStream out = new NormalizingDigestOutputStream(digest);
JsonTransformer.toJsonLines(scanner, Util.iterToStream(scanner), true, false,
sstable.metadata(), DUMP_NOW_SEC, out);
out.flushTail();
json = "sha256:" + org.apache.cassandra.utils.Hex.bytesToHex(digest.digest()) +
" (" + out.bytesSeen + " bytes)";
}
catch (java.security.NoSuchAlgorithmException e)
{
throw new AssertionError(e);
}
}
else
{
ByteArrayOutputStream baos = new ByteArrayOutputStream();
try (ISSTableScanner scanner = sstable.getScanner())
{
JsonTransformer.toJsonLines(scanner, Util.iterToStream(scanner), true, false,
sstable.metadata(), DUMP_NOW_SEC, baos);
}
json = baos.toString(StandardCharsets.UTF_8)
.replaceAll("\"expired\"\\s*:\\s*(true|false)", "\"expired\":\"normalized\"");
}
// 3. stats spot-check summary
StatsMetadata stats = sstable.getSSTableMetadata();
@ -421,38 +455,76 @@ public abstract class DifferentialCompactionTester extends CQLTester
CapturedSSTable it = iterator.sstables.get(i);
CapturedSSTable cu = cursor.sstables.get(i);
assertEquals("stats summary divergence in output sstable " + i, it.statsSummary, cu.statsSummary);
if (!it.json.equals(cu.json))
// logical first: a row-level diff is far more debuggable than a stats mismatch.
// In scale mode the dump is a digest — defer it below the byte comparison, which
// still localizes divergences to exact offsets.
boolean digestMode = it.json.startsWith("sha256:");
if (!digestMode && !it.json.equals(cu.json))
fail("LOGICAL divergence in output sstable " + i + " (iterator vs cursor):\n" + firstJsonDiff(it.json, cu.json) +
"\niterator stats: " + it.statsSummary + "\ncursor stats: " + cu.statsSummary);
assertEquals("stats summary divergence in output sstable " + i, it.statsSummary, cu.statsSummary);
SortedSet<String> components = new TreeSet<>();
components.addAll(it.componentSizes.keySet());
components.addAll(cu.componentSizes.keySet());
List<String> divergences = new ArrayList<>();
for (String comp : components)
{
byte[] a = readComponent(it.dir, comp);
byte[] b = readComponent(cu.dir, comp);
if (java.util.Arrays.equals(a, b))
Path a = it.dir.resolve(comp);
Path b = cu.dir.resolve(comp);
boolean hasA = Files.exists(a);
boolean hasB = Files.exists(b);
if (hasA != hasB)
{
if (!byteDiffAllowlist.contains(comp))
divergences.add(String.format(" %s: present only in %s path", comp, hasA ? "iterator" : "cursor"));
continue;
}
if (!hasA)
continue;
long firstDiff = firstFileDifference(a, b);
if (firstDiff < 0)
continue;
if (byteDiffAllowlist.contains(comp))
continue; // logical equivalence already asserted above
divergences.add(describeByteDiff(comp, a, b));
continue; // logical equivalence still asserted
divergences.add(describeFileDiff(comp, a, b, firstDiff));
}
if (!divergences.isEmpty())
fail("BYTE divergence in output sstable " + i + " (iterator vs cursor):\n" + String.join("\n", divergences) +
"\nIf a divergence is benign it must be added to the allowlist with a justifying comment next to the allowlist entry");
if (digestMode)
assertEquals("logical dump digest divergence in output sstable " + i +
" (scale mode; rerun a reduced scenario without scale mode for a row-level diff)",
it.json, cu.json);
}
}
private static byte[] readComponent(Path dir, String component)
/** Streaming comparison: -1 if byte-identical, else the offset of the first difference
* (the shorter length when one file is a prefix of the other). */
private static long firstFileDifference(Path a, Path b)
{
try
try (java.io.InputStream ia = new java.io.BufferedInputStream(Files.newInputStream(a), 1 << 16);
java.io.InputStream ib = new java.io.BufferedInputStream(Files.newInputStream(b), 1 << 16))
{
Path p = dir.resolve(component);
return Files.exists(p) ? Files.readAllBytes(p) : null;
byte[] bufA = new byte[1 << 16];
byte[] bufB = new byte[1 << 16];
long offset = 0;
while (true)
{
int readA = ia.readNBytes(bufA, 0, bufA.length);
int readB = ib.readNBytes(bufB, 0, bufB.length);
int common = Math.min(readA, readB);
int mismatch = java.util.Arrays.mismatch(bufA, 0, common, bufB, 0, common);
if (mismatch >= 0)
return offset + mismatch;
if (readA != readB)
return offset + common; // same prefix, different length
if (readA == 0)
return -1;
offset += readA;
}
}
catch (IOException e)
{
@ -460,42 +532,99 @@ public abstract class DifferentialCompactionTester extends CQLTester
}
}
private static String describeByteDiff(String component, byte[] a, byte[] b)
private static String describeFileDiff(String component, Path a, Path b, long firstDiff)
{
if (a == null || b == null)
return String.format(" %s: present only in %s path", component, a == null ? "cursor" : "iterator");
int firstDiff = -1;
int max = Math.min(a.length, b.length);
for (int i = 0; i < max; i++)
try
{
if (a[i] != b[i]) { firstDiff = i; break; }
return String.format(" %s: lengths %d vs %d, first divergence at offset %d%n iterator: %s%n cursor: %s",
component, Files.size(a), Files.size(b), firstDiff,
hexContext(a, firstDiff), hexContext(b, firstDiff));
}
catch (IOException e)
{
throw new UncheckedIOException(e);
}
if (firstDiff == -1)
firstDiff = max; // same prefix, different length
return String.format(" %s: lengths %d vs %d, first divergence at offset %d%n iterator: %s%n cursor: %s",
component, a.length, b.length, firstDiff,
hexContext(a, firstDiff), hexContext(b, firstDiff));
}
private static String hexContext(byte[] bytes, int offset)
private static String hexContext(Path file, long offset) throws IOException
{
int from = Math.max(0, offset - 8);
int to = Math.min(bytes.length, offset + 24);
StringBuilder sb = new StringBuilder();
for (int i = from; i < to; i++)
long size = Files.size(file);
long from = Math.max(0, offset - 8);
int len = (int) Math.min(size - from, 32);
byte[] window = new byte[Math.max(len, 0)];
try (java.io.InputStream in = Files.newInputStream(file))
{
if (i == offset)
in.skipNBytes(from);
in.readNBytes(window, 0, window.length);
}
StringBuilder sb = new StringBuilder();
for (int i = 0; i < window.length; i++)
{
long abs = from + i;
if (abs == offset)
sb.append('[');
sb.append(String.format("%02x", bytes[i]));
if (i == offset)
sb.append(String.format("%02x", window[i]));
if (abs == offset)
sb.append(']');
sb.append(' ');
}
if (to < bytes.length)
if (from + window.length < size)
sb.append("...");
return sb.toString();
}
/** Streams a JSON dump into a digest, normalizing the wall-clock-derived "expired"
* fields line by line (toJsonLines emits one partition per line). */
private static final class NormalizingDigestOutputStream extends java.io.OutputStream
{
private final java.security.MessageDigest digest;
private final ByteArrayOutputStream line = new ByteArrayOutputStream();
long bytesSeen;
NormalizingDigestOutputStream(java.security.MessageDigest digest)
{
this.digest = digest;
}
@Override
public void write(int b)
{
line.write(b);
if (b == '\n')
flushTail();
}
@Override
public void write(byte[] b, int off, int len)
{
int start = off;
int end = off + len;
for (int i = off; i < end; i++)
{
if (b[i] == '\n')
{
line.write(b, start, i - start + 1);
flushTail();
start = i + 1;
}
}
if (start < end)
line.write(b, start, end - start);
}
void flushTail()
{
if (line.size() == 0)
return;
byte[] normalized = line.toString(StandardCharsets.UTF_8)
.replaceAll("\"expired\"\\s*:\\s*(true|false)", "\"expired\":\"normalized\"")
.getBytes(StandardCharsets.UTF_8);
digest.update(normalized);
bytesSeen += normalized.length;
line.reset();
}
}
private static String firstJsonDiff(String a, String b)
{
String[] linesA = a.split("\n", -1);