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 000000000000..6dd0c5bc8e12 --- /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'; + + + +# 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 + + + + + +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. + + + + + +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. + + + + + +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). + + + + + +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. + + + + + +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. + + + + + +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. + + + + + +## Create a Table + +The following table uses all three families on different columns: Vector for `embedding`, BTree +for `amount`, and Bitmap for `status`. + + + + + +```sql +CREATE TABLE items ( + id BIGINT, + status STRING, + amount DECIMAL(12, 2), + embedding ARRAY 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` column to Paimon's fixed-length +`VECTOR` type. Java API users can define it directly with +`DataTypes.VECTOR(3, DataTypes.FLOAT())`. + + + + + +```sql +CREATE TABLE items ( + id BIGINT, + status STRING, + amount DECIMAL(12, 2), + embedding ARRAY 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"}' +); +``` + + + + + +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..pk-vector.index.type` | Required | ANN implementation, such as `ivf-flat`, `ivf-pq`, `ivf-hnsw-flat`, `ivf-hnsw-sq`, or `lumina`. | +| `fields..pk-vector.distance.metric` | `inner_product` | Distance metric: `l2`, `cosine`, or `inner_product`. | +| `fields..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..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..pk-bitmap.index.options` | Not set | JSON object containing Bitmap build options. Unqualified keys are scoped to `bitmap-index`. | +| `fields..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..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. + + + + + +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. + + + + + +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' +); +``` + + + + + +```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 reader = readBuilder.newRead().createReader(plan)) { + reader.forEachRemaining(row -> consume(row)); +} +``` + + + + + +### 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 54e33325d5d0..000000000000 --- a/docs/docs/primary-key-table/vector-index.md +++ /dev/null @@ -1,255 +0,0 @@ ---- -title: "Vector Index" -sidebar_position: 9 ---- - - - -# 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 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 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` column to Paimon's fixed-length -`VECTOR` 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..pk-vector.index.type` | Yes | ANN implementation, such as `ivf-flat`, `ivf-pq`, `ivf-hnsw-flat`, `ivf-hnsw-sq`, or `lumina`. | -| `fields..pk-vector.distance.metric` | No | `l2`, `cosine`, or `inner_product`. The default is `inner_product`. | -| `fields..pk-vector.index.options` | No | JSON object containing build options for the selected ANN implementation. Unqualified keys are scoped to that implementation. | -| `fields..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..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 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 c1516812420f..57077ff895cb 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 7e3967dadea2..e88793b7fe27 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 ac25c7692093..f739c86f1bef 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.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 @@ protected Plan postProcessPlan(Plan dataPlan) { || 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 35de4f3145c0..0979f874dbaa 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.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 @@ void testOrdinaryBatchScanUsesSnapshotScopedSortedIndex() { 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 @@ void testDisabledGlobalIndexUsesOrdinaryDataPlan() { assertThat(result.splits()).singleElement().isInstanceOf(DataSplit.class); } - private static TableSchema tableSchema(boolean globalIndexEnabled) { + private static TableSchema tableSchema(boolean globalIndexEnabled, int bucket) { Map 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 @@ private static ScanFixture fixture(GlobalIndexReader reader) { } 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 000000000000..0fadbf9b0606 --- /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 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 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 000000000000..4a09bfc18b2b --- /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")) + ) + } + } +}