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

Merge 9d264a0fb3 into 209ef2b6c4

This commit is contained in:
Andrés de la Peña 2026-07-29 13:35:24 +08:00 committed by GitHub
parent 209ef2b6c4 9d264a0fb3
commit 3ff70c32fa
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
11 changed files with 425 additions and 242 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

566
.circleci/config.yml
View File

File diff suppressed because it is too large Load Diff

8
src/java/org/apache/cassandra/index/sai/disk/format/IndexDescriptor.java
View File

@ -202,7 +202,7 @@ public class IndexDescriptor
file,
FBUtilities.prettyPrintMemory(file.length()));
return IndexFileUtils.instance.openBlockingInput(file);
return IndexFileUtils.instance().openBlockingInput(file);
}
public IndexInput openPerIndexInput(IndexComponent indexComponent, IndexIdentifier indexIdentifier)
@ -213,7 +213,7 @@ public class IndexDescriptor
file,
FBUtilities.prettyPrintMemory(file.length()));
return IndexFileUtils.instance.openBlockingInput(file);
return IndexFileUtils.instance().openBlockingInput(file);
}
public IndexOutputWriter openPerSSTableOutput(IndexComponent component) throws IOException
@ -230,7 +230,7 @@ public class IndexDescriptor
component,
file);
IndexOutputWriter writer = IndexFileUtils.instance.openOutput(file);
IndexOutputWriter writer = IndexFileUtils.instance().openOutput(file);
if (append)
{
@ -252,7 +252,7 @@ public class IndexDescriptor
if (logger.isTraceEnabled())
logger.trace(logMessage("Creating sstable attached index output for component {} on file {}..."), component, file);
IndexOutputWriter writer = IndexFileUtils.instance.openOutput(file);
IndexOutputWriter writer = IndexFileUtils.instance().openOutput(file);
if (append)
{

13
src/java/org/apache/cassandra/index/sai/disk/io/IndexFileUtils.java
View File

@ -46,11 +46,22 @@ public class IndexFileUtils
.finishOnClose(true)
.build();
public static final IndexFileUtils instance = new IndexFileUtils(DEFAULT_WRITER_OPTION);
private static final IndexFileUtils instance = new IndexFileUtils(DEFAULT_WRITER_OPTION);
private static IndexFileUtils overrideInstance = null;
private static final Supplier<Checksum> CHECKSUM_FACTORY = CRC32C::new;
private final SequentialWriterOption writerOption;
public static synchronized void setOverrideInstance(IndexFileUtils overrideInstance)
{
IndexFileUtils.overrideInstance = overrideInstance;
}
public static IndexFileUtils instance()
{
return overrideInstance == null ? instance : overrideInstance;
}
@VisibleForTesting
protected IndexFileUtils(SequentialWriterOption writerOption)
{

6
src/java/org/apache/cassandra/index/sai/disk/v1/bbtree/BlockBalancedTreeReader.java
View File

@ -109,9 +109,9 @@ public class BlockBalancedTreeReader extends BlockBalancedTreeWalker implements
}
listener.onSegmentHit();
IndexInput treeInput = IndexFileUtils.instance.openInput(treeIndexFile);
IndexInput postingsInput = IndexFileUtils.instance.openInput(postingsFile);
IndexInput postingsSummaryInput = IndexFileUtils.instance.openInput(postingsFile);
IndexInput treeInput = IndexFileUtils.instance().openInput(treeIndexFile);
IndexInput postingsInput = IndexFileUtils.instance().openInput(postingsFile);
IndexInput postingsSummaryInput = IndexFileUtils.instance().openInput(postingsFile);
Intersection intersection = relation == Relation.CELL_INSIDE_QUERY
? new Intersection(treeInput, postingsInput, postingsSummaryInput, listener, context)

2
src/java/org/apache/cassandra/index/sai/disk/v1/bitpack/BlockPackedReader.java
View File

@ -98,7 +98,7 @@ public class BlockPackedReader implements LongArray.Factory
@Override
public LongArray open()
{
IndexInput indexInput = IndexFileUtils.instance.openInput(file);
IndexInput indexInput = IndexFileUtils.instance().openInput(file);
return new AbstractBlockPackedReader(indexInput, blockBitsPerValue, blockShift, blockMask, valueCount)
{
@Override

2
src/java/org/apache/cassandra/index/sai/disk/v1/bitpack/MonotonicBlockPackedReader.java
View File

@ -84,7 +84,7 @@ public class MonotonicBlockPackedReader implements LongArray.Factory
@Override
public LongArray open()
{
final IndexInput indexInput = IndexFileUtils.instance.openInput(file);
final IndexInput indexInput = IndexFileUtils.instance().openInput(file);
return new AbstractBlockPackedReader(indexInput, blockBitsPerValue, blockShift, blockMask, valueCount)
{
@Override

8
src/java/org/apache/cassandra/index/sai/disk/v1/segment/LiteralIndexSegmentTermsReader.java
View File

@ -73,12 +73,12 @@ public class LiteralIndexSegmentTermsReader implements Closeable
postingsFile = postingLists;
termDictionaryRoot = root;
try (final IndexInput indexInput = IndexFileUtils.instance.openInput(termDictionaryFile))
try (final IndexInput indexInput = IndexFileUtils.instance().openInput(termDictionaryFile))
{
validate(indexInput, termsFooterPointer);
}
try (final IndexInput indexInput = IndexFileUtils.instance.openInput(postingsFile))
try (final IndexInput indexInput = IndexFileUtils.instance().openInput(postingsFile))
{
validate(indexInput);
}
@ -110,8 +110,8 @@ public class LiteralIndexSegmentTermsReader implements Closeable
TermQuery(ByteComparable term, QueryEventListener.TrieIndexEventListener listener, QueryContext context)
{
this.listener = listener;
postingsInput = IndexFileUtils.instance.openInput(postingsFile);
postingsSummaryInput = IndexFileUtils.instance.openInput(postingsFile);
postingsInput = IndexFileUtils.instance().openInput(postingsFile);
postingsSummaryInput = IndexFileUtils.instance().openInput(postingsFile);
this.term = term;
lookupStartTime = Clock.Global.nanoTime();
this.context = context;

6
src/java/org/apache/cassandra/index/sai/disk/v1/vector/OnHeapGraph.java
View File

@ -315,9 +315,9 @@ public class OnHeapGraph<T>
postingsMap.keySet().size(), vectorValues.size());
logger.debug("Writing graph with {} rows and {} distinct vectors", postingsMap.values().stream().mapToInt(VectorPostings::size).sum(), vectorValues.size());
try (var pqOutput = IndexFileUtils.instance.openOutput(indexDescriptor.fileFor(IndexComponent.COMPRESSED_VECTORS, indexIdentifier), true);
var postingsOutput = IndexFileUtils.instance.openOutput(indexDescriptor.fileFor(IndexComponent.POSTING_LISTS, indexIdentifier), true);
var indexOutput = IndexFileUtils.instance.openOutput(indexDescriptor.fileFor(IndexComponent.TERMS_DATA, indexIdentifier), true))
try (var pqOutput = IndexFileUtils.instance().openOutput(indexDescriptor.fileFor(IndexComponent.COMPRESSED_VECTORS, indexIdentifier), true);
var postingsOutput = IndexFileUtils.instance().openOutput(indexDescriptor.fileFor(IndexComponent.POSTING_LISTS, indexIdentifier), true);
var indexOutput = IndexFileUtils.instance().openOutput(indexDescriptor.fileFor(IndexComponent.TERMS_DATA, indexIdentifier), true))
{
SAICodecUtils.writeHeader(pqOutput);
SAICodecUtils.writeHeader(postingsOutput);

52
test/unit/org/apache/cassandra/index/sai/disk/v1/SAICodecUtilsTest.java
View File

@ -52,12 +52,12 @@ public class SAICodecUtilsTest extends SAIRandomizedTester
@Test
public void checkHeaderDoesNotFailWithValidHeader() throws Exception
{
try (IndexOutputWriter writer = IndexFileUtils.instance.openOutput(file))
try (IndexOutputWriter writer = IndexFileUtils.instance().openOutput(file))
{
SAICodecUtils.writeHeader(writer);
}
try (IndexInput input = IndexFileUtils.instance.openBlockingInput(file))
try (IndexInput input = IndexFileUtils.instance().openBlockingInput(file))
{
SAICodecUtils.checkHeader(input);
}
@ -66,12 +66,12 @@ public class SAICodecUtilsTest extends SAIRandomizedTester
@Test
public void checkHeaderFailsOnInvalidMagicValue() throws Exception
{
try (IndexOutputWriter writer = IndexFileUtils.instance.openOutput(file))
try (IndexOutputWriter writer = IndexFileUtils.instance().openOutput(file))
{
writeBEInt(writer, 1234);
}
try (IndexInput input = IndexFileUtils.instance.openBlockingInput(file))
try (IndexInput input = IndexFileUtils.instance().openBlockingInput(file))
{
assertThatThrownBy(() -> SAICodecUtils.checkHeader(input))
.isInstanceOf(CorruptIndexException.class)
@ -82,13 +82,13 @@ public class SAICodecUtilsTest extends SAIRandomizedTester
@Test
public void checkHeaderFailsOnInvalidVersion() throws Exception
{
try (IndexOutputWriter writer = IndexFileUtils.instance.openOutput(file))
try (IndexOutputWriter writer = IndexFileUtils.instance().openOutput(file))
{
writeBEInt(writer, CODEC_MAGIC);
writer.writeString("zz");
}
try (IndexInput input = IndexFileUtils.instance.openBlockingInput(file))
try (IndexInput input = IndexFileUtils.instance().openBlockingInput(file))
{
assertThatThrownBy(() -> SAICodecUtils.checkHeader(input))
.isInstanceOf(IllegalArgumentException.class)
@ -100,7 +100,7 @@ public class SAICodecUtilsTest extends SAIRandomizedTester
public void checkFooterDoesNotFailWithValidFooter() throws Exception
{
int numBytes = nextInt(1000, 10000);
try (IndexOutputWriter writer = IndexFileUtils.instance.openOutput(file))
try (IndexOutputWriter writer = IndexFileUtils.instance().openOutput(file))
{
SAICodecUtils.writeHeader(writer);
for (int value = 0; value < numBytes; value++)
@ -108,7 +108,7 @@ public class SAICodecUtilsTest extends SAIRandomizedTester
SAICodecUtils.writeFooter(writer);
}
try (IndexInput input = IndexFileUtils.instance.openBlockingInput(file);
try (IndexInput input = IndexFileUtils.instance().openBlockingInput(file);
ChecksumIndexInput checksumIndexInput = IndexFileUtils.getBufferedChecksumIndexInput(input))
{
SAICodecUtils.checkHeader(checksumIndexInput);
@ -122,14 +122,14 @@ public class SAICodecUtilsTest extends SAIRandomizedTester
public void checkFooterFailsWithMissingFooter() throws Exception
{
int numBytes = nextInt(1000, 10000);
try (IndexOutputWriter writer = IndexFileUtils.instance.openOutput(file))
try (IndexOutputWriter writer = IndexFileUtils.instance().openOutput(file))
{
SAICodecUtils.writeHeader(writer);
for (int value = 0; value < numBytes; value++)
writer.writeByte(getRandom().nextByte());
}
try (IndexInput input = IndexFileUtils.instance.openBlockingInput(file);
try (IndexInput input = IndexFileUtils.instance().openBlockingInput(file);
ChecksumIndexInput checksumIndexInput = IndexFileUtils.getBufferedChecksumIndexInput(input))
{
SAICodecUtils.checkHeader(checksumIndexInput);
@ -145,7 +145,7 @@ public class SAICodecUtilsTest extends SAIRandomizedTester
public void checkFooterFailsWithExtendedFooter() throws Exception
{
int numBytes = nextInt(1000, 10000);
try (IndexOutputWriter writer = IndexFileUtils.instance.openOutput(file))
try (IndexOutputWriter writer = IndexFileUtils.instance().openOutput(file))
{
SAICodecUtils.writeHeader(writer);
for (int value = 0; value < numBytes; value++)
@ -154,7 +154,7 @@ public class SAICodecUtilsTest extends SAIRandomizedTester
writeBEInt(writer, 1234);
}
try (IndexInput input = IndexFileUtils.instance.openBlockingInput(file);
try (IndexInput input = IndexFileUtils.instance().openBlockingInput(file);
ChecksumIndexInput checksumIndexInput = IndexFileUtils.getBufferedChecksumIndexInput(input))
{
SAICodecUtils.checkHeader(checksumIndexInput);
@ -170,7 +170,7 @@ public class SAICodecUtilsTest extends SAIRandomizedTester
public void checkFooterFailsWithInvalidFooter() throws Exception
{
int numBytes = nextInt(1000, 10000);
try (IndexOutputWriter writer = IndexFileUtils.instance.openOutput(file))
try (IndexOutputWriter writer = IndexFileUtils.instance().openOutput(file))
{
SAICodecUtils.writeHeader(writer);
for (int value = 0; value < numBytes; value++)
@ -181,7 +181,7 @@ public class SAICodecUtilsTest extends SAIRandomizedTester
writeBELong(writer, writer.getChecksum());
}
try (IndexInput input = IndexFileUtils.instance.openBlockingInput(file);
try (IndexInput input = IndexFileUtils.instance().openBlockingInput(file);
ChecksumIndexInput checksumIndexInput = IndexFileUtils.getBufferedChecksumIndexInput(input))
{
SAICodecUtils.checkHeader(checksumIndexInput);
@ -197,7 +197,7 @@ public class SAICodecUtilsTest extends SAIRandomizedTester
public void checkFooterFailsWithInvalidAlgorithmId() throws Exception
{
int numBytes = nextInt(1000, 10000);
try (IndexOutputWriter writer = IndexFileUtils.instance.openOutput(file))
try (IndexOutputWriter writer = IndexFileUtils.instance().openOutput(file))
{
SAICodecUtils.writeHeader(writer);
for (int value = 0; value < numBytes; value++)
@ -208,7 +208,7 @@ public class SAICodecUtilsTest extends SAIRandomizedTester
writeBELong(writer, writer.getChecksum());
}
try (IndexInput input = IndexFileUtils.instance.openBlockingInput(file);
try (IndexInput input = IndexFileUtils.instance().openBlockingInput(file);
ChecksumIndexInput checksumIndexInput = IndexFileUtils.getBufferedChecksumIndexInput(input))
{
SAICodecUtils.checkHeader(checksumIndexInput);
@ -224,7 +224,7 @@ public class SAICodecUtilsTest extends SAIRandomizedTester
public void checkFooterFailsWithInvalidChecksum() throws Exception
{
int numBytes = nextInt(1000, 10000);
try (IndexOutputWriter writer = IndexFileUtils.instance.openOutput(file))
try (IndexOutputWriter writer = IndexFileUtils.instance().openOutput(file))
{
SAICodecUtils.writeHeader(writer);
for (int value = 0; value < numBytes; value++)
@ -235,7 +235,7 @@ public class SAICodecUtilsTest extends SAIRandomizedTester
writeBELong(writer, 0);
}
try (IndexInput input = IndexFileUtils.instance.openBlockingInput(file);
try (IndexInput input = IndexFileUtils.instance().openBlockingInput(file);
ChecksumIndexInput checksumIndexInput = IndexFileUtils.getBufferedChecksumIndexInput(input))
{
SAICodecUtils.checkHeader(checksumIndexInput);
@ -251,7 +251,7 @@ public class SAICodecUtilsTest extends SAIRandomizedTester
public void checkFooterFailsWithIllegalChecksum() throws Exception
{
int numBytes = nextInt(1000, 10000);
try (IndexOutputWriter writer = IndexFileUtils.instance.openOutput(file))
try (IndexOutputWriter writer = IndexFileUtils.instance().openOutput(file))
{
SAICodecUtils.writeHeader(writer);
for (int value = 0; value < numBytes; value++)
@ -262,7 +262,7 @@ public class SAICodecUtilsTest extends SAIRandomizedTester
writeBELong(writer, 0xFFFFFFFF00000000L);
}
try (IndexInput input = IndexFileUtils.instance.openBlockingInput(file);
try (IndexInput input = IndexFileUtils.instance().openBlockingInput(file);
ChecksumIndexInput checksumIndexInput = IndexFileUtils.getBufferedChecksumIndexInput(input))
{
SAICodecUtils.checkHeader(checksumIndexInput);
@ -277,12 +277,12 @@ public class SAICodecUtilsTest extends SAIRandomizedTester
@Test
public void validateFooterAndResetPositionFailsWithShortFile() throws Exception
{
try (IndexOutputWriter writer = IndexFileUtils.instance.openOutput(file))
try (IndexOutputWriter writer = IndexFileUtils.instance().openOutput(file))
{
SAICodecUtils.writeHeader(writer);
}
try (IndexInput input = IndexFileUtils.instance.openBlockingInput(file))
try (IndexInput input = IndexFileUtils.instance().openBlockingInput(file))
{
assertThatThrownBy(() -> SAICodecUtils.validateFooterAndResetPosition(input))
.isInstanceOf(CorruptIndexException.class)
@ -294,7 +294,7 @@ public class SAICodecUtilsTest extends SAIRandomizedTester
public void validateChecksumFailsWithInvalidChecksum() throws Exception
{
int numBytes = nextInt(1000, 10000);
try (IndexOutputWriter writer = IndexFileUtils.instance.openOutput(file))
try (IndexOutputWriter writer = IndexFileUtils.instance().openOutput(file))
{
SAICodecUtils.writeHeader(writer);
for (int value = 0; value < numBytes; value++)
@ -305,7 +305,7 @@ public class SAICodecUtilsTest extends SAIRandomizedTester
writeBELong(writer, 0);
}
try (IndexInput input = IndexFileUtils.instance.openBlockingInput(file))
try (IndexInput input = IndexFileUtils.instance().openBlockingInput(file))
{
SAICodecUtils.checkHeader(input);
for (int value = 0; value < numBytes; value++)
@ -320,7 +320,7 @@ public class SAICodecUtilsTest extends SAIRandomizedTester
public void validateChecksumFailsWithIllegalChecksum() throws Exception
{
int numBytes = nextInt(1000, 10000);
try (IndexOutputWriter writer = IndexFileUtils.instance.openOutput(file))
try (IndexOutputWriter writer = IndexFileUtils.instance().openOutput(file))
{
SAICodecUtils.writeHeader(writer);
for (int value = 0; value < numBytes; value++)
@ -331,7 +331,7 @@ public class SAICodecUtilsTest extends SAIRandomizedTester
writeBELong(writer, 0xFFFFFFFF00000000L);
}
try (IndexInput input = IndexFileUtils.instance.openBlockingInput(file))
try (IndexInput input = IndexFileUtils.instance().openBlockingInput(file))
{
SAICodecUtils.checkHeader(input);
for (int value = 0; value < numBytes; value++)

2
test/unit/org/apache/cassandra/index/sai/disk/v1/TermsScanner.java
View File

@ -92,7 +92,7 @@ public class TermsScanner implements TermsIterator
private PostingList postings()
{
assert entry != null;
final IndexInput input = IndexFileUtils.instance.openInput(postingsFile);
final IndexInput input = IndexFileUtils.instance().openInput(postingsFile);
try
{
return new ScanningPostingsReader(input, new PostingsReader.BlocksSummary(input, entry.right));

2
test/unit/org/apache/cassandra/index/sai/utils/IndexInputLeakDetector.java
View File

@ -25,6 +25,7 @@ import java.util.Set;
import com.carrotsearch.randomizedtesting.rules.TestRuleAdapter;
import org.apache.cassandra.index.sai.disk.format.IndexDescriptor;
import org.apache.cassandra.index.sai.disk.io.IndexFileUtils;
import org.apache.cassandra.index.sai.disk.io.TrackingIndexFileUtils;
import org.apache.cassandra.io.sstable.Descriptor;
import org.apache.cassandra.io.util.SequentialWriterOption;
@ -41,6 +42,7 @@ public class IndexInputLeakDetector extends TestRuleAdapter
{
TrackingIndexFileUtils trackingIndexFileUtils = new TrackingIndexFileUtils(sequentialWriterOption);
trackedIndexFileUtils.add(trackingIndexFileUtils);
IndexFileUtils.setOverrideInstance(trackingIndexFileUtils);
return IndexDescriptor.create(descriptor, tableMetadata.partitioner, tableMetadata.comparator);
}