vinothchandar commented on code in PR #19309: URL: https://github.com/apache/hudi/pull/19309#discussion_r3799866142
########## rfc/rfc-109/rfc-109.md: ########## @@ -0,0 +1,714 @@ +<!-- + 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. +--> + +# RFC-109: Native Vector Search Support in Apache Hudi + +## Proposers + +@chrevanthreddy + +## Approvers + +- TBD + +## Status + +Umbrella issue: [apache/hudi#19094](https://github.com/apache/hudi/issues/19094) + +Related: [apache/hudi#18676](https://github.com/apache/hudi/issues/18676) + +State: UNDER REVIEW + +--- + +## Table of Contents Review Comment: Small thing. but the ToC makes it easy to follow. thanks ########## rfc/rfc-109/rfc-109.md: ########## @@ -0,0 +1,714 @@ +<!-- + 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. +--> + +# RFC-109: Native Vector Search Support in Apache Hudi + +## Proposers + +@chrevanthreddy + +## Approvers + +- TBD + +## Status + +Umbrella issue: [apache/hudi#19094](https://github.com/apache/hudi/issues/19094) + +Related: [apache/hudi#18676](https://github.com/apache/hudi/issues/18676) + +State: UNDER REVIEW + +--- + +## Table of Contents + +- [Abstract](#abstract) +- [1. Goals and Non-Goals](#1-goals-and-non-goals) +- [2. Architecture](#2-architecture) +- [3. IVF + RaBitQ Index Algorithm](#3-ivf--rabitq-index-algorithm) +- [4. Metadata Table Storage Model: the Posting Block](#4-metadata-table-storage-model-the-posting-block) +- [5. Bootstrap and Write Path](#5-bootstrap-and-write-path) +- [6. Read Path](#6-read-path) +- [7. Maintenance, Rebalancing, and Cleaner](#7-maintenance-rebalancing-and-cleaner) +- [8. Spark API Surface](#8-spark-api-surface) +- [9. Correctness, Compatibility, and Freshness](#9-correctness-compatibility-and-freshness) +- [10. Test Plan](#10-test-plan) +- [11. Benchmark Evidence, Rollout, and MVP Scope](#11-benchmark-evidence-rollout-and-mvp-scope) +- [12. References](#12-references) + +--- + +## Abstract + +This RFC proposes native approximate nearest-neighbor (ANN) vector search in Apache Hudi. +Tables increasingly carry embedding columns represented by Hudi's fixed-dimension +`VECTOR(D[, elementType])` logical type, and users want to ask *"find the K rows most +similar to this query vector"* — for semantic search, recommendations, RAG, and +deduplication — without copying data into a separate vector database. + +Today the only option on a Hudi table is a brute-force scan: read every vector, compute +every distance. That is correct but scales linearly with table size (tens of seconds at a +billion rows). This RFC adds an index so that vector queries read only a small, targeted +fraction of the index and the table, return results with high recall, and stay +transactionally consistent with the table under upserts and deletes — all with **no new +storage system**. The index lives in the Hudi Metadata Table (MDT), like Hudi's existing +record-level and secondary indexes, and is maintained by the same table services. + +The design combines three well-understood pieces — IVF clustering, RaBitQ quantization, and +exact re-ranking — with one storage innovation that makes them practical on an immutable, +columnar, object-store-resident lakehouse: + +> **The posting block.** Instead of one MDT record per indexed vector, the index packs +> ~1–4K vectors into a single MDT record laid out column-wise (structure-of-arrays), keyed +> so that one IVF cluster forms one contiguous, prefix-scannable key range. This reduces MDT +> record count by roughly three orders of magnitude, turns "scan a cluster" into a single +> contiguous range read, and lets a query touch only the columns a given scan pass needs. + +The base table remains the source of truth for exact vector values. The MDT stores only +routing, pruning, and approximate-scoring metadata; final ranking always reads exact vectors +from the base table. + +Artifacted measurements on a **1-billion-row, 128-dimensional table** show approximate-only +recall@10 of 0.824, 0.843, and 0.845 at `nprobe=16`, `32`, and `64`; exact reranking reaches +0.983 at `nprobe=32` with `refineFactor=50` (§11). These modes serve different goals: +approximate-only provides the latency floor, while exact reranking is the recommended +quality mode. + +--- + +## 1. Goals and Non-Goals + +### 1.1 Goals + +1. Keep authoritative vector values in a top-level base-table `VECTOR(D[, elementType])` column. +2. Store the vector index in the MDT, maintained by Hudi metadata-table commits, compaction, + and cleaning — no hidden or generated columns in base-table files. +3. Make candidate discovery cheap and targeted: probe a few clusters, scan contiguous key + ranges, score on compressed codes with a provable pruning bound. +4. Make results trustworthy: approximate math selects candidates; **exact** distance on + base-table vectors ranks them. +5. Stay transactionally consistent: snapshot-pinned reads, correct behavior under inserts, + updates, deletes, and clustering. +6. Be engine-neutral in design; Spark is the first implementation. +7. Maintain the index incrementally (no global rebuild for normal churn) and support + versioned, zero-downtime rebuilds. + +### 1.2 Non-Goals (initial landing) Review Comment: Let's add a top-level `## Limitations` section to this RFC, tracking what's knowingly not addressed in this first cut — so we can land the MVP and iterate with eyes open. First entry: **VECTOR schema evolution**. What happens when the indexed column's dimension or element type changes ########## rfc/rfc-109/rfc-109.md: ########## @@ -0,0 +1,714 @@ +<!-- + 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. +--> + +# RFC-109: Native Vector Search Support in Apache Hudi + +## Proposers + +@chrevanthreddy + +## Approvers + +- TBD + +## Status + +Umbrella issue: [apache/hudi#19094](https://github.com/apache/hudi/issues/19094) + +Related: [apache/hudi#18676](https://github.com/apache/hudi/issues/18676) + +State: UNDER REVIEW + +--- + +## Table of Contents + +- [Abstract](#abstract) +- [1. Goals and Non-Goals](#1-goals-and-non-goals) +- [2. Architecture](#2-architecture) +- [3. IVF + RaBitQ Index Algorithm](#3-ivf--rabitq-index-algorithm) +- [4. Metadata Table Storage Model: the Posting Block](#4-metadata-table-storage-model-the-posting-block) +- [5. Bootstrap and Write Path](#5-bootstrap-and-write-path) +- [6. Read Path](#6-read-path) +- [7. Maintenance, Rebalancing, and Cleaner](#7-maintenance-rebalancing-and-cleaner) +- [8. Spark API Surface](#8-spark-api-surface) +- [9. Correctness, Compatibility, and Freshness](#9-correctness-compatibility-and-freshness) +- [10. Test Plan](#10-test-plan) +- [11. Benchmark Evidence, Rollout, and MVP Scope](#11-benchmark-evidence-rollout-and-mvp-scope) +- [12. References](#12-references) + +--- + +## Abstract + +This RFC proposes native approximate nearest-neighbor (ANN) vector search in Apache Hudi. +Tables increasingly carry embedding columns represented by Hudi's fixed-dimension +`VECTOR(D[, elementType])` logical type, and users want to ask *"find the K rows most +similar to this query vector"* — for semantic search, recommendations, RAG, and +deduplication — without copying data into a separate vector database. + +Today the only option on a Hudi table is a brute-force scan: read every vector, compute +every distance. That is correct but scales linearly with table size (tens of seconds at a +billion rows). This RFC adds an index so that vector queries read only a small, targeted +fraction of the index and the table, return results with high recall, and stay +transactionally consistent with the table under upserts and deletes — all with **no new +storage system**. The index lives in the Hudi Metadata Table (MDT), like Hudi's existing +record-level and secondary indexes, and is maintained by the same table services. + +The design combines three well-understood pieces — IVF clustering, RaBitQ quantization, and +exact re-ranking — with one storage innovation that makes them practical on an immutable, +columnar, object-store-resident lakehouse: + +> **The posting block.** Instead of one MDT record per indexed vector, the index packs +> ~1–4K vectors into a single MDT record laid out column-wise (structure-of-arrays), keyed +> so that one IVF cluster forms one contiguous, prefix-scannable key range. This reduces MDT +> record count by roughly three orders of magnitude, turns "scan a cluster" into a single +> contiguous range read, and lets a query touch only the columns a given scan pass needs. + +The base table remains the source of truth for exact vector values. The MDT stores only +routing, pruning, and approximate-scoring metadata; final ranking always reads exact vectors +from the base table. + +Artifacted measurements on a **1-billion-row, 128-dimensional table** show approximate-only +recall@10 of 0.824, 0.843, and 0.845 at `nprobe=16`, `32`, and `64`; exact reranking reaches +0.983 at `nprobe=32` with `refineFactor=50` (§11). These modes serve different goals: +approximate-only provides the latency floor, while exact reranking is the recommended +quality mode. + +--- + +## 1. Goals and Non-Goals + +### 1.1 Goals + +1. Keep authoritative vector values in a top-level base-table `VECTOR(D[, elementType])` column. +2. Store the vector index in the MDT, maintained by Hudi metadata-table commits, compaction, + and cleaning — no hidden or generated columns in base-table files. +3. Make candidate discovery cheap and targeted: probe a few clusters, scan contiguous key + ranges, score on compressed codes with a provable pruning bound. +4. Make results trustworthy: approximate math selects candidates; **exact** distance on + base-table vectors ranks them. +5. Stay transactionally consistent: snapshot-pinned reads, correct behavior under inserts, + updates, deletes, and clustering. +6. Be engine-neutral in design; Spark is the first implementation. +7. Maintain the index incrementally (no global rebuild for normal churn) and support + versioned, zero-downtime rebuilds. + +### 1.2 Non-Goals (initial landing) + +- ANN families beyond IVF + RaBitQ (e.g. HNSW, DiskANN). +- Filtered search (arbitrary predicate + kNN) as a first-class planned operation. +- Time-travel-consistent index reads for historical snapshots. +- Native non-Spark generation construction, GPU encoding, and workload-specific auto-tuning. + +### 1.3 Alternatives considered + +- **One index record per vector** is simple but creates billions of MDT records and excessive + write, compaction, and scan amplification; posting blocks preserve MDT ownership while + amortizing that overhead. +- **Dedicated index files in the table** permit specialized layouts but introduce a second + commit, cleaning, and snapshot protocol. MDT records reuse Hudi's existing transaction and + table-service machinery. +- **External sidecar indexes or vector databases** may offer richer serving features but lose + atomic Hudi snapshot semantics and require another storage system. They remain valid when + independent serving infrastructure is desired. +- **Native ANN libraries** improve local kernels but do not define durable object-store layout, + multi-writer maintenance, or snapshot visibility. They may be used behind the interfaces in + this RFC without changing the persisted contract. + +--- + +## 2. Architecture + + + +The design splits responsibilities the way Hudi already does between the data table and the +metadata table: + +```text +DATA TABLE (parquet/orc) METADATA TABLE (vector_index partition) + authoritative vectors + payload ←── the index: centroids, quantizer, posting blocks, + read only for final re-ranking cluster manifests, generation manifest Review Comment: Readability: "generation" is used here (and throughout §2–§3) but only gets defined in §4.5. Add a one-liner up front — e.g. "a *generation* is an immutable, atomically-published version of the entire index (centroids + quantizer + routing + postings)" — so first-time readers aren't guessing until §4.5. ########## rfc/rfc-109/rfc-109.md: ########## @@ -0,0 +1,714 @@ +<!-- + 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. +--> + +# RFC-109: Native Vector Search Support in Apache Hudi + +## Proposers + +@chrevanthreddy + +## Approvers + +- TBD + +## Status + +Umbrella issue: [apache/hudi#19094](https://github.com/apache/hudi/issues/19094) + +Related: [apache/hudi#18676](https://github.com/apache/hudi/issues/18676) + +State: UNDER REVIEW + +--- + +## Table of Contents + +- [Abstract](#abstract) +- [1. Goals and Non-Goals](#1-goals-and-non-goals) +- [2. Architecture](#2-architecture) +- [3. IVF + RaBitQ Index Algorithm](#3-ivf--rabitq-index-algorithm) +- [4. Metadata Table Storage Model: the Posting Block](#4-metadata-table-storage-model-the-posting-block) +- [5. Bootstrap and Write Path](#5-bootstrap-and-write-path) +- [6. Read Path](#6-read-path) +- [7. Maintenance, Rebalancing, and Cleaner](#7-maintenance-rebalancing-and-cleaner) +- [8. Spark API Surface](#8-spark-api-surface) +- [9. Correctness, Compatibility, and Freshness](#9-correctness-compatibility-and-freshness) +- [10. Test Plan](#10-test-plan) +- [11. Benchmark Evidence, Rollout, and MVP Scope](#11-benchmark-evidence-rollout-and-mvp-scope) +- [12. References](#12-references) + +--- + +## Abstract + +This RFC proposes native approximate nearest-neighbor (ANN) vector search in Apache Hudi. +Tables increasingly carry embedding columns represented by Hudi's fixed-dimension +`VECTOR(D[, elementType])` logical type, and users want to ask *"find the K rows most +similar to this query vector"* — for semantic search, recommendations, RAG, and +deduplication — without copying data into a separate vector database. + +Today the only option on a Hudi table is a brute-force scan: read every vector, compute +every distance. That is correct but scales linearly with table size (tens of seconds at a +billion rows). This RFC adds an index so that vector queries read only a small, targeted +fraction of the index and the table, return results with high recall, and stay +transactionally consistent with the table under upserts and deletes — all with **no new +storage system**. The index lives in the Hudi Metadata Table (MDT), like Hudi's existing +record-level and secondary indexes, and is maintained by the same table services. + +The design combines three well-understood pieces — IVF clustering, RaBitQ quantization, and +exact re-ranking — with one storage innovation that makes them practical on an immutable, +columnar, object-store-resident lakehouse: + +> **The posting block.** Instead of one MDT record per indexed vector, the index packs +> ~1–4K vectors into a single MDT record laid out column-wise (structure-of-arrays), keyed +> so that one IVF cluster forms one contiguous, prefix-scannable key range. This reduces MDT +> record count by roughly three orders of magnitude, turns "scan a cluster" into a single +> contiguous range read, and lets a query touch only the columns a given scan pass needs. + +The base table remains the source of truth for exact vector values. The MDT stores only +routing, pruning, and approximate-scoring metadata; final ranking always reads exact vectors +from the base table. + +Artifacted measurements on a **1-billion-row, 128-dimensional table** show approximate-only +recall@10 of 0.824, 0.843, and 0.845 at `nprobe=16`, `32`, and `64`; exact reranking reaches +0.983 at `nprobe=32` with `refineFactor=50` (§11). These modes serve different goals: +approximate-only provides the latency floor, while exact reranking is the recommended +quality mode. + +--- + +## 1. Goals and Non-Goals + +### 1.1 Goals + +1. Keep authoritative vector values in a top-level base-table `VECTOR(D[, elementType])` column. +2. Store the vector index in the MDT, maintained by Hudi metadata-table commits, compaction, + and cleaning — no hidden or generated columns in base-table files. +3. Make candidate discovery cheap and targeted: probe a few clusters, scan contiguous key + ranges, score on compressed codes with a provable pruning bound. +4. Make results trustworthy: approximate math selects candidates; **exact** distance on + base-table vectors ranks them. +5. Stay transactionally consistent: snapshot-pinned reads, correct behavior under inserts, + updates, deletes, and clustering. +6. Be engine-neutral in design; Spark is the first implementation. +7. Maintain the index incrementally (no global rebuild for normal churn) and support + versioned, zero-downtime rebuilds. + +### 1.2 Non-Goals (initial landing) + +- ANN families beyond IVF + RaBitQ (e.g. HNSW, DiskANN). +- Filtered search (arbitrary predicate + kNN) as a first-class planned operation. +- Time-travel-consistent index reads for historical snapshots. +- Native non-Spark generation construction, GPU encoding, and workload-specific auto-tuning. + +### 1.3 Alternatives considered + +- **One index record per vector** is simple but creates billions of MDT records and excessive + write, compaction, and scan amplification; posting blocks preserve MDT ownership while + amortizing that overhead. +- **Dedicated index files in the table** permit specialized layouts but introduce a second + commit, cleaning, and snapshot protocol. MDT records reuse Hudi's existing transaction and + table-service machinery. +- **External sidecar indexes or vector databases** may offer richer serving features but lose + atomic Hudi snapshot semantics and require another storage system. They remain valid when + independent serving infrastructure is desired. +- **Native ANN libraries** improve local kernels but do not define durable object-store layout, + multi-writer maintenance, or snapshot visibility. They may be used behind the interfaces in + this RFC without changing the persisted contract. + +--- + +## 2. Architecture + + + +The design splits responsibilities the way Hudi already does between the data table and the +metadata table: + +```text +DATA TABLE (parquet/orc) METADATA TABLE (vector_index partition) + authoritative vectors + payload ←── the index: centroids, quantizer, posting blocks, + read only for final re-ranking cluster manifests, generation manifest + read for candidate generation +``` + +Each vector index is one MDT partition. RFC-109 adds no table properties, common writer- +dispatch changes, timeline-semantics changes, or behavior visible to non-vector readers or +writers. Every consistency mechanism is either a record in the vector-index MDT partition or +logic in the RFC-owned vector indexer and query planner. Creating an index: + +```sql +CREATE INDEX embedding_idx +ON products +USING VECTOR (embedding) +OPTIONS ( + 'vector.metric' = 'cosine', + 'vector.quantizer' = 'IVF_RABITQ', + 'vector.num_clusters'= '4096' +); +``` + +creates: + +```text +.hoodie/metadata/vector_index_embedding_idx/ +``` + +and does not add generated columns to the base-table schema. RFC-109 consumes rather than +redefines the RFC-99 vector contract: the source must be a top-level Hudi +`VECTOR(D[, elementType])`. The table schema is authoritative for `D` and element type; +index definitions and generation manifests repeat them only for integrity validation. + +The current RFC-99 storage backing is fixed-width bytes: Avro `FIXED` and Parquet +`FIXED_LEN_BYTE_ARRAY(D × elementWidth)`. Engine adapters expose idiomatic values—Spark uses +an annotated `ArrayType(FloatType|DoubleType|ByteType)`—and convert at the storage boundary. +This fixed width also avoids Parquet LIST repetition-level traversal during positional exact +fetch. A plain `ARRAY<FLOAT>` is not implicitly indexable and requires explicit migration or +backfill to `VECTOR(D)`; index creation neither reinterprets nor rewrites it. + +The query path uses the MDT first to discover candidates, then reads base-table vectors for +exact re-ranking: + +```text +query vector + → compare to centroids, pick nprobe clusters (in-memory, ms) + → MDT prefix-scan those clusters' posting blocks (targeted range reads) + → two-pass RaBitQ scoring, keep refineFactor·K best (bit math + error bounds) + → validate candidate freshness via Record Level Index (batched point lookups) + → fetch ONLY those rows from the base table by position (page-level reads) + → exact distance on real vectors → final top-K +``` + +--- + +## 3. IVF + RaBitQ Index Algorithm + +Every practical ANN index answers two questions: **where to look** (avoid scanning +everything) and **how to compare cheaply** (avoid full-precision math on what is scanned). + +### 3.1 Where to look: IVF routing + +Inverted File (IVF) indexing clusters vectors with KMeans into `numClusters` groups (e.g. +~4K–64K). Each vector belongs to its nearest centroid. A query compares against the centroids +only (thousands, not billions), selects the `nprobe` nearest clusters, and scans only those +clusters' entries. `nprobe` is the recall dial. + +IVF is the right fit for a lakehouse-resident index because a cluster's entries can be stored +**contiguously**, which maps directly onto sorted key ranges in Hudi's MDT (§4). Graph +indexes (HNSW) give excellent in-memory recall but require random traversal of the whole +graph, which fights columnar, immutable, object-store storage. + +### 3.2 How to compare cheaply: RaBitQ quantization + +Inside a probed cluster there are still thousands of full vectors. Quantization stores a +small *code* per vector plus a few correction scalars, so most comparisons run on compressed +codes and only the best few hundred candidates are re-checked against real vectors. + +RaBitQ is chosen over scalar (SQ), product (PQ), and plain binary quantization for four +reasons: + +1. **Unbiased estimator with a provable per-vector error bound.** Each code carries scalars + that turn a cheap bit-level dot product into an *unbiased* estimate of the true distance, + plus a bound on how wrong it can be. The bound enables **safe pruning**: skip a vector + only when even its best plausible distance cannot make top-K. +2. **No codebooks.** RaBitQ needs only a random rotation (a seed) and the centroids — both + tiny, both versioned in the index metadata. Nothing to retrain when data drifts. +3. **Tunable precision.** B = 1 bit/dim is a fast coarse filter; B = 4 bits/dim gives + near-SQ quality at ~8× less space. Both are used in a two-pass scan (§6.2). +4. **Metric-flexible.** One stored code serves L2, cosine, and dot-product; the metric is + applied at query time. + +The four ideas, precisely: + +- **Residual.** Store each vector as its difference from its centroid, `r = v − c`. + Residuals are small and centered, which is what lets few bits go far. +- **Rotation.** Apply one fixed random orthonormal rotation `R` (derived from a per-generation + seed) to everything first: `x = R·r`. This spreads information evenly across dimensions, + which is what makes the error bound hold for any data distribution. Only the seed is stored. +- **Code.** Quantize `x` to B bits per dimension, stored as **bit planes** — plane 0 holds + bit 0 of every dimension, plane 1 bit 1, etc. The top plane alone is a 1-bit sign sketch. + Scoring a plane against the (transformed) query is `AND`/`XOR` + `popcount` — a few CPU + instructions per 64 dimensions. +- **Factors.** A handful of small scalars per vector (residual norm, two rescale factors, an + error term, a centroid correction) that convert plane math into an unbiased distance + estimate plus its confidence interval. + +At query time the query vector is transformed the same way (`R·(q − c)` per probed cluster), +and the per-plane popcounts combine with the stored factors into the estimate and its bound. +The estimate builds the shortlist; exact base-table distances produce the final ranking (§6). + +### 3.3 Why this fits Hudi + +- Centroids are small enough to load at planning time: `K × D` floats (~12 MB for K=4096, + D=768). +- Codes are compact: a 1B × 128-dim float table's raw vectors are ~512 GB; the RaBitQ index + including keys and locators is ~136 GB, and the scanned portion per query is tens of MB. +- Posting keys are prefix-scannable by generation, cluster, and shard (§4). +- Quantizer state is stable: a seed and centroids, no per-generation learned codebook. +- Exact re-ranking preserves correctness for returned candidates. + +--- + +## 4. Metadata Table Storage Model: the Posting Block + +This section describes the core storage contribution of this RFC. + +### 4.1 The posting block + +A naive index would write one MDT record per indexed vector. At a billion rows that is a +billion MDT records per generation — prohibitive to write, compact, scan, and clean. + +Instead, the index sorts entries by `(cluster, fileGroup, rowPosition)` and packs +~1–4K vectors' worth of codes and metadata into a single MDT record — a **posting block** — Review Comment: Minor: let's define "posting" for the average reader when it first appears — it's inverted-index jargon (a cluster's posting list = the set of entries assigned to it) and many Hudi devs won't have that context. one line here would do it ########## rfc/rfc-109/rfc-109.md: ########## @@ -0,0 +1,714 @@ +<!-- + 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. +--> + +# RFC-109: Native Vector Search Support in Apache Hudi + +## Proposers + +@chrevanthreddy + +## Approvers + +- TBD + +## Status + +Umbrella issue: [apache/hudi#19094](https://github.com/apache/hudi/issues/19094) + +Related: [apache/hudi#18676](https://github.com/apache/hudi/issues/18676) + +State: UNDER REVIEW + +--- + +## Table of Contents + +- [Abstract](#abstract) +- [1. Goals and Non-Goals](#1-goals-and-non-goals) +- [2. Architecture](#2-architecture) +- [3. IVF + RaBitQ Index Algorithm](#3-ivf--rabitq-index-algorithm) +- [4. Metadata Table Storage Model: the Posting Block](#4-metadata-table-storage-model-the-posting-block) +- [5. Bootstrap and Write Path](#5-bootstrap-and-write-path) +- [6. Read Path](#6-read-path) +- [7. Maintenance, Rebalancing, and Cleaner](#7-maintenance-rebalancing-and-cleaner) +- [8. Spark API Surface](#8-spark-api-surface) +- [9. Correctness, Compatibility, and Freshness](#9-correctness-compatibility-and-freshness) +- [10. Test Plan](#10-test-plan) +- [11. Benchmark Evidence, Rollout, and MVP Scope](#11-benchmark-evidence-rollout-and-mvp-scope) +- [12. References](#12-references) + +--- + +## Abstract + +This RFC proposes native approximate nearest-neighbor (ANN) vector search in Apache Hudi. +Tables increasingly carry embedding columns represented by Hudi's fixed-dimension +`VECTOR(D[, elementType])` logical type, and users want to ask *"find the K rows most +similar to this query vector"* — for semantic search, recommendations, RAG, and +deduplication — without copying data into a separate vector database. + +Today the only option on a Hudi table is a brute-force scan: read every vector, compute +every distance. That is correct but scales linearly with table size (tens of seconds at a +billion rows). This RFC adds an index so that vector queries read only a small, targeted +fraction of the index and the table, return results with high recall, and stay +transactionally consistent with the table under upserts and deletes — all with **no new +storage system**. The index lives in the Hudi Metadata Table (MDT), like Hudi's existing +record-level and secondary indexes, and is maintained by the same table services. + +The design combines three well-understood pieces — IVF clustering, RaBitQ quantization, and +exact re-ranking — with one storage innovation that makes them practical on an immutable, +columnar, object-store-resident lakehouse: + +> **The posting block.** Instead of one MDT record per indexed vector, the index packs +> ~1–4K vectors into a single MDT record laid out column-wise (structure-of-arrays), keyed +> so that one IVF cluster forms one contiguous, prefix-scannable key range. This reduces MDT +> record count by roughly three orders of magnitude, turns "scan a cluster" into a single +> contiguous range read, and lets a query touch only the columns a given scan pass needs. + +The base table remains the source of truth for exact vector values. The MDT stores only +routing, pruning, and approximate-scoring metadata; final ranking always reads exact vectors +from the base table. + +Artifacted measurements on a **1-billion-row, 128-dimensional table** show approximate-only +recall@10 of 0.824, 0.843, and 0.845 at `nprobe=16`, `32`, and `64`; exact reranking reaches +0.983 at `nprobe=32` with `refineFactor=50` (§11). These modes serve different goals: +approximate-only provides the latency floor, while exact reranking is the recommended +quality mode. + +--- + +## 1. Goals and Non-Goals + +### 1.1 Goals + +1. Keep authoritative vector values in a top-level base-table `VECTOR(D[, elementType])` column. +2. Store the vector index in the MDT, maintained by Hudi metadata-table commits, compaction, + and cleaning — no hidden or generated columns in base-table files. +3. Make candidate discovery cheap and targeted: probe a few clusters, scan contiguous key + ranges, score on compressed codes with a provable pruning bound. +4. Make results trustworthy: approximate math selects candidates; **exact** distance on + base-table vectors ranks them. +5. Stay transactionally consistent: snapshot-pinned reads, correct behavior under inserts, + updates, deletes, and clustering. +6. Be engine-neutral in design; Spark is the first implementation. +7. Maintain the index incrementally (no global rebuild for normal churn) and support + versioned, zero-downtime rebuilds. + +### 1.2 Non-Goals (initial landing) + +- ANN families beyond IVF + RaBitQ (e.g. HNSW, DiskANN). +- Filtered search (arbitrary predicate + kNN) as a first-class planned operation. +- Time-travel-consistent index reads for historical snapshots. +- Native non-Spark generation construction, GPU encoding, and workload-specific auto-tuning. + +### 1.3 Alternatives considered + +- **One index record per vector** is simple but creates billions of MDT records and excessive + write, compaction, and scan amplification; posting blocks preserve MDT ownership while + amortizing that overhead. +- **Dedicated index files in the table** permit specialized layouts but introduce a second + commit, cleaning, and snapshot protocol. MDT records reuse Hudi's existing transaction and + table-service machinery. +- **External sidecar indexes or vector databases** may offer richer serving features but lose + atomic Hudi snapshot semantics and require another storage system. They remain valid when + independent serving infrastructure is desired. +- **Native ANN libraries** improve local kernels but do not define durable object-store layout, + multi-writer maintenance, or snapshot visibility. They may be used behind the interfaces in + this RFC without changing the persisted contract. + +--- + +## 2. Architecture + + + +The design splits responsibilities the way Hudi already does between the data table and the +metadata table: + +```text +DATA TABLE (parquet/orc) METADATA TABLE (vector_index partition) + authoritative vectors + payload ←── the index: centroids, quantizer, posting blocks, + read only for final re-ranking cluster manifests, generation manifest + read for candidate generation +``` + +Each vector index is one MDT partition. RFC-109 adds no table properties, common writer- +dispatch changes, timeline-semantics changes, or behavior visible to non-vector readers or +writers. Every consistency mechanism is either a record in the vector-index MDT partition or +logic in the RFC-owned vector indexer and query planner. Creating an index: + +```sql +CREATE INDEX embedding_idx +ON products +USING VECTOR (embedding) +OPTIONS ( + 'vector.metric' = 'cosine', + 'vector.quantizer' = 'IVF_RABITQ', + 'vector.num_clusters'= '4096' +); +``` + +creates: + +```text +.hoodie/metadata/vector_index_embedding_idx/ +``` + +and does not add generated columns to the base-table schema. RFC-109 consumes rather than +redefines the RFC-99 vector contract: the source must be a top-level Hudi +`VECTOR(D[, elementType])`. The table schema is authoritative for `D` and element type; +index definitions and generation manifests repeat them only for integrity validation. + +The current RFC-99 storage backing is fixed-width bytes: Avro `FIXED` and Parquet +`FIXED_LEN_BYTE_ARRAY(D × elementWidth)`. Engine adapters expose idiomatic values—Spark uses +an annotated `ArrayType(FloatType|DoubleType|ByteType)`—and convert at the storage boundary. +This fixed width also avoids Parquet LIST repetition-level traversal during positional exact +fetch. A plain `ARRAY<FLOAT>` is not implicitly indexable and requires explicit migration or +backfill to `VECTOR(D)`; index creation neither reinterprets nor rewrites it. + +The query path uses the MDT first to discover candidates, then reads base-table vectors for +exact re-ranking: + +```text +query vector + → compare to centroids, pick nprobe clusters (in-memory, ms) + → MDT prefix-scan those clusters' posting blocks (targeted range reads) + → two-pass RaBitQ scoring, keep refineFactor·K best (bit math + error bounds) + → validate candidate freshness via Record Level Index (batched point lookups) + → fetch ONLY those rows from the base table by position (page-level reads) + → exact distance on real vectors → final top-K +``` + +--- + +## 3. IVF + RaBitQ Index Algorithm + +Every practical ANN index answers two questions: **where to look** (avoid scanning +everything) and **how to compare cheaply** (avoid full-precision math on what is scanned). + +### 3.1 Where to look: IVF routing + +Inverted File (IVF) indexing clusters vectors with KMeans into `numClusters` groups (e.g. +~4K–64K). Each vector belongs to its nearest centroid. A query compares against the centroids +only (thousands, not billions), selects the `nprobe` nearest clusters, and scans only those +clusters' entries. `nprobe` is the recall dial. + +IVF is the right fit for a lakehouse-resident index because a cluster's entries can be stored +**contiguously**, which maps directly onto sorted key ranges in Hudi's MDT (§4). Graph +indexes (HNSW) give excellent in-memory recall but require random traversal of the whole +graph, which fights columnar, immutable, object-store storage. + +### 3.2 How to compare cheaply: RaBitQ quantization + +Inside a probed cluster there are still thousands of full vectors. Quantization stores a +small *code* per vector plus a few correction scalars, so most comparisons run on compressed +codes and only the best few hundred candidates are re-checked against real vectors. + +RaBitQ is chosen over scalar (SQ), product (PQ), and plain binary quantization for four +reasons: + +1. **Unbiased estimator with a provable per-vector error bound.** Each code carries scalars + that turn a cheap bit-level dot product into an *unbiased* estimate of the true distance, + plus a bound on how wrong it can be. The bound enables **safe pruning**: skip a vector + only when even its best plausible distance cannot make top-K. +2. **No codebooks.** RaBitQ needs only a random rotation (a seed) and the centroids — both + tiny, both versioned in the index metadata. Nothing to retrain when data drifts. +3. **Tunable precision.** B = 1 bit/dim is a fast coarse filter; B = 4 bits/dim gives + near-SQ quality at ~8× less space. Both are used in a two-pass scan (§6.2). +4. **Metric-flexible.** One stored code serves L2, cosine, and dot-product; the metric is + applied at query time. + +The four ideas, precisely: + +- **Residual.** Store each vector as its difference from its centroid, `r = v − c`. + Residuals are small and centered, which is what lets few bits go far. +- **Rotation.** Apply one fixed random orthonormal rotation `R` (derived from a per-generation + seed) to everything first: `x = R·r`. This spreads information evenly across dimensions, + which is what makes the error bound hold for any data distribution. Only the seed is stored. +- **Code.** Quantize `x` to B bits per dimension, stored as **bit planes** — plane 0 holds + bit 0 of every dimension, plane 1 bit 1, etc. The top plane alone is a 1-bit sign sketch. + Scoring a plane against the (transformed) query is `AND`/`XOR` + `popcount` — a few CPU + instructions per 64 dimensions. +- **Factors.** A handful of small scalars per vector (residual norm, two rescale factors, an + error term, a centroid correction) that convert plane math into an unbiased distance + estimate plus its confidence interval. + +At query time the query vector is transformed the same way (`R·(q − c)` per probed cluster), +and the per-plane popcounts combine with the stored factors into the estimate and its bound. +The estimate builds the shortlist; exact base-table distances produce the final ranking (§6). + +### 3.3 Why this fits Hudi + +- Centroids are small enough to load at planning time: `K × D` floats (~12 MB for K=4096, + D=768). +- Codes are compact: a 1B × 128-dim float table's raw vectors are ~512 GB; the RaBitQ index + including keys and locators is ~136 GB, and the scanned portion per query is tens of MB. +- Posting keys are prefix-scannable by generation, cluster, and shard (§4). +- Quantizer state is stable: a seed and centroids, no per-generation learned codebook. +- Exact re-ranking preserves correctness for returned candidates. + +--- + +## 4. Metadata Table Storage Model: the Posting Block + +This section describes the core storage contribution of this RFC. + +### 4.1 The posting block + +A naive index would write one MDT record per indexed vector. At a billion rows that is a +billion MDT records per generation — prohibitive to write, compact, scan, and clean. + +Instead, the index sorts entries by `(cluster, fileGroup, rowPosition)` and packs +~1–4K vectors' worth of codes and metadata into a single MDT record — a **posting block** — +laid out column-wise (structure-of-arrays) so each scan pass touches only the columns it +needs: + +```text +POSTING BLOCK (~512 KB target, one MDT record) +┌───────────────────────────────────────────────┐ +│ S1 sign planes ← pass 1 touches this │ +│ S2 extra bit planes ← pass 2, survivors │ +│ S3 factor arrays ← both passes │ +│ S4 row locators ← finalists only │ +│ S5 dictionaries (file groups, partitions) │ Review Comment: Since the partition dictionary already lives in the posting block, partition pushdown (bounding the ANN search to a subset of table partitions) looks within reach — filter candidates against the dictionary during the scan pass. Fine to defer for v1, but let's call it out as a known gap in the Limitations section given partition predicates are the most common filter in practice? ########## rfc/rfc-109/rfc-109.md: ########## @@ -0,0 +1,714 @@ +<!-- + 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. +--> + +# RFC-109: Native Vector Search Support in Apache Hudi + +## Proposers + +@chrevanthreddy + +## Approvers + +- TBD + +## Status + +Umbrella issue: [apache/hudi#19094](https://github.com/apache/hudi/issues/19094) + +Related: [apache/hudi#18676](https://github.com/apache/hudi/issues/18676) + +State: UNDER REVIEW + +--- + +## Table of Contents + +- [Abstract](#abstract) +- [1. Goals and Non-Goals](#1-goals-and-non-goals) +- [2. Architecture](#2-architecture) +- [3. IVF + RaBitQ Index Algorithm](#3-ivf--rabitq-index-algorithm) +- [4. Metadata Table Storage Model: the Posting Block](#4-metadata-table-storage-model-the-posting-block) +- [5. Bootstrap and Write Path](#5-bootstrap-and-write-path) +- [6. Read Path](#6-read-path) +- [7. Maintenance, Rebalancing, and Cleaner](#7-maintenance-rebalancing-and-cleaner) +- [8. Spark API Surface](#8-spark-api-surface) +- [9. Correctness, Compatibility, and Freshness](#9-correctness-compatibility-and-freshness) +- [10. Test Plan](#10-test-plan) +- [11. Benchmark Evidence, Rollout, and MVP Scope](#11-benchmark-evidence-rollout-and-mvp-scope) +- [12. References](#12-references) + +--- + +## Abstract + +This RFC proposes native approximate nearest-neighbor (ANN) vector search in Apache Hudi. +Tables increasingly carry embedding columns represented by Hudi's fixed-dimension +`VECTOR(D[, elementType])` logical type, and users want to ask *"find the K rows most +similar to this query vector"* — for semantic search, recommendations, RAG, and +deduplication — without copying data into a separate vector database. + +Today the only option on a Hudi table is a brute-force scan: read every vector, compute +every distance. That is correct but scales linearly with table size (tens of seconds at a +billion rows). This RFC adds an index so that vector queries read only a small, targeted +fraction of the index and the table, return results with high recall, and stay +transactionally consistent with the table under upserts and deletes — all with **no new +storage system**. The index lives in the Hudi Metadata Table (MDT), like Hudi's existing +record-level and secondary indexes, and is maintained by the same table services. + +The design combines three well-understood pieces — IVF clustering, RaBitQ quantization, and +exact re-ranking — with one storage innovation that makes them practical on an immutable, +columnar, object-store-resident lakehouse: + +> **The posting block.** Instead of one MDT record per indexed vector, the index packs +> ~1–4K vectors into a single MDT record laid out column-wise (structure-of-arrays), keyed +> so that one IVF cluster forms one contiguous, prefix-scannable key range. This reduces MDT +> record count by roughly three orders of magnitude, turns "scan a cluster" into a single +> contiguous range read, and lets a query touch only the columns a given scan pass needs. + +The base table remains the source of truth for exact vector values. The MDT stores only +routing, pruning, and approximate-scoring metadata; final ranking always reads exact vectors +from the base table. + +Artifacted measurements on a **1-billion-row, 128-dimensional table** show approximate-only +recall@10 of 0.824, 0.843, and 0.845 at `nprobe=16`, `32`, and `64`; exact reranking reaches +0.983 at `nprobe=32` with `refineFactor=50` (§11). These modes serve different goals: +approximate-only provides the latency floor, while exact reranking is the recommended +quality mode. + +--- + +## 1. Goals and Non-Goals + +### 1.1 Goals + +1. Keep authoritative vector values in a top-level base-table `VECTOR(D[, elementType])` column. +2. Store the vector index in the MDT, maintained by Hudi metadata-table commits, compaction, + and cleaning — no hidden or generated columns in base-table files. +3. Make candidate discovery cheap and targeted: probe a few clusters, scan contiguous key + ranges, score on compressed codes with a provable pruning bound. +4. Make results trustworthy: approximate math selects candidates; **exact** distance on + base-table vectors ranks them. +5. Stay transactionally consistent: snapshot-pinned reads, correct behavior under inserts, + updates, deletes, and clustering. +6. Be engine-neutral in design; Spark is the first implementation. +7. Maintain the index incrementally (no global rebuild for normal churn) and support + versioned, zero-downtime rebuilds. + +### 1.2 Non-Goals (initial landing) + +- ANN families beyond IVF + RaBitQ (e.g. HNSW, DiskANN). +- Filtered search (arbitrary predicate + kNN) as a first-class planned operation. +- Time-travel-consistent index reads for historical snapshots. +- Native non-Spark generation construction, GPU encoding, and workload-specific auto-tuning. + +### 1.3 Alternatives considered + +- **One index record per vector** is simple but creates billions of MDT records and excessive + write, compaction, and scan amplification; posting blocks preserve MDT ownership while + amortizing that overhead. +- **Dedicated index files in the table** permit specialized layouts but introduce a second + commit, cleaning, and snapshot protocol. MDT records reuse Hudi's existing transaction and + table-service machinery. +- **External sidecar indexes or vector databases** may offer richer serving features but lose + atomic Hudi snapshot semantics and require another storage system. They remain valid when + independent serving infrastructure is desired. +- **Native ANN libraries** improve local kernels but do not define durable object-store layout, + multi-writer maintenance, or snapshot visibility. They may be used behind the interfaces in + this RFC without changing the persisted contract. + +--- + +## 2. Architecture + + + +The design splits responsibilities the way Hudi already does between the data table and the +metadata table: + +```text +DATA TABLE (parquet/orc) METADATA TABLE (vector_index partition) + authoritative vectors + payload ←── the index: centroids, quantizer, posting blocks, + read only for final re-ranking cluster manifests, generation manifest + read for candidate generation +``` + +Each vector index is one MDT partition. RFC-109 adds no table properties, common writer- +dispatch changes, timeline-semantics changes, or behavior visible to non-vector readers or +writers. Every consistency mechanism is either a record in the vector-index MDT partition or +logic in the RFC-owned vector indexer and query planner. Creating an index: + +```sql +CREATE INDEX embedding_idx +ON products +USING VECTOR (embedding) +OPTIONS ( + 'vector.metric' = 'cosine', + 'vector.quantizer' = 'IVF_RABITQ', + 'vector.num_clusters'= '4096' +); +``` + +creates: + +```text +.hoodie/metadata/vector_index_embedding_idx/ +``` + +and does not add generated columns to the base-table schema. RFC-109 consumes rather than +redefines the RFC-99 vector contract: the source must be a top-level Hudi +`VECTOR(D[, elementType])`. The table schema is authoritative for `D` and element type; +index definitions and generation manifests repeat them only for integrity validation. + +The current RFC-99 storage backing is fixed-width bytes: Avro `FIXED` and Parquet +`FIXED_LEN_BYTE_ARRAY(D × elementWidth)`. Engine adapters expose idiomatic values—Spark uses +an annotated `ArrayType(FloatType|DoubleType|ByteType)`—and convert at the storage boundary. +This fixed width also avoids Parquet LIST repetition-level traversal during positional exact +fetch. A plain `ARRAY<FLOAT>` is not implicitly indexable and requires explicit migration or +backfill to `VECTOR(D)`; index creation neither reinterprets nor rewrites it. + +The query path uses the MDT first to discover candidates, then reads base-table vectors for +exact re-ranking: + +```text +query vector + → compare to centroids, pick nprobe clusters (in-memory, ms) + → MDT prefix-scan those clusters' posting blocks (targeted range reads) + → two-pass RaBitQ scoring, keep refineFactor·K best (bit math + error bounds) + → validate candidate freshness via Record Level Index (batched point lookups) + → fetch ONLY those rows from the base table by position (page-level reads) + → exact distance on real vectors → final top-K +``` + +--- + +## 3. IVF + RaBitQ Index Algorithm + +Every practical ANN index answers two questions: **where to look** (avoid scanning +everything) and **how to compare cheaply** (avoid full-precision math on what is scanned). + +### 3.1 Where to look: IVF routing + +Inverted File (IVF) indexing clusters vectors with KMeans into `numClusters` groups (e.g. +~4K–64K). Each vector belongs to its nearest centroid. A query compares against the centroids +only (thousands, not billions), selects the `nprobe` nearest clusters, and scans only those +clusters' entries. `nprobe` is the recall dial. + +IVF is the right fit for a lakehouse-resident index because a cluster's entries can be stored +**contiguously**, which maps directly onto sorted key ranges in Hudi's MDT (§4). Graph +indexes (HNSW) give excellent in-memory recall but require random traversal of the whole +graph, which fights columnar, immutable, object-store storage. + +### 3.2 How to compare cheaply: RaBitQ quantization + +Inside a probed cluster there are still thousands of full vectors. Quantization stores a +small *code* per vector plus a few correction scalars, so most comparisons run on compressed +codes and only the best few hundred candidates are re-checked against real vectors. + +RaBitQ is chosen over scalar (SQ), product (PQ), and plain binary quantization for four +reasons: + +1. **Unbiased estimator with a provable per-vector error bound.** Each code carries scalars + that turn a cheap bit-level dot product into an *unbiased* estimate of the true distance, + plus a bound on how wrong it can be. The bound enables **safe pruning**: skip a vector + only when even its best plausible distance cannot make top-K. +2. **No codebooks.** RaBitQ needs only a random rotation (a seed) and the centroids — both + tiny, both versioned in the index metadata. Nothing to retrain when data drifts. +3. **Tunable precision.** B = 1 bit/dim is a fast coarse filter; B = 4 bits/dim gives + near-SQ quality at ~8× less space. Both are used in a two-pass scan (§6.2). +4. **Metric-flexible.** One stored code serves L2, cosine, and dot-product; the metric is + applied at query time. + +The four ideas, precisely: + +- **Residual.** Store each vector as its difference from its centroid, `r = v − c`. + Residuals are small and centered, which is what lets few bits go far. +- **Rotation.** Apply one fixed random orthonormal rotation `R` (derived from a per-generation + seed) to everything first: `x = R·r`. This spreads information evenly across dimensions, + which is what makes the error bound hold for any data distribution. Only the seed is stored. +- **Code.** Quantize `x` to B bits per dimension, stored as **bit planes** — plane 0 holds + bit 0 of every dimension, plane 1 bit 1, etc. The top plane alone is a 1-bit sign sketch. + Scoring a plane against the (transformed) query is `AND`/`XOR` + `popcount` — a few CPU + instructions per 64 dimensions. +- **Factors.** A handful of small scalars per vector (residual norm, two rescale factors, an + error term, a centroid correction) that convert plane math into an unbiased distance + estimate plus its confidence interval. + +At query time the query vector is transformed the same way (`R·(q − c)` per probed cluster), +and the per-plane popcounts combine with the stored factors into the estimate and its bound. +The estimate builds the shortlist; exact base-table distances produce the final ranking (§6). + +### 3.3 Why this fits Hudi + +- Centroids are small enough to load at planning time: `K × D` floats (~12 MB for K=4096, + D=768). +- Codes are compact: a 1B × 128-dim float table's raw vectors are ~512 GB; the RaBitQ index + including keys and locators is ~136 GB, and the scanned portion per query is tens of MB. +- Posting keys are prefix-scannable by generation, cluster, and shard (§4). +- Quantizer state is stable: a seed and centroids, no per-generation learned codebook. +- Exact re-ranking preserves correctness for returned candidates. + +--- + +## 4. Metadata Table Storage Model: the Posting Block + +This section describes the core storage contribution of this RFC. + +### 4.1 The posting block + +A naive index would write one MDT record per indexed vector. At a billion rows that is a +billion MDT records per generation — prohibitive to write, compact, scan, and clean. + +Instead, the index sorts entries by `(cluster, fileGroup, rowPosition)` and packs +~1–4K vectors' worth of codes and metadata into a single MDT record — a **posting block** — +laid out column-wise (structure-of-arrays) so each scan pass touches only the columns it +needs: + +```text +POSTING BLOCK (~512 KB target, one MDT record) +┌───────────────────────────────────────────────┐ +│ S1 sign planes ← pass 1 touches this │ +│ S2 extra bit planes ← pass 2, survivors │ +│ S3 factor arrays ← both passes │ +│ S4 row locators ← finalists only │ +│ S5 dictionaries (file groups, partitions) │ +│ S6 record keys ← finalists only │ +└───────────────────────────────────────────────┘ +key: 0x10 | generation | clusterId | shardId | blockId +``` + +Consequences: + +- **~1000× fewer MDT records.** One block record replaces ~1–4K per-vector records, cutting + write amplification, compaction cost, and cleaner load by three orders of magnitude. +- **Contiguous cluster scans.** The binary key scheme makes "scan cluster 12345" a single + contiguous HFile range read rather than thousands of point lookups. +- **Pay only for promise.** Column-wise layout means pass 1 reads only sign planes + factors; + only survivors touch extra planes; only finalists touch locators and keys (§6.2). + +### 4.2 Row families + +The `vector_index_<name>` partition holds several record families under one binary-sorted key +scheme, so one prefix scan of a cluster returns its blocks and any fresh deltas together: + +| Key family | Cardinality | Purpose | +|---|---:|---| +| `__manifest__` | 1 | Active generation pointer and persisted-format version. | +| `M\|<generation>` | generations | Generation state, vector schema, bootstrap baseline, verified-contiguous frontier, chunk counts/checksums, and quantizer metadata. | +| `T\|<generation>\|<chunk>` | chunks per generation | Size-bounded chunks of the serialized `K × D` centroid matrix. | +| `C\|<generation>\|<cluster>` | K per generation | Cluster manifest: routing version, shard count, vector count, candidate file groups, and counters. | +| `P\|<generation>\|<cluster>\|<shard>\|<blockId>` | blocks per generation | **Posting block** (packed codes, factors, locators, keys). | +| `P\|...\|<DELTA>` | deltas | Small per-record delta records appended between compactions. | +| `F\|<generation>\|<dataInstant>` | data writes after baseline | Atomic proof that the generation incorporated one source data-write instant. | + +### 4.3 Posting shards + +Large clusters are split into posting shards so one hot cluster does not become one oversized Review Comment: Terminology: "shard" is overloaded — we already use file groups/buckets elsewhere and "shard" in the RLI context means something different. Add a clarifying line that a posting shard is purely an intra-cluster key-range split of the MDT prefix (nothing to do with RLI sharding or data-table bucketing). Consider renaming to something like "sub-range" to avoid the collision. ########## rfc/rfc-109/rfc-109.md: ########## @@ -0,0 +1,714 @@ +<!-- + 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. +--> + +# RFC-109: Native Vector Search Support in Apache Hudi + +## Proposers + +@chrevanthreddy + +## Approvers + +- TBD + +## Status + +Umbrella issue: [apache/hudi#19094](https://github.com/apache/hudi/issues/19094) + +Related: [apache/hudi#18676](https://github.com/apache/hudi/issues/18676) + +State: UNDER REVIEW + +--- + +## Table of Contents + +- [Abstract](#abstract) +- [1. Goals and Non-Goals](#1-goals-and-non-goals) +- [2. Architecture](#2-architecture) +- [3. IVF + RaBitQ Index Algorithm](#3-ivf--rabitq-index-algorithm) +- [4. Metadata Table Storage Model: the Posting Block](#4-metadata-table-storage-model-the-posting-block) +- [5. Bootstrap and Write Path](#5-bootstrap-and-write-path) +- [6. Read Path](#6-read-path) +- [7. Maintenance, Rebalancing, and Cleaner](#7-maintenance-rebalancing-and-cleaner) +- [8. Spark API Surface](#8-spark-api-surface) +- [9. Correctness, Compatibility, and Freshness](#9-correctness-compatibility-and-freshness) +- [10. Test Plan](#10-test-plan) +- [11. Benchmark Evidence, Rollout, and MVP Scope](#11-benchmark-evidence-rollout-and-mvp-scope) +- [12. References](#12-references) + +--- + +## Abstract + +This RFC proposes native approximate nearest-neighbor (ANN) vector search in Apache Hudi. +Tables increasingly carry embedding columns represented by Hudi's fixed-dimension +`VECTOR(D[, elementType])` logical type, and users want to ask *"find the K rows most +similar to this query vector"* — for semantic search, recommendations, RAG, and +deduplication — without copying data into a separate vector database. + +Today the only option on a Hudi table is a brute-force scan: read every vector, compute +every distance. That is correct but scales linearly with table size (tens of seconds at a +billion rows). This RFC adds an index so that vector queries read only a small, targeted +fraction of the index and the table, return results with high recall, and stay +transactionally consistent with the table under upserts and deletes — all with **no new +storage system**. The index lives in the Hudi Metadata Table (MDT), like Hudi's existing +record-level and secondary indexes, and is maintained by the same table services. + +The design combines three well-understood pieces — IVF clustering, RaBitQ quantization, and +exact re-ranking — with one storage innovation that makes them practical on an immutable, +columnar, object-store-resident lakehouse: + +> **The posting block.** Instead of one MDT record per indexed vector, the index packs Review Comment: this was key observation, from your runs. Thanks for being thorough. Can you dump any research numbers into an Appendix. So the data lives with the repo ########## rfc/rfc-109/rfc-109.md: ########## @@ -0,0 +1,714 @@ +<!-- + 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. +--> + +# RFC-109: Native Vector Search Support in Apache Hudi + +## Proposers + +@chrevanthreddy + +## Approvers + +- TBD + +## Status + +Umbrella issue: [apache/hudi#19094](https://github.com/apache/hudi/issues/19094) + +Related: [apache/hudi#18676](https://github.com/apache/hudi/issues/18676) + +State: UNDER REVIEW + +--- + +## Table of Contents + +- [Abstract](#abstract) +- [1. Goals and Non-Goals](#1-goals-and-non-goals) +- [2. Architecture](#2-architecture) +- [3. IVF + RaBitQ Index Algorithm](#3-ivf--rabitq-index-algorithm) +- [4. Metadata Table Storage Model: the Posting Block](#4-metadata-table-storage-model-the-posting-block) +- [5. Bootstrap and Write Path](#5-bootstrap-and-write-path) +- [6. Read Path](#6-read-path) +- [7. Maintenance, Rebalancing, and Cleaner](#7-maintenance-rebalancing-and-cleaner) +- [8. Spark API Surface](#8-spark-api-surface) +- [9. Correctness, Compatibility, and Freshness](#9-correctness-compatibility-and-freshness) +- [10. Test Plan](#10-test-plan) +- [11. Benchmark Evidence, Rollout, and MVP Scope](#11-benchmark-evidence-rollout-and-mvp-scope) +- [12. References](#12-references) + +--- + +## Abstract + +This RFC proposes native approximate nearest-neighbor (ANN) vector search in Apache Hudi. +Tables increasingly carry embedding columns represented by Hudi's fixed-dimension +`VECTOR(D[, elementType])` logical type, and users want to ask *"find the K rows most +similar to this query vector"* — for semantic search, recommendations, RAG, and +deduplication — without copying data into a separate vector database. + +Today the only option on a Hudi table is a brute-force scan: read every vector, compute +every distance. That is correct but scales linearly with table size (tens of seconds at a +billion rows). This RFC adds an index so that vector queries read only a small, targeted +fraction of the index and the table, return results with high recall, and stay +transactionally consistent with the table under upserts and deletes — all with **no new +storage system**. The index lives in the Hudi Metadata Table (MDT), like Hudi's existing +record-level and secondary indexes, and is maintained by the same table services. + +The design combines three well-understood pieces — IVF clustering, RaBitQ quantization, and +exact re-ranking — with one storage innovation that makes them practical on an immutable, +columnar, object-store-resident lakehouse: + +> **The posting block.** Instead of one MDT record per indexed vector, the index packs +> ~1–4K vectors into a single MDT record laid out column-wise (structure-of-arrays), keyed +> so that one IVF cluster forms one contiguous, prefix-scannable key range. This reduces MDT +> record count by roughly three orders of magnitude, turns "scan a cluster" into a single +> contiguous range read, and lets a query touch only the columns a given scan pass needs. + +The base table remains the source of truth for exact vector values. The MDT stores only +routing, pruning, and approximate-scoring metadata; final ranking always reads exact vectors +from the base table. + +Artifacted measurements on a **1-billion-row, 128-dimensional table** show approximate-only +recall@10 of 0.824, 0.843, and 0.845 at `nprobe=16`, `32`, and `64`; exact reranking reaches +0.983 at `nprobe=32` with `refineFactor=50` (§11). These modes serve different goals: +approximate-only provides the latency floor, while exact reranking is the recommended +quality mode. + +--- + +## 1. Goals and Non-Goals + +### 1.1 Goals + +1. Keep authoritative vector values in a top-level base-table `VECTOR(D[, elementType])` column. +2. Store the vector index in the MDT, maintained by Hudi metadata-table commits, compaction, + and cleaning — no hidden or generated columns in base-table files. +3. Make candidate discovery cheap and targeted: probe a few clusters, scan contiguous key + ranges, score on compressed codes with a provable pruning bound. +4. Make results trustworthy: approximate math selects candidates; **exact** distance on + base-table vectors ranks them. +5. Stay transactionally consistent: snapshot-pinned reads, correct behavior under inserts, + updates, deletes, and clustering. +6. Be engine-neutral in design; Spark is the first implementation. +7. Maintain the index incrementally (no global rebuild for normal churn) and support + versioned, zero-downtime rebuilds. + +### 1.2 Non-Goals (initial landing) + +- ANN families beyond IVF + RaBitQ (e.g. HNSW, DiskANN). +- Filtered search (arbitrary predicate + kNN) as a first-class planned operation. +- Time-travel-consistent index reads for historical snapshots. +- Native non-Spark generation construction, GPU encoding, and workload-specific auto-tuning. + +### 1.3 Alternatives considered + +- **One index record per vector** is simple but creates billions of MDT records and excessive + write, compaction, and scan amplification; posting blocks preserve MDT ownership while + amortizing that overhead. +- **Dedicated index files in the table** permit specialized layouts but introduce a second + commit, cleaning, and snapshot protocol. MDT records reuse Hudi's existing transaction and + table-service machinery. +- **External sidecar indexes or vector databases** may offer richer serving features but lose + atomic Hudi snapshot semantics and require another storage system. They remain valid when + independent serving infrastructure is desired. +- **Native ANN libraries** improve local kernels but do not define durable object-store layout, + multi-writer maintenance, or snapshot visibility. They may be used behind the interfaces in + this RFC without changing the persisted contract. + +--- + +## 2. Architecture + + + +The design splits responsibilities the way Hudi already does between the data table and the +metadata table: + +```text +DATA TABLE (parquet/orc) METADATA TABLE (vector_index partition) + authoritative vectors + payload ←── the index: centroids, quantizer, posting blocks, + read only for final re-ranking cluster manifests, generation manifest + read for candidate generation +``` + +Each vector index is one MDT partition. RFC-109 adds no table properties, common writer- +dispatch changes, timeline-semantics changes, or behavior visible to non-vector readers or +writers. Every consistency mechanism is either a record in the vector-index MDT partition or +logic in the RFC-owned vector indexer and query planner. Creating an index: + +```sql +CREATE INDEX embedding_idx +ON products +USING VECTOR (embedding) +OPTIONS ( + 'vector.metric' = 'cosine', + 'vector.quantizer' = 'IVF_RABITQ', + 'vector.num_clusters'= '4096' +); +``` + +creates: + +```text +.hoodie/metadata/vector_index_embedding_idx/ +``` + +and does not add generated columns to the base-table schema. RFC-109 consumes rather than +redefines the RFC-99 vector contract: the source must be a top-level Hudi +`VECTOR(D[, elementType])`. The table schema is authoritative for `D` and element type; +index definitions and generation manifests repeat them only for integrity validation. + +The current RFC-99 storage backing is fixed-width bytes: Avro `FIXED` and Parquet +`FIXED_LEN_BYTE_ARRAY(D × elementWidth)`. Engine adapters expose idiomatic values—Spark uses +an annotated `ArrayType(FloatType|DoubleType|ByteType)`—and convert at the storage boundary. +This fixed width also avoids Parquet LIST repetition-level traversal during positional exact +fetch. A plain `ARRAY<FLOAT>` is not implicitly indexable and requires explicit migration or +backfill to `VECTOR(D)`; index creation neither reinterprets nor rewrites it. + +The query path uses the MDT first to discover candidates, then reads base-table vectors for +exact re-ranking: + +```text +query vector + → compare to centroids, pick nprobe clusters (in-memory, ms) + → MDT prefix-scan those clusters' posting blocks (targeted range reads) + → two-pass RaBitQ scoring, keep refineFactor·K best (bit math + error bounds) + → validate candidate freshness via Record Level Index (batched point lookups) + → fetch ONLY those rows from the base table by position (page-level reads) + → exact distance on real vectors → final top-K +``` + +--- + +## 3. IVF + RaBitQ Index Algorithm + +Every practical ANN index answers two questions: **where to look** (avoid scanning +everything) and **how to compare cheaply** (avoid full-precision math on what is scanned). + +### 3.1 Where to look: IVF routing + +Inverted File (IVF) indexing clusters vectors with KMeans into `numClusters` groups (e.g. +~4K–64K). Each vector belongs to its nearest centroid. A query compares against the centroids +only (thousands, not billions), selects the `nprobe` nearest clusters, and scans only those +clusters' entries. `nprobe` is the recall dial. + +IVF is the right fit for a lakehouse-resident index because a cluster's entries can be stored +**contiguously**, which maps directly onto sorted key ranges in Hudi's MDT (§4). Graph +indexes (HNSW) give excellent in-memory recall but require random traversal of the whole +graph, which fights columnar, immutable, object-store storage. + +### 3.2 How to compare cheaply: RaBitQ quantization + +Inside a probed cluster there are still thousands of full vectors. Quantization stores a +small *code* per vector plus a few correction scalars, so most comparisons run on compressed +codes and only the best few hundred candidates are re-checked against real vectors. + +RaBitQ is chosen over scalar (SQ), product (PQ), and plain binary quantization for four +reasons: + +1. **Unbiased estimator with a provable per-vector error bound.** Each code carries scalars + that turn a cheap bit-level dot product into an *unbiased* estimate of the true distance, + plus a bound on how wrong it can be. The bound enables **safe pruning**: skip a vector + only when even its best plausible distance cannot make top-K. +2. **No codebooks.** RaBitQ needs only a random rotation (a seed) and the centroids — both + tiny, both versioned in the index metadata. Nothing to retrain when data drifts. +3. **Tunable precision.** B = 1 bit/dim is a fast coarse filter; B = 4 bits/dim gives + near-SQ quality at ~8× less space. Both are used in a two-pass scan (§6.2). +4. **Metric-flexible.** One stored code serves L2, cosine, and dot-product; the metric is + applied at query time. + +The four ideas, precisely: + +- **Residual.** Store each vector as its difference from its centroid, `r = v − c`. + Residuals are small and centered, which is what lets few bits go far. +- **Rotation.** Apply one fixed random orthonormal rotation `R` (derived from a per-generation + seed) to everything first: `x = R·r`. This spreads information evenly across dimensions, + which is what makes the error bound hold for any data distribution. Only the seed is stored. +- **Code.** Quantize `x` to B bits per dimension, stored as **bit planes** — plane 0 holds + bit 0 of every dimension, plane 1 bit 1, etc. The top plane alone is a 1-bit sign sketch. + Scoring a plane against the (transformed) query is `AND`/`XOR` + `popcount` — a few CPU + instructions per 64 dimensions. +- **Factors.** A handful of small scalars per vector (residual norm, two rescale factors, an + error term, a centroid correction) that convert plane math into an unbiased distance + estimate plus its confidence interval. + +At query time the query vector is transformed the same way (`R·(q − c)` per probed cluster), +and the per-plane popcounts combine with the stored factors into the estimate and its bound. +The estimate builds the shortlist; exact base-table distances produce the final ranking (§6). + +### 3.3 Why this fits Hudi + +- Centroids are small enough to load at planning time: `K × D` floats (~12 MB for K=4096, + D=768). +- Codes are compact: a 1B × 128-dim float table's raw vectors are ~512 GB; the RaBitQ index + including keys and locators is ~136 GB, and the scanned portion per query is tens of MB. +- Posting keys are prefix-scannable by generation, cluster, and shard (§4). +- Quantizer state is stable: a seed and centroids, no per-generation learned codebook. +- Exact re-ranking preserves correctness for returned candidates. + +--- + +## 4. Metadata Table Storage Model: the Posting Block + +This section describes the core storage contribution of this RFC. + +### 4.1 The posting block + +A naive index would write one MDT record per indexed vector. At a billion rows that is a +billion MDT records per generation — prohibitive to write, compact, scan, and clean. + +Instead, the index sorts entries by `(cluster, fileGroup, rowPosition)` and packs +~1–4K vectors' worth of codes and metadata into a single MDT record — a **posting block** — +laid out column-wise (structure-of-arrays) so each scan pass touches only the columns it +needs: + +```text +POSTING BLOCK (~512 KB target, one MDT record) +┌───────────────────────────────────────────────┐ +│ S1 sign planes ← pass 1 touches this │ +│ S2 extra bit planes ← pass 2, survivors │ +│ S3 factor arrays ← both passes │ +│ S4 row locators ← finalists only │ +│ S5 dictionaries (file groups, partitions) │ +│ S6 record keys ← finalists only │ +└───────────────────────────────────────────────┘ +key: 0x10 | generation | clusterId | shardId | blockId +``` + +Consequences: + +- **~1000× fewer MDT records.** One block record replaces ~1–4K per-vector records, cutting + write amplification, compaction cost, and cleaner load by three orders of magnitude. +- **Contiguous cluster scans.** The binary key scheme makes "scan cluster 12345" a single + contiguous HFile range read rather than thousands of point lookups. +- **Pay only for promise.** Column-wise layout means pass 1 reads only sign planes + factors; + only survivors touch extra planes; only finalists touch locators and keys (§6.2). + +### 4.2 Row families + +The `vector_index_<name>` partition holds several record families under one binary-sorted key +scheme, so one prefix scan of a cluster returns its blocks and any fresh deltas together: + +| Key family | Cardinality | Purpose | +|---|---:|---| +| `__manifest__` | 1 | Active generation pointer and persisted-format version. | +| `M\|<generation>` | generations | Generation state, vector schema, bootstrap baseline, verified-contiguous frontier, chunk counts/checksums, and quantizer metadata. | +| `T\|<generation>\|<chunk>` | chunks per generation | Size-bounded chunks of the serialized `K × D` centroid matrix. | +| `C\|<generation>\|<cluster>` | K per generation | Cluster manifest: routing version, shard count, vector count, candidate file groups, and counters. | +| `P\|<generation>\|<cluster>\|<shard>\|<blockId>` | blocks per generation | **Posting block** (packed codes, factors, locators, keys). | +| `P\|...\|<DELTA>` | deltas | Small per-record delta records appended between compactions. | +| `F\|<generation>\|<dataInstant>` | data writes after baseline | Atomic proof that the generation incorporated one source data-write instant. | + +### 4.3 Posting shards + +Large clusters are split into posting shards so one hot cluster does not become one oversized +prefix range. The cluster manifest stores `shardCount` and `routingVersion`; writers compute +`shardId = hash(record_key) % shardCount`. Changing `shardCount` changes every key's mapping, +so maintenance must rewrite the whole cluster under a new routing version and publish the +manifest change atomically with that rewrite. Large remaps use a new generation rather than +mixing routing versions in one cluster range. Within a shard, entries are packed by `blockId`. + +### 4.4 Delta records + +Between compactions, per-commit vector writes append small **delta records** (`blockId` +marked `DELTA`) at the end of the same cluster key range. Because they share the prefix, a +single cluster prefix scan sees packed blocks and fresh deltas in one pass (§6, §7). + +### 4.5 Generation model + +A generation is a consistent set of centroid, quantizer, cluster, and posting-block metadata: + +```text +__manifest__ -> active generation id + persisted-format version +M|<gen> -> generation metadata + schema + verified frontier + chunk integrity +T|<gen>|... -> size-bounded centroid chunks +C|<gen>|... -> cluster manifests +P|<gen>|... -> posting blocks + deltas +F|<gen>|... -> incorporated data-write markers +``` + +The builder allocates a generation id once and writes deterministic keys, so retrying a +partially completed build overwrites the same records. A `BUILDING` generation is invisible +to readers; bootstrap validates expected chunks, checksums, cluster/block counts, schema, and +memory budgets before one MDT commit changes it to `ACTIVE` and flips `__manifest__`. +Abandoned or invalid `BUILDING` generations are never activated and may be garbage-collected. +Old generations become `RETIRED` and remain until no retained snapshot or pinned reader needs +them. Generation is the only independent version axis: centroid, quantizer, routing, and +posting changes that cannot be published atomically in place create a new generation. + +--- + +## 5. Bootstrap and Write Path + + + +### 5.1 Spark bootstrap + +Bootstrap and full rebuild are Spark-only in v1. They build an invisible generation from one +pinned table snapshot: + +```text +1. Read latest file slices; extract key, partition, file group, base instant, row position, + and vector bytes; train IVF centroids from a bounded sample. +2. Validate K × D × elementWidth and driver/executor memory budgets before materialization. +3. Broadcast centroids; assign each vector, encode RaBitQ metadata, sort by + (cluster, fileGroup, rowPosition), and pack posting blocks. +4. Write M|, size-bounded T| centroid chunks, C|, and P| records under deterministic keys. +5. Validate chunk counts/checksums and posting counts, then atomically activate the generation. +``` + +The training sample satisfies percentage and per-cluster floors, for example +`min(N, max(1M, 256*K, min(10M, 0.5%–1% of N)))`. Chunking keeps individual MDT records +bounded; readers reconstruct the matrix only after validating manifest integrity metadata. + +### 5.2 Incremental inserts and vector updates + +For each relevant row, the metadata writer's vector hook computes the cluster, shard, RaBitQ +code, factors, and base-table location and appends a posting delta. If an update moves a +record, delta supersession and query-time RLI arbitration hide the stale entry (§6.3). +The hook also emits the commit's freshness marker in the same MDT commit as its vector deltas. +This uses the existing per-data-commit `Indexer.buildUpdate` dispatch; `VectorIndexer` returns +an `F|...` record even without deltas, so its update is never empty and needs no new seam. + +### 5.3 Non-vector updates and deletes + +A non-vector update can retain its code and placement; a moved locator is re-resolved through +RLI and refreshed by maintenance before the old slice is cleaned. Deletes require no posting +delta because RLI arbitration removes deleted candidates. The hook still emits one freshness +marker when a data write produces no vector records—including schema-only or allowed-empty +commits—so an empty update cannot create a permanent false-stale gap. + +### 5.4 Engine support + +Snapshot extraction and centroid training are engine-specific; routing, encoding, record +construction, freshness markers, and publication are common contracts. Spark supplies +bootstrap/rebuild in v1. Feature-aware Spark, Flink, and Java writers may maintain an active +generation through the common metadata-writer hook; unaware writers keep standard MDT +layer-2 skip semantics, with their commits detected by the query-time freshness gate (§9). + +### 5.5 Multi-writer rebuild and catch-up + +Rebuilds are replay-only: writers continue updating the active generation and never dual-write Review Comment: Concurrency behavior is currently scattered across this section and §9.1, and it's hard to convince ourselves it's complete. Can we add a table spelling out the scenarios across OCC and NBCC explicitly? Rows to cover : (a) two OCC writers, both vector-aware; (b) OCC writer + concurrent rebuild; (c) NBCC/multi-writer deltacommits with vector deltas landing out of order; (d) vector-aware + legacy writer interleaved; (e) rebuild racing clustering/replacecommit; (f) async MDT compaction racing the vector hook. For each: what wins, what the marker frontier does, and what the reader observes (fresh / stale-gap / FAIL). Even if the answers all fall out of TransactionManager + the metadata lock + markers, having the matrix in one place will save every reviewer re-deriving it. I am happy to take this on myself as well and push a commit. please lmk ########## rfc/rfc-109/rfc-109.md: ########## @@ -0,0 +1,714 @@ +<!-- + 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. +--> + +# RFC-109: Native Vector Search Support in Apache Hudi + +## Proposers + +@chrevanthreddy + +## Approvers + +- TBD + +## Status + +Umbrella issue: [apache/hudi#19094](https://github.com/apache/hudi/issues/19094) + +Related: [apache/hudi#18676](https://github.com/apache/hudi/issues/18676) + +State: UNDER REVIEW + +--- + +## Table of Contents + +- [Abstract](#abstract) +- [1. Goals and Non-Goals](#1-goals-and-non-goals) +- [2. Architecture](#2-architecture) +- [3. IVF + RaBitQ Index Algorithm](#3-ivf--rabitq-index-algorithm) +- [4. Metadata Table Storage Model: the Posting Block](#4-metadata-table-storage-model-the-posting-block) +- [5. Bootstrap and Write Path](#5-bootstrap-and-write-path) +- [6. Read Path](#6-read-path) +- [7. Maintenance, Rebalancing, and Cleaner](#7-maintenance-rebalancing-and-cleaner) +- [8. Spark API Surface](#8-spark-api-surface) +- [9. Correctness, Compatibility, and Freshness](#9-correctness-compatibility-and-freshness) +- [10. Test Plan](#10-test-plan) +- [11. Benchmark Evidence, Rollout, and MVP Scope](#11-benchmark-evidence-rollout-and-mvp-scope) +- [12. References](#12-references) + +--- + +## Abstract + +This RFC proposes native approximate nearest-neighbor (ANN) vector search in Apache Hudi. +Tables increasingly carry embedding columns represented by Hudi's fixed-dimension +`VECTOR(D[, elementType])` logical type, and users want to ask *"find the K rows most +similar to this query vector"* — for semantic search, recommendations, RAG, and +deduplication — without copying data into a separate vector database. + +Today the only option on a Hudi table is a brute-force scan: read every vector, compute +every distance. That is correct but scales linearly with table size (tens of seconds at a +billion rows). This RFC adds an index so that vector queries read only a small, targeted +fraction of the index and the table, return results with high recall, and stay +transactionally consistent with the table under upserts and deletes — all with **no new +storage system**. The index lives in the Hudi Metadata Table (MDT), like Hudi's existing +record-level and secondary indexes, and is maintained by the same table services. + +The design combines three well-understood pieces — IVF clustering, RaBitQ quantization, and +exact re-ranking — with one storage innovation that makes them practical on an immutable, +columnar, object-store-resident lakehouse: + +> **The posting block.** Instead of one MDT record per indexed vector, the index packs +> ~1–4K vectors into a single MDT record laid out column-wise (structure-of-arrays), keyed +> so that one IVF cluster forms one contiguous, prefix-scannable key range. This reduces MDT +> record count by roughly three orders of magnitude, turns "scan a cluster" into a single +> contiguous range read, and lets a query touch only the columns a given scan pass needs. + +The base table remains the source of truth for exact vector values. The MDT stores only +routing, pruning, and approximate-scoring metadata; final ranking always reads exact vectors +from the base table. + +Artifacted measurements on a **1-billion-row, 128-dimensional table** show approximate-only +recall@10 of 0.824, 0.843, and 0.845 at `nprobe=16`, `32`, and `64`; exact reranking reaches +0.983 at `nprobe=32` with `refineFactor=50` (§11). These modes serve different goals: +approximate-only provides the latency floor, while exact reranking is the recommended +quality mode. + +--- + +## 1. Goals and Non-Goals + +### 1.1 Goals + +1. Keep authoritative vector values in a top-level base-table `VECTOR(D[, elementType])` column. +2. Store the vector index in the MDT, maintained by Hudi metadata-table commits, compaction, + and cleaning — no hidden or generated columns in base-table files. +3. Make candidate discovery cheap and targeted: probe a few clusters, scan contiguous key + ranges, score on compressed codes with a provable pruning bound. +4. Make results trustworthy: approximate math selects candidates; **exact** distance on + base-table vectors ranks them. +5. Stay transactionally consistent: snapshot-pinned reads, correct behavior under inserts, + updates, deletes, and clustering. +6. Be engine-neutral in design; Spark is the first implementation. +7. Maintain the index incrementally (no global rebuild for normal churn) and support + versioned, zero-downtime rebuilds. + +### 1.2 Non-Goals (initial landing) + +- ANN families beyond IVF + RaBitQ (e.g. HNSW, DiskANN). +- Filtered search (arbitrary predicate + kNN) as a first-class planned operation. +- Time-travel-consistent index reads for historical snapshots. +- Native non-Spark generation construction, GPU encoding, and workload-specific auto-tuning. + +### 1.3 Alternatives considered + +- **One index record per vector** is simple but creates billions of MDT records and excessive + write, compaction, and scan amplification; posting blocks preserve MDT ownership while + amortizing that overhead. +- **Dedicated index files in the table** permit specialized layouts but introduce a second + commit, cleaning, and snapshot protocol. MDT records reuse Hudi's existing transaction and + table-service machinery. +- **External sidecar indexes or vector databases** may offer richer serving features but lose + atomic Hudi snapshot semantics and require another storage system. They remain valid when + independent serving infrastructure is desired. +- **Native ANN libraries** improve local kernels but do not define durable object-store layout, + multi-writer maintenance, or snapshot visibility. They may be used behind the interfaces in + this RFC without changing the persisted contract. + +--- + +## 2. Architecture + + + +The design splits responsibilities the way Hudi already does between the data table and the +metadata table: + +```text +DATA TABLE (parquet/orc) METADATA TABLE (vector_index partition) + authoritative vectors + payload ←── the index: centroids, quantizer, posting blocks, + read only for final re-ranking cluster manifests, generation manifest + read for candidate generation +``` + +Each vector index is one MDT partition. RFC-109 adds no table properties, common writer- +dispatch changes, timeline-semantics changes, or behavior visible to non-vector readers or +writers. Every consistency mechanism is either a record in the vector-index MDT partition or +logic in the RFC-owned vector indexer and query planner. Creating an index: + +```sql +CREATE INDEX embedding_idx +ON products +USING VECTOR (embedding) +OPTIONS ( + 'vector.metric' = 'cosine', + 'vector.quantizer' = 'IVF_RABITQ', + 'vector.num_clusters'= '4096' +); +``` + +creates: + +```text +.hoodie/metadata/vector_index_embedding_idx/ +``` + +and does not add generated columns to the base-table schema. RFC-109 consumes rather than +redefines the RFC-99 vector contract: the source must be a top-level Hudi +`VECTOR(D[, elementType])`. The table schema is authoritative for `D` and element type; +index definitions and generation manifests repeat them only for integrity validation. + +The current RFC-99 storage backing is fixed-width bytes: Avro `FIXED` and Parquet +`FIXED_LEN_BYTE_ARRAY(D × elementWidth)`. Engine adapters expose idiomatic values—Spark uses +an annotated `ArrayType(FloatType|DoubleType|ByteType)`—and convert at the storage boundary. +This fixed width also avoids Parquet LIST repetition-level traversal during positional exact +fetch. A plain `ARRAY<FLOAT>` is not implicitly indexable and requires explicit migration or +backfill to `VECTOR(D)`; index creation neither reinterprets nor rewrites it. + +The query path uses the MDT first to discover candidates, then reads base-table vectors for +exact re-ranking: + +```text +query vector + → compare to centroids, pick nprobe clusters (in-memory, ms) + → MDT prefix-scan those clusters' posting blocks (targeted range reads) + → two-pass RaBitQ scoring, keep refineFactor·K best (bit math + error bounds) + → validate candidate freshness via Record Level Index (batched point lookups) + → fetch ONLY those rows from the base table by position (page-level reads) + → exact distance on real vectors → final top-K +``` + +--- + +## 3. IVF + RaBitQ Index Algorithm + +Every practical ANN index answers two questions: **where to look** (avoid scanning +everything) and **how to compare cheaply** (avoid full-precision math on what is scanned). + +### 3.1 Where to look: IVF routing + +Inverted File (IVF) indexing clusters vectors with KMeans into `numClusters` groups (e.g. +~4K–64K). Each vector belongs to its nearest centroid. A query compares against the centroids +only (thousands, not billions), selects the `nprobe` nearest clusters, and scans only those +clusters' entries. `nprobe` is the recall dial. + +IVF is the right fit for a lakehouse-resident index because a cluster's entries can be stored +**contiguously**, which maps directly onto sorted key ranges in Hudi's MDT (§4). Graph +indexes (HNSW) give excellent in-memory recall but require random traversal of the whole +graph, which fights columnar, immutable, object-store storage. + +### 3.2 How to compare cheaply: RaBitQ quantization + +Inside a probed cluster there are still thousands of full vectors. Quantization stores a +small *code* per vector plus a few correction scalars, so most comparisons run on compressed +codes and only the best few hundred candidates are re-checked against real vectors. + +RaBitQ is chosen over scalar (SQ), product (PQ), and plain binary quantization for four +reasons: + +1. **Unbiased estimator with a provable per-vector error bound.** Each code carries scalars + that turn a cheap bit-level dot product into an *unbiased* estimate of the true distance, + plus a bound on how wrong it can be. The bound enables **safe pruning**: skip a vector + only when even its best plausible distance cannot make top-K. +2. **No codebooks.** RaBitQ needs only a random rotation (a seed) and the centroids — both + tiny, both versioned in the index metadata. Nothing to retrain when data drifts. +3. **Tunable precision.** B = 1 bit/dim is a fast coarse filter; B = 4 bits/dim gives + near-SQ quality at ~8× less space. Both are used in a two-pass scan (§6.2). +4. **Metric-flexible.** One stored code serves L2, cosine, and dot-product; the metric is + applied at query time. + +The four ideas, precisely: + +- **Residual.** Store each vector as its difference from its centroid, `r = v − c`. + Residuals are small and centered, which is what lets few bits go far. +- **Rotation.** Apply one fixed random orthonormal rotation `R` (derived from a per-generation + seed) to everything first: `x = R·r`. This spreads information evenly across dimensions, + which is what makes the error bound hold for any data distribution. Only the seed is stored. +- **Code.** Quantize `x` to B bits per dimension, stored as **bit planes** — plane 0 holds + bit 0 of every dimension, plane 1 bit 1, etc. The top plane alone is a 1-bit sign sketch. + Scoring a plane against the (transformed) query is `AND`/`XOR` + `popcount` — a few CPU + instructions per 64 dimensions. +- **Factors.** A handful of small scalars per vector (residual norm, two rescale factors, an + error term, a centroid correction) that convert plane math into an unbiased distance + estimate plus its confidence interval. + +At query time the query vector is transformed the same way (`R·(q − c)` per probed cluster), +and the per-plane popcounts combine with the stored factors into the estimate and its bound. +The estimate builds the shortlist; exact base-table distances produce the final ranking (§6). + +### 3.3 Why this fits Hudi + +- Centroids are small enough to load at planning time: `K × D` floats (~12 MB for K=4096, + D=768). +- Codes are compact: a 1B × 128-dim float table's raw vectors are ~512 GB; the RaBitQ index + including keys and locators is ~136 GB, and the scanned portion per query is tens of MB. +- Posting keys are prefix-scannable by generation, cluster, and shard (§4). +- Quantizer state is stable: a seed and centroids, no per-generation learned codebook. +- Exact re-ranking preserves correctness for returned candidates. + +--- + +## 4. Metadata Table Storage Model: the Posting Block + +This section describes the core storage contribution of this RFC. + +### 4.1 The posting block + +A naive index would write one MDT record per indexed vector. At a billion rows that is a +billion MDT records per generation — prohibitive to write, compact, scan, and clean. + +Instead, the index sorts entries by `(cluster, fileGroup, rowPosition)` and packs +~1–4K vectors' worth of codes and metadata into a single MDT record — a **posting block** — +laid out column-wise (structure-of-arrays) so each scan pass touches only the columns it +needs: + +```text +POSTING BLOCK (~512 KB target, one MDT record) +┌───────────────────────────────────────────────┐ +│ S1 sign planes ← pass 1 touches this │ +│ S2 extra bit planes ← pass 2, survivors │ +│ S3 factor arrays ← both passes │ +│ S4 row locators ← finalists only │ +│ S5 dictionaries (file groups, partitions) │ +│ S6 record keys ← finalists only │ +└───────────────────────────────────────────────┘ +key: 0x10 | generation | clusterId | shardId | blockId +``` + +Consequences: + +- **~1000× fewer MDT records.** One block record replaces ~1–4K per-vector records, cutting + write amplification, compaction cost, and cleaner load by three orders of magnitude. +- **Contiguous cluster scans.** The binary key scheme makes "scan cluster 12345" a single + contiguous HFile range read rather than thousands of point lookups. +- **Pay only for promise.** Column-wise layout means pass 1 reads only sign planes + factors; + only survivors touch extra planes; only finalists touch locators and keys (§6.2). + +### 4.2 Row families + +The `vector_index_<name>` partition holds several record families under one binary-sorted key +scheme, so one prefix scan of a cluster returns its blocks and any fresh deltas together: + +| Key family | Cardinality | Purpose | +|---|---:|---| +| `__manifest__` | 1 | Active generation pointer and persisted-format version. | +| `M\|<generation>` | generations | Generation state, vector schema, bootstrap baseline, verified-contiguous frontier, chunk counts/checksums, and quantizer metadata. | +| `T\|<generation>\|<chunk>` | chunks per generation | Size-bounded chunks of the serialized `K × D` centroid matrix. | +| `C\|<generation>\|<cluster>` | K per generation | Cluster manifest: routing version, shard count, vector count, candidate file groups, and counters. | +| `P\|<generation>\|<cluster>\|<shard>\|<blockId>` | blocks per generation | **Posting block** (packed codes, factors, locators, keys). | +| `P\|...\|<DELTA>` | deltas | Small per-record delta records appended between compactions. | +| `F\|<generation>\|<dataInstant>` | data writes after baseline | Atomic proof that the generation incorporated one source data-write instant. | + +### 4.3 Posting shards + +Large clusters are split into posting shards so one hot cluster does not become one oversized +prefix range. The cluster manifest stores `shardCount` and `routingVersion`; writers compute +`shardId = hash(record_key) % shardCount`. Changing `shardCount` changes every key's mapping, +so maintenance must rewrite the whole cluster under a new routing version and publish the +manifest change atomically with that rewrite. Large remaps use a new generation rather than +mixing routing versions in one cluster range. Within a shard, entries are packed by `blockId`. + +### 4.4 Delta records + +Between compactions, per-commit vector writes append small **delta records** (`blockId` +marked `DELTA`) at the end of the same cluster key range. Because they share the prefix, a +single cluster prefix scan sees packed blocks and fresh deltas in one pass (§6, §7). + +### 4.5 Generation model + +A generation is a consistent set of centroid, quantizer, cluster, and posting-block metadata: + +```text +__manifest__ -> active generation id + persisted-format version +M|<gen> -> generation metadata + schema + verified frontier + chunk integrity +T|<gen>|... -> size-bounded centroid chunks +C|<gen>|... -> cluster manifests +P|<gen>|... -> posting blocks + deltas +F|<gen>|... -> incorporated data-write markers +``` + +The builder allocates a generation id once and writes deterministic keys, so retrying a +partially completed build overwrites the same records. A `BUILDING` generation is invisible +to readers; bootstrap validates expected chunks, checksums, cluster/block counts, schema, and +memory budgets before one MDT commit changes it to `ACTIVE` and flips `__manifest__`. +Abandoned or invalid `BUILDING` generations are never activated and may be garbage-collected. +Old generations become `RETIRED` and remain until no retained snapshot or pinned reader needs +them. Generation is the only independent version axis: centroid, quantizer, routing, and +posting changes that cannot be published atomically in place create a new generation. + +--- + +## 5. Bootstrap and Write Path + + + +### 5.1 Spark bootstrap + +Bootstrap and full rebuild are Spark-only in v1. They build an invisible generation from one +pinned table snapshot: + +```text +1. Read latest file slices; extract key, partition, file group, base instant, row position, + and vector bytes; train IVF centroids from a bounded sample. +2. Validate K × D × elementWidth and driver/executor memory budgets before materialization. +3. Broadcast centroids; assign each vector, encode RaBitQ metadata, sort by + (cluster, fileGroup, rowPosition), and pack posting blocks. +4. Write M|, size-bounded T| centroid chunks, C|, and P| records under deterministic keys. +5. Validate chunk counts/checksums and posting counts, then atomically activate the generation. +``` + +The training sample satisfies percentage and per-cluster floors, for example +`min(N, max(1M, 256*K, min(10M, 0.5%–1% of N)))`. Chunking keeps individual MDT records +bounded; readers reconstruct the matrix only after validating manifest integrity metadata. + +### 5.2 Incremental inserts and vector updates + +For each relevant row, the metadata writer's vector hook computes the cluster, shard, RaBitQ +code, factors, and base-table location and appends a posting delta. If an update moves a +record, delta supersession and query-time RLI arbitration hide the stale entry (§6.3). +The hook also emits the commit's freshness marker in the same MDT commit as its vector deltas. +This uses the existing per-data-commit `Indexer.buildUpdate` dispatch; `VectorIndexer` returns +an `F|...` record even without deltas, so its update is never empty and needs no new seam. + +### 5.3 Non-vector updates and deletes + +A non-vector update can retain its code and placement; a moved locator is re-resolved through Review Comment: This section assumes we can tell a "non-vector update" apart from a vector update — but how do we actually detect that the vector column is unchanged? Needs fleshing out or a way to safely deal with it with slower perf (treat as vector update when in doubt?). ########## rfc/rfc-109/rfc-109.md: ########## @@ -0,0 +1,714 @@ +<!-- + 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. +--> + +# RFC-109: Native Vector Search Support in Apache Hudi + +## Proposers + +@chrevanthreddy + +## Approvers + +- TBD + +## Status + +Umbrella issue: [apache/hudi#19094](https://github.com/apache/hudi/issues/19094) + +Related: [apache/hudi#18676](https://github.com/apache/hudi/issues/18676) + +State: UNDER REVIEW + +--- + +## Table of Contents + +- [Abstract](#abstract) +- [1. Goals and Non-Goals](#1-goals-and-non-goals) +- [2. Architecture](#2-architecture) +- [3. IVF + RaBitQ Index Algorithm](#3-ivf--rabitq-index-algorithm) +- [4. Metadata Table Storage Model: the Posting Block](#4-metadata-table-storage-model-the-posting-block) +- [5. Bootstrap and Write Path](#5-bootstrap-and-write-path) +- [6. Read Path](#6-read-path) +- [7. Maintenance, Rebalancing, and Cleaner](#7-maintenance-rebalancing-and-cleaner) +- [8. Spark API Surface](#8-spark-api-surface) +- [9. Correctness, Compatibility, and Freshness](#9-correctness-compatibility-and-freshness) +- [10. Test Plan](#10-test-plan) +- [11. Benchmark Evidence, Rollout, and MVP Scope](#11-benchmark-evidence-rollout-and-mvp-scope) +- [12. References](#12-references) + +--- + +## Abstract + +This RFC proposes native approximate nearest-neighbor (ANN) vector search in Apache Hudi. +Tables increasingly carry embedding columns represented by Hudi's fixed-dimension +`VECTOR(D[, elementType])` logical type, and users want to ask *"find the K rows most +similar to this query vector"* — for semantic search, recommendations, RAG, and +deduplication — without copying data into a separate vector database. + +Today the only option on a Hudi table is a brute-force scan: read every vector, compute +every distance. That is correct but scales linearly with table size (tens of seconds at a +billion rows). This RFC adds an index so that vector queries read only a small, targeted +fraction of the index and the table, return results with high recall, and stay +transactionally consistent with the table under upserts and deletes — all with **no new +storage system**. The index lives in the Hudi Metadata Table (MDT), like Hudi's existing +record-level and secondary indexes, and is maintained by the same table services. + +The design combines three well-understood pieces — IVF clustering, RaBitQ quantization, and +exact re-ranking — with one storage innovation that makes them practical on an immutable, +columnar, object-store-resident lakehouse: + +> **The posting block.** Instead of one MDT record per indexed vector, the index packs +> ~1–4K vectors into a single MDT record laid out column-wise (structure-of-arrays), keyed +> so that one IVF cluster forms one contiguous, prefix-scannable key range. This reduces MDT +> record count by roughly three orders of magnitude, turns "scan a cluster" into a single +> contiguous range read, and lets a query touch only the columns a given scan pass needs. + +The base table remains the source of truth for exact vector values. The MDT stores only +routing, pruning, and approximate-scoring metadata; final ranking always reads exact vectors +from the base table. + +Artifacted measurements on a **1-billion-row, 128-dimensional table** show approximate-only +recall@10 of 0.824, 0.843, and 0.845 at `nprobe=16`, `32`, and `64`; exact reranking reaches +0.983 at `nprobe=32` with `refineFactor=50` (§11). These modes serve different goals: +approximate-only provides the latency floor, while exact reranking is the recommended +quality mode. + +--- + +## 1. Goals and Non-Goals + +### 1.1 Goals + +1. Keep authoritative vector values in a top-level base-table `VECTOR(D[, elementType])` column. +2. Store the vector index in the MDT, maintained by Hudi metadata-table commits, compaction, + and cleaning — no hidden or generated columns in base-table files. +3. Make candidate discovery cheap and targeted: probe a few clusters, scan contiguous key + ranges, score on compressed codes with a provable pruning bound. +4. Make results trustworthy: approximate math selects candidates; **exact** distance on + base-table vectors ranks them. +5. Stay transactionally consistent: snapshot-pinned reads, correct behavior under inserts, + updates, deletes, and clustering. +6. Be engine-neutral in design; Spark is the first implementation. +7. Maintain the index incrementally (no global rebuild for normal churn) and support + versioned, zero-downtime rebuilds. + +### 1.2 Non-Goals (initial landing) + +- ANN families beyond IVF + RaBitQ (e.g. HNSW, DiskANN). +- Filtered search (arbitrary predicate + kNN) as a first-class planned operation. +- Time-travel-consistent index reads for historical snapshots. +- Native non-Spark generation construction, GPU encoding, and workload-specific auto-tuning. + +### 1.3 Alternatives considered + +- **One index record per vector** is simple but creates billions of MDT records and excessive + write, compaction, and scan amplification; posting blocks preserve MDT ownership while + amortizing that overhead. +- **Dedicated index files in the table** permit specialized layouts but introduce a second + commit, cleaning, and snapshot protocol. MDT records reuse Hudi's existing transaction and + table-service machinery. +- **External sidecar indexes or vector databases** may offer richer serving features but lose + atomic Hudi snapshot semantics and require another storage system. They remain valid when + independent serving infrastructure is desired. +- **Native ANN libraries** improve local kernels but do not define durable object-store layout, + multi-writer maintenance, or snapshot visibility. They may be used behind the interfaces in + this RFC without changing the persisted contract. + +--- + +## 2. Architecture + + + +The design splits responsibilities the way Hudi already does between the data table and the +metadata table: + +```text +DATA TABLE (parquet/orc) METADATA TABLE (vector_index partition) + authoritative vectors + payload ←── the index: centroids, quantizer, posting blocks, + read only for final re-ranking cluster manifests, generation manifest + read for candidate generation +``` + +Each vector index is one MDT partition. RFC-109 adds no table properties, common writer- +dispatch changes, timeline-semantics changes, or behavior visible to non-vector readers or +writers. Every consistency mechanism is either a record in the vector-index MDT partition or +logic in the RFC-owned vector indexer and query planner. Creating an index: + +```sql +CREATE INDEX embedding_idx +ON products +USING VECTOR (embedding) +OPTIONS ( + 'vector.metric' = 'cosine', + 'vector.quantizer' = 'IVF_RABITQ', + 'vector.num_clusters'= '4096' +); +``` + +creates: + +```text +.hoodie/metadata/vector_index_embedding_idx/ +``` + +and does not add generated columns to the base-table schema. RFC-109 consumes rather than +redefines the RFC-99 vector contract: the source must be a top-level Hudi +`VECTOR(D[, elementType])`. The table schema is authoritative for `D` and element type; +index definitions and generation manifests repeat them only for integrity validation. + +The current RFC-99 storage backing is fixed-width bytes: Avro `FIXED` and Parquet +`FIXED_LEN_BYTE_ARRAY(D × elementWidth)`. Engine adapters expose idiomatic values—Spark uses +an annotated `ArrayType(FloatType|DoubleType|ByteType)`—and convert at the storage boundary. +This fixed width also avoids Parquet LIST repetition-level traversal during positional exact +fetch. A plain `ARRAY<FLOAT>` is not implicitly indexable and requires explicit migration or +backfill to `VECTOR(D)`; index creation neither reinterprets nor rewrites it. + +The query path uses the MDT first to discover candidates, then reads base-table vectors for +exact re-ranking: + +```text +query vector + → compare to centroids, pick nprobe clusters (in-memory, ms) + → MDT prefix-scan those clusters' posting blocks (targeted range reads) + → two-pass RaBitQ scoring, keep refineFactor·K best (bit math + error bounds) + → validate candidate freshness via Record Level Index (batched point lookups) + → fetch ONLY those rows from the base table by position (page-level reads) + → exact distance on real vectors → final top-K +``` + +--- + +## 3. IVF + RaBitQ Index Algorithm + +Every practical ANN index answers two questions: **where to look** (avoid scanning +everything) and **how to compare cheaply** (avoid full-precision math on what is scanned). + +### 3.1 Where to look: IVF routing + +Inverted File (IVF) indexing clusters vectors with KMeans into `numClusters` groups (e.g. +~4K–64K). Each vector belongs to its nearest centroid. A query compares against the centroids +only (thousands, not billions), selects the `nprobe` nearest clusters, and scans only those +clusters' entries. `nprobe` is the recall dial. + +IVF is the right fit for a lakehouse-resident index because a cluster's entries can be stored +**contiguously**, which maps directly onto sorted key ranges in Hudi's MDT (§4). Graph +indexes (HNSW) give excellent in-memory recall but require random traversal of the whole +graph, which fights columnar, immutable, object-store storage. + +### 3.2 How to compare cheaply: RaBitQ quantization + +Inside a probed cluster there are still thousands of full vectors. Quantization stores a +small *code* per vector plus a few correction scalars, so most comparisons run on compressed +codes and only the best few hundred candidates are re-checked against real vectors. + +RaBitQ is chosen over scalar (SQ), product (PQ), and plain binary quantization for four +reasons: + +1. **Unbiased estimator with a provable per-vector error bound.** Each code carries scalars + that turn a cheap bit-level dot product into an *unbiased* estimate of the true distance, + plus a bound on how wrong it can be. The bound enables **safe pruning**: skip a vector + only when even its best plausible distance cannot make top-K. +2. **No codebooks.** RaBitQ needs only a random rotation (a seed) and the centroids — both + tiny, both versioned in the index metadata. Nothing to retrain when data drifts. +3. **Tunable precision.** B = 1 bit/dim is a fast coarse filter; B = 4 bits/dim gives + near-SQ quality at ~8× less space. Both are used in a two-pass scan (§6.2). +4. **Metric-flexible.** One stored code serves L2, cosine, and dot-product; the metric is + applied at query time. + +The four ideas, precisely: + +- **Residual.** Store each vector as its difference from its centroid, `r = v − c`. + Residuals are small and centered, which is what lets few bits go far. +- **Rotation.** Apply one fixed random orthonormal rotation `R` (derived from a per-generation + seed) to everything first: `x = R·r`. This spreads information evenly across dimensions, + which is what makes the error bound hold for any data distribution. Only the seed is stored. +- **Code.** Quantize `x` to B bits per dimension, stored as **bit planes** — plane 0 holds + bit 0 of every dimension, plane 1 bit 1, etc. The top plane alone is a 1-bit sign sketch. + Scoring a plane against the (transformed) query is `AND`/`XOR` + `popcount` — a few CPU + instructions per 64 dimensions. +- **Factors.** A handful of small scalars per vector (residual norm, two rescale factors, an + error term, a centroid correction) that convert plane math into an unbiased distance + estimate plus its confidence interval. + +At query time the query vector is transformed the same way (`R·(q − c)` per probed cluster), +and the per-plane popcounts combine with the stored factors into the estimate and its bound. +The estimate builds the shortlist; exact base-table distances produce the final ranking (§6). + +### 3.3 Why this fits Hudi + +- Centroids are small enough to load at planning time: `K × D` floats (~12 MB for K=4096, + D=768). +- Codes are compact: a 1B × 128-dim float table's raw vectors are ~512 GB; the RaBitQ index + including keys and locators is ~136 GB, and the scanned portion per query is tens of MB. +- Posting keys are prefix-scannable by generation, cluster, and shard (§4). +- Quantizer state is stable: a seed and centroids, no per-generation learned codebook. +- Exact re-ranking preserves correctness for returned candidates. + +--- + +## 4. Metadata Table Storage Model: the Posting Block + +This section describes the core storage contribution of this RFC. + +### 4.1 The posting block + +A naive index would write one MDT record per indexed vector. At a billion rows that is a +billion MDT records per generation — prohibitive to write, compact, scan, and clean. + +Instead, the index sorts entries by `(cluster, fileGroup, rowPosition)` and packs +~1–4K vectors' worth of codes and metadata into a single MDT record — a **posting block** — +laid out column-wise (structure-of-arrays) so each scan pass touches only the columns it +needs: + +```text +POSTING BLOCK (~512 KB target, one MDT record) +┌───────────────────────────────────────────────┐ +│ S1 sign planes ← pass 1 touches this │ +│ S2 extra bit planes ← pass 2, survivors │ +│ S3 factor arrays ← both passes │ +│ S4 row locators ← finalists only │ +│ S5 dictionaries (file groups, partitions) │ +│ S6 record keys ← finalists only │ +└───────────────────────────────────────────────┘ +key: 0x10 | generation | clusterId | shardId | blockId +``` + +Consequences: + +- **~1000× fewer MDT records.** One block record replaces ~1–4K per-vector records, cutting + write amplification, compaction cost, and cleaner load by three orders of magnitude. +- **Contiguous cluster scans.** The binary key scheme makes "scan cluster 12345" a single + contiguous HFile range read rather than thousands of point lookups. +- **Pay only for promise.** Column-wise layout means pass 1 reads only sign planes + factors; + only survivors touch extra planes; only finalists touch locators and keys (§6.2). + +### 4.2 Row families + +The `vector_index_<name>` partition holds several record families under one binary-sorted key +scheme, so one prefix scan of a cluster returns its blocks and any fresh deltas together: + +| Key family | Cardinality | Purpose | +|---|---:|---| +| `__manifest__` | 1 | Active generation pointer and persisted-format version. | +| `M\|<generation>` | generations | Generation state, vector schema, bootstrap baseline, verified-contiguous frontier, chunk counts/checksums, and quantizer metadata. | +| `T\|<generation>\|<chunk>` | chunks per generation | Size-bounded chunks of the serialized `K × D` centroid matrix. | +| `C\|<generation>\|<cluster>` | K per generation | Cluster manifest: routing version, shard count, vector count, candidate file groups, and counters. | +| `P\|<generation>\|<cluster>\|<shard>\|<blockId>` | blocks per generation | **Posting block** (packed codes, factors, locators, keys). | +| `P\|...\|<DELTA>` | deltas | Small per-record delta records appended between compactions. | +| `F\|<generation>\|<dataInstant>` | data writes after baseline | Atomic proof that the generation incorporated one source data-write instant. | + +### 4.3 Posting shards + +Large clusters are split into posting shards so one hot cluster does not become one oversized +prefix range. The cluster manifest stores `shardCount` and `routingVersion`; writers compute +`shardId = hash(record_key) % shardCount`. Changing `shardCount` changes every key's mapping, +so maintenance must rewrite the whole cluster under a new routing version and publish the +manifest change atomically with that rewrite. Large remaps use a new generation rather than +mixing routing versions in one cluster range. Within a shard, entries are packed by `blockId`. + +### 4.4 Delta records + +Between compactions, per-commit vector writes append small **delta records** (`blockId` +marked `DELTA`) at the end of the same cluster key range. Because they share the prefix, a +single cluster prefix scan sees packed blocks and fresh deltas in one pass (§6, §7). + +### 4.5 Generation model + +A generation is a consistent set of centroid, quantizer, cluster, and posting-block metadata: + +```text +__manifest__ -> active generation id + persisted-format version +M|<gen> -> generation metadata + schema + verified frontier + chunk integrity +T|<gen>|... -> size-bounded centroid chunks +C|<gen>|... -> cluster manifests +P|<gen>|... -> posting blocks + deltas +F|<gen>|... -> incorporated data-write markers +``` + +The builder allocates a generation id once and writes deterministic keys, so retrying a +partially completed build overwrites the same records. A `BUILDING` generation is invisible +to readers; bootstrap validates expected chunks, checksums, cluster/block counts, schema, and +memory budgets before one MDT commit changes it to `ACTIVE` and flips `__manifest__`. +Abandoned or invalid `BUILDING` generations are never activated and may be garbage-collected. +Old generations become `RETIRED` and remain until no retained snapshot or pinned reader needs +them. Generation is the only independent version axis: centroid, quantizer, routing, and +posting changes that cannot be published atomically in place create a new generation. + +--- + +## 5. Bootstrap and Write Path + + + +### 5.1 Spark bootstrap + +Bootstrap and full rebuild are Spark-only in v1. They build an invisible generation from one +pinned table snapshot: + +```text +1. Read latest file slices; extract key, partition, file group, base instant, row position, + and vector bytes; train IVF centroids from a bounded sample. +2. Validate K × D × elementWidth and driver/executor memory budgets before materialization. +3. Broadcast centroids; assign each vector, encode RaBitQ metadata, sort by + (cluster, fileGroup, rowPosition), and pack posting blocks. +4. Write M|, size-bounded T| centroid chunks, C|, and P| records under deterministic keys. +5. Validate chunk counts/checksums and posting counts, then atomically activate the generation. +``` + +The training sample satisfies percentage and per-cluster floors, for example +`min(N, max(1M, 256*K, min(10M, 0.5%–1% of N)))`. Chunking keeps individual MDT records +bounded; readers reconstruct the matrix only after validating manifest integrity metadata. + +### 5.2 Incremental inserts and vector updates + +For each relevant row, the metadata writer's vector hook computes the cluster, shard, RaBitQ +code, factors, and base-table location and appends a posting delta. If an update moves a +record, delta supersession and query-time RLI arbitration hide the stale entry (§6.3). +The hook also emits the commit's freshness marker in the same MDT commit as its vector deltas. +This uses the existing per-data-commit `Indexer.buildUpdate` dispatch; `VectorIndexer` returns +an `F|...` record even without deltas, so its update is never empty and needs no new seam. + +### 5.3 Non-vector updates and deletes + +A non-vector update can retain its code and placement; a moved locator is re-resolved through +RLI and refreshed by maintenance before the old slice is cleaned. Deletes require no posting +delta because RLI arbitration removes deleted candidates. The hook still emits one freshness +marker when a data write produces no vector records—including schema-only or allowed-empty +commits—so an empty update cannot create a permanent false-stale gap. + +### 5.4 Engine support + +Snapshot extraction and centroid training are engine-specific; routing, encoding, record +construction, freshness markers, and publication are common contracts. Spark supplies +bootstrap/rebuild in v1. Feature-aware Spark, Flink, and Java writers may maintain an active +generation through the common metadata-writer hook; unaware writers keep standard MDT +layer-2 skip semantics, with their commits detected by the query-time freshness gate (§9). + +### 5.5 Multi-writer rebuild and catch-up + +Rebuilds are replay-only: writers continue updating the active generation and never dual-write +to a `BUILDING` generation. After constructing at baseline `T_boot`, the builder replays +completed data writes in timeline order outside the coordination lock. Replay uses the active +timeline and, when repairing a gap older than its retained window, the existing archived- +timeline APIs. Each replay writes the new generation's deltas and source-instant marker +atomically. When the gap is bounded, +`TransactionManager` and the metadata-index lock select `T_cut`, perform a final micro-catch-up, +verify contiguous coverage, and publish the new generation in one MDT commit. If replay cannot +converge within its round/time budget, the attempt exits without activation; it never holds an +unbounded lock or publishes partial state. + +--- + +## 6. Read Path + + + +### 6.1 Query planning + +```text +1. Pin the table snapshot and validate its vector schema against the active manifest. +2. Enforce the generation's marker frontier at the pinned data-write instant (§9). +3. Load and validate generation metadata, centroid chunks, quantizer, and cluster manifests. +4. Probe centroids, select top-nprobe clusters, and resolve shard prefixes/file groups. +``` + +### 6.2 Two-pass scan over posting blocks + +For each selected `(cluster, shard)`, prefix-scan `P|<gen>|<cluster>|<shard>|*` (blocks + +deltas) and score column-wise: + +```text +Pass 1 (bound + prune): read S1 sign planes + S3 factors. For every vector compute an + optimistic bound — the best distance it could possibly achieve. Skip vectors whose + best case cannot beat the current K-th candidate (typically 85–95% pruned). +Pass 2 (refine): for survivors, read S2 extra bit planes and compute the full + multibit unbiased estimate; keep the refineFactor·K best. +Finalize: only the surviving finalists touch S4 locators and S6 keys to learn + where they live and who they are. +``` + +Cost is proportional to promise, not to table size. Delta records encountered in the same +scan are scored identically and supersede matching packed entries. + +### 6.3 Freshness arbitration and exact re-rank + +The Record Level Index (RLI) is a hard prerequisite. Vector-index creation rejects a table +without RLI, and the vector planner resolves the dependency again for every query. If RLI was +subsequently dropped by a path unaware of vector indexes, planning fails with the missing +dependency and remediation rather than serving results. RFC-109 does not change the common +index-drop path. Posting locators are hints; finalist keys are validated by batched RLI lookup +on the pinned snapshot. Moved keys are re-resolved and deleted keys are removed. + +Candidate generation retains a surplus before arbitration. If stale/deleted candidates leave +fewer than K live rows, staged continuation scans the next candidates/ranges and repeats RLI +arbitration until K live rows are available or the request budget is exhausted. Exact mode +then positionally reads those vectors and ranks by exact distance. Budget exhaustion invokes +exact-scan fallback or fails explicitly; it never silently returns fewer than K as a complete +answer. `indexLagInstants`, when present, is marker-frontier lag as defined in §9. + +### 6.4 Batched queries + +For a relation of query vectors, the plan shares MDT work: load generation metadata once, +encode all queries, probe per query, group by selected `(cluster, shard)`, scan each range +once while maintaining a per-query top-R heap, then read the union of candidate rows and +exact re-rank per query. + +### 6.5 Fallback + +Missing, incompatible, stale, or budget-exhausted index state must not produce a silently +incorrect answer. Per policy, Hudi fails, emits a warning with `indexLagInstants`, or bypasses +the index for an exact table scan. Exact-rerank defaults to `FAIL` for stale state; callers +that require strict availability may configure exact fallback. + +--- + +## 7. Maintenance, Rebalancing, and Cleaner + +Vectors change. Index health is kept without global retraining or re-encoding through three +tiers of increasing rarity (LIRE — Local Incremental Rebalancing). Review Comment: Is LIRE in v1? §11 scopes the MVP to create/drop/rebuild + §7.1/§7.2-style maintenance, but this section fully specs §7.3 split/merge as well. Let's mark clearly here what ships in the first cut and note in the Limitations section that until then, sustained skew/drift is handled by full generation rebuild only? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
