From adc152462766ca1c4127aeb50442219d4dbb5268 Mon Sep 17 00:00:00 2001 From: Mykyta Bozhenko <21245729+cheeeee@users.noreply.github.com> Date: Fri, 11 Sep 2026 18:50:21 -0400 Subject: [PATCH] Advance the offline SSTable rewrite cache advice position Record the current output byte position after an offline cache-skip request so later partitions wait for the next preemptive interval. Exercise request ranges across populated and empty writer switches. CASSANDRA-19569 Generated-by: Claude (Anthropic) --- CHANGES.txt | 1 + .../cassandra/io/sstable/SSTableRewriter.java | 3 +- .../io/sstable/SSTableRewriterTest.java | 115 ++++++++++++++++++ 3 files changed, 118 insertions(+), 1 deletion(-) diff --git a/CHANGES.txt b/CHANGES.txt index aa356a2da163..97e5d9785709 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 7.0 + * Advance the offline SSTable rewrite position after cache advice to preserve the preemptive interval (CASSANDRA-19569) * Allow CQLSSTableWriter to specify SSTable id generator to use (CASSANDRA-21012) * Reject LIKE patterns with a wildcard (%) anywhere other than the start or end (CASSANDRA-21068) * Support pluggable default role initialization (CASSANDRA-21546) diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableRewriter.java b/src/java/org/apache/cassandra/io/sstable/SSTableRewriter.java index aaae5ecab876..254b4df7540f 100644 --- a/src/java/org/apache/cassandra/io/sstable/SSTableRewriter.java +++ b/src/java/org/apache/cassandra/io/sstable/SSTableRewriter.java @@ -59,7 +59,7 @@ public class SSTableRewriter extends Transactional.AbstractTransactional impleme private final ILifecycleTransaction transaction; // the readers we are rewriting (updated as they are replaced) private final List preparedForCommit = new ArrayList<>(); - private long currentlyOpenedEarlyAt; // the position (in MiB) in the target file we last (re)opened at + private long currentlyOpenedEarlyAt; // the byte position in the target file we last reopened or skipped cache at private long bytesWritten; // the bytes written by previous writers, or zero if the current writer is the first writer private final List writers = new ArrayList<>(); @@ -166,6 +166,7 @@ private void maybeReopenEarly(DecoratedKey key) { reader.trySkipFileCacheBefore(key); } + currentlyOpenedEarlyAt = writer.getFilePointer(); } else { diff --git a/test/unit/org/apache/cassandra/io/sstable/SSTableRewriterTest.java b/test/unit/org/apache/cassandra/io/sstable/SSTableRewriterTest.java index 69414329cce3..337ddb743705 100644 --- a/test/unit/org/apache/cassandra/io/sstable/SSTableRewriterTest.java +++ b/test/unit/org/apache/cassandra/io/sstable/SSTableRewriterTest.java @@ -19,11 +19,13 @@ package org.apache.cassandra.io.sstable; import java.nio.ByteBuffer; +import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; import java.util.HashSet; import java.util.LinkedHashSet; import java.util.List; +import java.util.Map; import java.util.Set; import java.util.concurrent.ExecutionException; import java.util.concurrent.atomic.AtomicBoolean; @@ -31,7 +33,10 @@ import com.google.common.collect.Iterables; import com.google.common.collect.Sets; +import org.jboss.byteman.contrib.bmunit.BMRule; +import org.jboss.byteman.contrib.bmunit.BMUnitRunner; import org.junit.Test; +import org.junit.runner.RunWith; import org.apache.cassandra.UpdateBuilder; import org.apache.cassandra.Util; @@ -56,6 +61,7 @@ import org.apache.cassandra.dht.ByteOrderedPartitioner; import org.apache.cassandra.dht.Range; import org.apache.cassandra.dht.Token; +import org.apache.cassandra.io.sstable.format.SSTableFormat.Components; import org.apache.cassandra.io.sstable.format.SSTableReader; import org.apache.cassandra.io.sstable.format.SSTableWriter; import org.apache.cassandra.io.util.File; @@ -65,6 +71,7 @@ import org.apache.cassandra.utils.FBUtilities; import static java.util.Collections.singletonList; +import static java.util.Collections.singletonMap; import static org.apache.cassandra.db.compaction.OperationType.COMPACTION; import static org.apache.cassandra.utils.FBUtilities.nowInSeconds; import static org.apache.cassandra.utils.TimeUUID.Generator.nextTimeUUID; @@ -75,8 +82,11 @@ import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; +@RunWith(BMUnitRunner.class) public class SSTableRewriterTest extends SSTableWriterTestBase { + private static final ThreadLocal>> cacheSkipRequests = new ThreadLocal<>(); + @Test public void basicTest() { @@ -955,4 +965,109 @@ public static Set writeFiles(ColumnFamilyStore cfs, int fileCount } return result; } + + @Test + @BMRule(name = "record offline rewriter cache skip ranges", + targetClass = "org.apache.cassandra.utils.NativeLibrary", + targetMethod = "trySkipCache(java.lang.String, long, long)", + targetLocation = "AT ENTRY", + action = "org.apache.cassandra.io.sstable.SSTableRewriterTest.recordCacheSkip($1, $2, $3)") + public void testOfflineCacheSkippingAcrossWriterSwitches() throws Exception + { + ColumnFamilyStore cfs = Keyspace.open(KEYSPACE).getColumnFamilyStore(CF); + truncate(cfs); + + int partitionsPerWriter = 16; + int writerCount = 3; + List keys = new ArrayList<>(); + for (int i = 0; i < partitionsPerWriter * writerCount; i++) + keys.add(cfs.decorateKey(ByteBufferUtil.bytes(Integer.toString(1000 + i)))); + keys.sort(null); + + File tempDir = new File(com.google.common.io.Files.createTempDir()); + try + { + SSTableReader source; + Descriptor desc = cfs.newSSTableDescriptor(tempDir); + try (SSTableTxnWriter sstableWriter = SSTableTxnWriter.create(cfs, desc, 0, 0, null, false, new SerializationHeader(true, cfs.metadata(), cfs.metadata().regularAndStaticColumns(), EncodingStats.NO_STATS))) + { + for (DecoratedKey key : keys) + { + UpdateBuilder builder = UpdateBuilder.create(cfs.metadata(), key.getKey()).withTimestamp(1); + builder.newRow("0").add("val", ByteBufferUtil.bytes("value")); + try (UnfilteredRowIterator partition = builder.build().unfilteredIterator()) + { + sstableWriter.append(partition); + } + } + source = sstableWriter.finish(true).iterator().next(); + } + + try + { + // Fixed-width keys and identical rows give equal-sized partitions in either SSTable format. + long partitionSize = source.getPosition(keys.get(1), SSTableReader.Operator.EQ); + assertTrue(partitionSize > 0); + for (int i = 0; i < keys.size(); i++) + assertEquals(i * partitionSize, source.getPosition(keys.get(i), SSTableReader.Operator.EQ)); + + List actual = new ArrayList<>(); + List expected = new ArrayList<>(); + cacheSkipRequests.set(singletonMap(source.descriptor.fileFor(Components.DATA).absolutePath(), actual)); + // Keep the input caller-owned on both commit and abort; release it in the finally block. + try (LifecycleTransaction txn = LifecycleTransaction.offline(OperationType.UPGRADE_SSTABLES, source); + SSTableRewriter rewriter = new SSTableRewriter(txn, 1000, 4 * partitionSize, true, true); + ISSTableScanner scanner = source.getScanner()) + { + for (int w = 0; w < writerCount; w++) + { + rewriter.switchWriter(getWriter(cfs, tempDir, txn)); + assertEquals(expected, actual); + for (int p = 0; p < partitionsPerWriter; p++) + { + int index = w * partitionsPerWriter + p; + assertTrue(scanner.hasNext()); + try (UnfilteredRowIterator partition = scanner.next()) + { + rewriter.append(partition); + } + + // The check precedes append and is strictly greater than four partitions. + // Expect skips before partitions 5, 10 and 15, then restart after each switch. + if (p > 0 && p % 5 == 0) + expected.add(index * partitionSize); + assertEquals("cache skip ranges at writer " + w + ", partition " + p, expected, actual); + } + } + assertFalse(scanner.hasNext()); + for (SSTableReader reader : rewriter.finish()) + reader.selfRef().release(); + } + } + finally + { + cacheSkipRequests.remove(); + source.selfRef().release(); + } + } + finally + { + LifecycleTransaction.waitForDeletions(); + FileUtils.deleteRecursive(tempDir); + truncate(cfs); + } + } + + public static void recordCacheSkip(String path, long offset, long length) + { + Map> requests = cacheSkipRequests.get(); + if (requests == null) + return; + List ranges = requests.get(path); + if (ranges != null) + { + assertEquals(0, offset); + ranges.add(length); + } + } }