natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'cassandra-5.0' into cassandra-6.0 

* cassandra-5.0:
  putShortVolatile is not volatile in InMemoryTrie
This commit is contained in:
Francisco Guerrero 2026-06-17 06:47:41 -05:00
parent 802ce7f8b2 659c0a526b
commit c9ffe11645
8 changed files with 232 additions and 76 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -50,6 +50,7 @@
* 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:
* 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)
* Use estimated compressed size for tables to check if there is enough free space for a compaction (CASSANDRA-21245)

2
src/java/org/apache/cassandra/db/tries/InMemoryTrie.java
View File

@ -128,7 +128,7 @@ public class InMemoryTrie<T> extends InMemoryReadTrie<T>
final void putShortVolatile(int pos, short value)
{
getChunk(pos).putShort(inChunkPointer(pos), value);
getChunk(pos).putShortVolatile(inChunkPointer(pos), value);
}
final void putByte(int pos, byte value)

28
src/java/org/apache/cassandra/io/sstable/AbstractSSTableIterator.java
View File

@ -97,6 +97,7 @@ public abstract class AbstractSSTableIterator<RIE extends AbstractRowIndexEntry>
}
else
{
Reader reader = null;
boolean shouldCloseFile = file == null;
try
{
@ -119,15 +120,18 @@ public abstract class AbstractSSTableIterator<RIE extends AbstractRowIndexEntry>
// Note that this needs to be called after file != null and after the partitionDeletion has been set, but before readStaticRow
// (since it uses it) so we can't move that up (but we'll be able to simplify as soon as we drop support for the old file format).
this.reader = createReader(indexEntry, file, shouldCloseFile);
reader = createReader(indexEntry, file, shouldCloseFile);
this.staticRow = readStaticRow(sstable, file, helper, columns.fetchedColumns().statics);
}
else
{
this.partitionLevelDeletion = indexEntry.deletionTime();
this.staticRow = Rows.EMPTY_STATIC_ROW;
this.reader = createReader(indexEntry, file, shouldCloseFile);
reader = createReader(indexEntry, file, shouldCloseFile);
}
this.reader = reader;
if (!partitionLevelDeletion.validate())
UnfilteredValidation.handleInvalid(metadata(), key, sstable, "partitionLevelDeletion="+partitionLevelDeletion.toString());
@ -137,9 +141,25 @@ public abstract class AbstractSSTableIterator<RIE extends AbstractRowIndexEntry>
if (reader == null && file != null && shouldCloseFile)
file.close();
}
catch (IOException e)
catch (CorruptSSTableException | IOException e)
{
sstable.markSuspect();
if (reader != null)
{
try
{
reader.close();
// reader will close the file internally, so there's no
// need to close it in the next block
shouldCloseFile = false;
}
catch (IOException suppressed)
{
e.addSuppressed(suppressed);
}
}
String filePath = file.getPath();
if (shouldCloseFile)
{
@ -152,6 +172,8 @@ public abstract class AbstractSSTableIterator<RIE extends AbstractRowIndexEntry>
e.addSuppressed(suppressed);
}
}
if (e instanceof CorruptSSTableException)
throw (CorruptSSTableException)e;
throw new CorruptSSTableException(e, filePath);
}
}

17
src/java/org/apache/cassandra/io/sstable/format/bti/BtiTableWriter.java
View File

@ -330,13 +330,16 @@ public class BtiTableWriter extends SortedTableWriter<BtiFormatPartitionWriter,
{
checkState(!dataWriterOpened, "Data writer has been already opened.");
return DataComponent.buildWriter(descriptor,
getTableMetadataRef().getLocal(),
getIOOptions().writerOptions,
getMetadataCollector(),
ensuringInBuildInternalContext(operationType),
getIOOptions().flushCompression,
getCompressionDictionaryManager());
SequentialWriter sequentialWriter = DataComponent.buildWriter(descriptor,
getTableMetadataRef().getLocal(),
getIOOptions().writerOptions,
getMetadataCollector(),
ensuringInBuildInternalContext(operationType),
getIOOptions().flushCompression,
getCompressionDictionaryManager());
dataWriterOpened = true;
return sequentialWriter;
}
@Override

27
src/java/org/apache/cassandra/io/tries/ReverseValueIterator.java
View File

