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 trunk

This commit is contained in:
Stefan Miklosovic 2024-10-31 16:24:54 +01:00
parent 32030e4fa3 194e632ab3
commit 15ed18e9d4
No known key found for this signature in database
GPG Key ID: 32F35CB2F546D93E
18 changed files with 88 additions and 75 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
.build/checkstyle.xml
View File

@ -183,6 +183,10 @@
<property name="message" value="Use the CassandraRelevantProperties or CassandraRelevantEnv instead." />
</module>
<module name="IllegalType"> <!-- usage of var check -->
<property name="illegalClassNames" value="var"/>
</module>
<module name="RedundantImport"/>
<module name="UnusedImports"/>

1
CHANGES.txt
View File

@ -94,6 +94,7 @@
* Add the ability to disable bulk loading of SSTables (CASSANDRA-18781)
* Clean up obsolete functions and simplify cql_version handling in cqlsh (CASSANDRA-18787)
Merged from 5.0:
* Ban the usage of "var" instead of full types in the production code (CASSANDRA-20038)
* Suppress CVE-2024-45772, upgrade to lucene-core-9.12.0.jar (CASSANDRA-20024)
* Use SinglePartitionReadCommand for index queries that use strict filtering (CASSANDRA-19968)
* Always write local expiration time as an int to LivenessInfo digest (CASSANDRA-19989)

2
src/java/org/apache/cassandra/cql3/statements/SelectStatement.java
View File

