diff --git a/test/unit/org/apache/cassandra/db/compaction/differential/BigVolumeDifferentialCompactionTest.java b/test/unit/org/apache/cassandra/db/compaction/differential/BigVolumeDifferentialCompactionTest.java new file mode 100644 index 0000000000..38486dfcc3 --- /dev/null +++ b/test/unit/org/apache/cassandra/db/compaction/differential/BigVolumeDifferentialCompactionTest.java @@ -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 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); + } +} 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 c08cf5fff8..07a33982af 100644 --- a/test/unit/org/apache/cassandra/db/compaction/differential/DifferentialCompactionTester.java +++ b/test/unit/org/apache/cassandra/db/compaction/differential/DifferentialCompactionTester.java @@ -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 components = new TreeSet<>(); components.addAll(it.componentSizes.keySet()); components.addAll(cu.componentSizes.keySet()); List 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);