@ -61,7 +61,16 @@ public class ReverseValueIterator<Concrete extends ReverseValueIterator<Concrete
{
super(source, root);
limit = null;
initializeNoRightBound(root, NOT_AT_LIMIT, false);
try
{
initializeNoRightBound(root, NOT_AT_LIMIT, false);
}
catch (Throwable t)
{
super.close();
throw t;
}
}
/**
@ -80,10 +89,18 @@ public class ReverseValueIterator<Concrete extends ReverseValueIterator<Concrete
super(source, root);
limit = start != null ? start.asComparableBytes(BYTE_COMPARABLE_VERSION) : null;
if (end != null)
initializeWithRightBound(root, end.asComparableBytes(BYTE_COMPARABLE_VERSION), admitPrefix, limit != null);
else
initializeNoRightBound(root, limit != null ? limit.next() : NOT_AT_LIMIT, admitPrefix);
try
{
if (end != null)
initializeWithRightBound(root, end.asComparableBytes(BYTE_COMPARABLE_VERSION), admitPrefix, limit != null);
else
initializeNoRightBound(root, limit != null ? limit.next() : NOT_AT_LIMIT, admitPrefix);
}
catch (Throwable t)
{
super.close();
throw t;
}
}
void initializeWithRightBound(long root, ByteSource endStream, boolean admitPrefix, boolean hasLimit)

119
src/java/org/apache/cassandra/io/tries/ValueIterator.java
View File

@ -72,7 +72,16 @@ public class ValueIterator<CONCRETE extends ValueIterator<CONCRETE>> extends Wal
super(source, root);
limit = null;
collector = collecting ? new TransitionBytesCollector() : null;
initializeNoLeftBound(root, 256);
try
{
initializeNoLeftBound(root, 256);
}
catch (Throwable t)
{
super.close();
throw t;
}
}
protected ValueIterator(Rebufferer source, long root, ByteComparable start, ByteComparable end, boolean admitPrefix)
@ -97,10 +106,18 @@ public class ValueIterator<CONCRETE extends ValueIterator<CONCRETE>> extends Wal
limit = end != null ? end.asComparableBytes(BYTE_COMPARABLE_VERSION) : null;
collector = collecting ? new TransitionBytesCollector() : null;
if (start != null)
initializeWithLeftBound(root, start.asComparableBytes(BYTE_COMPARABLE_VERSION), admitPrefix, limit != null);
else
initializeNoLeftBound(root, limit != null ? limit.next() : 256);
try
{
if (start != null)
initializeWithLeftBound(root, start.asComparableBytes(BYTE_COMPARABLE_VERSION), admitPrefix, limit != null);
else
initializeNoLeftBound(root, limit != null ? limit.next() : 256);
}
catch (Throwable t)
{
super.close();
throw t;
}
}
private void initializeWithLeftBound(long root, ByteSource startStream, boolean admitPrefix, boolean atLimit)
@ -110,76 +127,60 @@ public class ValueIterator<CONCRETE extends ValueIterator<CONCRETE>> extends Wal
int limitByte;
long payloadedNode = -1;
try
// Follow start position while we still have a prefix, stacking path and saving prefixes.
go(root);
while (true)
{
// Follow start position while we still have a prefix, stacking path and saving prefixes.
go(root);
while (true)
int s = startStream.next();
childIndex = search(s);
// For a separator trie the latest payload met along the prefix is a potential match for start
if (admitPrefix)
{
int s = startStream.next();
childIndex = search(s);
// For a separator trie the latest payload met along the prefix is a potential match for start
if (admitPrefix)
if (childIndex == 0 || childIndex == -1)
{
if (childIndex == 0 || childIndex == -1)
{
if (hasPayload())
payloadedNode = position;
}
else
{
payloadedNode = -1;
}
if (hasPayload())
payloadedNode = position;
}
limitByte = 256;
if (atLimit)
else
{
limitByte = limit.next();
if (s < limitByte)
atLimit = false;
payloadedNode = -1;
}
if (childIndex < 0)
break;
prev = new IterationPosition(position, childIndex, limitByte, prev);
go(transition(childIndex)); // child index is positive, transition must exist
}
childIndex = -1 - childIndex - 1;
stack = new IterationPosition(position, childIndex, limitByte, prev);
limitByte = 256;
if (atLimit)
{
limitByte = limit.next();
if (s < limitByte)
atLimit = false;
}
if (childIndex < 0)
break;
// Advancing now gives us first match if we didn't find one already.
if (payloadedNode != -1)
next = payloadedNode;
else
next = advanceNode();
}
catch (Throwable t)
{
super.close();
throw t;
prev = new IterationPosition(position, childIndex, limitByte, prev);
go(transition(childIndex)); // child index is positive, transition must exist
}
childIndex = -1 - childIndex - 1;
stack = new IterationPosition(position, childIndex, limitByte, prev);
// Advancing now gives us first match if we didn't find one already.
if (payloadedNode != -1)
next = payloadedNode;
else
next = advanceNode();
}
private void initializeNoLeftBound(long root, int limitByte)
{
stack = new IterationPosition(root, -1, limitByte, null);
try
{
go(root);
if (hasPayload())
next = root;
else
next = advanceNode();
}
catch (Throwable t)
{
super.close();
throw t;
}
go(root);
if (hasPayload())
next = root;
else
next = advanceNode();
}
/**

2
src/java/org/apache/cassandra/io/tries/Walker.java
View File

@ -46,7 +46,7 @@ import org.apache.cassandra.utils.bytecomparable.ByteSource;
public class Walker<CONCRETE extends Walker<CONCRETE>> implements AutoCloseable
{
/** Value used to indicate a branch (e.g. lesser/greaterBranch) does not exist. */
public static int NONE = TrieNode.NONE;
public static final int NONE = TrieNode.NONE;
private final Rebufferer source;
protected final long root;

