From d6393e217f0edff6782e851d5d377f1fe6422772 Mon Sep 17 00:00:00 2001 From: Jon Haddad Date: Thu, 11 Jun 2026 16:16:19 -0700 Subject: [PATCH] Widen index block start offset to long; partitions past 2GiB corrupted the index MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CursorIndexWriter.indexBlockStartOffset was an int, unlike the reference SortedTablePartitionWriter's long. Partitions larger than 2GiB wrap the offset negative past Integer.MAX_VALUE, and the block-cut arithmetic then cuts every row — corrupting the serialized offsets in the promoted index. Pinned by the giant-partition boundary run (ck_stride = rows_per_sstable, forcing disjoint windows so the merged partition size is maximized rather than halved by overlap). Also: the differential harness's structural verifier debug-logs every index block during its extended walk (~560K lines for a >2GiB partition), and ant's junit formatter buffers all test output in memory — at that log volume the buffering, not the verification itself, OOMs the test fork. The verifier's debug stream is now silenced in scale mode. Also corrects the giant-partition scenario's own math: the logged/documented partition size used sstables * rows_per_sstable (the total PRE-MERGE input rows), not the actual MERGED partition size, which with the default half-window overlap is (sstables-1) * ck_stride + rows_per_sstable — a smaller number. A prior "2.6GiB" boundary run under this miscount actually reached only ~1.4GiB. ck_stride is now a knob (defaulting to the existing half-window overlap); setting it equal to rows_per_sstable produces disjoint windows and the true worst-case merged partition size, needed to genuinely exercise the 2GiB boundary. --- .../io/sstable/CursorIndexWriter.java | 12 ++++-- .../io/sstable/SSTableCursorWriter.java | 3 +- .../DifferentialCompactionTester.java | 10 ++++- ...gePartitionDifferentialCompactionTest.java | 37 ++++++++++++------- 4 files changed, 42 insertions(+), 20 deletions(-) diff --git a/src/java/org/apache/cassandra/io/sstable/CursorIndexWriter.java b/src/java/org/apache/cassandra/io/sstable/CursorIndexWriter.java index 62cbec6d64..9a18a04a7f 100644 --- a/src/java/org/apache/cassandra/io/sstable/CursorIndexWriter.java +++ b/src/java/org/apache/cassandra/io/sstable/CursorIndexWriter.java @@ -35,8 +35,12 @@ import org.apache.cassandra.db.DeletionTime; public abstract class CursorIndexWriter { protected long partitionStart; - // Offset within the current partition where the current index block begins. - protected int indexBlockStartOffset; + // Offset within the current partition where the current index block begins. MUST be a + // long, like the reference SortedTablePartitionWriter.indexBlockStartOffset: partitions + // can exceed 2GiB, and an int here wraps negative past Integer.MAX_VALUE — the block-cut + // arithmetic then cuts every row and the serialized offsets corrupt the promoted index + // (pinned by the giant-partition boundary run, ck_stride = rows_per_sstable). + protected long indexBlockStartOffset; public final void startPartition(long partitionStartPosition, long positionAfterHeader) { @@ -51,14 +55,14 @@ public abstract class CursorIndexWriter notePosition(position); } - public final int indexBlockStartOffset() + public final long indexBlockStartOffset() { return indexBlockStartOffset; } protected final void notePosition(long endOfRowPosition) { - indexBlockStartOffset = (int) (endOfRowPosition - partitionStart); + indexBlockStartOffset = endOfRowPosition - partitionStart; } protected final long currentOffsetInPartition(long position) diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableCursorWriter.java b/src/java/org/apache/cassandra/io/sstable/SSTableCursorWriter.java index 0e6c35b4e7..4fbce2cb6c 100644 --- a/src/java/org/apache/cassandra/io/sstable/SSTableCursorWriter.java +++ b/src/java/org/apache/cassandra/io/sstable/SSTableCursorWriter.java @@ -162,7 +162,8 @@ public class SSTableCursorWriter implements AutoCloseable previousRowStartOffset = 0; writePartitionHeader(partitionKey, partitionKeyLength, partitionDeletionTime); cursorIndexWriter.startPartition(partitionStart, dataWriter.position()); - return cursorIndexWriter.indexBlockStartOffset(); + // immediately after startPartition this is the partition header length — always small + return Math.toIntExact(cursorIndexWriter.indexBlockStartOffset()); } public void writePartitionEnd(byte[] partitionKey, int partitionKeyLength, DeletionTime partitionDeletionTime, int headerLength) throws IOException 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 54aaf61321..ff03dc82ee 100644 --- a/test/unit/org/apache/cassandra/db/compaction/differential/DifferentialCompactionTester.java +++ b/test/unit/org/apache/cassandra/db/compaction/differential/DifferentialCompactionTester.java @@ -373,8 +373,14 @@ public abstract class DifferentialCompactionTester extends CQLTester private CapturedSSTable capture(ColumnFamilyStore cfs, SSTableReader sstable, Path dir) throws IOException { - // 1. structural verification of the output - try (IVerifier verifier = sstable.getVerifier(cfs, new OutputHandler.LogOutput(), false, + // 1. structural verification of the output. In scale mode the verifier's debug + // stream must be silenced: the extended index walk debug-logs EVERY index block + // (~560K lines for a >2GiB partition), and ant's junit formatter buffers all test + // output in memory — the log volume, not the verification, OOMs the fork. + OutputHandler verifyOutput = scaleCapture() + ? new OutputHandler.LogOutput() { @Override public void debug(String msg) {} } + : new OutputHandler.LogOutput(); + try (IVerifier verifier = sstable.getVerifier(cfs, verifyOutput, false, IVerifier.options().invokeDiskFailurePolicy(true) .extendedVerification(true).build())) { diff --git a/test/unit/org/apache/cassandra/db/compaction/differential/LargePartitionDifferentialCompactionTest.java b/test/unit/org/apache/cassandra/db/compaction/differential/LargePartitionDifferentialCompactionTest.java index 5985974d60..501ce7ed00 100644 --- a/test/unit/org/apache/cassandra/db/compaction/differential/LargePartitionDifferentialCompactionTest.java +++ b/test/unit/org/apache/cassandra/db/compaction/differential/LargePartitionDifferentialCompactionTest.java @@ -36,17 +36,22 @@ import org.apache.cassandra.db.ColumnFamilyStore; * resets after the monster. * * Parameters (defaults are CI-sane: ~1M rows, ~160MB partition, still ~40K index blocks at - * the 4KiB test column_index_size). The 2GiB boundary run: + * the 4KiB test column_index_size). The MERGED partition size is what crosses boundaries: + * with the default half-window overlap, distinct rows = (sstables-1) * ck_stride + + * rows_per_sstable — NOT sstables * rows_per_sstable (an earlier "2.6GiB" boundary run + * conflated pre-merge input rows with merged output rows and only reached ~1.4GiB). + * The true 2GiB boundary run uses DISJOINT windows (ck_stride = rows_per_sstable): * * ant testsome -Dtest.name=...LargePartitionDifferentialCompactionTest \ * -Dtest.timeout=14400000 \ * -Dtest.jvm.args="-Dcassandra.test.differential.largepartition.sstables=8 - * -Dcassandra.test.differential.largepartition.rows_per_sstable=1200000 - * -Dcassandra.test.differential.largepartition.value_padding=240" + * -Dcassandra.test.differential.largepartition.rows_per_sstable=1100000 + * -Dcassandra.test.differential.largepartition.value_padding=240 + * -Dcassandra.test.differential.largepartition.ck_stride=1100000" * - * That is ~9.6M rows at ~280B/row: a ~2.6GiB single partition. Peak disk for the full - * two-generation differential (inputs + live output + four captured output copies) is - * roughly 7x the partition size. The memtable may auto-flush large rounds, so the sstables + * That is ~8.8M distinct rows at ~280B/row: a ~2.4GiB single MERGED partition. Peak disk for + * the full two-generation differential (inputs + live output + four captured output copies) + * is roughly 7x the partition size. The memtable may auto-flush large rounds, so the sstables * parameter is a minimum input count, which the differential does not care about. */ public class LargePartitionDifferentialCompactionTest extends DifferentialCompactionTester @@ -58,8 +63,13 @@ public class LargePartitionDifferentialCompactionTest extends DifferentialCompac Integer.getInteger("cassandra.test.differential.largepartition.rows_per_sstable", 250_000); private static final String VALUE_PADDING = "p".repeat(Integer.getInteger("cassandra.test.differential.largepartition.value_padding", 120)); - /** Half-window overlap between rounds: every output row in the overlap merges from two inputs. */ - private static final long CK_STRIDE = Math.max(1, ROWS_PER_SSTABLE / 2); + /** + * Window stride between rounds. Default: half-window overlap, so every output row in the + * overlap merges from two inputs. Set equal to rows_per_sstable for DISJOINT windows, + * which maximizes the merged partition size (the 2GiB boundary run). + */ + private static final long CK_STRIDE = Math.max(1, + Integer.getInteger("cassandra.test.differential.largepartition.ck_stride", ROWS_PER_SSTABLE / 2)); @Override protected boolean scaleCapture() @@ -70,11 +80,12 @@ public class LargePartitionDifferentialCompactionTest extends DifferentialCompac @Test public void giantPartition() throws Throwable { - logger.info("large-partition parameters: sstables={} rowsPerSSTable={} valuePadding={}B " + - "-> ~{} rows in one partition, ~{}MB serialized", - SSTABLES, ROWS_PER_SSTABLE, VALUE_PADDING.length(), - (long) SSTABLES * ROWS_PER_SSTABLE, - (long) SSTABLES * ROWS_PER_SSTABLE * (40 + VALUE_PADDING.length()) / (1 << 20)); + long distinctRows = (long) (SSTABLES - 1) * CK_STRIDE + ROWS_PER_SSTABLE; + logger.info("large-partition parameters: sstables={} rowsPerSSTable={} ckStride={} valuePadding={}B " + + "-> ~{} distinct rows in one merged partition, ~{}MB serialized", + SSTABLES, ROWS_PER_SSTABLE, CK_STRIDE, VALUE_PADDING.length(), + distinctRows, + distinctRows * (40 + VALUE_PADDING.length()) / (1 << 20)); createTable("CREATE TABLE %s (pk bigint, ck bigint, v1 bigint, v2 text, " + "PRIMARY KEY (pk, ck)) WITH gc_grace_seconds = 864000");