You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by ma...@apache.org on 2026/08/04 03:55:05 UTC
(cassandra) 01/01: Merge branch 'cassandra-5.0' into cassandra-6.0
This is an automated email from the ASF dual-hosted git repository.
asf-gitbox-commits pushed a commit to branch cassandra-6.0
in repository https://gitbox.apache.org/repos/asf/cassandra.git
commit 793d0001ec6f1eef12addf674e2e6cad956a49b7
Merge: 8fd77ffea3 58863f59db
Author: Caleb Rackliffe <ca...@gmail.com>
AuthorDate: Mon Aug 3 22:28:40 2026 -0500
Merge branch 'cassandra-5.0' into cassandra-6.0
* cassandra-5.0:
Fix ThreadLocalReadAheadBuffer#fill() to throw a CorruptBlockException if chunk metadata and file size are out of sync
CHANGES.txt | 1 +
.../io/util/ThreadLocalReadAheadBuffer.java | 11 +++--
.../io/util/StandardCompressedChunkReaderTest.java | 54 ++++++++++++++++++++++
.../io/util/ThreadLocalReadAheadBufferTest.java | 3 +-
4 files changed, 64 insertions(+), 5 deletions(-)
diff --cc CHANGES.txt
index 5fe13105fd,3d12be0786..2d6041c04b
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@@ -1,81 -1,9 +1,82 @@@
-5.0.9
- * Return CorruptSSTableException if chunk metadata and file size are out of sync (CASSANDRA-21519)
+6.0-alpha2
+ * Apply performance optimizations for rows merging logic (CASSANDRA-21524)
+ * Fix operationMode reporting DECOMMISSION_FAILED instead of LEAVING when resuming a failed decommission (CASSANDRA-21493)
+ * Avoid megamorphic calls when serializing and deserializing fixed-length values (CASSANDRA-21536)
+ * Avoid megamorphic calls for Cell.timestamp/ttl/path/localDeletionTimeAsUnsignedInt methods (CASSANDRA-21526)
+ * Allow unreserved keywords as user and identity names in USER and IDENTITY statements (CASSANDRA-21510)
+ * Reduce allocations in DefaultQueryOptions (CASSANDRA-21467)
* Coordinator load-shedding returns OverloadedException without setting streamId, misrouting query responses (CASSANDRA-21508)
+ * Reduce number of scheduledTasks on metric id release in ThreadLocalMetrics (CASSANDRA-21475)
+ * Cache various Enum.values() used in deserialization to avoid per-read array allocation (CASSANDRA-21528)
+ * Fix Accord transaction error message when altering a table (CASSANDRA-20580)
+ * Depend only on platform-specific Zstd JNI native libraries (CASSANDRA-21483)
+ * Expose immediately-executed tasks in the queries virtual table (CASSANDRA-21471)
+ * Avoid potential deadlock between GlobalLogFollower and GossipStage (CASSANDRA-21384)
+ * Add CMS membership as a field in ClusterMetadata (CASSANDRA-20736)
+ * Fix maven remote publishing of Accord artifacts (CASSANDRA-21261)
+ * Move long running TCM operations to a longer timout (CASSANDRA-21453)
+ * Offline nodetool commands should not print network options in help (CASSANDRA-20876)
+ * Defer creation of the system_cluster_metadata keyspace until CMS initialization (CASSANDRA-21477)
+ * Support direct I/O for background SSTable writes (CASSANDRA-21134)
+ * Relax assertion on partitioner instances in SinglePartitionReadCommand (CASSANDRA-21251)
+ * Report cancelled read command execution to coordinator as a RequestFailure.TIMEOUT (CASSANDRA-21468)
+ * Fix TCM log catchup from peer with snapshots and gaps in the log sequence (CASSANDRA-21455)
+ * Speed up nodetool doc generation by producing all command help in a single jvm (CASSANDRA-21444)
+ * Always send TCM commit failures as Messaging failures (CASSANDRA-21457)
+ * Fix ReadCommand serializedSize() using incorrect epoch (CASSANDRA-21438)
+ * Allocation improvements in ProtocolVersion, StorageProxy and MerkleTree (CASSANDRA-21199)
+ * Don’t leave autocompaction disabled during bootstrap and replace (CASSANDRA-21236)
+ * Make nodetool abortbootstrap more robust (CASSANDRA-21235)
+ * Don't clear prepared statement cache on nodetool cms initialize (CASSANDRA-21234)
+ * Improve performance when deserializing cluster metadata (CASSANDRA-21224)
+ * Minor TokenMap performance improvement (CASSANDRA-21223)
+ * Handle lost response when committing PrepareMove (CASSANDRA-21222)
+ * SEPExecutor.maybeExecuteImmediately does not always execute tasks immediately despite available worker capacity (CASSANDRA-21429)
+ * Safely regain ranges and delete retired command stores (CASSANDRA-21212)
+ * Reduce memory allocations in miscellaneous places along read path (CASSANDRA-21360)
+ * Avoid ByteBuffer wrapping in cql3.selection.Selector.InputRow to reduce memory allocation rate (CASSANDRA-21362)
+ * Reduce cost to calculate BTreeRow.minDeletionTime (CASSANDRA-21414)
+ * Implement custom CassandraThread to keep direct references to frequently used thread local objects (CASSANDRA-21020)
+ * Avoid megamorphic call overhead at RandomAccessReader#current (CASSANDRA-21399)
+ * Enable async GC logging for JDK versions which support it to avoid potential hiccups caused by GC log file I/O blocking (CASSANDRA-21372)
+ * Add rowsMutatedPerWriteHistogram metric to track rows mutated per write request (CASSANDRA-21320)
+ * Add TotalRowsRead and TotalRowsMutated counters to TableMetrics for accurate per-table row throughput tracking (CASSANDRA-21321)
+ * Move exception handling of SPI startup checks when iterating over them (CASSANDRA-21409)
+ * Avoid type lookup in SerializationHeader#getType if schema and SSTable are aligned (CASSANDRA-21402)
+ * Replace LongAdder in metric-like logic with ThreadLocalCounter (CASSANDRA-21400)
+ * BTreeRow.hasLiveData: avoid Cell iteration if there are no cell tombstones (CASSANDRA-21363)
+ * Reduce memory allocations in row merge logic (CASSANDRA-21359)
+ * Restore option to avoid hint transfer during decommission (CASSANDRA-21341)
+ * Add an offline cluster metadata tool (CASSANDRA-19151)
+ * Accord: Tail Latency Improvements (CASSANDRA-21361)
+ * Artificial Latency Injection (CASSANDRA-17024)
+ * Accord: Clean Shutdown/Restart, Rebootstrap, et al (CASSANDRA-21355)
+ * Reduce memory allocations in SelectStatement.getQuery (CASSANDRA-21351)
+ * Avoid allocation by getFunctions in SelectStatement.authorize (CASSANDRA-21347)
+ * Avoid unit conversion in DatabaseDescriptor.getMaxValueSize() for every deserializing Cell (CASSANDRA-21295)
+ * Fix single token batch atomicity with Accord/non-Accord batches by using the batch log (CASSANDRA-20588)
+ * Avoid CompactionOptions parsing for every read by WithoutPurgeableTombstones (CASSANDRA-21294)
+ * Ensure schema created before 2.1 without tableId in folder name can be loaded in SnapshotLoader (CASSANDRA-21246)
+ * Differentiate between legitimate cases where the first entry is the same as the last entry and empty bounds in SSTableCursorWriter#addIndexBlock() (CASSANDRA-21255)
+ * Introduce minimum_threshold for data resurrection startup check (CASSANDRA-21293)
+ * Synchronously publish changes to local gossip state following metadata updates (CASSANDRA-21239)
+ * Change default for cassandra.set_sep_thread_name to false to reduce CPU usage (CASSANDRA-21089)
+ * Avoid permission checks for masked columns when the table doesn't have any (CASSANDRA-21299)
+ * Reduce allocations and array copies due to buffer resizing in LocalDataResponse during row serialization (CASSANDRA-21285)
+ * Implement a guardrail for client driver versions (CASSANDRA-21146)
+ * Enable IAuthenticator to declare supported and alterable role options (CASSANDRA-20834)
+ * Avoid capturing lambda allocation in UnfilteredSerializer.deserializeRowBody (CASSANDRA-21289)
+ * Avoid Cell iterator allocation for alive rows in MetricsRecording transformation of ReadCommand (CASSANDRA-21288)
+ * Reuse a single TrackedDataInputPlus instance per UnfilteredSerializer (CASSANDRA-21296)
+ * In-tree docs are included in binary artifacts (tarball, deb, rpm) as html and man pages (CASSANDRA-17260)
+ * Add tooling to repair system peers and peers_v2 if inconsistent with cluster metadata (CASSANDRA-21187)
+ * Fix a removed TTLed row re-appearance in a materialized view after a cursor compaction (CASSANDRA-21152)
+ * Rework ZSTD dictionary compression logic to create a trainer per training (CASSANDRA-21209)
+Merged from 5.0:
++ * Fix ThreadLocalReadAheadBuffer#fill() to throw a CorruptBlockException if chunk metadata and file size are out of sync (CASSANDRA-21519)
* SAI Component Checksum Validation Should be Segment-Aware (CASSANDRA-21516)
* Support Python 3.12 and 3.13 in cqlsh (CASSANDRA-20997)
- * Fix AssertionError in hasReplicaWithOngoingRepair when parallel_repair_count > 1 (CASSANDRA-21426)
+ * Make synchronization on VectorMemoryIndex inserts more granular (CASSANDRA-21160)
* putShortVolatile is not volatile in InMemoryTrie (CASSANDRA-21353)
* Fix RequestFailureReason serializer and nits in a few others (CASSANDRA-21437)
* Remove golang dependency in gen-doc and replace with python implementation (CASSANDRA-21432)
diff --cc src/java/org/apache/cassandra/io/util/ThreadLocalReadAheadBuffer.java
index c8ef6cf1e3,5e78697686..ff59f7cf96
--- a/src/java/org/apache/cassandra/io/util/ThreadLocalReadAheadBuffer.java
+++ b/src/java/org/apache/cassandra/io/util/ThreadLocalReadAheadBuffer.java
@@@ -21,18 -21,14 +21,19 @@@ package org.apache.cassandra.io.util
import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.Map;
+import java.util.function.Supplier;
-import io.netty.util.concurrent.FastThreadLocal;
import org.apache.cassandra.io.compress.BufferType;
+ import org.apache.cassandra.io.compress.CorruptBlockException;
import org.apache.cassandra.io.sstable.CorruptSSTableException;
+import org.apache.cassandra.utils.Closeable;
+import org.apache.cassandra.utils.memory.MemoryUtil;
+
+import io.netty.util.concurrent.FastThreadLocal;
-public final class ThreadLocalReadAheadBuffer
+public class ThreadLocalReadAheadBuffer implements Closeable
{
+
private static class Block
{
ByteBuffer buffer = null;
@@@ -119,16 -117,9 +122,16 @@@
blockBuffer.flip();
blockBuffer.limit(sizeToRead);
- blockBuffer.position((int) (realPosition - blockPosition));
+ blockBuffer.position((int) (position - blockPosition));
}
+ protected void loadBlock(ByteBuffer blockBuffer, long blockPosition, int sizeToRead)
+ {
+ blockBuffer.limit(sizeToRead);
+ if (channel.read(blockBuffer, blockPosition) != sizeToRead)
+ throw new CorruptSSTableException(null, channel.filePath());
+ }
+
public int read(ByteBuffer dest, int length)
{
Block block = getBlock();
diff --cc test/unit/org/apache/cassandra/io/util/StandardCompressedChunkReaderTest.java
index afad1cd117,0000000000..436a3e3aa2
mode 100644,000000..100644
--- a/test/unit/org/apache/cassandra/io/util/StandardCompressedChunkReaderTest.java
+++ b/test/unit/org/apache/cassandra/io/util/StandardCompressedChunkReaderTest.java
@@@ -1,115 -1,0 +1,169 @@@
+/*
+ * 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.io.util;
+
+import java.nio.ByteBuffer;
++import java.nio.channels.FileChannel;
+import java.nio.file.Files;
++import java.nio.file.StandardOpenOption;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.assertj.core.api.Assertions;
+import org.junit.Assert;
+import org.junit.Test;
+
+import accord.utils.Gens;
+
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.db.ClusteringComparator;
+import org.apache.cassandra.io.compress.CompressedSequentialWriter;
+import org.apache.cassandra.io.compress.CompressionMetadata;
+import org.apache.cassandra.io.filesystem.ListenableFileSystem;
++import org.apache.cassandra.io.sstable.CorruptSSTableException;
+import org.apache.cassandra.io.sstable.metadata.MetadataCollector;
+import org.apache.cassandra.schema.CompressionParams;
+import org.apache.cassandra.utils.memory.MemoryUtil;
+
+import static accord.utils.Property.qt;
+
+public class StandardCompressedChunkReaderTest extends CompressedChunkReaderTestBase
+{
+ static
+ {
+ DatabaseDescriptor.clientInitialization();
+ }
+
+ @Test
+ public void scanReaderReadsLessThanRAReader()
+ {
+ var optionGen = writerOptions();
+ var paramsGen = compressionParams(Gens.constant(CompressionParams.DEFAULT_CHUNK_LENGTH));
+ var lengthGen = Gens.longs().between(1, 1 << 16);
+
+ qt().forAll(Gens.random(), optionGen, paramsGen).check((rs, option, params) -> {
+ ListenableFileSystem fs = FileSystems.newGlobalInMemoryFileSystem();
+
+ File f = new File("/file.bin");
+ AtomicInteger reads = new AtomicInteger();
+ fs.onPostRead(f.path::equals, (p, c, pos, dst, r) -> reads.incrementAndGet());
+ long length = lengthGen.nextLong(rs);
+ CompressionMetadata metadata1, metadata2;
+ try (CompressedSequentialWriter writer = new CompressedSequentialWriter(f, new File("/file.offset"), new File("/file.digest"), option, params, new MetadataCollector(new ClusteringComparator())))
+ {
+ for (long i = 0; i < length; i++)
+ writer.writeLong(i);
+
+ writer.sync();
+ metadata1 = writer.open(0);
+ metadata2 = writer.open(0);
+ }
+
+ doReads(f, metadata1, length, false);
+ int raReads = reads.getAndSet(0);
+
+ doReads(f, metadata2, length, true);
+ int scanReads = reads.getAndSet(0);
+
+ if (Files.size(f.toPath()) > DatabaseDescriptor.getCompressedReadAheadBufferSize())
+ Assert.assertTrue(scanReads <= raReads);
+ });
+ }
+
+ protected void doReads(File f, CompressionMetadata metadata, long length, boolean useReadAhead)
+ {
+ ByteBuffer buffer = ByteBuffer.allocateDirect(metadata.chunkLength());
+
+ try (ChannelProxy channel = new ChannelProxy(f))
+ {
+ try (CompressedChunkReader reader = new CompressedChunkReader.Standard(channel, metadata, () -> 1d);
+ metadata)
+ {
+ if (useReadAhead)
+ reader.forScan();
+
+ long offset = 0;
+ long maxOffset = length * Long.BYTES;
+ do
+ {
+ reader.readChunk(offset, buffer);
+ for (long expected = offset / Long.BYTES; buffer.hasRemaining(); expected++)
+ Assertions.assertThat(buffer.getLong()).isEqualTo(expected);
+
+ offset += metadata.chunkLength();
+ }
+ while (offset < maxOffset);
+ }
+ }
+ finally
+ {
+ MemoryUtil.clean(buffer);
+ }
+ }
++
++ @Test(timeout = 10_000)
++ public void scanReaderShouldNotHangOnTruncatedFile() throws Exception
++ {
++ SequentialWriterOption writerOption = SequentialWriterOption.newBuilder().finishOnClose(false).bufferSize(1 << 10).build();
++ CompressionParams params = CompressionParams.snappy(4096, 1.1);
++
++ FileSystems.newGlobalInMemoryFileSystem();
++ File f = new File("/truncated_hang_repro.db");
++ File offsets = new File("/truncated_hang_repro.offset");
++ File digest = new File("/truncated_hang_repro.digest");
++
++ long longsToWrite = 600; // 4800 uncompressed bytes -> 2 compressed chunks (second one partial)
++ CompressionMetadata metadata;
++ try (CompressedSequentialWriter writer = new CompressedSequentialWriter(f, offsets, digest, writerOption, params, new MetadataCollector(new ClusteringComparator())))
++ {
++ for (long i = 0; i < longsToWrite; i++)
++ writer.writeLong(i);
++
++ writer.sync();
++ metadata = writer.open(0);
++ }
++
++ DatabaseDescriptor.setCompressedReadAheadBufferSizeInKb(256);
++
++ // Truncate file so that chunk metadata expects a chunk to extend further than the actual file size
++ long originalSize = Files.size(f.toPath());
++ long truncatedSize = originalSize - (params.chunkLength() / 2);
++ try (FileChannel fc = FileChannel.open(f.toPath(), StandardOpenOption.WRITE))
++ {
++ fc.truncate(truncatedSize);
++ }
++ long uncompressedTotal = longsToWrite * Long.BYTES;
++ long lastChunkUncompressedStart = ((uncompressedTotal - 1) / metadata.chunkLength()) * metadata.chunkLength();
++
++ ByteBuffer buffer = ByteBuffer.allocateDirect(metadata.chunkLength());
++ try (ChannelProxy channel = new ChannelProxy(f);
++ CompressedChunkReader reader = new CompressedChunkReader.Standard(channel, metadata, () -> 1.1);
++ metadata)
++ {
++ reader.forScan();
++
++ Assertions.assertThatThrownBy(() -> reader.readChunk(lastChunkUncompressedStart, buffer))
++ .as("readChunk() reading past truncated EOF via the scan path")
++ .isInstanceOf(CorruptSSTableException.class);
++ }
++ finally
++ {
++ MemoryUtil.clean(buffer);
++ }
++ }
+}
diff --cc test/unit/org/apache/cassandra/io/util/ThreadLocalReadAheadBufferTest.java
index a4a7f8d519,2d29504d6f..47bba7b6c5
--- a/test/unit/org/apache/cassandra/io/util/ThreadLocalReadAheadBufferTest.java
+++ b/test/unit/org/apache/cassandra/io/util/ThreadLocalReadAheadBufferTest.java
@@@ -37,8 -36,11 +37,9 @@@ import org.slf4j.LoggerFactory
import org.apache.cassandra.config.DataStorageSpec;
import org.apache.cassandra.io.compress.BufferType;
+ import org.apache.cassandra.io.compress.CorruptBlockException;
import org.apache.cassandra.io.sstable.CorruptSSTableException;
import org.apache.cassandra.utils.Pair;
-import org.quicktheories.WithQuickTheories;
-import org.quicktheories.core.Gen;
import static java.lang.Math.max;
import static org.apache.cassandra.config.CassandraRelevantProperties.JAVA_IO_TMPDIR;
@@@ -105,35 -130,7 +106,35 @@@ public class ThreadLocalReadAheadBuffer
}
}
- private Gen<InputData> reads()
+ protected static void testRead(Pair<Long, Integer> read, ChannelProxy bufferedChannel, ThreadLocalReadAheadBuffer tlrab)
+ {
+ int readSize = Math.min(read.right, (int) (bufferedChannel.size() - read.left));
+ ByteBuffer buf1 = ByteBuffer.allocate(readSize);
+ bufferedChannel.read(buf1, read.left);
+
+ ByteBuffer buf2 = ByteBuffer.allocate(readSize);
+ try
+ {
+ int copied = 0;
+ while (copied < readSize)
+ {
+ tlrab.fill(read.left + copied);
+ int leftToRead = readSize - copied;
+ if (tlrab.remaining() >= leftToRead)
+ copied += tlrab.read(buf2, leftToRead);
+ else
+ copied += tlrab.read(buf2, tlrab.remaining());
+ }
+ }
- catch (CorruptSSTableException e)
++ catch (CorruptSSTableException | CorruptBlockException e)
+ {
+ throw new RuntimeException(e);
+ }
+
+ Assert.assertEquals(buf1, buf2);
+ }
+
+ protected Gen<InputData> reads()
{
return arbitrary().pick(List.of(files))
.flatMap((file) ->
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org