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 <[email protected]> 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: [email protected] For additional commands, e-mail: [email protected]
