hudi-agent commented on code in PR #19318: URL: https://github.com/apache/hudi/pull/19318#discussion_r3716246606
########## hudi-common/src/main/java/org/apache/hudi/metadata/VectorPostingPrefixRawKey.java: ########## @@ -0,0 +1,40 @@ +/* + * 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.metadata; + +import lombok.Value; + +/** + * Raw key prefix for vector posting scans. + */ +@Value +public class VectorPostingPrefixRawKey implements RawKey { + + int generationId; + int clusterId; + Integer shardId; + + @Override + public String encode() { + return shardId == null + ? VectorIndexMetadataKey.postingPrefix(generationId, clusterId, 0).substring(0, 9) Review Comment: ๐ค nit: the literal `9` in `.substring(0, 9)` is opaque โ could you extract a named constant (e.g. `POSTING_PREFIX_SCAN_LENGTH`) or add a short comment explaining what those 9 bytes cover (generation + cluster id, no shard)? A future developer changing the key layout will have no signal that this number needs updating. <sub><i>โ ๏ธ AI-generated; verify before applying. React ๐/๐ to flag quality.</i></sub> ########## hudi-common/src/test/java/org/apache/hudi/common/index/vector/TestRaBitQResidualRecall.java: ########## @@ -0,0 +1,156 @@ +/* + * 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; + +import org.junit.jupiter.api.Test; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.Random; + +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * End-to-end recall validation of the corrected math path (RFC-109 ยง3 + ยง3), simulating a + * re-bootstrap: encodes an IVF-residual multibit index with the new {@link RaBitQNeutralFactors} + * (posting-block format 1), scores queries through the rotate-once {@link MetricQueryState} + + * {@link RaBitQEncoder#multibitDotTerm} estimator, and measures recall@10 against brute-force exact + * L2 truth. + * + * <p>This is deliberately non-circular: the pass/fail comparison is against exact L2, not against + * the estimator itself. It confirms the rotate-once query math and the corrected factors do not + * regress candidate-set recall on SIFT-like clustered data. + */ +public class TestRaBitQResidualRecall { + + private static final int DIM = 128; + private static final int NUM_CLUSTERS = 50; + private static final int PER_CLUSTER = 200; + private static final int N = NUM_CLUSTERS * PER_CLUSTER; // 10k base vectors + private static final int NUM_QUERIES = 100; + private static final int K = 10; + private static final int BITS = 4; + private static final long SEED = 7L; + + @Test + public void residualMultibitRecallMeetsFloorOnCorrectedPath() { + Random rng = new Random(SEED); + + // SIFT-like clustered, non-negative, large-norm data. + float[][] centers = new float[NUM_CLUSTERS][DIM]; + for (int c = 0; c < NUM_CLUSTERS; c++) { + for (int d = 0; d < DIM; d++) { + centers[c][d] = 20f + rng.nextFloat() * 160f; + } + } + float[][] data = new float[N][DIM]; + int[] assign = new int[N]; + int idx = 0; + for (int c = 0; c < NUM_CLUSTERS; c++) { + for (int p = 0; p < PER_CLUSTER; p++) { + float[] v = new float[DIM]; + for (int d = 0; d < DIM; d++) { + v[d] = Math.max(0f, centers[c][d] + (float) (rng.nextGaussian() * 15.0)); + } + data[idx] = v; + assign[idx] = c; + idx++; + } + } + + // Re-bootstrap the residual multibit index with the corrected factors. + RaBitQEncoder enc = new RaBitQEncoder(DIM, BITS, 42L, false); + QuantizedVector[] codes = new QuantizedVector[N]; + for (int i = 0; i < N; i++) { + codes[i] = enc.encodeResidual(data[i], centers[assign[i]]); + } + // Confirm the new factor layout is actually being produced. + assertTrue(codes[0].rescaleFactor != null && codes[0].additiveFactor != null, + "residual multibit encoding must produce neutral factors"); + + double recallSum = 0; + for (int qi = 0; qi < NUM_QUERIES; qi++) { + int c = rng.nextInt(NUM_CLUSTERS); + float[] q = new float[DIM]; + for (int d = 0; d < DIM; d++) { + q[d] = Math.max(0f, centers[c][d] + (float) (rng.nextGaussian() * 15.0)); + } + int[] truth = exactTopK(q, data, K); + + // Rotate-once query state; rotate each centroid once (corrected ยง2 path). + MetricQueryState state = + MetricQueryState.create(VectorDistanceMetric.L2, enc::rotateVector, q, false); + MetricQueryState.ClusterQuery[] cqByCluster = new MetricQueryState.ClusterQuery[NUM_CLUSTERS]; + for (int cc = 0; cc < NUM_CLUSTERS; cc++) { + cqByCluster[cc] = state.forRotatedCentroid(state.rotateCentroid(centers[cc])); + } + + float[] approx = new float[N]; + for (int i = 0; i < N; i++) { + MetricQueryState.ClusterQuery cq = cqByCluster[assign[i]]; + float dotTerm = RaBitQEncoder.multibitDotTerm( + cq.rotatedQuery, cq.querySum, codes[i].code, codes[i].extendedCode, DIM, BITS); + double rip = (codes[i].rescaleFactor == null ? 0.0 : codes[i].rescaleFactor) * (double) dotTerm; + float centerRip = codes[i].additiveFactor == null ? 0f : codes[i].additiveFactor; + float residualNorm = codes[i].scalar; + float vectorNorm = codes[i].vectorNorm == null ? Float.NaN : codes[i].vectorNorm; + approx[i] = (float) state.rankingDistance(rip, centerRip, residualNorm, vectorNorm, cq); + } + recallSum += recall(topKByDist(approx, K), truth); + } + double recall = recallSum / NUM_QUERIES; + System.out.printf("[RFC-109] residual multibit recall@%d (B=%d) = %.3f%n", K, BITS, recall); + + // Floor for 4-bit residual multibit on this easy synthetic corpus (all clusters probed). + assertTrue(recall >= 0.85, + "corrected-path recall@" + K + " regressed below floor: " + recall); + } + + private static float l2sq(float[] a, float[] b) { + float s = 0; + for (int i = 0; i < a.length; i++) { + float d = a[i] - b[i]; + s += d * d; + } + return s; + } + + private static int[] exactTopK(float[] q, float[][] data, int k) { + float[] d = new float[data.length]; Review Comment: ๐ค nit: `System.out.printf` in a unit test gets lost or interleaved in CI logs and is inconsistent with how Hudi surfaces diagnostics โ have you considered `LOG.debug(...)` or just dropping this line? <sub><i>โ ๏ธ AI-generated; verify before applying. React ๐/๐ to flag quality.</i></sub> ########## hudi-common/src/main/java/org/apache/hudi/common/index/vector/RaBitQByteLutScorer.java: ########## @@ -0,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.hudi.common.index.vector; + +import java.nio.ByteBuffer; + +/** + * Byte-lookup-table posting scorer (RFC-109 ยง3A, fixes 1 and 2). + * + * <p><b>Exact float-query semantics.</b> This scorer scores the same float rotated (residual) + * query used by {@link RaBitQEncoder#dotPackedBinary} and {@link RaBitQEncoder#multibitDotTerm}. + * It introduces <em>no</em> query quantization (unlike {@link RaBitQPlaneKernel}, which quantizes + * the query to {@code Bq} planes and therefore changes recall math). It only re-associates the + * per-dimension sum {@code dot(q, code)} into per-byte partial sums precomputed once per probed + * cluster, which lets the scan replace: + * <ul> + * <li>the pass-1 per-dimension sign loop ({@code dotSignRow}); and</li> + * <li>the pass-2 per-survivor {@code copyBuffer(signRow)} + {@code repackExtendedLevels} + + * {@code multibitDotTerm} allocation dance</li> + * </ul> + * with table lookups and zero per-survivor allocation. Results match the scalar path up to + * floating-point re-association (byte grouping), never a semantic (quantization) change. + * + * <p><b>LUT layout.</b> {@code lut[bytePos][pattern]} holds the sum of {@code query[bytePos*8 + b]} + * over the set bits {@code b} of {@code pattern}, with padding dimensions ({@code >= dimension}) + * contributing zero so the byte-grouped sum equals the {@code [0, dimension)} scalar sum. A packed + * plane dot is then {@code sum(lut[bytePos][planeByte])} over byte positions โ {@code codeRowBytes} + * lookups instead of {@code dimension} branchy bit tests (an ~8x op reduction at any dimension). + */ +public final class RaBitQByteLutScorer { + + private final double[][] lut; // [codeRowBytes][256] + private final int codeRowBytes; + private final float querySum; + + private RaBitQByteLutScorer(double[][] lut, int codeRowBytes, float querySum) { + this.lut = lut; + this.codeRowBytes = codeRowBytes; + this.querySum = querySum; + } + + /** + * Build the per-cluster LUT from the (residual) rotated query. Called at most once per distinct + * probed cluster; build cost is {@code codeRowBytes * 256} adds, amortized over the cluster's + * posting scan. + * + * @param rotatedQuery the rotated residual query {@code wRot} (length {@code >= dimension}) + * @param querySum {@code sum(wRot)}; folded into the pass-1/pass-2 centering terms + * @param dimension raw dimension scored (padding dims contribute zero) + * @param codeRowBytes long-aligned per-plane row width from the block layout + */ + public static RaBitQByteLutScorer forQuery(float[] rotatedQuery, float querySum, + int dimension, int codeRowBytes) { + if (rotatedQuery == null || dimension <= 0 || rotatedQuery.length < dimension) { + throw new IllegalArgumentException("rotatedQuery must contain every scored dimension"); + } + if (!Float.isFinite(querySum) || codeRowBytes < (dimension + Byte.SIZE - 1) / Byte.SIZE) { + throw new IllegalArgumentException("querySum and codeRowBytes must match the scored query"); + } + for (int i = 0; i < dimension; i++) { + if (!Float.isFinite(rotatedQuery[i])) { + throw new IllegalArgumentException("rotatedQuery contains a non-finite value at dimension " + i); + } + } + double[][] lut = new double[codeRowBytes][256]; + for (int bytePos = 0; bytePos < codeRowBytes; bytePos++) { + int baseDim = bytePos << 3; + double[] bitContribution = new double[8]; + for (int b = 0; b < 8; b++) { + int dim = baseDim + b; + bitContribution[b] = dim < dimension ? rotatedQuery[dim] : 0.0; + } + double[] table = lut[bytePos]; + for (int pattern = 0; pattern < 256; pattern++) { + double sum = 0.0; + // Ascending bit order mirrors dotSignRow's ascending-dimension accumulation. + for (int b = 0; b < 8; b++) { + if ((pattern & (1 << b)) != 0) { + sum += bitContribution[b]; + } + } + table[pattern] = sum; + } + } + return new RaBitQByteLutScorer(lut, codeRowBytes, querySum); + } + + /** + * Raw packed-plane inner product {@code dot(query, plane)} read directly from a plane buffer at an + * absolute byte offset. Equivalent to {@link RaBitQEncoder#dotPackedBinary} for the sign plane. + */ + public double planeDot(ByteBuffer planeBuffer, int offset) { + if (planeBuffer == null || offset < 0 || offset > planeBuffer.limit() - codeRowBytes) { + throw new IllegalArgumentException("Plane row exceeds the supplied buffer"); + } + double sum = 0.0; + for (int bytePos = 0; bytePos < codeRowBytes; bytePos++) { + sum += lut[bytePos][planeBuffer.get(offset + bytePos) & 0xFF]; + } + return sum; + } + + /** + * Pass-1 sign-only score {@code dot(query, sign) - 0.5*sumQuery} (== {@code dotSignRow}). Callers + * that also run pass-2 should keep the {@link #planeDot} sign value and reuse it via + * {@link #pass1FromDot(double)} and {@link #pass2(double, PostingBlockView, ByteBuffer, int, int, int)} + * rather than recomputing the sign dot. + */ + public float pass1(ByteBuffer signBuffer, int signOffset) { + return pass1FromDot(planeDot(signBuffer, signOffset)); + } + + /** Pass-1 score from an already-computed sign-plane dot (see {@link #planeDot}). */ + public float pass1FromDot(double signDot) { + return (float) (signDot + querySum * -0.5f); + } + + /** + * Pass-2 full multibit dot term, scored directly from the sign plane and extended bit-planes + * with no repacking and no per-survivor allocation. Bit-plane decomposition of the centered + * code makes this algebraically identical to {@link RaBitQEncoder#multibitDotTerm}: + * <pre> + * extendedDot = sum_p 2^(exBits-1-p) * dot(query, exPlane_p) + * dotTerm = 2^exBits * signDot + extendedDot + sumQuery * -((2^bits - 1)/2) + * </pre> + * + * @param signDot the raw sign-plane dot from {@link #planeDot} (reuse the pass-1 value) + * @param view the posting block view (for extended-plane offsets) + * @param exBuffer the extended-planes buffer ({@link PostingBlockView#exPlanesBuffer()}) + * @param vectorIndex the vector ordinal within the block + * @param exBits number of extended planes ({@code bits - 1}) + * @param bits total RaBitQ bits + */ + public float pass2(double signDot, PostingBlockView view, ByteBuffer exBuffer, + int vectorIndex, int exBits, int bits) { + if (view == null || exBuffer == null || bits < 1 || bits > 8 || exBits != bits - 1) { + throw new IllegalArgumentException("Posting view and a consistent 1-8 bit width are required"); + } + if (view.codeRowBytes() != codeRowBytes || view.numExPlanes() != exBits) { + throw new IllegalArgumentException("Scorer and posting block layouts do not match"); + } + view.signPlaneOffset(vectorIndex); // validates the vector index before any early return + if (exBits <= 0) { + return (float) (signDot + querySum * -0.5f); + } + double extendedDot = 0.0; + for (int p = 0; p < exBits; p++) { + extendedDot += (double) (1L << (exBits - 1 - p)) * planeDot(exBuffer, view.exPlaneOffset(vectorIndex, p)); Review Comment: ๐ค This weights extended plane `p` by `2^(exBits-1-p)` (plane 0 = MSB), but `RaBitQPlaneKernel.scorePass2` weights plane `b` by `2^b` (plane 0 = LSB). Both docs claim to reconstruct the same centered code as `multibitDotTerm`. Could you confirm the posting-block `exPlanes` writer and the `long[][]` plane writer really emit opposite bit orderings? If a single repack ever feeds both with the same plane index, one path would silently return wrong scores. <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( Review Comment: ๐ค nit: `createVectorIndexManifestRecord` has ~30 parameters, which makes call sites very hard to read and maintain. Have you considered introducing a `VectorIndexManifestSpec` value object (or a builder) to carry these fields? Even grouping the factor-config params (`kappa`, `gMin`, `eps1Max`, `epsNRel`) into the existing `RaBitQFactorConfig` would meaningfully reduce the arity here. <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, Review Comment: ๐ค nit: `fileGroupIds == null ? java.util.Collections.emptyList() : fileGroupIds.stream().collect(Collectors.toList())` โ the fully-qualified `java.util.Collections` is unusual (there are imports already), and when `fileGroupIds` is non-null `new ArrayList<>(fileGroupIds)` is simpler and avoids the stream allocation. <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, Review Comment: ๐ค nit: the 6-param overload accepts `quantizedCodeBytes` and `rabitqBits` but then silently ignores both and delegates to the 3-param version โ could you either use those arguments (pass `rabitqBits` instead of `1`) or remove the overload entirely? A caller who passes a custom `bits` value will be surprised to find their setting discarded. <sub><i>โ ๏ธ AI-generated; verify before applying. React ๐/๐ to flag quality.</i></sub> ########## hudi-common/src/main/java/org/apache/hudi/common/index/vector/PostingBlockBuilder.java: ########## @@ -0,0 +1,346 @@ +/* + * 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 org.apache.hudi.avro.model.HoodieVectorIndexPostingBlock; + +import java.nio.ByteBuffer; +import java.nio.ByteOrder; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +import static org.apache.hudi.common.util.ValidationUtils.checkArgument; + +/** + * Streaming-friendly builder for one immutable vector posting block. + */ +public final class PostingBlockBuilder { + + public static final int BLOCK_FORMAT_VERSION = 1; + public static final int SCALAR_FACTOR_COUNT = 6; + public static final int SCALAR_FACTOR_COUNT_WITH_VECTOR_NORM = 7; + public static final int ROW_LOCATOR_BYTES = 8; + + private final int codeRowBytes; + private final int numExPlanes; + private final boolean includeVectorNorm; + private final List<Row> rows = new ArrayList<>(); + private final Map<String, Integer> fileGroupDict = new LinkedHashMap<>(); + private final Map<String, Integer> instantTimeDict = new LinkedHashMap<>(); + private final List<String> partitionDict = new ArrayList<>(); + + public PostingBlockBuilder(int codeRowBytes, int numExPlanes) { + this(codeRowBytes, numExPlanes, false); + } + + public PostingBlockBuilder(int codeRowBytes, int numExPlanes, boolean includeVectorNorm) { + checkArgument(codeRowBytes > 0 && codeRowBytes % Long.BYTES == 0, + "codeRowBytes must be positive and long-aligned: " + codeRowBytes); + checkArgument(numExPlanes >= 0, "numExPlanes must be non-negative: " + numExPlanes); + this.codeRowBytes = codeRowBytes; + this.numExPlanes = numExPlanes; + this.includeVectorNorm = includeVectorNorm; + } + + public PostingBlockBuilder addRow(String recordKey, + byte[] signPlane, + byte[] exPlanes, + float fAdd1, + float fRescale1, + float err1, + float fAddEx, + float fRescaleEx, + float residualNorm, + String fileGroupId, + String instantTime, + String partitionPath, + long rowPosition) { + return addRow( + recordKey, + signPlane, + exPlanes, + fAdd1, + fRescale1, + err1, + fAddEx, + fRescaleEx, + residualNorm, + null, + fileGroupId, + instantTime, + partitionPath, + rowPosition); + } + + public PostingBlockBuilder addRow(String recordKey, Review Comment: ๐ค nit: the 14-parameter `addRow` overload is hard to call without a named-argument IDE. Have you considered introducing a small `RowSpec` (or builder) to bundle the scalar factors and location fields? Even just grouping the six scalar floats (`fAdd1`, `fRescale1`, `err1`, `fAddEx`, `fRescaleEx`, `residualNorm`) into an existing type like `QuantizedVector` would cut the arity significantly. <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]
