hudi-agent commented on code in PR #19513: URL: https://github.com/apache/hudi/pull/19513#discussion_r3716426251
########## hudi-common/src/main/java/org/apache/hudi/common/index/vector/search/VectorContinuationController.java: ########## @@ -0,0 +1,81 @@ +/* + * 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.hudi.common.index.vector.search; + +import java.util.List; + +/** + * Windows a single retained, distance-ordered candidate pool into continuation batches + * (RFC-109 ยง10). The candidate scan retains one ordered pool of at most + * {@code maxRerankCandidates} in a single MDT scan; this controller draws successive batches + * <em>from that retained pool</em> โ it never rescans MDT postings. + * + * <p>The first {@link #nextBatch()} returns up to {@code initialRerankCandidates}; subsequent calls + * return up to {@code rerankBatchSize}, until the pool (bounded by {@code maxRerankCandidates}) is + * exhausted. The reranker calls {@link #nextBatch()} while {@link #hasMore()} and the top-K + * accumulator still {@link VectorTopKAccumulator#needsMore() needs more} live results (and the + * deadline has not passed). + */ +public final class VectorContinuationController<T> { + + private final List<T> orderedPool; + private final int initialRerankCandidates; + private final int rerankBatchSize; + private final int effectiveMax; + private int cursor; + + public VectorContinuationController(List<T> orderedPool, + int initialRerankCandidates, + int rerankBatchSize, + int maxRerankCandidates) { + if (initialRerankCandidates <= 0 || rerankBatchSize <= 0) { + throw new IllegalArgumentException("batch sizes must be positive"); + } + this.orderedPool = orderedPool; + this.initialRerankCandidates = initialRerankCandidates; + this.rerankBatchSize = rerankBatchSize; + this.effectiveMax = Math.min(orderedPool.size(), Math.max(0, maxRerankCandidates)); + this.cursor = 0; + } + + /** Whether more retained candidates remain to draw (within {@code maxRerankCandidates}). */ + public boolean hasMore() { + return cursor < effectiveMax; + } + + /** Number of candidates drawn so far (monotonic; never exceeds the retained pool bound). */ + public int consumed() { + return cursor; + } + + /** + * Draw the next continuation batch as a window over the retained pool. First call returns up to + * {@code initialRerankCandidates}; later calls up to {@code rerankBatchSize}. Never rescans. + */ + public List<T> nextBatch() { + if (!hasMore()) { + return java.util.Collections.emptyList(); + } + int size = cursor == 0 ? initialRerankCandidates : rerankBatchSize; + int end = Math.min(cursor + size, effectiveMax); Review Comment: ๐ค `nextBatch()` returns `orderedPool.subList(...)`, which is a live view backed by the full pool, and `ListVectorCandidatePool` passes it straight into `HoodieListData.eager(...)` (which I confirmed keeps the list by reference, no copy). Two things I'd worry about: `ArrayList`'s SubList isn't `Serializable`, so if a batch ever gets shipped/serialized it'll throw; and since it's a mutable view, any downstream consumer that sorts/mutates the batch in place would corrupt the pool and later batches. Would it be safer to return `new ArrayList<>(orderedPool.subList(cursor, end))`? <sub><i>โ ๏ธ AI-generated; verify before applying. React ๐/๐ to flag quality.</i></sub> ########## hudi-common/src/main/java/org/apache/hudi/common/index/vector/VectorIndexPruner.java: ########## @@ -0,0 +1,154 @@ +/* + * 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.hudi.common.index.vector; + +import java.io.Serializable; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +/** + * Engine-agnostic IVF cluster pruner for vector queries. + * + * <p>Given a set of centroids and a file-group-to-cluster mapping, determines which + * file groups need to be scanned for an approximate nearest-neighbour query. + * + * <p>Used by both the Spark file index ({@code HoodieVectorAwareFileIndex}) and the + * Trino split manager ({@code HudiVectorSplitManager}), keeping all math in one place. + * + * <p>Instances are built from MDT data at query planning time and are short-lived + * (one per query or per query batch). They are not cached between queries because + * centroids can change after LIRE compaction. + */ +public final class VectorIndexPruner implements Serializable { + + private static final long serialVersionUID = 1L; + + /** Centroid vectors keyed by cluster id (0-based). */ + private final float[][] centroids; + + /** + * Mapping from cluster id โ set of file group ids containing vectors in that cluster. + * The inner sets are unmodifiable. + */ + private final Map<Integer, Set<String>> clusterToFileGroups; + + /** Distance metric for centroid scoring. */ + private final VectorDistanceMetric metric; + + /** + * @param centroids centroid vectors, indexed by cluster id + * @param clusterToFileGroups mapping built from the MDT fg_mapping partition + * @param metric distance metric matching the index definition + */ + public VectorIndexPruner( + float[][] centroids, + Map<Integer, Set<String>> clusterToFileGroups, + VectorDistanceMetric metric) { + this.centroids = centroids; + this.clusterToFileGroups = clusterToFileGroups; + this.metric = metric; + } + + /** + * Returns the set of file group ids that must be scanned to answer an ANN query. + * + * @param queryVector the query embedding + * @param numProbes number of clusters to probe (nProbes) + * @return file group ids; never null, may be empty if the index is not yet initialized + */ + public Set<String> probe(float[] queryVector, int numProbes) { + if (centroids == null || centroids.length == 0) { + return Collections.emptySet(); + } + int effectiveProbes = Math.min(numProbes, centroids.length); + int[] topClusters = findTopClusters(queryVector, effectiveProbes); + + Set<String> fileGroups = new HashSet<>(); + for (int clusterId : topClusters) { + Set<String> fgs = clusterToFileGroups.get(clusterId); + if (fgs != null) { + fileGroups.addAll(fgs); + } + } + return Collections.unmodifiableSet(fileGroups); + } + + /** + * Returns the cluster ids closest to the query (sorted best-first). + * Linear scan; HNSW routing replaces this in Phase 2. + */ + public int[] findTopClusters(float[] query, int numProbes) { Review Comment: ๐ค `findTopClusters` is public but will throw `IndexOutOfBoundsException` at `scored.get(i)` if `numProbes > centroids.length`. `probe()` guards this via `Math.min(...)`, but the class doc says this is also called directly by the Spark file index and Trino split manager. Since the actually-trained centroid count can be smaller than the configured `num_clusters`, would it be worth clamping `numProbes` to `centroids.length` here too, mirroring `probe`? <sub><i>โ ๏ธ AI-generated; verify before applying. React ๐/๐ to flag quality.</i></sub> ########## hudi-common/src/test/java/org/apache/hudi/common/index/vector/search/TestCommonVectorSearchExecutor.java: ########## @@ -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.hudi.common.index.vector.search; + +import org.apache.hudi.common.data.HoodieListData; +import org.apache.hudi.common.index.vector.VectorDistanceMetric; +import org.apache.hudi.common.model.HoodieRecordGlobalLocation; + +import org.junit.jupiter.api.Test; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.atomic.AtomicBoolean; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * End-to-end wiring test for {@link CommonVectorSearchExecutor} (RFC-109 ยง11): drives the full + * stage pipeline with a fake candidate source and reranker but the real + * {@link RecordIndexVectorCandidateArbiter} and {@link DefaultVectorFetchPlanner}, asserting the + * snapshot is resolved once, DELETED candidates are dropped end-to-end, and results flow through. + */ +public class TestCommonVectorSearchExecutor { + + private static VectorCandidate candidate(String key, String fileId) { + VectorPostingLocator loc = new VectorPostingLocator(1, 0, 0, 0L, 0, "p", fileId, "001", 7L); + return new VectorCandidate(key, 0, 0, 1.0, loc); + } + + private static VectorSearchRequest request() { + VectorSearchBudget budget = VectorSearchBudget.defaults(3, 5000L); + return new VectorSearchRequest("embedding", new float[] {1f, 2f}, VectorDistanceMetric.L2, + 3, 32, 50, true, null, budget); + } + + @Test + void drivesFullPipelineAndDropsDeletedEndToEnd() { + List<VectorCandidate> scanned = new ArrayList<>(); + scanned.add(candidate("k1", "fileA")); + scanned.add(candidate("k2", "fileA")); + scanned.add(candidate("k3", "fileB")); // will be DELETED via RLI miss + + AtomicBoolean snapshotResolved = new AtomicBoolean(false); + VectorSnapshotResolver resolver = req -> { + snapshotResolved.set(true); + return new VectorSearchSnapshot("001", + new VectorIndexSnapshot(1, 1, 1, "rot-v1", "quant-v1")); + }; + + // Real arbiter with a fake RLI: k1/k2 live & matching (SERVE), k3 absent (DELETED). + Map<String, HoodieRecordGlobalLocation> rli = new HashMap<>(); + rli.put("k1", new HoodieRecordGlobalLocation("p", "001", "fileA")); + rli.put("k2", new HoodieRecordGlobalLocation("p", "001", "fileA")); + RecordIndexLookup lookup = (keys, tableInstant) -> { + Map<String, HoodieRecordGlobalLocation> out = new HashMap<>(); + for (String k : keys) { + if (rli.containsKey(k)) { + out.put(k, rli.get(k)); + } + } + return out; + }; + + VectorCandidateSource source = (plan, ec) -> new ListVectorCandidatePool(scanned, plan.getRequest().getBudget()); + VectorCandidateArbiter arbiter = new RecordIndexVectorCandidateArbiter(lookup); + VectorFetchPlanner planner = new DefaultVectorFetchPlanner(); + // Fake reranker: emit one result per fetched row (distance = approx), preserving live location. + VectorExactReranker reranker = (tasks, req, snap, ec) -> { + List<VectorSearchResult> results = new ArrayList<>(); + for (VectorFetchTask task : tasks.collectAsList()) { + for (VectorRowRequest r : task.getRequests()) { + results.add(new VectorSearchResult(r.getRecordKey(), r.getApproximateDistance(), + new HoodieRecordGlobalLocation("p", task.getBaseInstant(), task.getFileId()))); + } + } + return HoodieListData.eager(results); + }; + + CommonVectorSearchExecutor executor = new CommonVectorSearchExecutor( + resolver, new ThresholdVectorExecutionModeSelector(), source, arbiter, planner, reranker); + + List<VectorSearchResult> results = executor.execute(request(), null).collectAsList(); + + assertTrue(snapshotResolved.get(), "snapshot must be resolved once at the top of the pipeline"); + assertEquals(2, results.size(), "k3 (DELETED via RLI miss) must be dropped end-to-end"); + List<String> keys = new ArrayList<>(); + for (VectorSearchResult r : results) { + keys.add(r.getRecordKey()); + } + assertTrue(keys.contains("k1") && keys.contains("k2")); Review Comment: ๐ค nit: `assertTrue(!keys.contains("k3"))` reads more clearly as `assertFalse(keys.contains("k3"))` โ matches how the assertion failure message reads and is consistent with the JUnit idiom. <sub><i>โ ๏ธ AI-generated; verify before applying. React ๐/๐ to flag quality.</i></sub> ########## hudi-common/src/main/java/org/apache/hudi/metadata/HoodieMetadataPayload.java: ########## @@ -253,6 +300,410 @@ protected HoodieMetadataPayload(String key, HoodieSecondaryIndexInfo secondaryIn this(key, MetadataPartitionType.SECONDARY_INDEX.getRecordType(), null, null, null, null, secondaryIndexMetadata, secondaryIndexMetadata.getIsDeleted()); } + protected HoodieMetadataPayload(String key, Object vectorIndexInfo) { + this.key = key; + this.type = MetadataPartitionType.VECTOR_INDEX.getRecordType(); + this.vectorIndexMetadata = vectorIndexInfo; + this.isDeletedRecord = vectorIndexInfo instanceof HoodieVectorIndexTombstone; + } + + /** + * Create the singleton reader-visible generation pointer. + */ + public static HoodieRecord<HoodieMetadataPayload> createVectorIndexActiveManifestRecord( + Integer activeGeneration, String metadataPartitionPath) { + String recordKey = VectorIndexMetadataKey.activeManifest(); + HoodieVectorIndexActiveManifest manifest = new HoodieVectorIndexActiveManifest(1, activeGeneration); + return new HoodieAvroRecord<>( + new HoodieKey(recordKey, metadataPartitionPath), + new HoodieMetadataPayload(recordKey, manifest)); + } + + /** + * Create the generation-one centroid record for the given index partition. + */ + public static HoodieRecord<HoodieMetadataPayload> createVectorIndexCentroidsRecord( + ByteBuffer centroidBytes, String partitionPath) { + HoodieVectorIndexCentroids centroids = new HoodieVectorIndexCentroids( + ByteBuffer.allocate(0), + centroidBytes, + ByteBuffer.allocate(0)); + String recordKey = VectorIndexMetadataKey.centroids(1, 0); + HoodieMetadataPayload payload = new HoodieMetadataPayload(recordKey, centroids); + HoodieKey key = new HoodieKey(recordKey, partitionPath); + return new HoodieAvroRecord<>(key, payload); + } + + public static HoodieRecord<HoodieMetadataPayload> createVectorIndexCentroidsRecord( + int generation, + int chunk, + ByteBuffer clusterIds, + ByteBuffer centroidBytes, + ByteBuffer clusterRadii, + String partitionPath) { + String recordKey = VectorIndexMetadataKey.centroids(generation, chunk); + HoodieVectorIndexCentroids centroids = new HoodieVectorIndexCentroids( + clusterIds, centroidBytes, clusterRadii); + return new HoodieAvroRecord<>( + new HoodieKey(recordKey, partitionPath), + new HoodieMetadataPayload(recordKey, centroids)); + } + + /** + * Create the generation-one quantizer metadata record for the given index partition. + */ + public static HoodieRecord<HoodieMetadataPayload> createVectorIndexQuantizerMetadataRecord( + String quantizerType, + int quantizedCodeBytes, + long randomSeed, + boolean assumeNormalized, + String partitionPath) { + return createVectorIndexQuantizerMetadataRecord( + quantizerType, + quantizedCodeBytes, + 1, + randomSeed, + assumeNormalized, + partitionPath); + } + + public static HoodieRecord<HoodieMetadataPayload> createVectorIndexQuantizerMetadataRecord( + String quantizerType, + int quantizedCodeBytes, + int rabitqBits, + long randomSeed, + boolean assumeNormalized, + String partitionPath) { + return createVectorIndexQuantizerMetadataRecord(1, 0, quantizerType, randomSeed, null, partitionPath); + } + + public static HoodieRecord<HoodieMetadataPayload> createVectorIndexQuantizerMetadataRecord( + int generation, + int chunk, + String quantizerType, + long randomSeed, + ByteBuffer rotationBytes, + String partitionPath) { + String recordKey = VectorIndexMetadataKey.quantizer(generation, chunk); + HoodieVectorIndexQuantizer quantizer = new HoodieVectorIndexQuantizer(quantizerType, randomSeed, rotationBytes); + return new HoodieAvroRecord<>( + new HoodieKey(recordKey, partitionPath), + new HoodieMetadataPayload(recordKey, quantizer)); + } + + public static HoodieRecord<HoodieMetadataPayload> createVectorIndexManifestRecord( + int generation, + String generationOrdinalText, + String state, + int dim, + int dimPadded, + int codeRowBytes, + int bitsTotal, + int numExPlanes, + int numClusters, + int shardCount, + int fileGroupCount, + String metric, + boolean assumeNormalized, + boolean residualEncoding, + String vectorColumn, + int targetBlockBytes, + int vectorsPerBlock, + int blockFormatVersion, + int factorVersion, + double kappa, + double gMin, + double eps1Max, + double epsNRel, + int centroidChunkCount, + String centroidChecksum, + int splitLimit, + int mergeFloor, + String bootstrapInstant, + String verifiedFrontier, + long createdTs, + String metadataPartitionPath) { + String recordKey = VectorIndexMetadataKey.manifest(generation); + HoodieVectorIndexManifest manifest = new HoodieVectorIndexManifest( + 1, + generationOrdinalText, + state, + dim, + dimPadded, + codeRowBytes, + bitsTotal, + numExPlanes, + numClusters, + shardCount, + fileGroupCount, + metric, + assumeNormalized, + residualEncoding, + vectorColumn == null ? "" : vectorColumn, + targetBlockBytes, + vectorsPerBlock, + blockFormatVersion, + factorVersion, + kappa, + gMin, + eps1Max, + epsNRel, + centroidChunkCount, + centroidChecksum, + splitLimit, + mergeFloor, + bootstrapInstant, + verifiedFrontier, + createdTs); + return new HoodieAvroRecord<>( + new HoodieKey(recordKey, metadataPartitionPath), + new HoodieMetadataPayload(recordKey, manifest)); + } + + public static HoodieRecord<HoodieMetadataPayload> createVectorIndexClusterManifestRecord( + int generation, + int clusterId, + int shardCount, + Collection<String> fileGroupIds, + long vectorCount, + long lastUpdatedTs, + String metadataPartitionPath) { + return createVectorIndexClusterManifestRecord( + generation, clusterId, 0, shardCount, fileGroupIds, vectorCount, lastUpdatedTs, metadataPartitionPath); + } + + public static HoodieRecord<HoodieMetadataPayload> createVectorIndexClusterManifestRecord( + int generation, + int clusterId, + int routingVersion, + int shardCount, + Collection<String> fileGroupIds, + long vectorCount, + long lastUpdatedTs, + String metadataPartitionPath) { + String recordKey = VectorIndexMetadataKey.clusterStats(generation, clusterId); + HoodieVectorIndexClusterStats stats = new HoodieVectorIndexClusterStats( + routingVersion, + shardCount, + fileGroupIds == null ? java.util.Collections.emptyList() : fileGroupIds.stream().collect(Collectors.toList()), + vectorCount, + 0L, + 0L, + null, + lastUpdatedTs); + return new HoodieAvroRecord<>( + new HoodieKey(recordKey, metadataPartitionPath), + new HoodieMetadataPayload(recordKey, stats)); + } + + public static HoodieRecord<HoodieMetadataPayload> createVectorIndexClusterStatsRecord( + int generation, + int clusterId, + long liveCount, + long deltaCount, + long tombstoneCount, + String metadataPartitionPath) { + String recordKey = VectorIndexMetadataKey.clusterStats(generation, clusterId); + HoodieVectorIndexClusterStats stats = new HoodieVectorIndexClusterStats( + 0, + 1, + java.util.Collections.emptyList(), Review Comment: ๐ค `createVectorIndexClusterStatsRecord` and `createVectorIndexClusterManifestRecord` above both write to the same key (`clusterStats(generation, clusterId)`), but this variant sets `fileGroupIds` to `emptyList()` and `shardCount` to `1`. Since `VECTOR_INDEX` doesn't override `combineMetadataPayloads` (the default returns `newer` wholesale), a stats update landing after the manifest record would wipe the reader-critical `fileGroupIds`/`shardCount`. Is the intent that these two writers merge field-by-field, or that only one is ever written per key? Looks latent for now (only the manifest variant is wired in), but worth pinning down before incremental maintenance lands. <sub><i>โ ๏ธ AI-generated; verify before applying. React ๐/๐ to flag quality.</i></sub> ########## hudi-common/src/test/java/org/apache/hudi/common/index/vector/search/TestCommonVectorSearchContinuation.java: ########## @@ -0,0 +1,85 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file for details. + */ + +package org.apache.hudi.common.index.vector.search; + +import org.apache.hudi.common.index.vector.VectorDistanceMetric; +import org.apache.hudi.common.index.vector.VectorStalePolicy; +import org.apache.hudi.common.model.HoodieRecordGlobalLocation; + +import org.junit.jupiter.api.Test; + +import java.util.Arrays; +import java.util.List; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +class TestCommonVectorSearchContinuation { + + @Test + void drawsAnotherRetainedWindowWithoutRescanning() { + VectorSearchBudget budget = new VectorSearchBudget( + 5000, 2, 2, 4, 10, 1, VectorExecutionMode.LOCAL, 10, DeadlinePolicy.FAIL); + VectorSearchRequest request = new VectorSearchRequest( + "embedding", new float[] {1f}, VectorDistanceMetric.L2, 2, 1, 1, true, + VectorStalePolicy.FALLBACK, "001", budget); + List<VectorCandidate> candidates = Arrays.asList( + candidate("deleted-1", 1), candidate("deleted-2", 2), + candidate("live-1", 3), candidate("live-2", 4)); + + AtomicInteger scans = new AtomicInteger(); + AtomicReference<ListVectorCandidatePool> retainedPool = new AtomicReference<>(); + VectorCandidateSource source = (plan, context) -> { + scans.incrementAndGet(); + ListVectorCandidatePool pool = new ListVectorCandidatePool(candidates, budget); + retainedPool.set(pool); + return pool; + }; + RecordIndexLookup lookup = (keys, instant) -> { + java.util.Map<String, HoodieRecordGlobalLocation> locations = new java.util.HashMap<>(); Review Comment: ๐ค nit: the inline `java.util.Map` / `java.util.HashMap` / `java.util.ArrayList` FQNs inside the lambda bodies are harder to scan than normal imports โ could you move these to the top-level import block instead? <sub><i>โ ๏ธ AI-generated; verify before applying. React ๐/๐ to flag quality.</i></sub> ########## hudi-common/src/main/java/org/apache/hudi/common/index/vector/VectorIndexOptions.java: ########## @@ -0,0 +1,243 @@ +/* + * 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.hudi.common.index.vector; + +import java.util.Arrays; +import java.util.Collections; +import java.util.HashSet; +import java.util.LinkedHashMap; +import java.util.Locale; +import java.util.Map; +import java.util.Set; + +/** + * Options accepted by {@code CREATE INDEX ... USING VECTOR}. + * + * <p>The indexed column's Hudi {@code VECTOR(D[, elementType])} schema is authoritative for + * dimension and element type. Index options configure only the acceleration structure. DDL + * implementations must call {@link #validateAndNormalize(Map)} before persisting an index + * definition; individual parsing helpers are intentionally private so aggregate validation cannot + * be bypassed. + */ +public final class VectorIndexOptions { + + public static final String METRIC = "vector.metric"; + public static final String QUANTIZER = "vector.quantizer"; Review Comment: ๐ค nit: these option keys use a `vector.xxx` prefix rather than the Hudi-wide `hoodie.xxx.yyy.zzz` dot-separated lowercase convention. Could they be `hoodie.vector.index.metric`, `hoodie.vector.index.num_clusters`, etc.? Using the standard prefix keeps them consistent with every other `hoodie.*` config key and makes them easier to discover alongside the rest of the table-config surface. <sub><i>โ ๏ธ AI-generated; verify before applying. React ๐/๐ to flag quality.</i></sub> ########## hudi-common/src/test/java/org/apache/hudi/common/index/vector/search/TestDefaultVectorFetchPlanner.java: ########## @@ -0,0 +1,106 @@ +/* + * 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.hudi.common.index.vector.search; + +import org.apache.hudi.common.data.HoodieData; +import org.apache.hudi.common.data.HoodieListData; +import org.apache.hudi.common.model.HoodieRecordGlobalLocation; + +import org.junit.jupiter.api.Test; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Verifies {@link DefaultVectorFetchPlanner} (RFC-109 ยง8): grouping by live file, positional row + * preservation for SERVE, key-fallback ({@code rowPosition = -1}) for STALE, and exclusion of DELETED. + */ +public class TestDefaultVectorFetchPlanner { + + private static VectorCandidate candidate(String key, int cluster, long rowPos, String partition, String fileId) { + VectorPostingLocator loc = new VectorPostingLocator( + 1, cluster, 0, 0L, 0, partition, fileId, "001", rowPos); + return new VectorCandidate(key, cluster, 0, 1.0, loc); + } + + private static ArbitratedVectorCandidate arb(VectorCandidate c, VectorCandidateState state, + String partition, String fileId) { + HoodieRecordGlobalLocation live = state == VectorCandidateState.DELETED + ? null : new HoodieRecordGlobalLocation(partition, "001", fileId); + return new ArbitratedVectorCandidate(c, state, live); + } + + @Test + void groupsByFileAndPreservesPositionsAndFallback() { + List<ArbitratedVectorCandidate> input = new ArrayList<>(); + // File A: two SERVE (positional) + one STALE (key fallback). + input.add(arb(candidate("k1", 0, 10L, "p", "fileA"), VectorCandidateState.SERVE, "p", "fileA")); + input.add(arb(candidate("k2", 0, 20L, "p", "fileA"), VectorCandidateState.SERVE, "p", "fileA")); + input.add(arb(candidate("k3", 0, 30L, "p", "fileA"), VectorCandidateState.STALE, "p", "fileA")); + // File B: one SERVE. + input.add(arb(candidate("k4", 1, 5L, "p", "fileB"), VectorCandidateState.SERVE, "p", "fileB")); + // DELETED: must be dropped. + input.add(arb(candidate("k5", 1, 7L, "p", "fileB"), VectorCandidateState.DELETED, "p", "fileB")); + + HoodieData<ArbitratedVectorCandidate> data = HoodieListData.eager(input); + List<VectorFetchTask> tasks = new DefaultVectorFetchPlanner().plan(data, null, null).collectAsList(); + + Map<String, VectorFetchTask> byFile = new HashMap<>(); + for (VectorFetchTask t : tasks) { + byFile.put(t.getFileId(), t); + } + assertEquals(2, tasks.size(), "expected one task per live file (A, B)"); + assertNull(byFile.get("fileB").getBaseFilePath(), "baseFilePath resolved later by read handle"); + + VectorFetchTask a = byFile.get("fileA"); + assertEquals(3, a.size(), "fileA must contain 3 rows (2 SERVE + 1 STALE), DELETED excluded"); + int positional = 0; + int fallback = 0; + for (VectorRowRequest r : a.getRequests()) { + if (r.getState() == VectorCandidateState.SERVE) { + assertTrue(r.isPositional() && r.getRowPosition() >= 0, "SERVE must keep its row position"); + positional++; + } else if (r.getState() == VectorCandidateState.STALE) { + assertEquals(-1L, r.getRowPosition(), "STALE must drop the row position for key fallback"); + assertTrue(!r.isPositional()); + fallback++; + } + } + assertEquals(2, positional); Review Comment: ๐ค nit: `assertTrue(!r.isPositional())` could be `assertFalse(r.isPositional())` โ same reasoning as the `assertTrue(!...)` pattern a few lines up with `isPositional` checks. <sub><i>โ ๏ธ AI-generated; verify before applying. React ๐/๐ to flag quality.</i></sub> -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