@ -1538,7 +1538,7 @@ public class SelectStatement implements CQLStatement.SingleKeyspaceCqlStatement
return false;
Boolean[] reversedMap = new Boolean[table.clusteringColumns().size()];
int i = 0;
for (var entry : orderingColumns.entrySet())
for (Map.Entry<ColumnMetadata, Ordering> entry : orderingColumns.entrySet())
{
ColumnMetadata def = entry.getKey();
Ordering ordering = entry.getValue();

3
src/java/org/apache/cassandra/index/sai/VectorQueryContext.java
View File

@ -19,6 +19,7 @@
package org.apache.cassandra.index.sai;
import java.io.IOException;
import java.util.Collection;
import java.util.Collections;
import java.util.HashSet;
import java.util.NavigableSet;
@ -181,7 +182,7 @@ public class VectorQueryContext
@Override
public boolean get(int ordinal)
{
var keys = graph.keysFromOrdinal(ordinal);
Collection<PrimaryKey> keys = graph.keysFromOrdinal(ordinal);
return keys.stream().anyMatch(k -> !ignored.contains(k));
}

2
src/java/org/apache/cassandra/index/sai/disk/v1/segment/SegmentTrieBuffer.java
View File

@ -79,7 +79,7 @@ public class SegmentTrieBuffer
public Iterator<IndexEntry> iterator()
{
var iterator = trie.entrySet().iterator();
Iterator<Map.Entry<ByteComparable, PackedLongValues.Builder>> iterator = trie.entrySet().iterator();
return new Iterator<>()
{

17
src/java/org/apache/cassandra/index/sai/disk/v1/segment/VectorIndexSegmentSearcher.java
View File

@ -38,6 +38,7 @@ import org.apache.cassandra.index.sai.StorageAttachedIndex;
import org.apache.cassandra.index.sai.VectorQueryContext;
import org.apache.cassandra.index.sai.disk.PrimaryKeyMap;
import org.apache.cassandra.index.sai.disk.v1.PerColumnIndexFiles;
import org.apache.cassandra.index.sai.disk.v1.postings.VectorPostingList;
import org.apache.cassandra.index.sai.disk.v1.vector.DiskAnn;
import org.apache.cassandra.index.sai.disk.v1.vector.OptimizeFor;
import org.apache.cassandra.index.sai.iterators.KeyRangeIterator;
@ -102,7 +103,7 @@ public class VectorIndexSegmentSearcher extends IndexSegmentSearcher
return toPrimaryKeyIterator(bitsOrPostingList.postingList(), context);
float[] queryVector = index.termType().decomposeVector(exp.lower().value.raw.duplicate());
var vectorPostings = graph.search(queryVector, topK, limit, bitsOrPostingList.getBits());
VectorPostingList vectorPostings = graph.search(queryVector, topK, limit, bitsOrPostingList.getBits());
if (bitsOrPostingList.expectedNodesVisited >= 0)
updateExpectedNodes(vectorPostings.getVisitedCount(), bitsOrPostingList.expectedNodesVisited);
return toPrimaryKeyIterator(vectorPostings, context);
@ -139,7 +140,7 @@ public class VectorIndexSegmentSearcher extends IndexSegmentSearcher
// If num of matches are not bigger than limit, skip ANN.
// (nRows should not include shadowed rows, but context doesn't break those out by segment,
// so we will live with the inaccuracy.)
var nRows = Math.toIntExact(maxSSTableRowId - minSSTableRowId + 1);
int nRows = Math.toIntExact(maxSSTableRowId - minSSTableRowId + 1);
int maxBruteForceRows = min(globalBruteForceRows, maxBruteForceRows(limit, nRows, graph.size()));
if (logger.isTraceEnabled())
logger.trace("Search range covers {} rows; max brute force rows is {} for sstable index with {} nodes, LIMIT {}",
@ -203,7 +204,7 @@ public class VectorIndexSegmentSearcher extends IndexSegmentSearcher
private SparseFixedBitSet bitSetForSearch()
{
var bits = cachedBitSets.get();
SparseFixedBitSet bits = cachedBitSets.get();
bits.clear();
return bits;
}
@ -227,9 +228,9 @@ public class VectorIndexSegmentSearcher extends IndexSegmentSearcher
{
// the iterator represents keys from the whole table -- we'll only pull of those that
// are from our own token range, so we can use row ids to order the results by vector similarity.
var maxSegmentRowId = metadata.toSegmentRowId(metadata.maxSSTableRowId);
int maxSegmentRowId = metadata.toSegmentRowId(metadata.maxSSTableRowId);
SparseFixedBitSet bits = bitSetForSearch();
var rowIds = new IntArrayList();
IntArrayList rowIds = new IntArrayList();
try (var ordinalsView = graph.getOrdinalsView())
{
for (PrimaryKey primaryKey : keysInRange)
@ -259,7 +260,7 @@ public class VectorIndexSegmentSearcher extends IndexSegmentSearcher
// else ask the index to perform a search limited to the bits we created
float[] queryVector = index.termType().decomposeVector(expression.lower().value.raw.duplicate());
var results = graph.search(queryVector, topK, limit, bits);
VectorPostingList results = graph.search(queryVector, topK, limit, bits);
updateExpectedNodes(results.getVisitedCount(), expectedNodesVisited(topK, maxSegmentRowId, graph.size()));
return toPrimaryKeyIterator(results, context);
}
@ -268,7 +269,7 @@ public class VectorIndexSegmentSearcher extends IndexSegmentSearcher
private boolean shouldUseBruteForce(int topK, int limit, int numRows)
{
// if we have a small number of results then let TopK processor do exact NN computation
var maxBruteForceRows = min(globalBruteForceRows, maxBruteForceRows(topK, numRows, graph.size()));
int maxBruteForceRows = min(globalBruteForceRows, maxBruteForceRows(topK, numRows, graph.size()));
if (logger.isTraceEnabled())
logger.trace("SAI materialized {} rows; max brute force rows is {} for sstable index with {} nodes, LIMIT {}",
numRows, maxBruteForceRows, graph.size(), limit);
@ -289,7 +290,7 @@ public class VectorIndexSegmentSearcher extends IndexSegmentSearcher
private int expectedNodesVisited(int limit, int nPermittedOrdinals, int graphSize)
{
var observedRatio = actualExpectedRatio.getUpdateCount() >= 10 ? actualExpectedRatio.get() : 1.0;
double observedRatio = actualExpectedRatio.getUpdateCount() >= 10 ? actualExpectedRatio.get() : 1.0;
return (int) (observedRatio * VectorMemoryIndex.expectedNodesVisited(limit, nPermittedOrdinals, graphSize));
}

4
src/java/org/apache/cassandra/index/sai/disk/v1/vector/BitsUtil.java
View File

@ -94,7 +94,7 @@ public class BitsUtil
@Override
public boolean get(int i)
{
var p = postings.get(i);
VectorPostings<T> p = postings.get(i);
assert p != null : "No postings for ordinal " + i;
return !p.isEmpty();
}
@ -114,7 +114,7 @@ public class BitsUtil
@Override
public boolean get(int i)
{
var p = postings.get(i);
VectorPostings<T> p = postings.get(i);
assert p != null : "No postings for ordinal " + i;
return !p.isEmpty() && toAccept.get(i);
}

4
src/java/org/apache/cassandra/index/sai/disk/v1/vector/CompactionVectorValues.java
View File

@ -79,8 +79,8 @@ public class CompactionVectorValues implements RamAwareVectorValues
writer.writeInt(size());
writer.writeInt(dimension());
for (var i = 0; i < size(); i++) {
var bb = values.get(i);
for (int i = 0; i < size(); i++) {
ByteBuffer bb = values.get(i);
assert bb != null : "null vector at index " + i + " of " + size();
writer.write(bb);
}

17
src/java/org/apache/cassandra/index/sai/disk/v1/vector/DiskAnn.java
View File

@ -27,6 +27,7 @@ import java.util.stream.IntStream;
import io.github.jbellis.jvector.disk.CachingGraphIndex;
import io.github.jbellis.jvector.disk.OnDiskGraphIndex;
import io.github.jbellis.jvector.graph.GraphIndex;
import io.github.jbellis.jvector.graph.GraphSearcher;
import io.github.jbellis.jvector.graph.NeighborSimilarity;
import io.github.jbellis.jvector.graph.SearchResult;
@ -64,7 +65,7 @@ public class DiskAnn implements AutoCloseable
try (var pqFileHandle = indexFiles.compressedVectors(); var reader = new RandomAccessReaderAdapter(pqFileHandle))
{
reader.seek(pqSegmentOffset);
var containsCompressedVectors = reader.readBoolean();
boolean containsCompressedVectors = reader.readBoolean();
if (containsCompressedVectors)
compressedVectors = CompressedVectors.load(reader, reader.getFilePointer());
else
@ -92,8 +93,8 @@ public class DiskAnn implements AutoCloseable
{
OnHeapGraph.validateIndexable(queryVector, similarityFunction);
var view = graph.getView();
var searcher = new GraphSearcher.Builder<>(view).build();
GraphIndex.View<float[]> view = graph.getView();
GraphSearcher<float[]> searcher = new GraphSearcher.Builder<>(view).build();
NeighborSimilarity.ScoreFunction scoreFunction;
NeighborSimilarity.ReRanker<float[]> reRanker;
if (compressedVectors == null)
@ -107,10 +108,10 @@ public class DiskAnn implements AutoCloseable
scoreFunction = compressedVectors.approximateScoreFunctionFor(queryVector, similarityFunction);
reRanker = (i, map) -> similarityFunction.compare(queryVector, map.get(i));
}
var result = searcher.search(scoreFunction,
reRanker,
topK,
ordinalsMap.ignoringDeleted(acceptBits));
SearchResult result = searcher.search(scoreFunction,
reRanker,
topK,
ordinalsMap.ignoringDeleted(acceptBits));
Tracing.trace("DiskANN search visited {} nodes to return {} results", result.getVisitedCount(), result.getNodes().length);
return annRowIdsToPostings(result, limit);
}
@ -134,7 +135,7 @@ public class DiskAnn implements AutoCloseable
{
try
{
var ordinal = it.next().node;
int ordinal = it.next().node;
segmentRowIdIterator = Arrays.stream(rowIdsView.getSegmentRowIdsMatching(ordinal)).iterator();
}
catch (IOException e)

10
src/java/org/apache/cassandra/index/sai/disk/v1/vector/OnDiskOrdinalsMap.java
View File

@ -48,7 +48,7 @@ public class OnDiskOrdinalsMap implements AutoCloseable
{
reader.seek(segmentOffset);
int deletedCount = reader.readInt();
for (var i = 0; i < deletedCount; i++)
for (int i = 0; i < deletedCount; i++)
{
deletedOrdinals.add(reader.readInt());
}
@ -93,7 +93,7 @@ public class OnDiskOrdinalsMap implements AutoCloseable
throw new RuntimeException(String.format("Error seeking to index offset for ordinal %d with ordToRowOffset %d",
vectorOrdinal, ordToRowOffset), e);
}
var offset = reader.readLong();
long offset = reader.readLong();
// seek to and read rowIds
try
{
@ -104,9 +104,9 @@ public class OnDiskOrdinalsMap implements AutoCloseable
throw new RuntimeException(String.format("Error seeking to rowIds offset for ordinal %d with ordToRowOffset %d",
vectorOrdinal, ordToRowOffset), e);
}
var postingsSize = reader.readInt();
var rowIds = new int[postingsSize];
for (var i = 0; i < rowIds.length; i++)
int postingsSize = reader.readInt();
int[] rowIds = new int[postingsSize];
for (int i = 0; i < rowIds.length; i++)
{
rowIds[i] = reader.readInt();
}

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

@ -25,6 +25,7 @@ import java.util.Collection;
import java.util.HashSet;
import java.util.Map;
import java.util.PriorityQueue;
import java.util.Set;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.atomic.AtomicInteger;
@ -41,6 +42,7 @@ import io.github.jbellis.jvector.graph.GraphIndexBuilder;
import io.github.jbellis.jvector.graph.GraphSearcher;
import io.github.jbellis.jvector.graph.NeighborSimilarity;
import io.github.jbellis.jvector.graph.RandomAccessVectorValues;
import io.github.jbellis.jvector.graph.SearchResult;
import io.github.jbellis.jvector.pq.CompressedVectors;
import io.github.jbellis.jvector.pq.ProductQuantization;
import io.github.jbellis.jvector.util.Bits;
@ -132,7 +134,7 @@ public class OnHeapGraph<T>
{
assert term != null && term.remaining() != 0;
var vector = vectorType.composeAsFloat(term);
float[] vector = vectorType.composeAsFloat(term);
if (behavior == InvalidVectorBehavior.IGNORE)
{
try
@ -151,7 +153,7 @@ public class OnHeapGraph<T>
validateIndexable(vector, similarityFunction);
}
var bytesUsed = 0L;
long bytesUsed = 0L;
VectorPostings<T> postings = postingsMap.get(vector);
// if the vector is already in the graph, all that happens is that the postings list is updated
// otherwise, we add the vector in this order:
@ -167,7 +169,7 @@ public class OnHeapGraph<T>
if (postingsMap.putIfAbsent(vector, postings) == null)
{
// we won the race to add the new entry; assign it an ordinal and add to the other structures
var ordinal = nextOrdinal.getAndIncrement();
int ordinal = nextOrdinal.getAndIncrement();
postings.setOrdinal(ordinal);
bytesUsed += RamEstimation.concurrentHashMapRamUsed(1); // the new posting Map entry
bytesUsed += (vectorValues instanceof ConcurrentVectorValues)
@ -242,8 +244,8 @@ public class OnHeapGraph<T>
{
assert term != null && term.remaining() != 0;
var vector = vectorType.composeAsFloat(term);
var postings = postingsMap.get(vector);
float[] vector = vectorType.composeAsFloat(term);
VectorPostings<T> postings = postingsMap.get(vector);
if (postings == null)
{
// it's possible for this to be called against a different memtable than the one
@ -269,11 +271,11 @@ public class OnHeapGraph<T>
Bits bits = hasDeletions ? BitsUtil.bitsIgnoringDeleted(toAccept, postingsByOrdinal) : toAccept;
GraphIndex<float[]> graph = builder.getGraph();
var searcher = new GraphSearcher.Builder<>(graph.getView()).withConcurrentUpdates().build();
GraphSearcher<float[]> searcher = new GraphSearcher.Builder<>(graph.getView()).withConcurrentUpdates().build();
NeighborSimilarity.ExactScoreFunction scoreFunction = node2 -> vectorCompareFunction(queryVector, node2);
var result = searcher.search(scoreFunction, null, limit, bits);
SearchResult result = searcher.search(scoreFunction, null, limit, bits);
Tracing.trace("ANN search visited {} in-memory nodes to return {} results", result.getVisitedCount(), result.getNodes().length);
var a = result.getNodes();
SearchResult.NodeScore[] a = result.getNodes();
PriorityQueue<T> keyQueue = new PriorityQueue<>();
for (int i = 0; i < a.length; i++)
keyQueue.addAll(keysFromOrdinal(a[i].node));
@ -305,7 +307,7 @@ public class OnHeapGraph<T>
long pqPosition = writePQ(pqOutput.asSequentialWriter());
long pqLength = pqPosition - pqOffset;
var deletedOrdinals = new HashSet<Integer>();
Set<Integer> deletedOrdinals = new HashSet<>();
postingsMap.values().stream().filter(VectorPostings::isEmpty).forEach(vectorPostings -> deletedOrdinals.add(vectorPostings.getOrdinal()));
// remove ordinals that don't have corresponding row ids due to partition/range deletion
for (VectorPostings<T> vectorPostings : postingsMap.values())
@ -371,7 +373,7 @@ public class OnHeapGraph<T>
.mapToObj(i -> pq.encode(vectorValues.vectorValue(i)))
.toArray(byte[][]::new);
}
var cv = new CompressedVectors(pq, encoded);
CompressedVectors cv = new CompressedVectors(pq, encoded);
// save
cv.write(writer);
return writer.position();

13
src/java/org/apache/cassandra/index/sai/disk/v1/vector/RandomAccessReaderAdapter.java
View File

@ -28,6 +28,7 @@ import com.google.common.primitives.Ints;
import io.github.jbellis.jvector.disk.ReaderSupplier;
import org.apache.cassandra.io.util.FileHandle;
import org.apache.cassandra.io.util.RandomAccessReader;
import org.apache.cassandra.io.util.Rebufferer.BufferHolder;
public class RandomAccessReaderAdapter extends RandomAccessReader implements io.github.jbellis.jvector.disk.RandomAccessReader
{
@ -44,7 +45,7 @@ public class RandomAccessReaderAdapter extends RandomAccessReader implements io.
@Override
public void readFully(float[] dest) throws IOException
{
var bh = bufferHolder;
BufferHolder bh = bufferHolder;
long position = getPosition();
FloatBuffer floatBuffer;
@ -60,7 +61,7 @@ public class RandomAccessReaderAdapter extends RandomAccessReader implements io.
{
// offset is non-zero, and probably not aligned to Float.BYTES, so
// set the position before converting to FloatBuffer.
var bb = bh.buffer();
ByteBuffer bb = bh.buffer();
bb.position(Ints.checkedCast(position - bh.offset()));
floatBuffer = bb.asFloatBuffer();
}
@ -68,7 +69,7 @@ public class RandomAccessReaderAdapter extends RandomAccessReader implements io.
if (dest.length > floatBuffer.remaining())
{
// slow path -- desired slice is across region boundaries
var bb = ByteBuffer.allocate(Float.BYTES * dest.length);
ByteBuffer bb = ByteBuffer.allocate(Float.BYTES * dest.length);
readFully(bb);
floatBuffer = bb.asFloatBuffer();
}
@ -92,7 +93,7 @@ public class RandomAccessReaderAdapter extends RandomAccessReader implements io.
if (count == 0)
return;
var bh = bufferHolder;
BufferHolder bh = bufferHolder;
long position = getPosition();
IntBuffer intBuffer;
@ -108,7 +109,7 @@ public class RandomAccessReaderAdapter extends RandomAccessReader implements io.
{
// offset is non-zero, and probably not aligned to Integer.BYTES, so
// set the position before converting to IntBuffer.
var bb = bh.buffer();
ByteBuffer bb = bh.buffer();
bb.position(Ints.checkedCast(position - bh.offset()));
intBuffer = bb.asIntBuffer();
}
@ -116,7 +117,7 @@ public class RandomAccessReaderAdapter extends RandomAccessReader implements io.
if (count > intBuffer.remaining())
{
// slow path -- desired slice is across region boundaries
var bb = ByteBuffer.allocate(Integer.BYTES * count);
ByteBuffer bb = ByteBuffer.allocate(Integer.BYTES * count);
readFully(bb);
intBuffer = bb.asIntBuffer();
}

21
src/java/org/apache/cassandra/index/sai/disk/v1/vector/VectorPostingsWriter.java
View File

@ -25,6 +25,7 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import org.agrona.collections.IntArrayList;
import org.apache.cassandra.io.util.SequentialWriter;
import org.apache.cassandra.utils.Pair;
@ -45,7 +46,7 @@ public class VectorPostingsWriter<T>
private void writeDeletedOrdinals(SequentialWriter writer, Set<Integer> deletedOrdinals) throws IOException
{
writer.writeInt(deletedOrdinals.size());
for (var ordinal : deletedOrdinals) {
for (int ordinal : deletedOrdinals) {
writer.writeInt(ordinal);
}
}
@ -60,21 +61,21 @@ public class VectorPostingsWriter<T>
writer.writeInt(vectorValues.size());
// Write the offsets of the postings for each ordinal
var offsetsStartAt = ordToRowOffset + 4L + 8L * vectorValues.size();
var nextOffset = offsetsStartAt;
for (var i = 0; i < vectorValues.size(); i++) {
long offsetsStartAt = ordToRowOffset + 4L + 8L * vectorValues.size();
long nextOffset = offsetsStartAt;
for (int i = 0; i < vectorValues.size(); i++) {
// (ordinal is implied; don't need to write it)
writer.writeLong(nextOffset);
var rowIds = postingsMap.get(vectorValues.vectorValue(i)).getRowIds();
IntArrayList rowIds = postingsMap.get(vectorValues.vectorValue(i)).getRowIds();
nextOffset += 4 + (rowIds.size() * 4L); // 4 bytes for size and 4 bytes for each integer in the list
}
assert writer.position() == offsetsStartAt : "writer.position()=" + writer.position() + " offsetsStartAt=" + offsetsStartAt;
// Write postings lists
for (var i = 0; i < vectorValues.size(); i++) {
for (int i = 0; i < vectorValues.size(); i++) {
VectorPostings<T> postings = postingsMap.get(vectorValues.vectorValue(i));
var rowIds = postings.getRowIds();
IntArrayList rowIds = postings.getRowIds();
writer.writeInt(rowIds.size());
for (int r = 0; r < rowIds.size(); r++)
writer.writeInt(rowIds.getInt(r));
@ -89,8 +90,8 @@ public class VectorPostingsWriter<T>
List<Pair<Integer, Integer>> pairs = new ArrayList<>();
// Collect all (rowId, vectorOrdinal) pairs
for (var i = 0; i < vectorValues.size(); i++) {
var rowIds = postingsMap.get(vectorValues.vectorValue(i)).getRowIds();
for (int i = 0; i < vectorValues.size(); i++) {
IntArrayList rowIds = postingsMap.get(vectorValues.vectorValue(i)).getRowIds();
for (int r = 0; r < rowIds.size(); r++)
pairs.add(Pair.create(rowIds.getInt(r), i));
}
@ -100,7 +101,7 @@ public class VectorPostingsWriter<T>
// Write the pairs to the file
long startOffset = writer.position();
for (var pair : pairs) {
for (Pair<Integer, Integer> pair : pairs) {
writer.writeInt(pair.left);
writer.writeInt(pair.right);
}

2
src/java/org/apache/cassandra/index/sai/iterators/KeyRangeOrderingIterator.java
View File

@ -65,7 +65,7 @@ public class KeyRangeOrderingIterator extends KeyRangeIterator
}
while (nextKeys.size() < chunkSize && input.hasNext());
// Get the next iterator before closing this one to prevent releasing the resource.
var previousIterator = nextIterator;
KeyRangeIterator previousIterator = nextIterator;
// If this results in an exception, previousIterator is closed in close() method.
nextIterator = nextRangeFunction.apply(nextKeys);
if (previousIterator != null)

25
src/java/org/apache/cassandra/index/sai/memory/VectorMemoryIndex.java
View File

@ -20,6 +20,7 @@ package org.apache.cassandra.index.sai.memory;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Collection;
import java.util.Iterator;
import java.util.List;
import java.util.NavigableSet;
@ -80,8 +81,8 @@ public class VectorMemoryIndex extends MemoryIndex
if (value == null || value.remaining() == 0 || !index.validateTermSize(key, value, false, null))
return 0;
var primaryKey = index.hasClustering() ? index.keyFactory().create(key, clustering)
: index.keyFactory().create(key);
PrimaryKey primaryKey = index.hasClustering() ? index.keyFactory().create(key, clustering)
: index.keyFactory().create(key);
return index(primaryKey, value);
}
@ -116,8 +117,8 @@ public class VectorMemoryIndex extends MemoryIndex
long bytesUsed = 0;
if (different)
{
var primaryKey = index.hasClustering() ? index.keyFactory().create(key, clustering)
: index.keyFactory().create(key);
PrimaryKey primaryKey = index.hasClustering() ? index.keyFactory().create(key, clustering)
: index.keyFactory().create(key);
// update bounds because only rows with vectors are included in the key bounds,
// so if the vector was null before, we won't have included it
updateKeyBounds(primaryKey);
@ -154,7 +155,7 @@ public class VectorMemoryIndex extends MemoryIndex
VectorQueryContext vectorQueryContext = queryContext.vectorContext();
var buffer = expr.lower().value.raw;
ByteBuffer buffer = expr.lower().value.raw;
float[] qv = index.termType().decomposeVector(buffer);
Bits bits;
@ -191,7 +192,7 @@ public class VectorMemoryIndex extends MemoryIndex
bits = queryContext.vectorContext().bitsetForShadowedPrimaryKeys(graph);
}
var keyQueue = graph.search(qv, queryContext.vectorContext().limit(), bits);
PriorityQueue<PrimaryKey> keyQueue = graph.search(qv, queryContext.vectorContext().limit(), bits);
if (keyQueue.isEmpty())
return KeyRangeIterator.empty();
return new ReorderingRangeIterator(keyQueue);
@ -221,8 +222,8 @@ public class VectorMemoryIndex extends MemoryIndex
ByteBuffer buffer = expression.lower().value.raw;
float[] qv = index.termType().decomposeVector(buffer);
var bits = new KeyFilteringBits(results);
var keyQueue = graph.search(qv, limit, bits);
KeyFilteringBits bits = new KeyFilteringBits(results);
PriorityQueue<PrimaryKey> keyQueue = graph.search(qv, limit, bits);
if (keyQueue.isEmpty())
return KeyRangeIterator.empty();
return new ReorderingRangeIterator(keyQueue);
@ -247,8 +248,8 @@ public class VectorMemoryIndex extends MemoryIndex
{
// constants are computed by Code Interpreter based on observed comparison counts in tests
// https://chat.openai.com/share/2b1d7195-b4cf-4a45-8dce-1b9b2f893c75
var sizeRestriction = min(nPermittedOrdinals, graphSize);
var raw = (int) (0.7 * pow(log(graphSize), 2) *
int sizeRestriction = min(nPermittedOrdinals, graphSize);
int raw = (int) (0.7 * pow(log(graphSize), 2) *
pow(graphSize, 0.33) *
pow(log(limit), 2) *
pow(log((double) graphSize / sizeRestriction), 2) / pow(sizeRestriction, 0.13));
@ -309,7 +310,7 @@ public class VectorMemoryIndex extends MemoryIndex
if (bits != null && !bits.get(ordinal))
return false;
var keys = graph.keysFromOrdinal(ordinal);
Collection<PrimaryKey> keys = graph.keysFromOrdinal(ordinal);
return keys.stream().anyMatch(k -> keyRange.contains(k.partitionKey()));
}
@ -362,7 +363,7 @@ public class VectorMemoryIndex extends MemoryIndex
@Override
public boolean get(int i)
{
var pk = graph.keysFromOrdinal(i);
Collection<PrimaryKey> pk = graph.keysFromOrdinal(i);
return results.stream().anyMatch(pk::contains);
}

6
src/java/org/apache/cassandra/index/sai/plan/Operation.java
View File

@ -258,13 +258,13 @@ public class Operation
*/
static KeyRangeIterator buildIterator(QueryController controller)
{
var orderings = controller.indexFilter().getExpressions()
.stream().filter(e -> e.operator() == Operator.ANN).collect(Collectors.toList());
List<RowFilter.Expression> orderings = controller.indexFilter().getExpressions()
.stream().filter(e -> e.operator() == Operator.ANN).collect(Collectors.toList());
assert orderings.size() <= 1;
if (controller.indexFilter().getExpressions().size() == 1 && orderings.size() == 1)
// If we only have one expression, we just use the ANN index to order and limit.
return controller.getTopKRows(orderings.get(0));
var iterator = Node.buildTree(controller.indexFilter()).analyzeTree(controller).rangeIterator(controller);
KeyRangeIterator iterator = Node.buildTree(controller.indexFilter()).analyzeTree(controller).rangeIterator(controller);
if (orderings.isEmpty())
return iterator;
return controller.getTopKRows(iterator, orderings.get(0));

8
src/java/org/apache/cassandra/index/sai/plan/QueryController.java
View File

@ -309,7 +309,7 @@ public class QueryController
assert expression.operator() == Operator.ANN;
StorageAttachedIndex index = indexFor(expression);
assert index != null;
var planExpression = Expression.create(index).add(Operator.ANN, expression.getIndexValue().duplicate());
Expression planExpression = Expression.create(index).add(Operator.ANN, expression.getIndexValue().duplicate());
// search memtable before referencing sstable indexes; otherwise we may miss newly flushed memtable index
KeyRangeIterator memtableResults = index.memtableIndexManager().searchMemtableIndexes(queryContext, planExpression, mergeRange);
@ -345,10 +345,10 @@ public class QueryController
// Filter out PKs now. Each PK is passed to every segment of the ANN index, so filtering shadowed keys
// eagerly can save some work when going from PK to row id for on disk segments.
// Since the result is shared with multiple streams, we use an unmodifiable list.
var sourceKeys = rawSourceKeys.stream().filter(vectorQueryContext::shouldInclude).collect(Collectors.toList());
List<PrimaryKey> sourceKeys = rawSourceKeys.stream().filter(vectorQueryContext::shouldInclude).collect(Collectors.toList());
StorageAttachedIndex index = indexFor(expression);
assert index != null : "Cannot do ANN ordering on an unindexed column";
var planExpression = Expression.create(index);
Expression planExpression = Expression.create(index);
planExpression.add(Operator.ANN, expression.getIndexValue().duplicate());
// search memtable before referencing sstable indexes; otherwise we may miss newly flushed memtable index
@ -388,7 +388,7 @@ public class QueryController
*/
private KeyRangeIterator createRowIdIterator(Pair<Expression, Collection<SSTableIndex>> indexExpression)
{
var subIterators = indexExpression.right
List<KeyRangeIterator> subIterators = indexExpression.right
.stream()
.map(index ->
{

2
src/java/org/apache/cassandra/index/sai/utils/AtomicRatio.java
View File

@ -54,7 +54,7 @@ public class AtomicRatio
public double get()
{
var current = ratio.get();
Ratio current = ratio.get();
return (double) current.numerator / current.denominator;
}