112
test/unit/org/apache/cassandra/io/tries/ReverseValueIteratorLeakTest.java Normal file
View File

@ -0,0 +1,112 @@
/*
* 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.tries;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.concurrent.atomic.AtomicBoolean;
import org.junit.Test;
import org.apache.cassandra.io.util.DataOutputBuffer;
import org.apache.cassandra.io.util.Rebufferer;
import org.apache.cassandra.utils.bytecomparable.ByteComparable;
import org.apache.cassandra.utils.bytecomparable.ByteSource;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
/**
* Reproducer: ReverseValueIterator leaks Rebufferer on constructor exception
*
* <p>If initializeWithRightBound() or initializeNoRightBound() throws after the Walker
* super-constructor acquires a Rebufferer, the Rebufferer is never closed. Compare with
* ValueIterator which wraps initialization in try-catch with super.close().
*
* Expected: Rebufferer.closeReader() is called when initialization throws.
* Actual (buggy): Rebufferer leaks — closeReader() never called.
* Failure criterion: closeReader() not called after constructor exception.
*/
@SuppressWarnings({"unchecked", "RedundantSuppression"})
public class ReverseValueIteratorLeakTest extends AbstractTrieTestBase
{
// ORACLE: Rebufferer.closeReader() must be called if constructor throws
@Test
public void testRebuffererClosedOnInitWithRightBoundFailure() throws IOException
{
// HARNESS: build a valid trie, wrap its Rebufferer with close tracking,
// then trigger a failure during initializeWithRightBound
DataOutputBuffer buf = new DataOutputBufferPaged();
IncrementalTrieWriter<Integer> builder = newTrieWriter(serializer, buf);
builder.add(source("aaa"), 1);
builder.add(source("bbb"), 2);
long root = builder.complete();
AtomicBoolean closeReaderCalled = new AtomicBoolean(false);
Rebufferer trackingSource = new TrackingRebufferer(buf.asNewBuffer(), closeReaderCalled);
// TRIGGER: provide a ByteComparable whose ByteSource throws during iteration,
// causing initializeWithRightBound to fail after Walker acquires the Rebufferer
ByteComparable throwingEnd = v -> new ByteSource()
{
private int calls = 0;
@Override
public int next()
{
if (++calls > 1)
throw new RuntimeException("simulated failure during trie traversal");
return 'b';
}
};
try
{
new ReverseValueIterator<>(trackingSource, root, source("aaa"), throwingEnd, false);
fail("Constructor should have thrown");
}
catch (RuntimeException e)
{
assertTrue("Expected simulated failure", e.getMessage().contains("simulated failure"));
}
// ORACLE: Rebufferer must have been closed despite the constructor failure
assertTrue("Rebufferer.closeReader() must be called when ReverseValueIterator " +
"constructor fails during initializeWithRightBound — without the fix, " +
"the Rebufferer leaks",
closeReaderCalled.get());
}
private static class TrackingRebufferer extends ByteBufRebufferer
{
private final AtomicBoolean closeReaderCalled;
TrackingRebufferer(ByteBuffer buffer, AtomicBoolean closeReaderCalled)
{
super(buffer);
this.closeReaderCalled = closeReaderCalled;
}
@Override
public void closeReader()
{
closeReaderCalled.set(true);
}
}
}