diff --git a/hudi-common/src/main/avro/HoodieMetadata.avsc b/hudi-common/src/main/avro/HoodieMetadata.avsc
index 84dc97dc67206..880c17bd4f9f2 100644
--- a/hudi-common/src/main/avro/HoodieMetadata.avsc
+++ b/hudi-common/src/main/avro/HoodieMetadata.avsc
@@ -554,6 +554,147 @@
}
],
"default" : null
+ },
+ {
+ "name": "VectorIndexMetadata",
+ "doc": "Typed metadata records for the MDT-backed IVF + RaBitQ vector index.",
+ "type": [
+ "null",
+ {
+ "type": "record",
+ "name": "HoodieVectorIndexActiveManifest",
+ "fields": [
+ {"name": "indexVersion", "type": "int", "doc": "Persisted vector-index format version."},
+ {"name": "activeGeneration", "type": ["null", "int"], "default": null,
+ "doc": "Reader-visible generation ordinal, or null before first activation."}
+ ]
+ },
+ {
+ "type": "record",
+ "name": "HoodieVectorIndexManifest",
+ "fields": [
+ {"name": "indexVersion", "type": "int"},
+ {"name": "generationId", "type": "string", "doc": "Creating Hudi instant + optional human tag. The gen ordinal lives in the key."},
+ {"name": "state", "type": "string", "doc": "BUILDING | ACTIVE | RETIRED. Readers serve only the generation referenced by the singleton active manifest."},
+ {"name": "dim", "type": "int"},
+ {"name": "dimPadded", "type": "int", "doc": "nextPow2(dim) for hadamard quantizers, else dim. Authoritative for codeRowBytes."},
+ {"name": "codeRowBytes", "type": "int", "doc": "ceil(dimPadded/64)*8. Long-aligned per-plane row width."},
+ {"name": "bitsTotal", "type": "int", "doc": "B = 1 + numExPlanes."},
+ {"name": "numExPlanes", "type": "int"},
+ {"name": "numClusters", "type": "int"},
+ {"name": "shardCount", "type": "int", "doc": "Initial per-cluster posting shard geometry frozen for this generation."},
+ {"name": "fileGroupCount", "type": "int", "doc": "MDT file-group geometry frozen for this generation; readers and writers never re-derive it."},
+ {"name": "metric", "type": "string", "doc": "L2 | DOT | COSINE."},
+ {"name": "assumeNormalized", "type": "boolean"},
+ {"name": "residualEncoding", "type": "boolean", "default": false, "doc": "True when posting codes encode x - centroid and readers must use residual scoring."},
+ {"name": "vectorColumn", "type": "string", "doc": "Authoritative base-table vector column used for bootstrap and exact rerank."},
+ {"name": "targetBlockBytes", "type": "int", "doc": "Config input, e.g. 524288."},
+ {"name": "vectorsPerBlock", "type": "int", "doc": "Resolved N. Frozen per generation; readers never re-derive."},
+ {"name": "blockFormatVersion", "type": "int", "doc": "Expected posting-block layout version for this generation."},
+ {"name": "factorVersion", "type": "int", "doc": "RaBitQ scalar-factor semantics version. Readers reject unsupported values."},
+ {"name": "kappa", "type": "double", "doc": "Generation-scoped pass-1 error scale."},
+ {"name": "gMin", "type": "double", "doc": "Generation-scoped normalized alignment floor."},
+ {"name": "eps1Max", "type": "double", "doc": "Maximum permitted relative pass-1 error."},
+ {"name": "epsNRel", "type": "double", "doc": "Relative residual-norm floor."},
+ {"name": "centroidChunkCount", "type": "int", "doc": "Expected number of size-bounded centroid chunks."},
+ {"name": "centroidChecksum", "type": ["null", "string"], "default": null,
+ "doc": "Checksum of the canonical centroid payload used to reject incomplete or mismatched builds."},
+ {"name": "splitLimit", "type": "int"},
+ {"name": "mergeFloor", "type": "int"},
+ {"name": "bootstrapInstant", "type": ["null", "string"], "default": null,
+ "doc": "Source data-write instant included by the generation bootstrap baseline."},
+ {"name": "verifiedFrontier", "type": ["null", "string"], "default": null,
+ "doc": "Latest source data-write instant with verified contiguous marker coverage from the bootstrap baseline."},
+ {"name": "createdTs", "type": "long"}
+ ]
+ },
+ {
+ "type": "record",
+ "name": "HoodieVectorIndexQuantizer",
+ "fields": [
+ {"name": "quantizerType", "type": "string", "doc": "rb-rotation-explicit | rb-rotation-hadamard."},
+ {"name": "randomSeed", "type": "long"},
+ {"name": "rotationBytes", "type": ["null", "bytes"], "default": null,
+ "doc": "Explicit path: rows of the D x D float32 LE orthogonal matrix. Null for hadamard."}
+ ]
+ },
+ {
+ "type": "record",
+ "name": "HoodieVectorIndexCentroids",
+ "fields": [
+ {"name": "clusterIds", "type": "bytes", "doc": "k x u32 LE cluster ids in this chunk, parallel to centroidBytes rows."},
+ {"name": "centroidBytes", "type": "bytes", "doc": "k x dimPadded float32 LE, row-major, in rotated space."},
+ {"name": "clusterRadii", "type": "bytes", "doc": "k x float32 LE max residual norm per cluster."}
+ ]
+ },
+ {
+ "type": "record",
+ "name": "HoodieVectorIndexPostingBlock",
+ "fields": [
+ {"name": "blockFormatVersion", "type": "int"},
+ {"name": "numVectors", "type": "int"},
+ {"name": "codeRowBytes", "type": "int", "doc": "Duplicated from manifest for self-description; MUST match."},
+ {"name": "signPlane", "type": "bytes", "doc": "S1: N x codeRowBytes, row-major. Bit plane B-1 (MSB)."},
+ {"name": "exPlanes", "type": "bytes", "doc": "S2: N x numExPlanes x codeRowBytes. Vector-major; planes MSB-first."},
+ {"name": "scalarFactors", "type": "bytes", "doc": "S3: fAdd1 | fRescale1 | err1 | fAddEx | fRescaleEx | residualNorm | optional vectorNorm, float32 LE SoA."},
+ {"name": "rowLocators", "type": "bytes", "doc": "S4: N x 8 B LE structs: fgDictIdx u16 | instantDictIdx u16 | rowPosition u32."},
+ {"name": "fileGroupDict", "type": {"type": "array", "items": "string"}},
+ {"name": "instantTimeDict", "type": {"type": "array", "items": "string"}},
+ {"name": "partitionDict", "type": {"type": "array", "items": "string"}, "doc": "Parallel indexing with fileGroupDict."},
+ {"name": "recordKeyOffsets", "type": "bytes", "doc": "S6a: (N+1) x u32 LE offsets into recordKeyBytes."},
+ {"name": "recordKeyBytes", "type": "bytes", "doc": "S6b: concatenated UTF-8 keys. Must remain final field."}
+ ]
+ },
+ {
+ "type": "record",
+ "name": "HoodieVectorIndexPostingDelta",
+ "fields": [
+ {"name": "recordKey", "type": "string"},
+ {"name": "binaryCode", "type": "bytes", "doc": "MSB row followed by ex rows, identical row encoding and plane order as blocks."},
+ {"name": "fAdd1", "type": "float"},
+ {"name": "fRescale1", "type": "float"},
+ {"name": "err1", "type": "float"},
+ {"name": "fAddEx", "type": "float"},
+ {"name": "fRescaleEx", "type": "float"},
+ {"name": "residualNorm", "type": "float"},
+ {"name": "vectorNorm", "type": ["null", "float"], "default": null},
+ {"name": "fileGroupId", "type": "string"},
+ {"name": "partitionPath", "type": "string"},
+ {"name": "baseInstantTime", "type": "string"},
+ {"name": "rowPosition", "type": "long"}
+ ]
+ },
+ {
+ "type": "record",
+ "name": "HoodieVectorIndexClusterStats",
+ "fields": [
+ {"name": "routingVersion", "type": "int", "doc": "Cache invalidation version for atomically published cluster routing changes."},
+ {"name": "shardCount", "type": "int"},
+ {"name": "fileGroupIds", "type": {"type": "array", "items": "string"}, "doc": "Candidate base-table file groups for this cluster."},
+ {"name": "liveCount", "type": "long"},
+ {"name": "deltaCount", "type": "long"},
+ {"name": "tombstoneCount", "type": "long"},
+ {"name": "lastRebalanceInstant", "type": ["null", "string"], "default": null},
+ {"name": "lastUpdatedTs", "type": "long"}
+ ]
+ },
+ {
+ "type": "record",
+ "name": "HoodieVectorIndexSourceInstantMarker",
+ "fields": [
+ {"name": "dataInstant", "type": "string", "doc": "Completed source data-write instant incorporated by this generation."}
+ ]
+ },
+ {
+ "type": "record",
+ "name": "HoodieVectorIndexTombstone",
+ "fields": [
+ {"name": "deleteInstant", "type": "string"},
+ {"name": "deleteReason", "type": "string", "doc": "SPLIT | MERGE | REBALANCE | GC | MANUAL."}
+ ]
+ }
+ ],
+ "default" : null
}
]
}
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/index/vector/MetricQueryState.java b/hudi-common/src/main/java/org/apache/hudi/common/index/vector/MetricQueryState.java
new file mode 100644
index 0000000000000..cb2586e56aac2
--- /dev/null
+++ b/hudi-common/src/main/java/org/apache/hudi/common/index/vector/MetricQueryState.java
@@ -0,0 +1,294 @@
+/*
+ * 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 java.io.Serializable;
+
+/**
+ * Per-metric query state using residual-query estimation in rotated space (RFC-109 §3).
+ *
+ *
The MDT stores metric-neutral residual factors. The padded query is rotated exactly
+ * once in the constructor to {@code qRot = P(qPad)}. For each probed IVF cluster the scorer
+ * consumes an already-rotated centroid {@code cRot} and forms the rotated residual query
+ * {@code wRot = qRot - cRot} by subtraction only — it never rotates {@code q - c} per cluster.
+ * Because rotation is a linear isometry, {@code P(q - c) = P(q) - P(c)} and
+ * the inner product {@code <q, c> = <qRot, cRot>}, so exact terms are computed directly
+ * in rotated space.
+ */
+public abstract class MetricQueryState implements Serializable {
+
+ private static final long serialVersionUID = 2L;
+ protected static final double EPS_NORM = 1.0e-9;
+
+ /** Rotated padded query {@code qRot = P(qPad)}; computed once. */
+ protected final float[] rotatedQuery;
+ /** ||q|| (rotation-invariant). */
+ protected final double queryNorm;
+ /** The shared query rotation {@code P}; retained only to rotate centroids once (cached). */
+ private final QueryRotation rotation;
+
+ protected MetricQueryState(QueryRotation rotation, float[] paddedQuery) {
+ this.rotation = rotation;
+ this.rotatedQuery = validateRotationResult(
+ rotation.apply(paddedQuery.clone()), paddedQuery.length, "query");
+ this.queryNorm = Math.sqrt(sqNorm(rotatedQuery));
+ }
+
+ /**
+ * Apply the shared rotation {@code P} to a raw padded centroid, yielding {@code cRot = P(c)}.
+ * Callers should invoke this at most once per distinct probed cluster (cached) — never per
+ * {@code q - c} — and pass the result to {@link #forRotatedCentroid(float[])}. Prefer feeding
+ * pre-rotated centroids from the index cache (RFC-109 §3) where available.
+ */
+ public final float[] rotateCentroid(float[] paddedCentroid) {
+ if (paddedCentroid == null) {
+ throw new IllegalArgumentException("paddedCentroid must not be null");
+ }
+ return validateRotationResult(
+ rotation.apply(paddedCentroid.clone()), paddedCentroid.length, "centroid");
+ }
+
+ /** Whether this metric requires the 7th (VECTOR_NORM) scalar array in blocks. */
+ public boolean requiresVectorNorm() {
+ return false;
+ }
+
+ /** Reject degenerate queries per the spec's normative rules. */
+ public void validate() {
+ // L2 permits any query (s ~ 0 handled per cluster); overridden where zero-q is invalid.
+ }
+
+ /** The rotated padded query {@code qRot}; exposed so callers may compute error bounds. */
+ public final float[] rotatedQuery() {
+ return rotatedQuery.clone();
+ }
+
+ /**
+ * Build per-cluster query state from an already-rotated centroid {@code cRot = P(c)}. The float
+ * kernel consumes {@code wRot = qRot - cRot} and its sum. No rotation is performed here — callers
+ * must rotate (and preferably cache) centroids once, never per {@code q - c}.
+ */
+ public final ClusterQuery forRotatedCentroid(float[] rotatedCentroid) {
+ int n = rotatedQuery.length;
+ if (rotatedCentroid.length != n) {
+ throw new IllegalArgumentException(
+ "Rotated centroid dimension mismatch: expected " + n + ", got " + rotatedCentroid.length);
+ }
+ float[] residualQuery = new float[n];
+ for (int i = 0; i < n; i++) {
+ residualQuery[i] = rotatedQuery[i] - rotatedCentroid[i];
+ }
+ double wNormSq = sqNorm(residualQuery);
+ return new ClusterQuery(
+ residualQuery,
+ sum(residualQuery),
+ exactTerms(rotatedCentroid, wNormSq),
+ wNormSq);
+ }
+
+ /** Exact per-metric terms; {@code rotatedCentroid} is {@code P(c)}, {@code wNormSq = ||q - c||^2}. */
+ protected abstract ExactTerms exactTerms(float[] rotatedCentroid, double wNormSq);
+
+ /**
+ * Metric-specific ranking value from the residual inner-product estimate {@code ripW}.
+ * L2 uses squared Euclidean distance to avoid a square root in the posting scan; cosine and
+ * dot-product use the same units as {@link VectorDistanceMetric}.
+ */
+ public abstract double rankingDistance(double ripW, float centerRip, float residualNorm,
+ float vectorNorm, ClusterQuery cq);
+
+ /** Optimistic ranking value for pass-1 pruning, using the residual-query norm. */
+ public final double optimisticRankingDistance(double ripW, float err1, float centerRip,
+ float residualNorm, float vectorNorm,
+ ClusterQuery cq) {
+ double errAbs = (double) err1 * Math.sqrt(cq.wNormSq);
+ return rankingDistance(ripW + errAbs, centerRip, residualNorm, vectorNorm, cq);
+ }
+
+ public static final class ClusterQuery implements Serializable {
+ private static final long serialVersionUID = 2L;
+ public final float[] rotatedQuery;
+ public final float querySum;
+ public final ExactTerms terms;
+ public final double wNormSq;
+ public final boolean queryAtCentroid;
+
+ ClusterQuery(float[] rotatedQuery, float querySum, ExactTerms terms, double wNormSq) {
+ this.rotatedQuery = rotatedQuery;
+ this.querySum = querySum;
+ this.terms = terms;
+ this.wNormSq = wNormSq;
+ this.queryAtCentroid = wNormSq < EPS_NORM * EPS_NORM;
+ }
+ }
+
+ public static final class ExactTerms implements Serializable {
+ private static final long serialVersionUID = 2L;
+ public final double qDotC;
+
+ ExactTerms(double qDotC) {
+ this.qDotC = qDotC;
+ }
+ }
+
+ // --------------------------------------------------------------------------------------
+
+ public static MetricQueryState create(VectorDistanceMetric metric, QueryRotation rotation,
+ float[] paddedQuery,
+ boolean assumeNormalized) {
+ if (metric == null || rotation == null || paddedQuery == null) {
+ throw new IllegalArgumentException("metric, rotation, and paddedQuery must not be null");
+ }
+ for (int i = 0; i < paddedQuery.length; i++) {
+ if (!Float.isFinite(paddedQuery[i])) {
+ throw new IllegalArgumentException("paddedQuery contains a non-finite value at dimension " + i);
+ }
+ }
+ MetricQueryState state;
+ switch (metric) {
+ case L2:
+ state = new L2QueryState(rotation, paddedQuery);
+ break;
+ case DOT_PRODUCT:
+ state = new DotQueryState(rotation, paddedQuery);
+ break;
+ case COSINE:
+ state = new CosineQueryState(rotation, paddedQuery, assumeNormalized);
+ break;
+ default:
+ throw new IllegalArgumentException("Unsupported metric: " + metric);
+ }
+ state.validate();
+ return state;
+ }
+
+ // --------------------------------------------------------------------------------------
+
+ /** d2 = ||q - c||^2 + n^2 - 2 * ripW. */
+ public static final class L2QueryState extends MetricQueryState {
+ L2QueryState(QueryRotation rotation, float[] paddedQuery) {
+ super(rotation, paddedQuery);
+ }
+
+ @Override
+ protected ExactTerms exactTerms(float[] rotatedCentroid, double wNormSq) {
+ return new ExactTerms(0.0);
+ }
+
+ @Override
+ public double rankingDistance(double ripW, float centerRip, float residualNorm,
+ float vectorNorm, ClusterQuery cq) {
+ if (cq.queryAtCentroid) {
+ return (double) residualNorm * residualNorm; // exact; scanner may skip the kernel
+ }
+ return cq.wNormSq + (double) residualNorm * residualNorm - 2.0 * ripW;
+ }
+ }
+
+ /** Dot-product distance finish. Zero query rejected. */
+ public static final class DotQueryState extends MetricQueryState {
+ DotQueryState(QueryRotation rotation, float[] paddedQuery) {
+ super(rotation, paddedQuery);
+ }
+
+ @Override
+ public void validate() {
+ if (queryNorm < 1.0e-9f) {
+ throw new IllegalArgumentException("DOT_PRODUCT query must be non-zero.");
+ }
+ }
+
+ @Override
+ protected ExactTerms exactTerms(float[] rotatedCentroid, double wNormSq) {
+ return new ExactTerms(dot(rotatedQuery, rotatedCentroid));
+ }
+
+ @Override
+ public double rankingDistance(double ripW, float centerRip, float residualNorm,
+ float vectorNorm, ClusterQuery cq) {
+ return -(cq.terms.qDotC + centerRip + ripW);
+ }
+ }
+
+ /** Cosine distance finish. Zero query rejected. */
+ public static final class CosineQueryState extends MetricQueryState {
+ private final boolean assumeNormalized;
+
+ CosineQueryState(QueryRotation rotation, float[] paddedQuery, boolean assumeNormalized) {
+ super(rotation, paddedQuery);
+ this.assumeNormalized = assumeNormalized;
+ }
+
+ @Override
+ public void validate() {
+ if (queryNorm < 1.0e-9f) {
+ throw new IllegalArgumentException("COSINE query must be non-zero.");
+ }
+ }
+
+ @Override
+ public boolean requiresVectorNorm() {
+ return !assumeNormalized;
+ }
+
+ @Override
+ protected ExactTerms exactTerms(float[] rotatedCentroid, double wNormSq) {
+ return new ExactTerms(dot(rotatedQuery, rotatedCentroid));
+ }
+
+ @Override
+ public double rankingDistance(double ripW, float centerRip, float residualNorm,
+ float vectorNorm, ClusterQuery cq) {
+ double xNorm = assumeNormalized ? 1.0 : vectorNorm;
+ if (xNorm < EPS_NORM) {
+ return 1.0; // zero vector: orthogonal by convention
+ }
+ return 1.0 - (cq.terms.qDotC + centerRip + ripW) / (queryNorm * xNorm);
+ }
+ }
+
+ // --------------------------------------------------------------------------------------
+
+ private static float[] validateRotationResult(float[] result, int expectedLength, String name) {
+ if (result == null || result.length != expectedLength) {
+ throw new IllegalArgumentException("Rotation returned an invalid " + name + " dimension");
+ }
+ for (int i = 0; i < result.length; i++) {
+ if (!Float.isFinite(result[i])) {
+ throw new IllegalArgumentException("Rotation returned a non-finite " + name + " value at " + i);
+ }
+ }
+ return result.clone();
+ }
+
+ private static double sqNorm(float[] v) {
+ double s = 0.0;
+ for (float x : v) {
+ s += (double) x * x;
+ }
+ return s;
+ }
+
+ private static double dot(float[] a, float[] b) {
+ double s = 0.0;
+ for (int i = 0; i < a.length; i++) {
+ s += (double) a[i] * b[i];
+ }
+ return s;
+ }
+
+ private static float sum(float[] values) {
+ float sum = 0.0f;
+ for (float value : values) {
+ sum += value;
+ }
+ return sum;
+ }
+
+ public interface QueryRotation extends Serializable {
+ float[] apply(float[] vector);
+ }
+}
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/index/vector/PostingBlockBuilder.java b/hudi-common/src/main/java/org/apache/hudi/common/index/vector/PostingBlockBuilder.java
new file mode 100644
index 0000000000000..16d913a2a1834
--- /dev/null
+++ b/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 rows = new ArrayList<>();
+ private final Map fileGroupDict = new LinkedHashMap<>();
+ private final Map instantTimeDict = new LinkedHashMap<>();
+ private final List 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,
+ byte[] signPlane,
+ byte[] exPlanes,
+ float fAdd1,
+ float fRescale1,
+ float err1,
+ float fAddEx,
+ float fRescaleEx,
+ float residualNorm,
+ Float vectorNorm,
+ String fileGroupId,
+ String instantTime,
+ String partitionPath,
+ long rowPosition) {
+ checkArgument(recordKey != null, "recordKey must not be null");
+ checkArgument(signPlane != null && signPlane.length == codeRowBytes,
+ "signPlane must be exactly codeRowBytes");
+ checkArgument(exPlanes != null && exPlanes.length == numExPlanes * codeRowBytes,
+ "exPlanes must be numExPlanes * codeRowBytes");
+ checkArgument(rowPosition >= 0 && rowPosition <= 0xFFFFFFFFL,
+ "rowPosition must fit in unsigned int: " + rowPosition);
+ checkArgument(includeVectorNorm == (vectorNorm != null),
+ "vectorNorm presence must match block scalar layout");
+
+ int fileGroupIdx = fileGroupDictionaryIndex(fileGroupId, partitionPath);
+ int instantIdx = dictionaryIndex(instantTimeDict, instantTime);
+ rows.add(new Row(
+ recordKey,
+ signPlane.clone(),
+ exPlanes.clone(),
+ fAdd1,
+ fRescale1,
+ err1,
+ fAddEx,
+ fRescaleEx,
+ residualNorm,
+ vectorNorm == null ? 0.0f : vectorNorm,
+ fileGroupIdx,
+ instantIdx,
+ rowPosition));
+ return this;
+ }
+
+ public HoodieVectorIndexPostingBlock build() {
+ int numVectors = rows.size();
+ ByteBuffer signPlane = allocateLittleEndian(numVectors * codeRowBytes);
+ ByteBuffer exPlanes = allocateLittleEndian(numVectors * numExPlanes * codeRowBytes);
+ ByteBuffer scalarFactors = allocateLittleEndian(numVectors * scalarFactorCount() * Float.BYTES);
+ ByteBuffer rowLocators = allocateLittleEndian(numVectors * ROW_LOCATOR_BYTES);
+ ByteBuffer recordKeyOffsets = allocateLittleEndian((numVectors + 1) * Integer.BYTES);
+ ByteBuffer recordKeyBytes = allocateLittleEndian(totalRecordKeyBytes());
+
+ for (Row row : rows) {
+ signPlane.put(row.signPlane);
+ exPlanes.put(row.exPlanes);
+ }
+
+ writeScalarArray(scalarFactors, ScalarFactor.F_ADD_1);
+ writeScalarArray(scalarFactors, ScalarFactor.F_RESCALE_1);
+ writeScalarArray(scalarFactors, ScalarFactor.ERR_1);
+ writeScalarArray(scalarFactors, ScalarFactor.F_ADD_EX);
+ writeScalarArray(scalarFactors, ScalarFactor.F_RESCALE_EX);
+ writeScalarArray(scalarFactors, ScalarFactor.RESIDUAL_NORM);
+ if (includeVectorNorm) {
+ writeScalarArray(scalarFactors, ScalarFactor.VECTOR_NORM);
+ }
+
+ int currentKeyOffset = 0;
+ recordKeyOffsets.putInt(currentKeyOffset);
+ for (Row row : rows) {
+ rowLocators.putShort((short) row.fileGroupIdx);
+ rowLocators.putShort((short) row.instantIdx);
+ rowLocators.putInt((int) row.rowPosition);
+
+ byte[] keyBytes = row.recordKey.getBytes(StandardCharsets.UTF_8);
+ recordKeyBytes.put(keyBytes);
+ currentKeyOffset += keyBytes.length;
+ recordKeyOffsets.putInt(currentKeyOffset);
+ }
+
+ return new HoodieVectorIndexPostingBlock(
+ BLOCK_FORMAT_VERSION,
+ numVectors,
+ codeRowBytes,
+ flip(signPlane),
+ flip(exPlanes),
+ flip(scalarFactors),
+ flip(rowLocators),
+ new ArrayList<>(fileGroupDict.keySet()),
+ new ArrayList<>(instantTimeDict.keySet()),
+ new ArrayList<>(partitionDict),
+ flip(recordKeyOffsets),
+ flip(recordKeyBytes));
+ }
+
+ public int rowCount() {
+ return rows.size();
+ }
+
+ public void reset() {
+ rows.clear();
+ fileGroupDict.clear();
+ instantTimeDict.clear();
+ partitionDict.clear();
+ }
+
+ public static int deriveVectorsPerBlock(int targetBlockBytes, int dimPadded, int bitsTotal, int avgKeyLen) {
+ return deriveVectorsPerBlock(targetBlockBytes, dimPadded, bitsTotal, avgKeyLen, false);
+ }
+
+ public static int deriveVectorsPerBlock(int targetBlockBytes, int dimPadded, int bitsTotal, int avgKeyLen, boolean includeVectorNorm) {
+ checkArgument(targetBlockBytes > 0, "targetBlockBytes must be positive");
+ checkArgument(dimPadded > 0, "dimPadded must be positive");
+ checkArgument(bitsTotal > 0, "bitsTotal must be positive");
+ int codeRowBytes = ((dimPadded + 63) / 64) * Long.BYTES;
+ int scalarFactorCount = includeVectorNorm ? SCALAR_FACTOR_COUNT_WITH_VECTOR_NORM : SCALAR_FACTOR_COUNT;
+ int perVectorBytes = bitsTotal * codeRowBytes + scalarFactorCount * Float.BYTES
+ + ROW_LOCATOR_BYTES + Math.max(0, avgKeyLen) + Integer.BYTES;
+ int raw = targetBlockBytes / perVectorBytes;
+ int prevPowerOfTwo = Integer.highestOneBit(Math.max(1, raw));
+ return Math.max(256, Math.min(4096, prevPowerOfTwo));
+ }
+
+ private void writeScalarArray(ByteBuffer scalarFactors, ScalarFactor factor) {
+ for (Row row : rows) {
+ scalarFactors.putFloat(row.scalar(factor));
+ }
+ }
+
+ private int scalarFactorCount() {
+ return includeVectorNorm ? SCALAR_FACTOR_COUNT_WITH_VECTOR_NORM : SCALAR_FACTOR_COUNT;
+ }
+
+ private int totalRecordKeyBytes() {
+ int total = 0;
+ for (Row row : rows) {
+ total += row.recordKey.getBytes(StandardCharsets.UTF_8).length;
+ }
+ return total;
+ }
+
+ private static int dictionaryIndex(Map dictionary, String value) {
+ checkArgument(value != null, "dictionary value must not be null");
+ Integer existing = dictionary.get(value);
+ if (existing != null) {
+ return existing;
+ }
+ int index = dictionary.size();
+ checkArgument(index <= 0xFFFF, "block-local dictionary cannot exceed unsigned short cardinality");
+ dictionary.put(value, index);
+ return index;
+ }
+
+ private int fileGroupDictionaryIndex(String fileGroupId, String partitionPath) {
+ checkArgument(fileGroupId != null, "fileGroupId must not be null");
+ checkArgument(partitionPath != null, "partitionPath must not be null");
+ Integer existing = fileGroupDict.get(fileGroupId);
+ if (existing != null) {
+ checkArgument(partitionDict.get(existing).equals(partitionPath),
+ "fileGroupId cannot map to multiple partition paths in one block: " + fileGroupId);
+ return existing;
+ }
+ int index = fileGroupDict.size();
+ checkArgument(index <= 0xFFFF, "block-local dictionary cannot exceed unsigned short cardinality");
+ fileGroupDict.put(fileGroupId, index);
+ partitionDict.add(partitionPath);
+ return index;
+ }
+
+ private static ByteBuffer allocateLittleEndian(int size) {
+ return ByteBuffer.allocate(size).order(ByteOrder.LITTLE_ENDIAN);
+ }
+
+ private static ByteBuffer flip(ByteBuffer buffer) {
+ buffer.flip();
+ return buffer;
+ }
+
+ private enum ScalarFactor {
+ F_ADD_1,
+ F_RESCALE_1,
+ ERR_1,
+ F_ADD_EX,
+ F_RESCALE_EX,
+ RESIDUAL_NORM,
+ VECTOR_NORM
+ }
+
+ private static final class Row {
+ private final String recordKey;
+ private final byte[] signPlane;
+ private final byte[] exPlanes;
+ private final float fAdd1;
+ private final float fRescale1;
+ private final float err1;
+ private final float fAddEx;
+ private final float fRescaleEx;
+ private final float residualNorm;
+ private final float vectorNorm;
+ private final int fileGroupIdx;
+ private final int instantIdx;
+ private final long rowPosition;
+
+ private Row(String recordKey,
+ byte[] signPlane,
+ byte[] exPlanes,
+ float fAdd1,
+ float fRescale1,
+ float err1,
+ float fAddEx,
+ float fRescaleEx,
+ float residualNorm,
+ float vectorNorm,
+ int fileGroupIdx,
+ int instantIdx,
+ long rowPosition) {
+ this.recordKey = recordKey;
+ this.signPlane = signPlane;
+ this.exPlanes = exPlanes;
+ this.fAdd1 = fAdd1;
+ this.fRescale1 = fRescale1;
+ this.err1 = err1;
+ this.fAddEx = fAddEx;
+ this.fRescaleEx = fRescaleEx;
+ this.residualNorm = residualNorm;
+ this.vectorNorm = vectorNorm;
+ this.fileGroupIdx = fileGroupIdx;
+ this.instantIdx = instantIdx;
+ this.rowPosition = rowPosition;
+ }
+
+ private float scalar(ScalarFactor factor) {
+ switch (factor) {
+ case F_ADD_1:
+ return fAdd1;
+ case F_RESCALE_1:
+ return fRescale1;
+ case ERR_1:
+ return err1;
+ case F_ADD_EX:
+ return fAddEx;
+ case F_RESCALE_EX:
+ return fRescaleEx;
+ case RESIDUAL_NORM:
+ return residualNorm;
+ case VECTOR_NORM:
+ return vectorNorm;
+ default:
+ throw new IllegalArgumentException("Unknown scalar factor: " + factor);
+ }
+ }
+ }
+}
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/index/vector/PostingBlockView.java b/hudi-common/src/main/java/org/apache/hudi/common/index/vector/PostingBlockView.java
new file mode 100644
index 0000000000000..069a96acc458d
--- /dev/null
+++ b/hudi-common/src/main/java/org/apache/hudi/common/index/vector/PostingBlockView.java
@@ -0,0 +1,277 @@
+/*
+ * 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.List;
+
+import static org.apache.hudi.common.util.ValidationUtils.checkArgument;
+
+/**
+ * Zero-copy section view over a vector posting block payload.
+ */
+public final class PostingBlockView {
+
+ private final HoodieVectorIndexPostingBlock block;
+ private final int numVectors;
+ private final int codeRowBytes;
+ private final int numExPlanes;
+ private final ByteBuffer signPlane;
+ private final ByteBuffer exPlanes;
+ private final ByteBuffer scalarFactors;
+ private final int scalarFactorCount;
+ private final ByteBuffer rowLocators;
+ private final ByteBuffer recordKeyOffsets;
+ private final ByteBuffer recordKeyBytes;
+
+ public PostingBlockView(HoodieVectorIndexPostingBlock block) {
+ checkArgument(block != null, "posting block must not be null");
+ this.block = block;
+ this.numVectors = block.getNumVectors();
+ this.codeRowBytes = block.getCodeRowBytes();
+ this.signPlane = littleEndian(block.getSignPlane());
+ this.exPlanes = littleEndian(block.getExPlanes());
+ this.scalarFactors = littleEndian(block.getScalarFactors());
+ this.rowLocators = littleEndian(block.getRowLocators());
+ this.recordKeyOffsets = littleEndian(block.getRecordKeyOffsets());
+ this.recordKeyBytes = littleEndian(block.getRecordKeyBytes());
+
+ checkArgument(block.getBlockFormatVersion() == PostingBlockBuilder.BLOCK_FORMAT_VERSION,
+ "Unsupported posting block format version: " + block.getBlockFormatVersion());
+ checkArgument(numVectors >= 0, "numVectors must be non-negative");
+ checkArgument(codeRowBytes > 0 && codeRowBytes % Long.BYTES == 0,
+ "codeRowBytes must be positive and long-aligned");
+ checkArgument(signPlane.remaining() == numVectors * codeRowBytes,
+ "signPlane length does not match numVectors * codeRowBytes");
+ checkArgument(exPlanes.remaining() % Math.max(1, numVectors * codeRowBytes) == 0,
+ "exPlanes length is not row aligned");
+ this.numExPlanes = numVectors == 0 ? 0 : exPlanes.remaining() / (numVectors * codeRowBytes);
+ int scalarArrayBytes = numVectors * Float.BYTES;
+ checkArgument(numVectors == 0 || scalarFactors.remaining() % scalarArrayBytes == 0,
+ "scalarFactors length must be a whole number of float arrays");
+ this.scalarFactorCount = numVectors == 0
+ ? PostingBlockBuilder.SCALAR_FACTOR_COUNT
+ : scalarFactors.remaining() / scalarArrayBytes;
+ checkArgument(scalarFactorCount == PostingBlockBuilder.SCALAR_FACTOR_COUNT
+ || scalarFactorCount == PostingBlockBuilder.SCALAR_FACTOR_COUNT_WITH_VECTOR_NORM,
+ "scalarFactors length must contain six or seven float arrays");
+ checkArgument(rowLocators.remaining() == numVectors * PostingBlockBuilder.ROW_LOCATOR_BYTES,
+ "rowLocators length must use 8-byte locator stride");
+ checkArgument(recordKeyOffsets.remaining() == (numVectors + 1) * Integer.BYTES,
+ "recordKeyOffsets length must be (numVectors + 1) * 4");
+ }
+
+ public int numVectors() {
+ return numVectors;
+ }
+
+ public int codeRowBytes() {
+ return codeRowBytes;
+ }
+
+ public int numExPlanes() {
+ return numExPlanes;
+ }
+
+ public boolean hasVectorNorm() {
+ return scalarFactorCount == PostingBlockBuilder.SCALAR_FACTOR_COUNT_WITH_VECTOR_NORM;
+ }
+
+ public int signPlaneOffset(int vectorIndex) {
+ checkVectorIndex(vectorIndex);
+ return vectorIndex * codeRowBytes;
+ }
+
+ public int exPlaneOffset(int vectorIndex, int exPlaneIndex) {
+ checkVectorIndex(vectorIndex);
+ checkArgument(exPlaneIndex >= 0 && exPlaneIndex < numExPlanes,
+ "exPlaneIndex out of bounds: " + exPlaneIndex);
+ return vectorIndex * numExPlanes * codeRowBytes + exPlaneIndex * codeRowBytes;
+ }
+
+ public ByteBuffer signPlaneRow(int vectorIndex) {
+ return slice(signPlane, signPlaneOffset(vectorIndex), codeRowBytes);
+ }
+
+ /**
+ * Returns a duplicate of the whole sign-plane section. Hot scan loops should
+ * use absolute {@code get(offset)} reads from this buffer instead of slicing
+ * per vector.
+ */
+ public ByteBuffer signPlaneBuffer() {
+ return signPlane.duplicate().order(ByteOrder.LITTLE_ENDIAN);
+ }
+
+ /**
+ * Returns a duplicate of the whole extended-plane section for survivor-only
+ * repacking.
+ */
+ public ByteBuffer exPlanesBuffer() {
+ return exPlanes.duplicate().order(ByteOrder.LITTLE_ENDIAN);
+ }
+
+ public ByteBuffer exPlaneRow(int vectorIndex, int exPlaneIndex) {
+ return slice(exPlanes, exPlaneOffset(vectorIndex, exPlaneIndex), codeRowBytes);
+ }
+
+ public float scalarFactor(ScalarFactor factor, int vectorIndex) {
+ checkVectorIndex(vectorIndex);
+ checkScalarFactor(factor);
+ int offset = scalarFactorOffset(factor, vectorIndex);
+ return scalarFactors.getFloat(offset);
+ }
+
+ public float vectorNormOrNaN(int vectorIndex) {
+ return hasVectorNorm() ? scalarFactor(ScalarFactor.VECTOR_NORM, vectorIndex) : Float.NaN;
+ }
+
+ public int scalarFactorOffset(ScalarFactor factor, int vectorIndex) {
+ checkVectorIndex(vectorIndex);
+ checkScalarFactor(factor);
+ return factor.ordinal() * numVectors * Float.BYTES + vectorIndex * Float.BYTES;
+ }
+
+ public RowLocator rowLocator(int vectorIndex) {
+ checkVectorIndex(vectorIndex);
+ int offset = vectorIndex * PostingBlockBuilder.ROW_LOCATOR_BYTES;
+ int fileGroupIdx = Short.toUnsignedInt(rowLocators.getShort(offset));
+ int instantIdx = Short.toUnsignedInt(rowLocators.getShort(offset + Short.BYTES));
+ long rowPosition = Integer.toUnsignedLong(rowLocators.getInt(offset + 2 * Short.BYTES));
+ return new RowLocator(
+ fileGroupIdx,
+ instantIdx,
+ rowPosition,
+ block.getFileGroupDict().get(fileGroupIdx),
+ block.getInstantTimeDict().get(instantIdx),
+ partitionForFileGroup(fileGroupIdx));
+ }
+
+ public String recordKey(int vectorIndex) {
+ checkVectorIndex(vectorIndex);
+ int start = recordKeyOffsets.getInt(vectorIndex * Integer.BYTES);
+ int end = recordKeyOffsets.getInt((vectorIndex + 1) * Integer.BYTES);
+ checkArgument(start >= 0 && end >= start && end <= recordKeyBytes.remaining(),
+ "Invalid record key offset range");
+ ByteBuffer keySlice = slice(recordKeyBytes, start, end - start);
+ byte[] bytes = new byte[keySlice.remaining()];
+ keySlice.get(bytes);
+ return new String(bytes, StandardCharsets.UTF_8);
+ }
+
+ public List fileGroupDict() {
+ return block.getFileGroupDict();
+ }
+
+ public List instantTimeDict() {
+ return block.getInstantTimeDict();
+ }
+
+ public List partitionDict() {
+ return block.getPartitionDict();
+ }
+
+ private String partitionForFileGroup(int fileGroupIdx) {
+ return block.getPartitionDict().isEmpty() ? "" : block.getPartitionDict().get(fileGroupIdx);
+ }
+
+ private void checkVectorIndex(int vectorIndex) {
+ checkArgument(vectorIndex >= 0 && vectorIndex < numVectors, "vectorIndex out of bounds: " + vectorIndex);
+ }
+
+ private void checkScalarFactor(ScalarFactor factor) {
+ checkArgument(factor.ordinal() < scalarFactorCount,
+ "Scalar factor is absent from this block layout: " + factor);
+ }
+
+ private static ByteBuffer littleEndian(ByteBuffer buffer) {
+ ByteBuffer duplicate = buffer.duplicate();
+ duplicate.order(ByteOrder.LITTLE_ENDIAN);
+ return duplicate;
+ }
+
+ private static ByteBuffer slice(ByteBuffer buffer, int offset, int length) {
+ ByteBuffer duplicate = buffer.duplicate();
+ duplicate.position(offset);
+ duplicate.limit(offset + length);
+ ByteBuffer slice = duplicate.slice();
+ slice.order(ByteOrder.LITTLE_ENDIAN);
+ return slice;
+ }
+
+ public enum ScalarFactor {
+ F_ADD_1,
+ F_RESCALE_1,
+ ERR_1,
+ F_ADD_EX,
+ F_RESCALE_EX,
+ RESIDUAL_NORM,
+ VECTOR_NORM
+ }
+
+ public static final class RowLocator {
+ private final int fileGroupDictIndex;
+ private final int instantTimeDictIndex;
+ private final long rowPosition;
+ private final String fileGroupId;
+ private final String instantTime;
+ private final String partitionPath;
+
+ private RowLocator(int fileGroupDictIndex,
+ int instantTimeDictIndex,
+ long rowPosition,
+ String fileGroupId,
+ String instantTime,
+ String partitionPath) {
+ this.fileGroupDictIndex = fileGroupDictIndex;
+ this.instantTimeDictIndex = instantTimeDictIndex;
+ this.rowPosition = rowPosition;
+ this.fileGroupId = fileGroupId;
+ this.instantTime = instantTime;
+ this.partitionPath = partitionPath;
+ }
+
+ public int getFileGroupDictIndex() {
+ return fileGroupDictIndex;
+ }
+
+ public int getInstantTimeDictIndex() {
+ return instantTimeDictIndex;
+ }
+
+ public long getRowPosition() {
+ return rowPosition;
+ }
+
+ public String getFileGroupId() {
+ return fileGroupId;
+ }
+
+ public String getInstantTime() {
+ return instantTime;
+ }
+
+ public String getPartitionPath() {
+ return partitionPath;
+ }
+ }
+}
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/index/vector/QuantizedVector.java b/hudi-common/src/main/java/org/apache/hudi/common/index/vector/QuantizedVector.java
new file mode 100644
index 0000000000000..962952990aa3e
--- /dev/null
+++ b/hudi-common/src/main/java/org/apache/hudi/common/index/vector/QuantizedVector.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.io.Serializable;
+
+/** Encoded vector and its persisted RaBitQ scoring factors. */
+public final class QuantizedVector implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ /** Primary packed sign/MSB code. */
+ final byte[] code;
+ /** Optional packed lower-bit code planes for multibit RaBitQ. */
+ final byte[] extendedCode;
+ /** Original or residual vector norm. */
+ final float scalar;
+ /** Optional L2-estimator additive factor used by multibit MDT postings. */
+ final Float additiveFactor;
+ /** Optional L2-estimator rescale factor used by multibit MDT postings. */
+ final Float rescaleFactor;
+ /** Sign-only L2-estimator additive factor used by packed-block pass 1. */
+ final Float additiveFactor1;
+ /** Sign-only L2-estimator rescale factor used by packed-block pass 1. */
+ final Float rescaleFactor1;
+ /** Query-independent pass-1 distance error factor. */
+ final Float error1;
+ /** Optional raw vector norm for metric-neutral residual scoring. */
+ final Float vectorNorm;
+ /** Total RaBitQ bit width represented by this payload. */
+ final int bits;
+
+ public QuantizedVector(byte[] code, float scalar) {
+ this(code, null, scalar, null, null, null, null, null, null, 1);
+ }
+
+ public QuantizedVector(byte[] code,
+ byte[] extendedCode,
+ float scalar,
+ Float additiveFactor,
+ Float rescaleFactor,
+ int bits) {
+ this(code, extendedCode, scalar, additiveFactor, rescaleFactor, null, null, null, null, bits);
+ }
+
+ public QuantizedVector(byte[] code,
+ byte[] extendedCode,
+ float scalar,
+ Float additiveFactor,
+ Float rescaleFactor,
+ Float additiveFactor1,
+ Float rescaleFactor1,
+ Float error1,
+ int bits) {
+ this(code, extendedCode, scalar, additiveFactor, rescaleFactor,
+ additiveFactor1, rescaleFactor1, error1, null, bits);
+ }
+
+ public QuantizedVector(byte[] code,
+ byte[] extendedCode,
+ float scalar,
+ Float additiveFactor,
+ Float rescaleFactor,
+ Float additiveFactor1,
+ Float rescaleFactor1,
+ Float error1,
+ Float vectorNorm,
+ int bits) {
+ if (code == null || code.length == 0) {
+ throw new IllegalArgumentException("Primary code must not be empty");
+ }
+ if (bits < 1 || bits > 8) {
+ throw new IllegalArgumentException("RaBitQ bits must be in [1, 8]: " + bits);
+ }
+ if (!Float.isFinite(scalar) || scalar < 0f) {
+ throw new IllegalArgumentException("Scalar must be finite and non-negative: " + scalar);
+ }
+ if ((bits == 1 && extendedCode != null && extendedCode.length != 0)
+ || (bits > 1 && (extendedCode == null || extendedCode.length == 0))) {
+ throw new IllegalArgumentException("Extended code presence must match the RaBitQ bit width");
+ }
+ validateOptionalFactor(additiveFactor, "additiveFactor");
+ validateNonNegativeFactor(rescaleFactor, "rescaleFactor");
+ validateOptionalFactor(additiveFactor1, "additiveFactor1");
+ validateNonNegativeFactor(rescaleFactor1, "rescaleFactor1");
+ validateNonNegativeFactor(error1, "error1");
+ validateNonNegativeFactor(vectorNorm, "vectorNorm");
+ this.code = code.clone();
+ this.extendedCode = extendedCode == null ? null : extendedCode.clone();
+ this.scalar = scalar;
+ this.additiveFactor = additiveFactor;
+ this.rescaleFactor = rescaleFactor;
+ this.additiveFactor1 = additiveFactor1;
+ this.rescaleFactor1 = rescaleFactor1;
+ this.error1 = error1;
+ this.vectorNorm = vectorNorm;
+ this.bits = bits;
+ }
+
+ public byte[] getCode() {
+ return code.clone();
+ }
+
+ public byte[] getExtendedCode() {
+ return extendedCode == null ? null : extendedCode.clone();
+ }
+
+ public float getScalar() {
+ return scalar;
+ }
+
+ public Float getAdditiveFactor() {
+ return additiveFactor;
+ }
+
+ public Float getRescaleFactor() {
+ return rescaleFactor;
+ }
+
+ public Float getAdditiveFactor1() {
+ return additiveFactor1;
+ }
+
+ public Float getRescaleFactor1() {
+ return rescaleFactor1;
+ }
+
+ public Float getError1() {
+ return error1;
+ }
+
+ public Float getVectorNorm() {
+ return vectorNorm;
+ }
+
+ public int getBits() {
+ return bits;
+ }
+
+ private static void validateOptionalFactor(Float value, String name) {
+ if (value != null && !Float.isFinite(value)) {
+ throw new IllegalArgumentException(name + " must be finite: " + value);
+ }
+ }
+
+ private static void validateNonNegativeFactor(Float value, String name) {
+ validateOptionalFactor(value, name);
+ if (value != null && value < 0f) {
+ throw new IllegalArgumentException(name + " must be non-negative: " + value);
+ }
+ }
+}
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/index/vector/RaBitQByteLutScorer.java b/hudi-common/src/main/java/org/apache/hudi/common/index/vector/RaBitQByteLutScorer.java
new file mode 100644
index 0000000000000..6454351fbcbc0
--- /dev/null
+++ b/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).
+ *
+ *
Exact float-query semantics. This scorer scores the same float rotated (residual)
+ * query used by {@link RaBitQEncoder#dotPackedBinary} and {@link RaBitQEncoder#multibitDotTerm}.
+ * It introduces no 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:
+ *
+ *
the pass-1 per-dimension sign loop ({@code dotSignRow}); and
+ * 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.
+ *
+ *
LUT layout. {@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}:
+ *
+ *
+ * @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));
+ }
+ double cBias = -((double) ((1 << bits) - 1)) / 2.0;
+ return (float) (((double) (1 << exBits)) * signDot + extendedDot + querySum * cBias);
+ }
+}
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/index/vector/RaBitQDistanceScorer.java b/hudi-common/src/main/java/org/apache/hudi/common/index/vector/RaBitQDistanceScorer.java
new file mode 100644
index 0000000000000..22c6d4b79ef03
--- /dev/null
+++ b/hudi-common/src/main/java/org/apache/hudi/common/index/vector/RaBitQDistanceScorer.java
@@ -0,0 +1,126 @@
+/*
+ * 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;
+
+/** Distance reconstruction for sign-encoded RaBitQ vectors. */
+public final class RaBitQDistanceScorer {
+
+ private RaBitQDistanceScorer() {
+ }
+
+ /** Estimates the legacy norm-weighted angular proxy from two packed sign codes. */
+ public static float estimateDistance(byte[] queryCode, byte[] encodedCode,
+ float scalar, int dimension) {
+ if (!Float.isFinite(scalar) || scalar < 0f) {
+ throw new IllegalArgumentException("scalar must be finite and non-negative: " + scalar);
+ }
+ float cosine = symmetricCosine(queryCode, encodedCode, dimension);
+ return scalar * (1.0f - cosine);
+ }
+
+ /** Cosine estimate from binary-vs-binary Hamming distance. */
+ public static float symmetricCosine(byte[] queryCode, byte[] encodedCode, int dimension) {
+ validatePackedCode(queryCode, dimension, "queryCode");
+ validatePackedCode(encodedCode, dimension, "encodedCode");
+ int fullBytes = dimension / Byte.SIZE;
+ int hamming = 0;
+ for (int i = 0; i < fullBytes; i++) {
+ hamming += Integer.bitCount((queryCode[i] ^ encodedCode[i]) & 0xFF);
+ }
+ int remainingBits = dimension % Byte.SIZE;
+ if (remainingBits != 0) {
+ int mask = (1 << remainingBits) - 1;
+ hamming += Integer.bitCount((queryCode[fullBytes] ^ encodedCode[fullBytes]) & mask);
+ }
+ return 1.0f - 2.0f * hamming / dimension;
+ }
+
+ /**
+ * Cosine estimate from the rotated float query and a packed sign code.
+ * This retains full query precision and therefore has lower variance than Hamming scoring.
+ */
+ public static float asymmetricCosine(float[] rotatedQuery, byte[] encodedCode, int dimension) {
+ if (rotatedQuery == null || rotatedQuery.length != dimension) {
+ throw new IllegalArgumentException("rotatedQuery length must equal dimension " + dimension);
+ }
+ validatePackedCode(encodedCode, dimension, "encodedCode");
+ double sum = 0.0;
+ for (int i = 0; i < dimension; i++) {
+ if (!Float.isFinite(rotatedQuery[i])) {
+ throw new IllegalArgumentException("rotatedQuery contains a non-finite value at dimension " + i);
+ }
+ boolean positive = (encodedCode[i >> 3] & (1 << (i & 7))) != 0;
+ sum += positive ? rotatedQuery[i] : -rotatedQuery[i];
+ }
+ return (float) (sum / Math.sqrt(dimension));
+ }
+
+ /** Reconstructs the configured distance from a cosine estimate and vector norms. */
+ public static float reconstructDistance(VectorDistanceMetric metric,
+ float cosineEstimate,
+ float queryNorm,
+ float vectorNorm) {
+ if (metric == null || !Float.isFinite(cosineEstimate)
+ || !Float.isFinite(queryNorm) || queryNorm < 0f
+ || !Float.isFinite(vectorNorm) || vectorNorm < 0f) {
+ throw new IllegalArgumentException("Metric, cosine estimate, and norms must be valid");
+ }
+ float cosine = Math.max(-1.0f, Math.min(1.0f, cosineEstimate));
+ switch (metric) {
+ case L2:
+ double squaredDistance = (double) queryNorm * queryNorm
+ + (double) vectorNorm * vectorNorm
+ - 2.0 * queryNorm * vectorNorm * cosine;
+ return (float) Math.sqrt(Math.max(0.0, squaredDistance));
+ case DOT_PRODUCT:
+ return -(queryNorm * vectorNorm * cosine);
+ case COSINE:
+ default:
+ return queryNorm == 0f || vectorNorm == 0f ? 1.0f : 1.0f - cosine;
+ }
+ }
+
+ /** Computes Hamming distance between equally sized packed binary codes. */
+ public static int hammingDistance(byte[] left, byte[] right) {
+ if (left == null || right == null) {
+ throw new IllegalArgumentException("Packed codes must not be null");
+ }
+ if (left.length != right.length) {
+ throw new IllegalArgumentException(
+ "Packed code length mismatch: " + left.length + " != " + right.length);
+ }
+ int count = 0;
+ for (int i = 0; i < left.length; i++) {
+ count += Integer.bitCount((left[i] ^ right[i]) & 0xFF);
+ }
+ return count;
+ }
+
+ private static void validatePackedCode(byte[] code, int dimension, String name) {
+ if (dimension <= 0) {
+ throw new IllegalArgumentException("dimension must be positive: " + dimension);
+ }
+ int expectedBytes = (dimension + Byte.SIZE - 1) / Byte.SIZE;
+ if (code == null || code.length != expectedBytes) {
+ throw new IllegalArgumentException(
+ name + " length must be " + expectedBytes + " bytes for dimension " + dimension);
+ }
+ }
+}
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/index/vector/RaBitQEncoder.java b/hudi-common/src/main/java/org/apache/hudi/common/index/vector/RaBitQEncoder.java
new file mode 100644
index 0000000000000..cc5ef0db2f6ca
--- /dev/null
+++ b/hudi-common/src/main/java/org/apache/hudi/common/index/vector/RaBitQEncoder.java
@@ -0,0 +1,588 @@
+/*
+ * 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.Map;
+import java.util.Objects;
+import java.util.Random;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * RaBitQ (Randomized Binary Quantization) encoder.
+ *
+ *
Encodes float vectors into 1-bit-per-dimension binary codes using a random
+ * orthogonal rotation matrix R. The rotation is deterministic given (seed, dimension),
+ * so R is stored as a seed — it never needs retraining.
+ *
+ *
Encoding (write path):
+ *
+ *
Normalize: v̂ = v / ||v||
+ *
Rotate: v_rot = R @ v̂
+ *
Binarize: code = pack(sign(v_rot)) — D bits → ceil(D/8) bytes
+ *
Scalar: s = ||v|| (stored alongside code; 1.0 if assume_normalized)
+ *
+ *
+ *
Query scan (read path):
+ *
+ *
Normalize + rotate the query vector (same R)
+ *
Binarize to get q_bin
+ *
For each database code b: Hamming(q_bin, b) → estimated cosine
+ *
Re-rank top-R candidates with exact distance
+ *
+ *
+ *
Memory: the rotation matrix is D×D floats = 4·D² bytes (~2.3 MB at D=768).
+ * It is built lazily on first use and reused across encode calls.
+ *
+ *
Thread-safe after the first call to {@link #encode} or {@link #encodeQuery}
+ * (the lazy init is synchronized).
+ */
+public final class RaBitQEncoder implements Serializable {
+ private static final long serialVersionUID = 1L;
+ private static final Map ROTATION_MATRIX_CACHE = new ConcurrentHashMap<>();
+ /** Neutral-factor thresholds for the current posting-block format (RFC-109 §3). */
+ private static final RaBitQFactorConfig FACTOR_CONFIG = RaBitQFactorConfig.defaults();
+ private final int dimension;
+ private final int bits;
+ private final long seed;
+ private final boolean assumeNormalized;
+
+ /** Rotation matrix, row-major. Populated lazily. */
+ private transient volatile float[][] rotMat;
+ public RaBitQEncoder(int dimension, long seed, boolean assumeNormalized) {
+ this(dimension, 1, seed, assumeNormalized);
+ }
+
+ public RaBitQEncoder(int dimension, int bits, long seed, boolean assumeNormalized) {
+ if (dimension <= 0) {
+ throw new IllegalArgumentException("Dimension must be positive, got: " + dimension);
+ }
+ if (bits <= 0 || bits > 8) {
+ throw new IllegalArgumentException("RaBitQ bits must be in [1, 8], got: " + bits);
+ }
+ this.dimension = dimension;
+ this.bits = bits;
+ this.seed = seed;
+ this.assumeNormalized = assumeNormalized;
+ }
+
+ /** Convenience constructor using default seed. */
+ public RaBitQEncoder(int dimension) {
+ this(dimension, 1, 42L, false);
+ }
+
+ // ---- encoding ----------------------------------------------------------
+
+ public QuantizedVector encode(float[] vector) {
+ validateVector(vector, "vector");
+ float norm = norm(vector);
+ float scalar = assumeNormalized ? 1.0f : norm;
+ float[] normalized = (norm == 0f || assumeNormalized)
+ ? vector
+ : normalize(vector, norm);
+ float[] rotated = rotate(normalized);
+ return new QuantizedVector(binarize(rotated), scalar);
+ }
+
+ /**
+ * Metric-neutral residual encoding used by MDT postings.
+ *
+ *
The optional {@code center} is the IVF centroid in the original vector space.
+ * When null, the vector is encoded relative to the origin.
+ */
+ public QuantizedVector encodeResidual(float[] vector, float[] center) {
+ validateVector(vector, "vector");
+ if (center != null) {
+ validateVector(center, "center");
+ }
+ float[] rotatedVector = rotate(vector);
+ float[] rotatedCenter = center == null ? new float[dimension] : rotate(center);
+ float[] residual = subtract(rotatedVector, rotatedCenter);
+ byte[] binaryCode = binarize(residual);
+
+ float residualNorm = norm(residual);
+ if (residualNorm == 0.0f) {
+ return new QuantizedVector(
+ binaryCode,
+ new byte[extendedCodeBytes()],
+ 0.0f,
+ 0.0f,
+ 0.0f,
+ 0.0f,
+ 0.0f,
+ 0.0f,
+ norm(vector),
+ bits);
+ }
+
+ if (bits <= 1) {
+ double ipResidual1 = 0.0d;
+ for (int i = 0; i < dimension; i++) {
+ ipResidual1 += residual[i] * (residual[i] > 0f ? 0.5d : -0.5d);
+ }
+ RaBitQNeutralFactors.Factors factors = RaBitQNeutralFactors.compute(
+ residual,
+ rotatedCenter,
+ rotatedVector,
+ ipResidual1,
+ ipResidual1,
+ dimension,
+ FACTOR_CONFIG);
+ return new QuantizedVector(
+ binaryCode,
+ new byte[0],
+ factors.residualNorm,
+ factors.centerRip,
+ factors.fRescale1,
+ factors.centerRip,
+ factors.fRescale1,
+ factors.err1,
+ factors.vectorNorm,
+ bits);
+ }
+
+ int exBits = bits - 1;
+
+ float[] absNormalizedResidual = new float[dimension];
+ for (int i = 0; i < dimension; i++) {
+ absNormalizedResidual[i] = Math.abs(residual[i] / residualNorm);
+ }
+
+ QuantizedLevels quantizedLevels = quantizeEx(absNormalizedResidual, exBits);
+ int[] exCode = quantizedLevels.levels;
+ int mask = (1 << exBits) - 1;
+ for (int i = 0; i < dimension; i++) {
+ if (residual[i] < 0f) {
+ exCode[i] = mask - exCode[i];
+ }
+ }
+
+ float cBias = (float) -((1 << exBits) - 0.5d);
+ double ipResidual1 = 0.0d;
+ double ipResidual = 0.0d;
+ for (int i = 0; i < dimension; i++) {
+ boolean positive = residual[i] > 0f;
+ double signOnlyCode = positive ? 0.5d : -0.5d;
+ float centeredCode = exCode[i] + (positive ? (1 << exBits) : 0) + cBias;
+ ipResidual1 += residual[i] * signOnlyCode;
+ ipResidual += residual[i] * centeredCode;
+ }
+
+ RaBitQNeutralFactors.Factors factors = RaBitQNeutralFactors.compute(
+ residual,
+ rotatedCenter,
+ rotatedVector,
+ ipResidual1,
+ ipResidual,
+ dimension,
+ FACTOR_CONFIG);
+ return new QuantizedVector(
+ binaryCode,
+ packUnsignedLevels(exCode, exBits),
+ factors.residualNorm,
+ factors.centerRip,
+ factors.fRescaleEx,
+ factors.centerRip,
+ factors.fRescale1,
+ factors.err1,
+ factors.vectorNorm,
+ bits);
+ }
+
+ /**
+ * Compatibility shim for old call sites. New postings must use the metric-neutral
+ * residual factor convention produced by {@link #encodeResidual(float[], float[])}.
+ */
+ public QuantizedVector encodeForL2(float[] vector, float[] center) {
+ return encodeResidual(vector, center);
+ }
+
+ public RaBitQQueryState encodeQuery(float[] queryVector) {
+ validateVector(queryVector, "query");
+ float norm = norm(queryVector);
+ float[] normalized = norm == 0f ? queryVector : normalize(queryVector, norm);
+ float[] rotated = rotate(normalized);
+ return new RaBitQQueryState(binarize(rotated), rotated, norm);
+ }
+
+ public RaBitQQueryState encodeQueryForL2(float[] queryVector) {
+ validateVector(queryVector, "query");
+ float[] rotated = rotate(queryVector);
+ return new RaBitQQueryState(binarize(rotated), rotated, norm(queryVector));
+ }
+
+ public float estimateDistance(RaBitQQueryState queryState, QuantizedVector encoded) {
+ return RaBitQDistanceScorer.estimateDistance(
+ queryState.binaryCodeUnsafe(), encoded.code, encoded.scalar, dimension);
+ }
+
+ /** Metric-aware approximate distance using symmetric or asymmetric cosine estimation. */
+ public float estimateDistance(RaBitQQueryState queryState,
+ QuantizedVector encoded,
+ VectorDistanceMetric metric,
+ boolean asymmetric) {
+ float cosine = asymmetric
+ ? RaBitQDistanceScorer.asymmetricCosine(queryState.rotatedQueryUnsafe(), encoded.code, dimension)
+ : RaBitQDistanceScorer.symmetricCosine(queryState.binaryCodeUnsafe(), encoded.code, dimension);
+ return RaBitQDistanceScorer.reconstructDistance(
+ metric, cosine, queryState.getQueryNorm(), encoded.scalar);
+ }
+
+ public int codeBytes() {
+ return (dimension + 7) / 8;
+ }
+
+ public int extendedCodeBytes() {
+ return bits <= 1 ? 0 : (dimension * (bits - 1) + 7) / 8;
+ }
+
+ public int totalCodeBytes() {
+ return codeBytes() + extendedCodeBytes();
+ }
+
+ public int getBits() {
+ return bits;
+ }
+
+ // ---- public utilities --------------------------------------------------
+
+ /** L2 norm of a float vector. */
+ public static float norm(float[] v) {
+ double sum = 0.0;
+ for (float x : v) {
+ sum += (double) x * x;
+ }
+ return (float) Math.sqrt(sum);
+ }
+
+ public static float l2Squared(float[] left, float[] right) {
+ double sum = 0.0;
+ for (int i = 0; i < left.length; i++) {
+ double delta = (double) left[i] - right[i];
+ sum += delta * delta;
+ }
+ return (float) sum;
+ }
+
+ public float[] rotateVector(float[] vector) {
+ validateVector(vector, "vector");
+ return rotate(vector);
+ }
+
+ public static float dotPackedBinary(byte[] binaryCode, float[] query, int dimension) {
+ double sum = 0.0;
+ for (int i = 0; i < dimension; i++) {
+ if ((binaryCode[i >> 3] & (1 << (i & 7))) != 0) {
+ sum += query[i];
+ }
+ }
+ return (float) sum;
+ }
+
+ public static float dotPackedUnsigned(byte[] packedLevels, int bitsPerValue, float[] query, int dimension) {
+ if (packedLevels == null || bitsPerValue <= 0) {
+ return 0.0f;
+ }
+ double sum = 0.0;
+ int bitOffset = 0;
+ for (int i = 0; i < dimension; i++) {
+ int value = 0;
+ for (int bit = 0; bit < bitsPerValue; bit++) {
+ int absoluteBit = bitOffset + bit;
+ int byteIndex = absoluteBit >> 3;
+ int bitIndex = absoluteBit & 7;
+ if ((packedLevels[byteIndex] & (1 << bitIndex)) != 0) {
+ value |= (1 << bit);
+ }
+ }
+ sum += (double) value * query[i];
+ bitOffset += bitsPerValue;
+ }
+ return (float) sum;
+ }
+
+ public static float multibitDotTerm(float[] rotatedQuery,
+ float sumQuery,
+ byte[] binaryCode,
+ byte[] extendedCode,
+ int dimension,
+ int bits) {
+ int exBits = bits - 1;
+ float binaryDot = dotPackedBinary(binaryCode, rotatedQuery, dimension);
+ if (exBits <= 0) {
+ return binaryDot + (sumQuery * -0.5f);
+ }
+ float extendedDot = dotPackedUnsigned(extendedCode, exBits, rotatedQuery, dimension);
+ float cBias = (float) -((1 << bits) - 1) / 2.0f;
+ return ((float) (1 << exBits) * binaryDot) + extendedDot + (sumQuery * cBias);
+ }
+
+ public static float multibitDotTerm(float[] rotatedQuery,
+ byte[] binaryCode,
+ byte[] extendedCode,
+ int dimension,
+ int bits) {
+ return multibitDotTerm(rotatedQuery, sum(rotatedQuery), binaryCode, extendedCode, dimension, bits);
+ }
+
+ // ---- private -----------------------------------------------------------
+
+ private void validateVector(float[] vector, String name) {
+ if (vector == null) {
+ throw new IllegalArgumentException(name + " must not be null");
+ }
+ if (vector.length != dimension) {
+ throw new IllegalArgumentException(
+ "Expected " + name + " dimension " + dimension + ", got " + vector.length);
+ }
+ for (int i = 0; i < vector.length; i++) {
+ if (!Float.isFinite(vector[i])) {
+ throw new IllegalArgumentException(
+ name + " contains a non-finite value at dimension " + i + ": " + vector[i]);
+ }
+ }
+ }
+
+ private float[] normalize(float[] v, float norm) {
+ float[] out = new float[dimension];
+ for (int i = 0; i < dimension; i++) {
+ out[i] = v[i] / norm;
+ }
+ return out;
+ }
+
+ /** Applies the D×D rotation matrix to the (normalized) input. */
+ private float[] rotate(float[] v) {
+ float[][] rotation = getRotationMatrix();
+ float[] out = new float[dimension];
+ for (int i = 0; i < dimension; i++) {
+ double acc = 0.0;
+ float[] row = rotation[i];
+ for (int j = 0; j < dimension; j++) {
+ acc += (double) row[j] * v[j];
+ }
+ out[i] = (float) acc;
+ }
+ return out;
+ }
+
+ /** Packs sign(v[i]) into ceil(D/8) bytes. Positive = bit 1, non-positive = bit 0. */
+ private byte[] binarize(float[] v) {
+ byte[] code = new byte[(dimension + 7) / 8];
+ for (int i = 0; i < dimension; i++) {
+ if (v[i] > 0f) {
+ code[i >> 3] |= (byte) (1 << (i & 7));
+ }
+ }
+ return code;
+ }
+
+ /** Lazy double-checked rotation matrix construction (Modified Gram-Schmidt). */
+ private float[][] getRotationMatrix() {
+ if (rotMat == null) {
+ synchronized (this) {
+ if (rotMat == null) {
+ rotMat = getOrBuildRotationMatrix(dimension, seed);
+ }
+ }
+ }
+ return rotMat;
+ }
+
+ private static float[][] getOrBuildRotationMatrix(int dimension, long seed) {
+ return ROTATION_MATRIX_CACHE.computeIfAbsent(new RotationKey(dimension, seed),
+ key -> buildRotationMatrix(key.dimension, key.seed));
+ }
+
+ /**
+ * Builds a random orthogonal D×D matrix using Modified Gram-Schmidt (MGS).
+ * Runtime: O(D³) — acceptable as a one-time cost (~0.5 s at D=768).
+ * Memory: 4 * D² bytes (~2.3 MB at D=768).
+ */
+ static float[][] buildRotationMatrix(int d, long seed) {
+ Random rng = new Random(seed);
+ // Gaussian random matrix
+ double[][] randMatrix = new double[d][d];
+ for (int i = 0; i < d; i++) {
+ for (int j = 0; j < d; j++) {
+ randMatrix[i][j] = rng.nextGaussian();
+ }
+ }
+ // Modified Gram-Schmidt orthogonalization (column-wise)
+ for (int j = 0; j < d; j++) {
+ // Orthogonalize column j against all previous columns
+ for (int k = 0; k < j; k++) {
+ double dot = 0.0;
+ for (int i = 0; i < d; i++) {
+ dot += randMatrix[i][k] * randMatrix[i][j];
+ }
+ for (int i = 0; i < d; i++) {
+ randMatrix[i][j] -= dot * randMatrix[i][k];
+ }
+ }
+ // Normalize column j
+ double colNorm = 0.0;
+ for (int i = 0; i < d; i++) {
+ colNorm += randMatrix[i][j] * randMatrix[i][j];
+ }
+ colNorm = Math.sqrt(colNorm);
+ if (colNorm > 1e-10) {
+ for (int i = 0; i < d; i++) {
+ randMatrix[i][j] /= colNorm;
+ }
+ }
+ }
+ // Transpose to row-major for efficient row-vector multiply
+ float[][] rotMat = new float[d][d];
+ for (int i = 0; i < d; i++) {
+ for (int j = 0; j < d; j++) {
+ rotMat[i][j] = (float) randMatrix[j][i];
+ }
+ }
+ return rotMat;
+ }
+
+ private float[] subtract(float[] left, float[] right) {
+ float[] out = new float[dimension];
+ for (int i = 0; i < dimension; i++) {
+ out[i] = left[i] - right[i];
+ }
+ return out;
+ }
+
+ private QuantizedLevels quantizeEx(float[] absValues, int exBits) {
+ double rescale = bestRescaleFactor(absValues, exBits);
+ int[] levels = new int[dimension];
+ double ipNorm = 0.0;
+ int maxLevel = (1 << exBits) - 1;
+ for (int i = 0; i < dimension; i++) {
+ int level = (int) Math.floor((rescale * absValues[i]) + 1.0e-5);
+ if (level < 0) {
+ level = 0;
+ } else if (level > maxLevel) {
+ level = maxLevel;
+ }
+ levels[i] = level;
+ ipNorm += (level + 0.5d) * absValues[i];
+ }
+ double ipNormInv = (Double.isFinite(ipNorm) && Math.abs(ipNorm) > 1.0e-12d) ? (1.0d / ipNorm) : 1.0d;
+ return new QuantizedLevels(levels, ipNormInv);
+ }
+
+ private static float sum(float[] values) {
+ float sum = 0.0f;
+ for (float value : values) {
+ sum += value;
+ }
+ return sum;
+ }
+
+ private double bestRescaleFactor(float[] absValues, int exBits) {
+ double max = 0.0d;
+ for (float value : absValues) {
+ max = Math.max(max, value);
+ }
+ if (max <= 1.0e-12d) {
+ return 1.0d;
+ }
+
+ int maxLevel = (1 << exBits) - 1;
+ double bestFactor = maxLevel / max;
+ double bestScore = Double.NEGATIVE_INFINITY;
+ for (int step = 1; step <= 64; step++) {
+ double factor = (step * maxLevel) / (64.0d * max);
+ double numerator = 0.0d;
+ double denominator = dimension * 0.25d;
+ for (float value : absValues) {
+ int level = (int) Math.floor((factor * value) + 1.0e-5d);
+ if (level < 0) {
+ level = 0;
+ } else if (level > maxLevel) {
+ level = maxLevel;
+ }
+ numerator += (level + 0.5d) * value;
+ denominator += (level * (double) level) + level;
+ }
+ double score = numerator / Math.sqrt(denominator);
+ if (score > bestScore) {
+ bestScore = score;
+ bestFactor = factor;
+ }
+ }
+ return bestFactor;
+ }
+
+ private byte[] packUnsignedLevels(int[] levels, int bitsPerValue) {
+ if (bitsPerValue <= 0) {
+ return new byte[0];
+ }
+ byte[] packed = new byte[(dimension * bitsPerValue + 7) / 8];
+ int bitOffset = 0;
+ for (int level : levels) {
+ for (int bit = 0; bit < bitsPerValue; bit++) {
+ if ((level & (1 << bit)) != 0) {
+ int absoluteBit = bitOffset + bit;
+ packed[absoluteBit >> 3] |= (byte) (1 << (absoluteBit & 7));
+ }
+ }
+ bitOffset += bitsPerValue;
+ }
+ return packed;
+ }
+
+ // ---- inner types -------------------------------------------------------
+
+ private static final class QuantizedLevels {
+ private final int[] levels;
+ private final double ipNormInv;
+
+ private QuantizedLevels(int[] levels, double ipNormInv) {
+ this.levels = levels;
+ this.ipNormInv = ipNormInv;
+ }
+ }
+
+ private static final class RotationKey {
+ private final int dimension;
+ private final long seed;
+
+ private RotationKey(int dimension, long seed) {
+ this.dimension = dimension;
+ this.seed = seed;
+ }
+
+ @Override
+ public boolean equals(Object other) {
+ if (this == other) {
+ return true;
+ }
+ if (other == null || getClass() != other.getClass()) {
+ return false;
+ }
+ RotationKey that = (RotationKey) other;
+ return dimension == that.dimension && seed == that.seed;
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(dimension, seed);
+ }
+ }
+}
\ No newline at end of file
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/index/vector/RaBitQFactorConfig.java b/hudi-common/src/main/java/org/apache/hudi/common/index/vector/RaBitQFactorConfig.java
new file mode 100644
index 0000000000000..509eb2102f2be
--- /dev/null
+++ b/hudi-common/src/main/java/org/apache/hudi/common/index/vector/RaBitQFactorConfig.java
@@ -0,0 +1,83 @@
+/*
+ * 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 java.io.Serializable;
+
+/**
+ * Generation-scoped configuration for RaBitQ neutral-factor computation (RFC-109 §3).
+ *
+ *
Replaces the previously static constants ({@code ERR_KAPPA}, absolute {@code EPS_IP}) with
+ * explicit, versioned thresholds. Readers reject factor versions they do not support instead of
+ * interpreting persisted factors with current-code constants.
+ *
+ *
{@code gMin} — normalized alignment floor: when the sign-code alignment
+ * {@code gHat1 < gMin} the pass-1 estimator is disabled (replaces the absolute
+ * {@code |ipResidual| < EPS_IP} gate).
+ *
{@code eps1Max} — maximum permitted relative pass-1 error; above it the estimator is
+ * disabled and the maximal valid bound {@code ERR_1 = residualNorm} is used, instead of the
+ * old (invalid) {@code min(1, eps1)} clamp that could falsely prune true neighbours.
+ *
{@code epsNRel} — relative residual-norm floor: a vector whose residual norm is in
+ * {@code (0, epsNRel * vectorNorm]} is treated as coincident-with-centroid; the estimator is
+ * disabled with {@code ERR_1 = residualNorm}. Only an exact-zero residual uses
+ * {@code ERR_1 = 0}.
+ *
+ */
+public final class RaBitQFactorConfig implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ public static final int FACTOR_VERSION = 1;
+ public static final double DEFAULT_KAPPA = 1.9;
+ public static final double DEFAULT_GMIN = 1.0e-3;
+ public static final double DEFAULT_EPS1_MAX = 1.0;
+ public static final double DEFAULT_EPS_N_REL = 1.0e-3;
+
+ private final int factorVersion;
+ private final double kappa;
+ private final double gMin;
+ private final double eps1Max;
+ private final double epsNRel;
+
+ public RaBitQFactorConfig(int factorVersion, double kappa, double gMin, double eps1Max, double epsNRel) {
+ if (factorVersion != FACTOR_VERSION) {
+ throw new IllegalArgumentException("Unsupported RaBitQ factor version: " + factorVersion);
+ }
+ this.factorVersion = factorVersion;
+ this.kappa = kappa;
+ this.gMin = gMin;
+ this.eps1Max = eps1Max;
+ this.epsNRel = epsNRel;
+ }
+
+ /** The defaults for the first persisted factor format (RFC-109 §3). */
+ public static RaBitQFactorConfig defaults() {
+ return new RaBitQFactorConfig(
+ FACTOR_VERSION, DEFAULT_KAPPA, DEFAULT_GMIN, DEFAULT_EPS1_MAX, DEFAULT_EPS_N_REL);
+ }
+
+ public int getFactorVersion() {
+ return factorVersion;
+ }
+
+ public double getKappa() {
+ return kappa;
+ }
+
+ public double getGMin() {
+ return gMin;
+ }
+
+ public double getEps1Max() {
+ return eps1Max;
+ }
+
+ public double getEpsNRel() {
+ return epsNRel;
+ }
+}
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/index/vector/RaBitQNeutralFactors.java b/hudi-common/src/main/java/org/apache/hudi/common/index/vector/RaBitQNeutralFactors.java
new file mode 100644
index 0000000000000..4faf8810d2815
--- /dev/null
+++ b/hudi-common/src/main/java/org/apache/hudi/common/index/vector/RaBitQNeutralFactors.java
@@ -0,0 +1,133 @@
+/*
+ * 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;
+
+/**
+ * Metric-NEUTRAL factor computation for residual RaBitQ encoding (RFC-109 §3).
+ *
+ *
The stored factors commit to NO metric; L2 / DOT / COSINE are composed at query time by
+ * {@link MetricQueryState}. The enclosing posting block's format version owns the factor layout:
+ *
No invalid error cap. The old {@code ERR_1 = n * min(1, eps1)} could report a
+ * bound smaller than the true estimation error and falsely prune true neighbours. When the
+ * relative error {@code eps1} exceeds {@link RaBitQFactorConfig#getEps1Max()} the estimator
+ * is disabled and the maximal valid bound {@code ERR_1 = n} is used instead.
+ *
Normalized quality gate. The absolute {@code |ipResidual| < EPS_IP} gate is replaced
+ * by a normalized alignment gate {@code gHat1 < gMin}.
+ *
Exact-zero vs small-nonzero residual. Only an exactly-zero residual uses
+ * {@code ERR_1 = 0}; a small nonzero residual (norm {@code <= epsNRel * ||x||}) disables the
+ * estimator with {@code ERR_1 = residualNorm}.
+ *
dimPadded loops. Norms, center-rip, and alignment use the padded dimension; padded
+ * coordinates are zero in {@code r} and codes so they contribute nothing but keep the code
+ * geometry ({@code ||code1|| = 0.5 * sqrt(dimPadded)}) correct.
+ *
+ */
+public final class RaBitQNeutralFactors {
+
+ public static final double EPS_NORM = 1.0e-9;
+
+ private RaBitQNeutralFactors() {
+ }
+
+ /** Immutable per-vector factor set in posting-block scalar-array order. */
+ public static final class Factors {
+ public final int factorVersion;
+ public final float centerRip;
+ public final float fRescale1;
+ public final float err1;
+ public final float fRescaleEx;
+ public final float residualNorm;
+ public final float vectorNorm; // consumed only for raw-cosine generations
+
+ Factors(int factorVersion, float centerRip, float fRescale1, float err1, float fRescaleEx,
+ float residualNorm, float vectorNorm) {
+ this.factorVersion = factorVersion;
+ this.centerRip = centerRip;
+ this.fRescale1 = fRescale1;
+ this.err1 = err1;
+ this.fRescaleEx = fRescaleEx;
+ this.residualNorm = residualNorm;
+ this.vectorNorm = vectorNorm;
+ }
+ }
+
+ /**
+ * @param residual r = rotated(x) - rotatedCenter, full precision (zero in padded dims)
+ * @param rotatedCenter P * c
+ * @param rotatedVector P * x (for ||x||; rotation preserves the norm)
+ * @param ipResidual1 <r, code1> accumulated in the encode loop
+ * @param ipResidualEx <r, codeEx> accumulated in the encode loop
+ * @param dimPadded padded dimension D'; padded coords are zero in r and codes
+ * @param config generation factor configuration (thresholds + version)
+ */
+ public static Factors compute(float[] residual, float[] rotatedCenter, float[] rotatedVector,
+ double ipResidual1, double ipResidualEx, int dimPadded,
+ RaBitQFactorConfig config) {
+ int factorVersion = config.getFactorVersion();
+ double nSq = 0.0;
+ double centerRip = 0.0;
+ double vSq = 0.0;
+ for (int i = 0; i < dimPadded; i++) {
+ nSq += (double) residual[i] * residual[i];
+ centerRip += (double) rotatedCenter[i] * residual[i];
+ vSq += (double) rotatedVector[i] * rotatedVector[i];
+ }
+ double n = Math.sqrt(nSq);
+ float vectorNorm = (float) Math.sqrt(vSq);
+ // Residual-norm tiers (RFC-109 §3).
+ if (n == 0.0) {
+ // Vector coincides with centroid exactly: composition is exact; ERR_1 = 0 is legitimate.
+ return new Factors(factorVersion, (float) centerRip, 0f, 0f, 0f, 0f, vectorNorm);
+ }
+ if (n <= config.getEpsNRel() * vectorNorm) {
+ // Tiny but nonzero residual: disable the estimator, ERR_1 = residualNorm (maximal valid bound).
+ return new Factors(factorVersion, (float) centerRip, 0f, (float) n, 0f, (float) n, vectorNorm);
+ }
+
+ // Normalized alignment of the residual with the sign code; ||code1|| = 0.5 * sqrt(dimPadded).
+ double gHat1 = ipResidual1 / (n * 0.5 * Math.sqrt(dimPadded));
+
+ float fRescale1;
+ float err1;
+ if (dimPadded == 1) {
+ // In one dimension the sign code reconstructs the residual exactly; the generic
+ // concentration bound is undefined because it contains (D - 1) in the denominator.
+ fRescale1 = (float) (nSq / ipResidual1);
+ err1 = 0f;
+ } else if (gHat1 < config.getGMin()) {
+ // Alignment too weak to trust the estimator (also guards the divide): disable, maximal bound.
+ fRescale1 = 0f;
+ err1 = (float) n;
+ } else {
+ double eps1 = config.getKappa() * Math.sqrt(
+ Math.max(0.0, 1.0 - gHat1 * gHat1) / (gHat1 * gHat1 * (dimPadded - 1)));
+ if (eps1 > config.getEps1Max()) {
+ // Relative error exceeds budget: disabling is the ONLY valid choice (never clamp to 1).
+ fRescale1 = 0f;
+ err1 = (float) n;
+ } else {
+ fRescale1 = (float) (nSq / ipResidual1);
+ err1 = (float) (n * eps1);
+ }
+ }
+
+ // Pass-2 rescale; relative guard against a degenerate extended inner product (avoids Inf/NaN).
+ float fRescaleEx = Math.abs(ipResidualEx) <= EPS_NORM * Math.max(1.0, nSq)
+ ? 0f : (float) (nSq / ipResidualEx);
+
+ return new Factors(factorVersion, (float) centerRip, fRescale1, err1, fRescaleEx, (float) n, vectorNorm);
+ }
+}
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/index/vector/RaBitQPlaneKernel.java b/hudi-common/src/main/java/org/apache/hudi/common/index/vector/RaBitQPlaneKernel.java
new file mode 100644
index 0000000000000..c0b1f43d9a3b9
--- /dev/null
+++ b/hudi-common/src/main/java/org/apache/hudi/common/index/vector/RaBitQPlaneKernel.java
@@ -0,0 +1,118 @@
+/*
+ * 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;
+
+/**
+ * Direct plane popcount kernel (RFC-109 §3). Scores a quantized query ({@link VectorQueryPlanes})
+ * against a data code stored as bit-planes (one sign plane + {@code exBits} extended planes, each a
+ * {@code long[]}), computing the inner product {@code } purely with
+ * {@code Long.bitCount(queryWord & dataWord)} — no per-dimension unpacking, sign-row copies, or
+ * extended-level repacking.
+ *
+ *
+ * where every {@code sum}/{@code <.,.>} term is a weighted popcount over plane words. The result is
+ * bit-exact with a scalar reference dot of the same {@code qPrime} against the same code.
+ */
+public final class RaBitQPlaneKernel {
+
+ private RaBitQPlaneKernel() {
+ }
+
+ /**
+ * Pass-1 inner product {@code } where {@code code1[i] = signBit[i] - 0.5}
+ * (sign-only 1-bit code). Uses only the sign plane.
+ */
+ public static double scorePass1(VectorQueryPlanes q, long[] signPlane) {
+ int words = q.words();
+ checkWords(signPlane, words, "signPlane");
+ double qDotSignFull = queryDotPlane(q, signPlane); //
+ return qDotSignFull - 0.5 * q.sumQPrime();
+ }
+
+ /**
+ * Pass-2 inner product {@code } for the full {@code (exBits+1)}-bit centered
+ * code. {@code exPlanes[b]} is the b-th extended bit-plane (LSB first), each {@code long[words]}.
+ */
+ public static double scorePass2(VectorQueryPlanes q, long[] signPlane, long[][] exPlanes, int exBits) {
+ int words = q.words();
+ checkWords(signPlane, words, "signPlane");
+ if (exBits <= 0) {
+ return scorePass1(q, signPlane);
+ }
+ if (exPlanes == null || exPlanes.length != exBits) {
+ throw new IllegalArgumentException("expected " + exBits + " extended planes");
+ }
+
+ double signScale = (double) (1L << exBits);
+ double cBias = -((double) (1L << (exBits + 1)) - 1.0) / 2.0; // -((2^bits - 1)/2), bits = exBits+1
+
+ // sumDataCode = 2^exBits * popcount(sign) + sum_b 2^b * popcount(exPlane_b) + cBias * dim
+ long popSign = popcount(signPlane);
+ double sumExLevels = 0.0;
+ for (int b = 0; b < exBits; b++) {
+ checkWords(exPlanes[b], words, "exPlane[" + b + "]");
+ sumExLevels += (double) (1L << b) * popcount(exPlanes[b]);
+ }
+ double sumDataCode = signScale * popSign + sumExLevels + cBias * q.dim();
+
+ // = 2^exBits * + + cBias * sumQLevel
+ double qlDotSign = queryLevelDotPlane(q, signPlane);
+ double qlDotEx = 0.0;
+ for (int b = 0; b < exBits; b++) {
+ qlDotEx += (double) (1L << b) * queryLevelDotPlane(q, exPlanes[b]);
+ }
+ double sumQLevel = queryLevelSum(q);
+ double qlDotDataCode = signScale * qlDotSign + qlDotEx + cBias * sumQLevel;
+
+ return q.queryMin() * sumDataCode + q.deltaQ() * qlDotDataCode;
+ }
+
+ // = qMin*popcount(plane) + deltaQ *
+ private static double queryDotPlane(VectorQueryPlanes q, long[] plane) {
+ return q.queryMin() * popcount(plane) + q.deltaQ() * queryLevelDotPlane(q, plane);
+ }
+
+ // = sum_a 2^a * popcount(queryPlane_a & plane)
+ private static double queryLevelDotPlane(VectorQueryPlanes q, long[] plane) {
+ double acc = 0.0;
+ for (int a = 0; a < q.bq(); a++) {
+ long[] qp = q.plane(a);
+ long pop = 0;
+ for (int w = 0; w < plane.length; w++) {
+ pop += Long.bitCount(qp[w] & plane[w]);
+ }
+ acc += (double) (1L << a) * pop;
+ }
+ return acc;
+ }
+
+ // sum_i qLevel[i] = sum_a 2^a * popcount(queryPlane_a)
+ private static double queryLevelSum(VectorQueryPlanes q) {
+ double acc = 0.0;
+ for (int a = 0; a < q.bq(); a++) {
+ acc += (double) (1L << a) * popcount(q.plane(a));
+ }
+ return acc;
+ }
+
+ private static long popcount(long[] words) {
+ long pop = 0;
+ for (long w : words) {
+ pop += Long.bitCount(w);
+ }
+ return pop;
+ }
+
+ private static void checkWords(long[] plane, int words, String name) {
+ if (plane == null || plane.length != words) {
+ throw new IllegalArgumentException(name + " must have " + words + " words");
+ }
+ }
+}
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/index/vector/RaBitQQueryState.java b/hudi-common/src/main/java/org/apache/hudi/common/index/vector/RaBitQQueryState.java
new file mode 100644
index 0000000000000..0e3992a3d6489
--- /dev/null
+++ b/hudi-common/src/main/java/org/apache/hudi/common/index/vector/RaBitQQueryState.java
@@ -0,0 +1,71 @@
+/*
+ * 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;
+
+/** Query state reused while scoring RaBitQ candidates. */
+public final class RaBitQQueryState implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ private final byte[] binaryCode;
+ /** Rotation of the unit query, used by the asymmetric estimator. */
+ private final float[] rotatedQuery;
+ /** Query norm before normalization, used for metric reconstruction. */
+ private final float queryNorm;
+ /** Sum of {@link #rotatedQuery}, reused by multibit scoring. */
+ private final float querySum;
+
+ RaBitQQueryState(byte[] binaryCode, float[] rotatedQuery, float queryNorm) {
+ this.binaryCode = binaryCode.clone();
+ this.rotatedQuery = rotatedQuery.clone();
+ this.queryNorm = queryNorm;
+ float sum = 0f;
+ for (float value : rotatedQuery) {
+ sum += value;
+ }
+ this.querySum = sum;
+ }
+
+ public byte[] getBinaryCode() {
+ return binaryCode.clone();
+ }
+
+ public float[] getRotatedQuery() {
+ return rotatedQuery.clone();
+ }
+
+ public float getQueryNorm() {
+ return queryNorm;
+ }
+
+ public float getQuerySum() {
+ return querySum;
+ }
+
+ byte[] binaryCodeUnsafe() {
+ return binaryCode;
+ }
+
+ float[] rotatedQueryUnsafe() {
+ return rotatedQuery;
+ }
+}
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/index/vector/VectorDistanceMetric.java b/hudi-common/src/main/java/org/apache/hudi/common/index/vector/VectorDistanceMetric.java
new file mode 100644
index 0000000000000..abbe51bf3c0f4
--- /dev/null
+++ b/hudi-common/src/main/java/org/apache/hudi/common/index/vector/VectorDistanceMetric.java
@@ -0,0 +1,113 @@
+/*
+ * 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.Locale;
+
+/**
+ * Distance metrics for vector similarity search.
+ *
+ *
All metrics are returned as distances (smaller = more similar),
+ * so they can be compared uniformly with a min-heap.
+ */
+public enum VectorDistanceMetric {
+
+ /**
+ * Cosine distance: 1 - cosine_similarity.
+ * Range: [0, 2]. 0 = identical direction, 2 = opposite.
+ */
+ COSINE {
+ @Override
+ public float compute(float[] a, float[] b) {
+ checkDimensions(a, b);
+ double dot = 0;
+ double normA = 0;
+ double normB = 0;
+ for (int i = 0; i < a.length; i++) {
+ dot += (double) a[i] * b[i];
+ normA += (double) a[i] * a[i];
+ normB += (double) b[i] * b[i];
+ }
+ double denom = Math.sqrt(normA) * Math.sqrt(normB);
+ return denom == 0.0 ? 1.0f : (float) (1.0 - dot / denom);
+ }
+ },
+
+ /**
+ * Euclidean (L2) distance.
+ * Range: [0, ∞). 0 = identical.
+ */
+ L2 {
+ @Override
+ public float compute(float[] a, float[] b) {
+ checkDimensions(a, b);
+ double sum = 0;
+ for (int i = 0; i < a.length; i++) {
+ double d = (double) a[i] - b[i];
+ sum += d * d;
+ }
+ return (float) Math.sqrt(sum);
+ }
+ },
+
+ /**
+ * Maximum inner product distance: negated dot product.
+ * Negated so smaller = higher similarity, consistent with the min-heap contract.
+ */
+ DOT_PRODUCT {
+ @Override
+ public float compute(float[] a, float[] b) {
+ checkDimensions(a, b);
+ double dot = 0;
+ for (int i = 0; i < a.length; i++) {
+ dot += (double) a[i] * b[i];
+ }
+ return (float) -dot;
+ }
+ };
+
+ /**
+ * Compute the distance between two float vectors.
+ *
+ * @param a first vector
+ * @param b second vector
+ * @return non-negative distance (smaller = more similar)
+ */
+ public abstract float compute(float[] a, float[] b);
+
+ /**
+ * Parse a metric name (case-insensitive).
+ *
+ * @param name e.g. "cosine", "l2", "dot_product"
+ * @return matching enum constant
+ */
+ static VectorDistanceMetric fromString(String name) {
+ return valueOf(name.toUpperCase(Locale.ROOT).replace(" ", "_").replace("-", "_"));
+ }
+
+ // ---- helpers -----------------------------------------------------------
+
+ private static void checkDimensions(float[] a, float[] b) {
+ if (a.length != b.length) {
+ throw new IllegalArgumentException(
+ "Vector dimension mismatch: " + a.length + " vs " + b.length);
+ }
+ }
+}
\ No newline at end of file
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/index/vector/VectorIndexArbiter.java b/hudi-common/src/main/java/org/apache/hudi/common/index/vector/VectorIndexArbiter.java
new file mode 100644
index 0000000000000..a46449759234f
--- /dev/null
+++ b/hudi-common/src/main/java/org/apache/hudi/common/index/vector/VectorIndexArbiter.java
@@ -0,0 +1,130 @@
+/*
+ * 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.common.model.HoodieRecordGlobalLocation;
+
+import java.util.Objects;
+
+/**
+ * The RLI finalist arbiter: resolves whether a vector-index posting still faithfully represents
+ * a live record, using the record-level index as the table's global version authority.
+ *
+ *
This is the pure classification core of RFC-109 "Upsert and Delete Support". It has no
+ * Spark or metadata-table dependency: callers resolve the current RLI location for a finalist
+ * key (a batched {@code readRecordIndexLocationsWithKeys} in the plan builder), then call
+ * {@link #classify} with the posting locator and that current location. The action taken per
+ * verdict is mode-specific and lives in the caller:
+ *
+ *
The classification itself is identical across modes so both report the same semantics and
+ * the same {@link ExclusionCounts} ({@code arbiterExclusions.stale} / {@code .deleted}).
+ */
+public final class VectorIndexArbiter {
+
+ /**
+ * Verdict for a single finalist posting.
+ *
+ *
+ *
{@code SERVE}: RLI hit and the current location matches the posting locator. The posting
+ * faithfully represents the live record; positional trust is preserved (subject to the
+ * positional validity gate for {@code rowPosition >= 0}).
+ *
{@code STALE}: RLI hit but the current location differs. The record was rewritten or
+ * moved; this posting's locator (and, for updates, its code) is no longer authoritative.
+ *
{@code DELETED}: RLI miss. The record no longer exists in the table.
+ *
+ */
+ public enum Decision {
+ SERVE,
+ STALE,
+ DELETED
+ }
+
+ private VectorIndexArbiter() {
+ }
+
+ /**
+ * Classify a single finalist posting against the record's current RLI location.
+ *
+ * @param postingPartitionPath partition path stored in the posting locator (may be null)
+ * @param postingFileGroupId file group id stored in the posting locator (may be null)
+ * @param postingBaseInstantTime base instant time stored in the posting locator (may be null)
+ * @param currentLocation current RLI location for the record key, or {@code null} for an
+ * RLI miss (deleted)
+ * @return the arbiter verdict
+ */
+ public static Decision classify(String postingPartitionPath,
+ String postingFileGroupId,
+ String postingBaseInstantTime,
+ HoodieRecordGlobalLocation currentLocation) {
+ if (currentLocation == null) {
+ return Decision.DELETED;
+ }
+ boolean matches = Objects.equals(postingPartitionPath, currentLocation.getPartitionPath())
+ && Objects.equals(postingFileGroupId, currentLocation.getFileId())
+ && Objects.equals(postingBaseInstantTime, currentLocation.getInstantTime());
+ return matches ? Decision.SERVE : Decision.STALE;
+ }
+
+ /**
+ * Mutable tally of arbiter exclusions, mirrored into both query modes' log lines as the
+ * {@code arbiterExclusions} freshness observability metric. Split into {@code stale} and
+ * {@code deleted} so the two upsert/delete effects are separable in dashboards.
+ */
+ public static final class ExclusionCounts {
+ private long stale;
+ private long deleted;
+
+ /** Record a verdict, incrementing the matching counter. {@code SERVE} is a no-op. */
+ public void record(Decision decision) {
+ switch (decision) {
+ case STALE:
+ stale++;
+ break;
+ case DELETED:
+ deleted++;
+ break;
+ default:
+ break;
+ }
+ }
+
+ public long stale() {
+ return stale;
+ }
+
+ public long deleted() {
+ return deleted;
+ }
+
+ public long total() {
+ return stale + deleted;
+ }
+
+ @Override
+ public String toString() {
+ return "arbiterExclusions{stale=" + stale + ", deleted=" + deleted + "}";
+ }
+ }
+}
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/index/vector/VectorIndexOptions.java b/hudi-common/src/main/java/org/apache/hudi/common/index/vector/VectorIndexOptions.java
new file mode 100644
index 0000000000000..44fd48cb5539b
--- /dev/null
+++ b/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}.
+ *
+ *
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";
+ public static final String NUM_CLUSTERS = "vector.num_clusters";
+ public static final String MAX_ITER = "vector.max_iter";
+ public static final String RABITQ_BITS = "vector.rabitq.bits";
+ public static final String RABITQ_SEED = "vector.rabitq.seed";
+ public static final String RABITQ_ASSUME_NORMALIZED = "vector.rabitq.assume_normalized";
+ public static final String QUERY_NUM_PROBES = "vector.query.nprobes";
+ public static final String QUERY_REFINE_FACTOR = "vector.query.refine_factor";
+ public static final String QUERY_MODE = "vector.query.mode";
+ public static final String QUERY_STALE_POLICY = "vector.query.stale_policy";
+
+ public static final VectorDistanceMetric DEFAULT_METRIC = VectorDistanceMetric.COSINE;
+ public static final VectorQuantizer DEFAULT_QUANTIZER = VectorQuantizer.IVF_RABITQ;
+ public static final int DEFAULT_NUM_CLUSTERS = 256;
+ public static final int DEFAULT_MAX_ITER = 20;
+ public static final int DEFAULT_RABITQ_BITS = 4;
+ public static final long DEFAULT_RABITQ_SEED = 42L;
+ public static final int DEFAULT_NUM_PROBES = 32;
+ public static final int DEFAULT_REFINE_FACTOR = 50;
+ public static final VectorQueryMode DEFAULT_QUERY_MODE = VectorQueryMode.EXACT_RERANK;
+ public static final VectorStalePolicy DEFAULT_STALE_POLICY = VectorStalePolicy.FAIL;
+
+ private static final Set SUPPORTED_OPTIONS = Collections.unmodifiableSet(
+ new HashSet<>(Arrays.asList(
+ METRIC,
+ QUANTIZER,
+ NUM_CLUSTERS,
+ MAX_ITER,
+ RABITQ_BITS,
+ RABITQ_SEED,
+ RABITQ_ASSUME_NORMALIZED,
+ QUERY_NUM_PROBES,
+ QUERY_REFINE_FACTOR,
+ QUERY_MODE,
+ QUERY_STALE_POLICY)));
+
+ private VectorIndexOptions() {
+ }
+
+ /**
+ * Validates the complete option map and returns canonical values for persistence.
+ *
+ *
The returned map contains every supported option, including explicit defaults. Unknown,
+ * retired, misspelled, and invalid options are rejected instead of being silently ignored.
+ */
+ public static Map validateAndNormalize(Map options) {
+ Set unknownOptions = new HashSet<>(options.keySet());
+ unknownOptions.removeAll(SUPPORTED_OPTIONS);
+ if (!unknownOptions.isEmpty()) {
+ throw new IllegalArgumentException("Unsupported vector index options: " + unknownOptions);
+ }
+
+ VectorDistanceMetric metric = getMetric(options);
+ VectorQuantizer quantizer = getQuantizer(options);
+ int numClusters = getNumClusters(options);
+ int maxIter = getMaxIter(options);
+ int bits = getRaBitQBits(options);
+ long seed = getRaBitQSeed(options);
+ boolean assumeNormalized = shouldAssumeNormalizedVectors(options);
+ int numProbes = getNumProbes(options);
+ int refineFactor = getRefineFactor(options);
+ VectorQueryMode queryMode = getQueryMode(options);
+ VectorStalePolicy stalePolicy = getStalePolicy(options);
+
+ if (numProbes > numClusters) {
+ throw new IllegalArgumentException(
+ "Option '" + QUERY_NUM_PROBES + "' must not exceed '" + NUM_CLUSTERS + "': "
+ + numProbes + " > " + numClusters);
+ }
+
+ Map normalized = new LinkedHashMap<>();
+ normalized.put(METRIC, metric.name().toLowerCase(Locale.ROOT));
+ normalized.put(QUANTIZER, quantizer.name());
+ normalized.put(NUM_CLUSTERS, String.valueOf(numClusters));
+ normalized.put(MAX_ITER, String.valueOf(maxIter));
+ normalized.put(RABITQ_BITS, String.valueOf(bits));
+ normalized.put(RABITQ_SEED, String.valueOf(seed));
+ normalized.put(RABITQ_ASSUME_NORMALIZED, String.valueOf(assumeNormalized));
+ normalized.put(QUERY_NUM_PROBES, String.valueOf(numProbes));
+ normalized.put(QUERY_REFINE_FACTOR, String.valueOf(refineFactor));
+ normalized.put(QUERY_MODE, queryMode.name().toLowerCase(Locale.ROOT));
+ normalized.put(QUERY_STALE_POLICY, stalePolicy.name().toLowerCase(Locale.ROOT));
+ return Collections.unmodifiableMap(normalized);
+ }
+
+ private static VectorDistanceMetric getMetric(Map options) {
+ String value = getOption(options, METRIC, DEFAULT_METRIC.name());
+ try {
+ return VectorDistanceMetric.fromString(value);
+ } catch (IllegalArgumentException e) {
+ throw unsupportedValue(METRIC, value, e);
+ }
+ }
+
+ private static VectorQuantizer getQuantizer(Map options) {
+ String value = getOption(options, QUANTIZER, DEFAULT_QUANTIZER.name());
+ try {
+ return VectorQuantizer.fromString(value);
+ } catch (IllegalArgumentException e) {
+ throw unsupportedValue(QUANTIZER, value, e);
+ }
+ }
+
+ private static int getNumClusters(Map options) {
+ return getPositiveInt(options, NUM_CLUSTERS, DEFAULT_NUM_CLUSTERS);
+ }
+
+ private static int getMaxIter(Map options) {
+ return getPositiveInt(options, MAX_ITER, DEFAULT_MAX_ITER);
+ }
+
+ private static int getRaBitQBits(Map options) {
+ int bits = getInt(options, RABITQ_BITS, DEFAULT_RABITQ_BITS);
+ if (bits < 1 || bits > 8) {
+ throw new IllegalArgumentException(
+ "Option '" + RABITQ_BITS + "' must be between 1 and 8: " + bits);
+ }
+ return bits;
+ }
+
+ private static long getRaBitQSeed(Map options) {
+ String value = getOption(options, RABITQ_SEED, String.valueOf(DEFAULT_RABITQ_SEED));
+ try {
+ return Long.parseLong(value);
+ } catch (NumberFormatException e) {
+ throw invalidNumber(RABITQ_SEED, value, e);
+ }
+ }
+
+ private static boolean shouldAssumeNormalizedVectors(Map options) {
+ String value = getOption(
+ options, RABITQ_ASSUME_NORMALIZED, String.valueOf(false)).toLowerCase(Locale.ROOT);
+ if (!"true".equals(value) && !"false".equals(value)) {
+ throw new IllegalArgumentException(
+ "Option '" + RABITQ_ASSUME_NORMALIZED + "' must be either 'true' or 'false': " + value);
+ }
+ return Boolean.parseBoolean(value);
+ }
+
+ private static int getNumProbes(Map options) {
+ return getPositiveInt(options, QUERY_NUM_PROBES, DEFAULT_NUM_PROBES);
+ }
+
+ private static int getRefineFactor(Map options) {
+ return getPositiveInt(options, QUERY_REFINE_FACTOR, DEFAULT_REFINE_FACTOR);
+ }
+
+ private static VectorQueryMode getQueryMode(Map options) {
+ String value = getOption(options, QUERY_MODE, DEFAULT_QUERY_MODE.name());
+ try {
+ return VectorQueryMode.fromString(value);
+ } catch (IllegalArgumentException e) {
+ throw unsupportedValue(QUERY_MODE, value, e);
+ }
+ }
+
+ private static VectorStalePolicy getStalePolicy(Map options) {
+ String value = getOption(options, QUERY_STALE_POLICY, DEFAULT_STALE_POLICY.name());
+ try {
+ return VectorStalePolicy.fromString(value);
+ } catch (IllegalArgumentException e) {
+ throw unsupportedValue(QUERY_STALE_POLICY, value, e);
+ }
+ }
+
+ private static int getPositiveInt(Map options, String key, int defaultValue) {
+ int value = getInt(options, key, defaultValue);
+ if (value <= 0) {
+ throw new IllegalArgumentException("Option '" + key + "' must be greater than 0: " + value);
+ }
+ return value;
+ }
+
+ private static int getInt(Map options, String key, int defaultValue) {
+ String value = getOption(options, key, String.valueOf(defaultValue));
+ try {
+ return Integer.parseInt(value);
+ } catch (NumberFormatException e) {
+ throw invalidNumber(key, value, e);
+ }
+ }
+
+ private static IllegalArgumentException invalidNumber(
+ String key, String value, NumberFormatException cause) {
+ return new IllegalArgumentException(
+ "Option '" + key + "' must be a valid number: " + value, cause);
+ }
+
+ private static IllegalArgumentException unsupportedValue(
+ String key, String value, IllegalArgumentException cause) {
+ return new IllegalArgumentException(
+ "Unsupported value for option '" + key + "': " + value, cause);
+ }
+
+ private static String getOption(Map options, String key, String defaultValue) {
+ String value = options.getOrDefault(key, defaultValue);
+ if (value == null || value.isEmpty()) {
+ throw new IllegalArgumentException("Option '" + key + "' must not be empty");
+ }
+ return value;
+ }
+}
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/index/vector/VectorIndexPruner.java b/hudi-common/src/main/java/org/apache/hudi/common/index/vector/VectorIndexPruner.java
new file mode 100644
index 0000000000000..97a640cc216b8
--- /dev/null
+++ b/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.
+ *
+ *
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.
+ *
+ *
Used by both the Spark file index ({@code HoodieVectorAwareFileIndex}) and the
+ * Trino split manager ({@code HudiVectorSplitManager}), keeping all math in one place.
+ *
+ *
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> 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> 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 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 fileGroups = new HashSet<>();
+ for (int clusterId : topClusters) {
+ Set 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) {
+ int k = centroids.length;
+ // scored[i] = (distance, cluster_id)
+ List scored = new ArrayList<>(k);
+ for (int i = 0; i < k; i++) {
+ float dist = metric.compute(query, centroids[i]);
+ scored.add(new float[]{dist, i});
+ }
+ scored.sort((a, b) -> Float.compare(a[0], b[0]));
+
+ int[] result = new int[numProbes];
+ for (int i = 0; i < numProbes; i++) {
+ result[i] = (int) scored.get(i)[1];
+ }
+ return result;
+ }
+
+ /** Returns the number of clusters (K). */
+ public int numClusters() {
+ return centroids == null ? 0 : centroids.length;
+ }
+
+ // ---- factory helpers ---------------------------------------------------
+
+ /**
+ * Builds a cluster→file-group map from a flat assignment list.
+ * Each entry in {@code assignments} must be a three-element array:
+ * {@code [clusterId (int), fileGroupId (String), partitionPath (String)]}.
+ *
+ *
The returned map is partition-aware: pass {@code partitionFilter = null}
+ * to include all partitions, or a non-null set to restrict to specific partitions.
+ *
+ * @param assignments rows from the MDT assignments/fg_mapping partition
+ * @param partitionFilter partition paths to include; null = all
+ * @return cluster → file-group mapping
+ */
+ public static Map> buildClusterMap(
+ Iterable