This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 711481bea2 [core][connectors][docs] Complete primary-key sorted-index
support (#8615)
711481bea2 is described below
commit 711481bea28f317f68a467123f783c68c2d1d6a0
Author: Jingsong Lee <[email protected]>
AuthorDate: Tue Jul 14 22:25:28 2026 +0800
[core][connectors][docs] Complete primary-key sorted-index support (#8615)
Complete source-backed primary-key BTree and Bitmap query support across
the core batch planner, Flink, and Spark, including postpone-bucket
compaction. Consolidate Vector, BTree, and Bitmap documentation into one
primary-key index guide aligned with the shared LSM maintenance now on
master.
---
docs/docs/primary-key-table/global-index.mdx | 380 +++++++++++++++++++++
docs/docs/primary-key-table/vector-index.md | 255 --------------
docs/redirects.js | 8 +
docs/sidebars.js | 2 +-
.../paimon/table/source/PrimaryKeyBatchScan.java | 3 +-
.../source/PrimaryKeySortedIndexBatchScanTest.java | 21 +-
.../paimon/flink/PrimaryKeySortedIndexITCase.java | 112 ++++++
.../spark/sql/PrimaryKeySortedIndexTest.scala | 128 +++++++
8 files changed, 649 insertions(+), 260 deletions(-)
diff --git a/docs/docs/primary-key-table/global-index.mdx
b/docs/docs/primary-key-table/global-index.mdx
new file mode 100644
index 0000000000..6dd0c5bc8e
--- /dev/null
+++ b/docs/docs/primary-key-table/global-index.mdx
@@ -0,0 +1,380 @@
+---
+title: "Primary-Key Indexes"
+sidebar_position: 9
+---
+
+import Tabs from '@theme/Tabs';
+import TabItem from '@theme/TabItem';
+
+<!--
+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.
+-->
+
+# Primary-Key Indexes
+
+Primary-key tables can maintain Vector, BTree, and Bitmap indexes together
with compact data
+files. These indexes are bucket-local and source-backed: every index group
records its source data
+files and maps matches back to physical row positions. Deletion vectors are
applied when indexed
+rows are read, so updates and deletes remain exact.
+
+Primary-key indexes differ from indexes created with `create_global_index`. A
regular
+[Global Index](../multimodal-table/global-index) addresses table-wide row IDs
and is built
+independently for a Data Evolution table. A primary-key index follows the
write and compaction
+lifecycle of a primary-key table and addresses rows inside its compact data
files.
+
+## Choose an Index
+
+<Tabs groupId="primary-key-index-family">
+
+<TabItem value="btree" label="BTree">
+
+Use BTree for selective scalar predicates such as equality, `IN`, comparisons,
ranges, and null
+checks. Normal Flink, Spark, and Java batch scans apply it automatically; no
index-specific query
+API is required.
+
+See [BTree Index](../multimodal-table/global-index/btree) for the built-in
format and supported
+data types.
+
+</TabItem>
+
+<TabItem value="bitmap" label="Bitmap">
+
+Use Bitmap for enum-like dimensions, tags, and other columns where compressed
bitmaps can evaluate
+equality, `IN`, null checks, complement predicates, and supported string
predicates efficiently.
+Normal batch scans apply it automatically.
+
+See [Bitmap Index](../multimodal-table/global-index/bitmap) for predicate and
format details.
+
+</TabItem>
+
+<TabItem value="vector" label="Vector">
+
+Use Vector for approximate nearest-neighbor (ANN) Top-K search on embeddings
that are updated with
+the primary-key table. The index implementation and distance metric are
configured per field, and
+search is exposed through Spark SQL, a Flink procedure, and the Java API.
+
+For an append-only or Data Evolution table whose vector index is built
independently, see
+[Global Vector Index](../multimodal-table/global-index/vector).
+
+</TabItem>
+
+</Tabs>
+
+Different columns in one table can use different index families. One column
can occur in at most
+one of `pk-vector.index.columns`, `pk-btree.index.columns`, and
`pk-bitmap.index.columns`.
+
+## Requirements
+
+All primary-key indexes require:
+
+- A primary-key table in fixed-bucket mode (`bucket > 0`) or postpone-bucket
mode
+ (`bucket = -2`).
+- `pk-clustering-override = false`.
+- An indexed column supported by the selected index implementation.
+
+Dynamic-bucket tables, append-only tables, compound indexes, and custom index
families are not
+supported by this configuration.
+
+<Tabs groupId="primary-key-index-requirements">
+
+<TabItem value="scalar" label="BTree and Bitmap">
+
+BTree and Bitmap additionally require:
+
+- `deletion-vectors.enabled = true`.
+- `deletion-vectors.merge-on-read = false`.
+- A supported scalar column type.
+
+Each entry creates an independent single-column index. Multiple BTree and
Bitmap columns are
+supported as long as a column is not listed by another primary-key index
family.
+
+</TabItem>
+
+<TabItem value="vector" label="Vector">
+
+Vector additionally requires:
+
+- A `VECTOR` column whose element type is `FLOAT`.
+- A merge engine of `deduplicate`, `partial-update`, `aggregation`, or
`first-row`.
+- `deletion-vectors.enabled = true`, except for `first-row`, where it must be
`false`.
+- The configured ANN implementation on every writer and reader classpath.
+
+Exactly one vector column is currently supported per table.
+
+</TabItem>
+
+</Tabs>
+
+## Create a Table
+
+The following table uses all three families on different columns: Vector for
`embedding`, BTree
+for `amount`, and Bitmap for `status`.
+
+<Tabs groupId="primary-key-index-create-table">
+
+<TabItem value="flink-sql" label="Flink SQL">
+
+```sql
+CREATE TABLE items (
+ id BIGINT,
+ status STRING,
+ amount DECIMAL(12, 2),
+ embedding ARRAY<FLOAT> COMMENT '__VECTOR_FIELD;3',
+ PRIMARY KEY (id) NOT ENFORCED
+) WITH (
+ 'bucket' = '8',
+ 'deletion-vectors.enabled' = 'true',
+ 'pk-vector.index.columns' = 'embedding',
+ 'fields.embedding.pk-vector.index.type' = 'ivf-flat',
+ 'fields.embedding.pk-vector.distance.metric' = 'cosine',
+ 'fields.embedding.pk-vector.index.options' = '{"nlist":"256"}',
+ 'pk-btree.index.columns' = 'amount',
+ 'fields.amount.pk-btree.index.options' = '{"block-size":"64 kb"}',
+ 'pk-bitmap.index.columns' = 'status',
+ 'fields.status.pk-bitmap.index.options' = '{"dictionary-block-size":"16
kb"}'
+);
+```
+
+The vector comment directive converts the SQL `ARRAY<FLOAT>` column to
Paimon's fixed-length
+`VECTOR<FLOAT>` type. Java API users can define it directly with
+`DataTypes.VECTOR(3, DataTypes.FLOAT())`.
+
+</TabItem>
+
+<TabItem value="spark-sql" label="Spark SQL">
+
+```sql
+CREATE TABLE items (
+ id BIGINT,
+ status STRING,
+ amount DECIMAL(12, 2),
+ embedding ARRAY<FLOAT> COMMENT '__VECTOR_FIELD;3'
+) USING paimon
+TBLPROPERTIES (
+ 'primary-key' = 'id',
+ 'bucket' = '8',
+ 'deletion-vectors.enabled' = 'true',
+ 'pk-vector.index.columns' = 'embedding',
+ 'fields.embedding.pk-vector.index.type' = 'ivf-flat',
+ 'fields.embedding.pk-vector.distance.metric' = 'cosine',
+ 'fields.embedding.pk-vector.index.options' = '{"nlist":"256"}',
+ 'pk-btree.index.columns' = 'amount',
+ 'fields.amount.pk-btree.index.options' = '{"block-size":"64 kb"}',
+ 'pk-bitmap.index.columns' = 'status',
+ 'fields.status.pk-bitmap.index.options' = '{"dictionary-block-size":"16
kb"}'
+);
+```
+
+</TabItem>
+
+</Tabs>
+
+Duplicate, empty, unknown, unsupported, or cross-family duplicate columns are
rejected during
+schema validation.
+
+### Options
+
+| Option | Default | Description |
+|---|---|---|
+| `pk-vector.index.columns` | Not set | Vector column to index. Exactly one
vector column is currently supported. |
+| `fields.<column>.pk-vector.index.type` | Required | ANN implementation, such
as `ivf-flat`, `ivf-pq`, `ivf-hnsw-flat`, `ivf-hnsw-sq`, or `lumina`. |
+| `fields.<column>.pk-vector.distance.metric` | `inner_product` | Distance
metric: `l2`, `cosine`, or `inner_product`. |
+| `fields.<column>.pk-vector.index.options` | Not set | JSON object containing
build options for the selected ANN implementation. |
+| `pk-btree.index.columns` | Not set | Comma-separated columns which own
independent BTree indexes. |
+| `fields.<column>.pk-btree.index.options` | Not set | JSON object containing
BTree build options. Unqualified keys are scoped to `btree-index`. |
+| `pk-bitmap.index.columns` | Not set | Comma-separated columns which own
independent Bitmap indexes. |
+| `fields.<column>.pk-bitmap.index.options` | Not set | JSON object containing
Bitmap build options. Unqualified keys are scoped to `bitmap-index`. |
+| `fields.<column>.pk-index.compaction.level-fanout` | `5` | Number of
similarly sized index groups which triggers a rebuild and maximum row-count
ratio within one size tier. Shared by all three families. Must be greater than
`1`. |
+| `fields.<column>.pk-index.compaction.stale-ratio-threshold` | `0.2` | Ratio
of rows from inactive source files which triggers a rebuild. Shared by all
three families. Must be in `(0, 1]`. |
+
+For algorithm-specific options, see the corresponding
+[BTree](../multimodal-table/global-index/btree),
+[Bitmap](../multimodal-table/global-index/bitmap), or
+[Vector](../multimodal-table/global-index/vector) index page.
+
+## Maintenance and Coverage
+
+Paimon builds primary-key indexes from complete `COMPACT` data files above
Level 0. An index group
+stores the source file names and row counts required to map index results back
to physical rows.
+Data-file and index-file changes are committed in the same snapshot.
+
+New Level-0 `APPEND` files are not index sources. They remain uncovered until
physical data
+compaction produces an eligible output. A metadata-only level upgrade keeps
its `APPEND` source
+and therefore remains uncovered as well.
+
+For a postpone-bucket table, foreground writes can land in bucket `-2` or in
pending Level-0 files
+inside real buckets. These rows become visible after batch compaction
publishes them to real
+buckets. Indexes are created when that process physically rewrites the rows
into eligible compact
+output; simply assigning or upgrading a pending file does not make it an index
source.
+
+### Index LSM Maintenance
+
+Each indexed column maintains its own immutable index groups. Maintenance uses
the shared
+field-scoped compaction options:
+
+- When at least `level-fanout` similarly sized groups exist, Paimon rebuilds
them into a larger
+ group. The largest selected group can contain at most `level-fanout` times
the rows of the
+ smallest group.
+- When the ratio of rows belonging to inactive source files reaches
+ `stale-ratio-threshold`, Paimon rebuilds the affected group from its
remaining active sources.
+- A rebuild atomically replaces its input groups after the new group is
complete.
+
+Index construction can execute asynchronously inside the writer. A writer
which waits for
+compaction also waits for active index maintenance; a non-blocking writer can
complete maintenance
+in a later commit. Coverage can therefore be temporarily partial.
+
+Partial coverage affects acceleration, not correctness:
+
+- BTree and Bitmap scans read uncovered files through the ordinary data path.
+- Vector search evaluates files without an active ANN group exactly.
+- The original scalar predicate and deletion vectors are applied after index
pruning.
+
+## BTree and Bitmap Queries
+
+BTree and Bitmap indexes are applied automatically to snapshot-scoped batch
scans. They can
+accelerate equality, `IN`, null checks, comparisons and ranges, complement
predicates, and
+supported string predicates.
+
+Paimon combines indexed scalar results as follows:
+
+- For `AND`, any usable indexed child can narrow a source file; unindexed
children remain residual
+ filters.
+- For `OR`, index positions are used only when every branch can be evaluated
safely. Otherwise,
+ the source file is scanned normally.
+
+Planning uses the data and index manifests from the selected snapshot. If an
index group is
+missing, incomplete, unreadable, corrupt, or returns an invalid physical
position, the affected
+source file falls back to the ordinary data path.
+
+## Vector Search
+
+A vector search captures one table snapshot, searches active ANN groups in the
selected buckets,
+and merges their candidates into a global Top-K. Candidates are materialized
from source data
+files by physical row position. Deletion vectors remove stale versions and
deleted rows.
+
+<Tabs groupId="primary-key-vector-search-api">
+
+<TabItem value="spark-sql" label="Spark SQL">
+
+Use the `vector_search` table-valued function. Spark exposes the ANN score
through the
+`__paimon_search_score` metadata column.
+
+```sql
+SELECT id, status, __paimon_search_score
+FROM vector_search(
+ 'items',
+ 'embedding',
+ array(0.1f, 0.2f, 0.3f),
+ 10,
+ map('ivf.nprobe', '32')
+);
+```
+
+The query dimension must match the indexed vector dimension. Partition
predicates are applied
+before ANN search. When `spark.paimon.vector-search.distribute.enabled` is
`true`, Spark can
+distribute sufficiently large groups of bucket-local searches and merge
task-local Top-K results
+on the driver.
+
+</TabItem>
+
+<TabItem value="flink-sql" label="Flink SQL">
+
+Flink exposes vector search as a procedure and returns JSON-serialized rows.
Use `projection` to
+avoid reading columns which are not needed.
+
+```sql
+CALL sys.vector_search(
+ `table` => 'default.items',
+ vector_column => 'embedding',
+ query_vector => '0.1,0.2,0.3',
+ top_k => 10,
+ projection => 'id,status',
+ options => 'ivf.nprobe=32'
+);
+```
+
+</TabItem>
+
+<TabItem value="java-api" label="Java API">
+
+```java
+GlobalIndexResult result = table.newVectorSearchBuilder()
+ .withVectorColumn("embedding")
+ .withVector(queryVector)
+ .withLimit(10)
+ .withOption("ivf.nprobe", "32")
+ .executeLocal();
+
+ReadBuilder readBuilder = table.newReadBuilder();
+TableScan.Plan plan =
readBuilder.newScan().withGlobalIndexResult(result).plan();
+try (RecordReader<InternalRow> reader =
readBuilder.newRead().createReader(plan)) {
+ reader.forEachRemaining(row -> consume(row));
+}
+```
+
+</TabItem>
+
+</Tabs>
+
+### Exact Rerank
+
+Primary-key vector search can retrieve additional ANN candidates and rerank
them with the original
+vectors. For example, this table option retrieves up to four times the
requested Top-K before
+computing exact distances:
+
+```sql
+'fields.embedding.ivf-flat.refine_factor' = '4'
+```
+
+`refine_factor`, `refine-factor`, `rerank_factor`, and `rerank-factor` are
accepted. Query options
+override table options, and field/index prefixes take precedence over
less-specific prefixes. The
+factor must be a positive integer. A factor of `1` performs exact reranking
without retrieving
+additional candidates.
+
+Only ANN candidates can win the rerank, so a larger factor can improve recall
but does not
+guarantee the exact global Top-K. It also increases ANN work and data-file
I/O. Uncovered files are
+searched exactly and merged separately with the ANN candidates.
+
+## Merge-Engine Behavior
+
+Vector maintenance follows the table's merge engine:
+
+- `deduplicate`: an update indexes the latest row and the deletion vector
hides the replaced
+ physical row. A delete removes the old row from search results through the
deletion vector.
+- `partial-update`: the vector index is built from the lookup-completed
compact-output row.
+- `aggregation`: the vector index is built from the aggregated compact-output
row.
+- `first-row`: the retained first row is indexed. Deletion vectors must be
disabled because later
+ rows with the same primary key are ignored rather than deleting the retained
row.
+
+## Schema Evolution
+
+An indexed column cannot be renamed, dropped, or have its type changed while
its definition is
+present. Treat index column lists, field-scoped implementation settings, and
build options as part
+of the index definition. To change the family or incompatible build options of
an indexed column,
+create a table with the desired definition and migrate the data.
+
+## Limitations
+
+- BTree and Bitmap definitions are single-column indexes.
+- Exactly one vector index column is currently supported per table.
+- Only `FLOAT` vectors are supported.
+- Indexes are built from eligible compact output, not directly from Level-0
appends.
+- Index acceleration and vector search are snapshot-scoped batch operations;
continuous streaming
+ and lateral vector search are not supported.
+- Flink vector search returns rows but does not expose the ANN score as a
separate column.
+- Online replacement between two definitions on the same column is not
supported.
diff --git a/docs/docs/primary-key-table/vector-index.md
b/docs/docs/primary-key-table/vector-index.md
deleted file mode 100644
index 54e33325d5..0000000000
--- a/docs/docs/primary-key-table/vector-index.md
+++ /dev/null
@@ -1,255 +0,0 @@
----
-title: "Vector Index"
-sidebar_position: 9
----
-
-<!--
-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.
--->
-
-# Vector Index
-
-Primary key tables can maintain a bucket-local approximate nearest neighbor
(ANN) index together
-with their data. Unlike a global vector index created by
`create_global_index`, a primary-key
-vector index is part of the normal write and compaction lifecycle. Paimon
builds it synchronously
-when complete compact-output files are produced and commits the index changes
together with those
-files.
-
-Use a primary-key vector index when vectors are frequently updated and the ANN
index should follow
-the primary-key table's compaction lifecycle. For append-only or Data
Evolution tables whose index
-is built separately from writes, see
-[Global Vector Index](../multimodal-table/global-index/vector).
-
-## Requirements
-
-A table with a primary-key vector index must satisfy all of the following:
-
-- It is a primary-key table in fixed-bucket mode (`bucket > 0`) or
postpone-bucket mode
- (`bucket = -2`).
-- `deletion-vectors.enabled` is `true`, except for `first-row`, where it must
be `false`.
-- Its merge engine is `deduplicate`, `partial-update`, `aggregation`, or
`first-row`.
-- The indexed column is a `VECTOR` whose element type is `FLOAT`.
-- `pk-clustering-override` is disabled.
-- The configured vector index implementation is available on every writer and
reader classpath.
-
-The first release supports exactly one indexed vector column per table. The
option layout is
-field-scoped so that more independently indexed vector columns can be
supported in a future
-release.
-
-## Create Table
-
-The following Flink SQL example creates a three-dimensional vector column and
maintains an
-IVF-Flat index for it. Use the dimension produced by your embedding model in
production.
-
-```sql
-CREATE TABLE item_embeddings (
- id BIGINT,
- payload STRING,
- embedding ARRAY<FLOAT> COMMENT '__VECTOR_FIELD;3',
- PRIMARY KEY (id) NOT ENFORCED
-) WITH (
- 'bucket' = '16',
- 'deletion-vectors.enabled' = 'true',
- 'pk-vector.index.columns' = 'embedding',
- 'fields.embedding.pk-vector.index.type' = 'ivf-flat',
- 'fields.embedding.pk-vector.distance.metric' = 'cosine',
- 'fields.embedding.pk-vector.index.options' = '{"nlist":"256"}'
-);
-```
-
-Use the same properties in Spark SQL:
-
-```sql
-CREATE TABLE item_embeddings (
- id BIGINT,
- payload STRING,
- embedding ARRAY<FLOAT> COMMENT '__VECTOR_FIELD;3'
-) USING paimon
-TBLPROPERTIES (
- 'primary-key' = 'id',
- 'bucket' = '16',
- 'deletion-vectors.enabled' = 'true',
- 'pk-vector.index.columns' = 'embedding',
- 'fields.embedding.pk-vector.index.type' = 'ivf-flat',
- 'fields.embedding.pk-vector.distance.metric' = 'cosine',
- 'fields.embedding.pk-vector.index.options' = '{"nlist":"256"}'
-);
-```
-
-The vector comment directive converts the SQL `ARRAY<FLOAT>` column to
Paimon's fixed-length
-`VECTOR<FLOAT>` type. Java API users can define the column directly with
-`DataTypes.VECTOR(3, DataTypes.FLOAT())`.
-
-### Options
-
-| Option | Required | Description |
-|---|---|---|
-| `pk-vector.index.columns` | Yes | Indexed vector column. Exactly one column
is supported in the first release. |
-| `fields.<column>.pk-vector.index.type` | Yes | ANN implementation, such as
`ivf-flat`, `ivf-pq`, `ivf-hnsw-flat`, `ivf-hnsw-sq`, or `lumina`. |
-| `fields.<column>.pk-vector.distance.metric` | No | `l2`, `cosine`, or
`inner_product`. The default is `inner_product`. |
-| `fields.<column>.pk-vector.index.options` | No | JSON object containing
build options for the selected ANN implementation. Unqualified keys are scoped
to that implementation. |
-| `fields.<column>.pk-index.compaction.level-fanout` | No | Number of
similarly sized index groups which triggers a rebuild and maximum row-count
ratio within one size tier. Shared by vector, BTree, and Bitmap primary-key
indexes. Default: `5`. |
-| `fields.<column>.pk-index.compaction.stale-ratio-threshold` | No | Ratio of
rows from inactive source files which triggers an index rebuild. Shared by
vector, BTree, and Bitmap primary-key indexes. Default: `0.2`. |
-
-For algorithm-specific build and search options, see
-[Vector Index](../multimodal-table/global-index/vector).
-
-## Index Maintenance
-
-Paimon builds immutable ANN segments from complete compact-output data files
inside each bucket.
-The index segment records the source data files and maps ANN ordinals back to
their physical row
-positions. Compact-output data-file and index-file changes are committed
atomically, so a reader
-never observes an index from a different compact-output snapshot.
-
-For a postpone-bucket table, foreground writes remain write-only. Fixed-bucket
batch writes produce
-Level-0 files in real buckets, while postponed writes produce files in bucket
`-2`. These rows do not
-become visible to normal reads or vector search until a batch compact runs.
The background compact
-processes both kinds of pending files, builds their ANN indexes, and publishes
the data and index
-changes in one atomic commit.
-
-ANN compaction is configured independently from data compaction. It does not
inherit
-`vector.target-file-size`, `num-sorted-run.compaction-trigger`, or
-`compaction.delete-ratio-threshold`.
-
-The maintenance behavior depends on the merge engine:
-
-- `deduplicate`: an update indexes the latest row and the deletion vector
hides the replaced
- physical row. A delete removes the old row from search results through the
deletion vector.
-- `partial-update`: Paimon builds the vector index from the lookup-completed
Level-1
- compact-output row.
-- `aggregation`: Paimon builds the vector index from the aggregated Level-1
compact-output row.
-- `first-row`: Paimon indexes the retained first row. Deletion vectors must be
disabled because
- later rows with the same primary key are ignored rather than deleting the
retained row.
-
-When compaction replaces source data files, Paimon removes ANN segments that
reference those files
-and creates replacement segments for the new compact-output files. Small
outputs are indexed as
-well; there is no minimum-row threshold before a new segment can be built.
-
-The index follows compaction freshness. Newly appended level-0 files are not
ANN sources, so a
-streaming write may not be searchable until compaction has produced and
committed its complete
-level-1 output. Wait for that compaction when read-after-write vector-search
visibility is
-required. Batch writes which wait for compaction can publish the data and its
index together.
-
-## Search
-
-### Exact Rerank
-
-Primary-key vector search can retrieve more ANN candidates and rerank them
with the original
-vectors stored in the table. For example, the following table option retrieves
up to four times
-the requested Top-K from the `ivf-flat` index before computing exact distances:
-
-```sql
-'fields.embedding.ivf-flat.refine_factor' = '4'
-```
-
-The option is disabled by default. Its configuration semantics are the same as
for a Data
-Evolution vector index:
-
-- `refine_factor`, `refine-factor`, `rerank_factor`, and `rerank-factor` are
accepted.
-- A query option overrides every table option. Within either set of options,
field and index
- prefixes take precedence over less specific prefixes. For example,
- `fields.embedding.ivf-flat.refine_factor` takes precedence over
- `fields.embedding.ivf.refine_factor`, which takes precedence over
`ivf.refine_factor` and then
- `refine_factor`. The normalized underscore form of an index type, such as
`ivf_flat`, is also
- accepted after the configured index name.
-- The factor must be a positive integer. A factor of `1` performs exact
reranking without
- retrieving additional ANN candidates.
-
-Only candidates returned by ANN can win the rerank, so a larger factor can
improve recall but does
-not guarantee the exact global Top-K. It also increases ANN work and data-file
I/O. Files without
-an active ANN segment are already searched exactly and are kept separate from
approximate
-candidates until the final Top-K merge.
-
-For distributed Spark searches, executors return bounded ANN and exact
candidate streams. The
-driver globally merges the ANN candidates and rereads their original vectors
for exact reranking;
-this does not start a second Spark job.
-
-### Spark SQL
-
-Use the `vector_search` table-valued function. Spark exposes the ANN score
through the
-`__paimon_search_score` metadata column.
-
-```sql
-SELECT id, payload, __paimon_search_score
-FROM vector_search(
- 'item_embeddings',
- 'embedding',
- array(0.1f, 0.2f, 0.3f),
- 10,
- map('ivf.nprobe', '32')
-);
-```
-
-The query vector dimension must match the indexed column dimension. For
partitioned tables, Spark
-applies a partition predicate before running ANN and merging the global Top-K.
-When `spark.paimon.vector-search.distribute.enabled` is `true`, Spark
distributes sufficiently
-large groups of bucket-local ANN searches across executors and merges their
task-local Top-K
-results on the driver. Small plans stay local to avoid Spark job startup
overhead.
-
-### Flink SQL
-
-Flink exposes vector search as a procedure and returns JSON-serialized rows.
Use `projection` to
-avoid reading columns that are not needed.
-
-```sql
-CALL sys.vector_search(
- `table` => 'default.item_embeddings',
- vector_column => 'embedding',
- query_vector => '0.1,0.2,0.3',
- top_k => 10,
- projection => 'id,payload',
- options => 'ivf.nprobe=32'
-);
-```
-
-### Java API
-
-```java
-GlobalIndexResult result = table.newVectorSearchBuilder()
- .withVectorColumn("embedding")
- .withVector(queryVector)
- .withLimit(10)
- .withOption("ivf.nprobe", "32")
- .executeLocal();
-
-ReadBuilder readBuilder = table.newReadBuilder();
-TableScan.Plan plan =
readBuilder.newScan().withGlobalIndexResult(result).plan();
-try (RecordReader<InternalRow> reader =
readBuilder.newRead().createReader(plan)) {
- reader.forEachRemaining(row -> consume(row));
-}
-```
-
-## Query Planning
-
-A search captures one table snapshot, plans the active ANN segments for every
selected bucket,
-searches those segments, and merges their candidates into one global Top-K.
The returned candidates
-are materialized from the source data files by physical row position. Deletion
vectors are applied
-while searching and reading, so stale versions and deleted rows are not
returned.
-
-For low latency on object storage, cache data files and ANN payloads with a
caching file system.
-The first query may still need to download index files; subsequent queries can
search the local
-cached payloads and fetch only the selected data-file positions.
-
-## Limitations
-
-- Exactly one vector index column is supported per table in the first release.
-- Only `FLOAT` vectors are supported.
-- Dynamic-bucket and `pk-clustering-override` tables are not supported.
-- Flink's procedure returns rows but does not expose the ANN score as a
separate column.
-- Vector search is snapshot-scoped batch reading; streaming search and lateral
vector search for
- primary-key tables are not supported.
diff --git a/docs/redirects.js b/docs/redirects.js
index c151681242..57077ff895 100644
--- a/docs/redirects.js
+++ b/docs/redirects.js
@@ -328,6 +328,14 @@ module.exports = [
"from": "/primary-key-table/query-performance.html",
"to": "/primary-key-table/query-performance"
},
+ {
+ "from": "/primary-key-table/vector-index.html",
+ "to": "/primary-key-table/global-index"
+ },
+ {
+ "from": "/primary-key-table/vector-index",
+ "to": "/primary-key-table/global-index"
+ },
{
"from": "/primary-key-table/sequence-rowkind.html",
"to": "/primary-key-table/sequence-rowkind"
diff --git a/docs/sidebars.js b/docs/sidebars.js
index 7e3967dade..e88793b7fe 100644
--- a/docs/sidebars.js
+++ b/docs/sidebars.js
@@ -84,7 +84,7 @@ const sidebars = {
"primary-key-table/sequence-rowkind",
"primary-key-table/compaction",
"primary-key-table/query-performance",
- "primary-key-table/vector-index",
+ "primary-key-table/global-index",
"primary-key-table/chain-table",
"primary-key-table/pk-clustering-override",
{
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeyBatchScan.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeyBatchScan.java
index ac25c76920..f739c86f1b 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeyBatchScan.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeyBatchScan.java
@@ -28,6 +28,7 @@ import org.apache.paimon.manifest.FileKind;
import org.apache.paimon.manifest.IndexManifestEntry;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.table.BucketMode;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.source.snapshot.SnapshotReader;
@@ -102,7 +103,7 @@ public class PrimaryKeyBatchScan extends
AbstractBatchTableScan {
|| table.schema().primaryKeys().isEmpty()
|| !options().deletionVectorsEnabled()
|| options().deletionVectorsMergeOnRead()
- || options().bucket() <= 0
+ || (options().bucket() <= 0 && options().bucket() !=
BucketMode.POSTPONE_BUCKET)
|| snapshotPlan.snapshotId() == null
|| snapshotPlan.splits().isEmpty()) {
return dataPlan;
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeySortedIndexBatchScanTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeySortedIndexBatchScanTest.java
index 35de4f3145..0979f874db 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeySortedIndexBatchScanTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeySortedIndexBatchScanTest.java
@@ -39,6 +39,7 @@ import org.apache.paimon.predicate.PredicateBuilder;
import org.apache.paimon.schema.SchemaManager;
import org.apache.paimon.schema.TableSchema;
import org.apache.paimon.stats.SimpleStats;
+import org.apache.paimon.table.BucketMode;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.source.snapshot.SnapshotReader;
import org.apache.paimon.types.DataField;
@@ -80,6 +81,15 @@ class PrimaryKeySortedIndexBatchScanTest {
assertThat(indexedSplit.rowRanges()).containsExactly(new Range(2, 2));
}
+ @Test
+ void testPostponeBucketBatchScanUsesSnapshotScopedSortedIndex() {
+ ScanFixture fixture = fixture(reader(2), true,
BucketMode.POSTPONE_BUCKET);
+
+ TableScan.Plan result = fixture.scan.plan();
+
+
assertThat(result.splits()).singleElement().isInstanceOf(IndexedSplit.class);
+ }
+
@Test
void testOrdinaryBatchScanFailsWhenApplyingSortedIndexFails() {
ScanFixture fixture = fixture(reader(2));
@@ -105,9 +115,9 @@ class PrimaryKeySortedIndexBatchScanTest {
assertThat(result.splits()).singleElement().isInstanceOf(DataSplit.class);
}
- private static TableSchema tableSchema(boolean globalIndexEnabled) {
+ private static TableSchema tableSchema(boolean globalIndexEnabled, int
bucket) {
Map<String, String> options = new HashMap<>();
- options.put(CoreOptions.BUCKET.key(), "2");
+ options.put(CoreOptions.BUCKET.key(), Integer.toString(bucket));
options.put(CoreOptions.DELETION_VECTORS_ENABLED.key(), "true");
options.put(CoreOptions.DELETION_VECTORS_MERGE_ON_READ.key(), "false");
options.put(CoreOptions.GLOBAL_INDEX_ENABLED.key(),
Boolean.toString(globalIndexEnabled));
@@ -129,7 +139,12 @@ class PrimaryKeySortedIndexBatchScanTest {
}
private static ScanFixture fixture(GlobalIndexReader reader, boolean
globalIndexEnabled) {
- TableSchema schema = tableSchema(globalIndexEnabled);
+ return fixture(reader, globalIndexEnabled, 2);
+ }
+
+ private static ScanFixture fixture(
+ GlobalIndexReader reader, boolean globalIndexEnabled, int bucket) {
+ TableSchema schema = tableSchema(globalIndexEnabled, bucket);
CoreOptions options = new CoreOptions(schema.options());
DataFileMeta dataFile = dataFile("data-1", 4);
DataSplit dataSplit = dataSplit(11, dataFile);
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PrimaryKeySortedIndexITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PrimaryKeySortedIndexITCase.java
new file mode 100644
index 0000000000..0fadbf9b06
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PrimaryKeySortedIndexITCase.java
@@ -0,0 +1,112 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink;
+
+import org.apache.paimon.globalindex.IndexedSplit;
+import org.apache.paimon.index.IndexFileMeta;
+import org.apache.paimon.manifest.IndexManifestEntry;
+import org.apache.paimon.predicate.PredicateBuilder;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.table.source.Split;
+
+import org.apache.flink.table.api.ExplainFormat;
+import org.apache.flink.types.Row;
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+import java.util.stream.Collectors;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** End-to-end Flink SQL tests for source-backed primary-key BTree and Bitmap
indexes. */
+public class PrimaryKeySortedIndexITCase extends CatalogITCaseBase {
+
+ @Test
+ public void testMixedSortedIndexesWithDeletionVectors() throws Exception {
+ sql(
+ "CREATE TABLE T ("
+ + "id INT PRIMARY KEY NOT ENFORCED, "
+ + "score INT, category STRING, note STRING"
+ + ") WITH ("
+ + "'bucket' = '1', "
+ + "'deletion-vectors.enabled' = 'true', "
+ + "'pk-btree.index.columns' = 'score', "
+ + "'pk-bitmap.index.columns' = 'category', "
+ + "'fields.score.pk-btree.index.options' = "
+ + "'{\"block-size\":\"4 kb\"}', "
+ + "'fields.category.pk-bitmap.index.options' = "
+ + "'{\"dictionary-block-size\":\"8 kb\"}'"
+ + ")");
+ sql("INSERT INTO T VALUES (1, 10, 'red', 'keep'), (2, 20, 'blue',
'drop')");
+ sql("INSERT INTO T VALUES (3, 30, 'red', 'keep'), (4, 40, 'green',
'keep')");
+ sql("CALL sys.compact(`table` => 'default.T')");
+
+ FileStoreTable table = paimonTable("T");
+ List<IndexFileMeta> sourceIndexes =
+ table.store().newIndexFileHandler().scanEntries().stream()
+ .map(IndexManifestEntry::indexFile)
+ .filter(
+ file ->
+ file.globalIndexMeta() != null
+ &&
file.globalIndexMeta().sourceMeta() != null)
+ .collect(Collectors.toList());
+
assertThat(sourceIndexes).extracting(IndexFileMeta::indexType).contains("btree",
"bitmap");
+
+ sql("UPDATE T SET score = 25, category = 'red', note = 'keep' WHERE id
= 2");
+ sql("DELETE FROM T WHERE id = 3");
+ sql("INSERT INTO T VALUES (5, 35, 'yellow', 'keep')");
+
+ String indexedQuery = "SELECT * FROM T WHERE score >= 20 AND score <
40";
+ assertThat(tEnv.explainSql(indexedQuery, ExplainFormat.TEXT))
+ .contains("TableSourceScan", "filter=[", ">=(score, 20)",
"<(score, 40)");
+ PredicateBuilder predicateBuilder = new
PredicateBuilder(table.rowType());
+ List<Split> splits =
+ table.newReadBuilder()
+ .withFilter(
+ PredicateBuilder.and(
+ predicateBuilder.greaterOrEqual(1, 20),
+ predicateBuilder.lessThan(1, 40)))
+ .newScan()
+ .plan()
+ .splits();
+ assertThat(splits).anyMatch(IndexedSplit.class::isInstance);
+ assertThat(splits).anyMatch(DataSplit.class::isInstance);
+
+ assertThat(sql("SELECT * FROM T WHERE score = 10"))
+ .containsExactly(Row.of(1, 10, "red", "keep"));
+ assertThat(sql(indexedQuery))
+ .containsExactlyInAnyOrder(
+ Row.of(2, 25, "red", "keep"), Row.of(5, 35, "yellow",
"keep"));
+ assertThat(sql("SELECT * FROM T WHERE score >= 10 AND category =
'red'"))
+ .containsExactlyInAnyOrder(
+ Row.of(1, 10, "red", "keep"), Row.of(2, 25, "red",
"keep"));
+ assertThat(sql("SELECT * FROM T WHERE score = 40 OR category = 'red'"))
+ .containsExactlyInAnyOrder(
+ Row.of(1, 10, "red", "keep"),
+ Row.of(2, 25, "red", "keep"),
+ Row.of(4, 40, "green", "keep"));
+ assertThat(sql("SELECT * FROM T WHERE note = 'keep'"))
+ .containsExactlyInAnyOrder(
+ Row.of(1, 10, "red", "keep"),
+ Row.of(2, 25, "red", "keep"),
+ Row.of(4, 40, "green", "keep"),
+ Row.of(5, 35, "yellow", "keep"));
+ }
+}
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PrimaryKeySortedIndexTest.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PrimaryKeySortedIndexTest.scala
new file mode 100644
index 0000000000..4a09bfc18b
--- /dev/null
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PrimaryKeySortedIndexTest.scala
@@ -0,0 +1,128 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.spark.sql
+
+import org.apache.paimon.globalindex.IndexedSplit
+import org.apache.paimon.spark.PaimonSparkTestBase
+import org.apache.paimon.table.source.DataSplit
+
+import org.apache.spark.sql.Row
+
+import scala.collection.JavaConverters._
+
+/** End-to-end Spark SQL tests for source-backed primary-key BTree and Bitmap
indexes. */
+class PrimaryKeySortedIndexTest extends PaimonSparkTestBase {
+
+ test("postpone bucket builds and applies sorted indexes during compact") {
+ withTable("t") {
+ spark.sql("""
+ |CREATE TABLE t (id INT, score INT, category STRING)
+ |TBLPROPERTIES (
+ | 'primary-key' = 'id',
+ | 'bucket' = '-2',
+ | 'postpone.batch-write-fixed-bucket' = 'false',
+ | 'compaction.force-up-level-0' = 'true',
+ | 'compaction.force-rewrite-all-files' = 'true',
+ | 'deletion-vectors.enabled' = 'true',
+ | 'pk-btree.index.columns' = 'score',
+ | 'pk-bitmap.index.columns' = 'category',
+ | 'fields.score.pk-btree.index.options' =
+ | '{"block-size":"4 kb"}',
+ | 'fields.category.pk-bitmap.index.options' =
+ | '{"dictionary-block-size":"8 kb"}'
+ |)
+ |""".stripMargin)
+ spark.sql("INSERT INTO t VALUES (1, 10, 'red'), (2, 20, 'blue'), (3, 30,
'red')")
+
+ assert(spark.sql("SELECT * FROM t").collect().isEmpty)
+ assert(spark.sql("SELECT bucket FROM
`t$buckets`").collect().exists(_.getInt(0) == -2))
+
+ // A single L0 run can be upgraded without a rewrite, so force an
eligible compact output.
+ spark.sql("CALL sys.compact(table => 't')")
+
+ assert(!spark.sql("SELECT bucket FROM
`t$buckets`").collect().exists(_.getInt(0) == -2))
+ assert(spark.sql("SELECT file_path FROM `t$files` WHERE level =
0").collect().isEmpty)
+ val sourceIndexes =
loadTable("t").store.newIndexFileHandler.scanEntries.asScala
+ .map(_.indexFile)
+ .filter(meta => meta.globalIndexMeta != null &&
meta.globalIndexMeta.sourceMeta != null)
+ assert(sourceIndexes.map(_.indexType).toSet == Set("btree", "bitmap"))
+
+ val indexedQuery = "SELECT * FROM t WHERE score >= 20 AND category =
'red'"
+ val indexedScan = getPaimonScan(indexedQuery)
+ assert(indexedScan.inputSplits.exists(_.isInstanceOf[IndexedSplit]))
+ checkAnswer(spark.sql(indexedQuery), Seq(Row(3, 30, "red")))
+ }
+ }
+
+ test("mixed sorted indexes with deletion vectors") {
+ withTable("t") {
+ spark.sql("""
+ |CREATE TABLE t (id INT, score INT, category STRING, note
STRING)
+ |TBLPROPERTIES (
+ | 'primary-key' = 'id',
+ | 'bucket' = '1',
+ | 'deletion-vectors.enabled' = 'true',
+ | 'pk-btree.index.columns' = 'score',
+ | 'pk-bitmap.index.columns' = 'category',
+ | 'fields.score.pk-btree.index.options' =
+ | '{"block-size":"4 kb"}',
+ | 'fields.category.pk-bitmap.index.options' =
+ | '{"dictionary-block-size":"8 kb"}'
+ |)
+ |""".stripMargin)
+ spark.sql("INSERT INTO t VALUES (1, 10, 'red', 'keep'), (2, 20, 'blue',
'drop')")
+ spark.sql("INSERT INTO t VALUES (3, 30, 'red', 'keep'), (4, 40, 'green',
'keep')")
+ spark.sql("CALL sys.compact(table => 't')")
+
+ val sourceIndexes =
loadTable("t").store.newIndexFileHandler.scanEntries.asScala
+ .map(_.indexFile)
+ .filter(meta => meta.globalIndexMeta != null &&
meta.globalIndexMeta.sourceMeta != null)
+ assert(sourceIndexes.map(_.indexType).toSet == Set("btree", "bitmap"))
+
+ spark.sql("UPDATE t SET score = 25, category = 'red', note = 'keep'
WHERE id = 2")
+ spark.sql("DELETE FROM t WHERE id = 3")
+ spark.sql("INSERT INTO t VALUES (5, 35, 'yellow', 'keep')")
+
+ val indexedQuery = "SELECT * FROM t WHERE score >= 20 AND score < 40"
+ val indexedScan = getPaimonScan(indexedQuery)
+ assert(indexedScan.pushedDataFilters.nonEmpty)
+ assert(indexedScan.inputSplits.exists(_.isInstanceOf[IndexedSplit]))
+ assert(indexedScan.inputSplits.exists(_.isInstanceOf[DataSplit]))
+
+ checkAnswer(spark.sql("SELECT * FROM t WHERE score = 10"), Seq(Row(1,
10, "red", "keep")))
+ checkAnswer(
+ spark.sql(indexedQuery),
+ Seq(Row(2, 25, "red", "keep"), Row(5, 35, "yellow", "keep")))
+ checkAnswer(
+ spark.sql("SELECT * FROM t WHERE score >= 10 AND category = 'red'"),
+ Seq(Row(1, 10, "red", "keep"), Row(2, 25, "red", "keep")))
+ checkAnswer(
+ spark.sql("SELECT * FROM t WHERE score = 40 OR category = 'red'"),
+ Seq(Row(1, 10, "red", "keep"), Row(2, 25, "red", "keep"), Row(4, 40,
"green", "keep")))
+ checkAnswer(
+ spark.sql("SELECT * FROM t WHERE note = 'keep'"),
+ Seq(
+ Row(1, 10, "red", "keep"),
+ Row(2, 25, "red", "keep"),
+ Row(4, 40, "green", "keep"),
+ Row(5, 35, "yellow", "keep"))
+ )
+ }
+ }
+}
