You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by lz...@apache.org on 2026/08/18 13:42:47 UTC
(paimon) branch master updated: [core] Add Multivalue global index for array membership (#9282)
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 e6e481a222 [core] Add Multivalue global index for array membership (#9282)
e6e481a222 is described below
commit e6e481a222e94ac82ace7ff673903feb280cb272
Author: Jingsong Lee <ji...@gmail.com>
AuthorDate: Tue Aug 18 21:42:39 2026 +0800
[core] Add Multivalue global index for array membership (#9282)
---
docs/docs/multimodal-table/global-index.mdx | 14 +-
docs/docs/multimodal-table/global-index/bitmap.mdx | 2 +-
docs/docs/multimodal-table/global-index/btree.mdx | 2 +-
.../multimodal-table/global-index/multivalue.mdx | 118 +++++++++++
docs/docs/multimodal-table/index.mdx | 2 +-
docs/docs/primary-key-table/global-index.mdx | 90 +++++---
docs/generated/core_configuration.html | 6 +
docs/sidebars.js | 1 +
.../main/java/org/apache/paimon/CoreOptions.java | 15 ++
....apache.paimon.globalindex.GlobalIndexerFactory | 1 +
.../bitmap/MultiValueBitmapIndexReaderTest.java | 7 +-
.../java/org/apache/paimon/KeyValueFileStore.java | 3 +-
.../globalindex/GlobalIndexBuilderUtils.java | 49 +++++
.../sorted/SortedGlobalIndexWriter.java | 27 ++-
.../sorted/SortedSingleColumnIndexWriter.java | 29 ++-
.../pk/BucketedPrimaryKeyIndexMaintainer.java | 1 +
.../paimon/index/pk/PrimaryKeyIndexDefinition.java | 1 +
.../index/pk/PrimaryKeyIndexDefinitions.java | 16 +-
.../pksorted/BucketedSortedIndexMaintainer.java | 2 +-
.../index/pksorted/PkSortedIndexBuilder.java | 47 +++--
.../paimon/index/pksorted/PkSortedIndexFile.java | 13 +-
.../org/apache/paimon/schema/SchemaManager.java | 1 +
.../org/apache/paimon/schema/SchemaValidation.java | 28 ++-
.../paimon/table/source/PrimaryKeyBatchScan.java | 3 +-
.../table/source/PrimaryKeySortedIndexScan.java | 23 ++-
.../globalindex/GlobalIndexBuilderUtilsTest.java | 23 +++
.../sorted/SortedGlobalIndexScannerTest.java | 11 +-
.../sorted/SortedGlobalIndexTestUtils.java | 28 ++-
.../sorted/SortedGlobalIndexWriterTest.java | 36 +++-
.../index/pk/PrimaryKeyIndexDefinitionsTest.java | 13 +-
.../index/pksorted/PkSortedIndexBuilderTest.java | 177 +++++++++++++++-
.../index/pksorted/PkSortedIndexFileTest.java | 27 +++
.../PrimaryKeySortedIndexMaintenanceTest.java | 2 +-
.../pksorted/PrimaryKeySortedIndexOptionsTest.java | 22 ++
.../paimon/operation/PrimaryKeyIndexWriteTest.java | 26 ++-
.../PrimaryKeySortedIndexValidationTest.java | 53 ++++-
.../apache/paimon/schema/SchemaManagerTest.java | 30 +++
.../table/MultiValueGlobalIndexTableTest.java | 164 +++++++++++++++
.../source/PrimaryKeySortedIndexReadTest.java | 53 ++++-
.../source/PrimaryKeySortedIndexScanTest.java | 60 +++++-
.../flink/action/DataEvolutionMergeIntoAction.java | 40 +++-
.../flink/globalindex/SortedIndexTopoBuilder.java | 213 +++++++++++++------
.../procedure/CreateGlobalIndexProcedure.java | 3 +
.../paimon/flink/SortedGlobalIndexITCase.java | 100 +++++++++
.../globalindex/SortedIndexTopoBuilderTest.java | 8 +-
.../globalindex/sorted/SortedIndexTopoBuilder.java | 230 +++++++++++++++++++--
.../procedure/CreateGlobalIndexProcedureTest.scala | 44 ++++
47 files changed, 1666 insertions(+), 198 deletions(-)
diff --git a/docs/docs/multimodal-table/global-index.mdx b/docs/docs/multimodal-table/global-index.mdx
index 434a248da4..a506a08759 100644
--- a/docs/docs/multimodal-table/global-index.mdx
+++ b/docs/docs/multimodal-table/global-index.mdx
@@ -34,6 +34,7 @@ without full-table scans. Paimon supports multiple global index types:
- **[BTree Index](./global-index/btree)**: A B-tree based index for scalar column lookups. Supports equality, IN, range predicates, and can be combined across multiple columns with AND/OR logic.
- **[Bitmap Index](./global-index/bitmap)**: A bitmap based index for enum-like scalar dimensions and tag columns. Supports equality, IN, prefix match on string columns, complement predicates, and null checks with compressed row-id bitmaps.
+- **[Multivalue Index](./global-index/multivalue)**: A bitmap-backed index for element-membership predicates on `ARRAY` columns.
- **[Vector Index](./global-index/vector)**: An approximate nearest neighbor (ANN) index powered by Paimon's vector index library for vector similarity search.
- **[Full-Text Index](./global-index/full-text)**: A full-text search index backed by the native full-text engine for text retrieval. Supports term matching and relevance scoring.
- **[Hybrid Search](./global-index/hybrid-search)**: A multi-route search API that combines results from multiple vector routes, multiple full-text routes, or both before reading table rows.
@@ -42,6 +43,7 @@ without full-table scans. Paimon supports multiple global index types:
|---|---|---|
| BTree | Scalar filters on numeric, string, date, and timestamp columns | Best when predicates are selective, such as equality, IN, range, and null checks. |
| Bitmap | Enum-like dimensions and tag columns | Best for equality, IN, string prefix match, complement predicates, and null checks over compressed row-id bitmaps. |
+| Multivalue | Membership tests on arrays of supported scalar elements | Best for `ARRAY_CONTAINS(array_column, element)`. |
| Vector | Top-K similarity search on embeddings | Uses ANN algorithms. Tune build-time and search-time options to balance recall, latency, and index size. |
| Full-Text | Keyword search over text columns | Uses full-text scoring and tokenizer configuration stored with each index file. |
| Hybrid Search | Combining multiple vector routes, multiple full-text routes, or vector and full-text retrieval together | Runs multiple scored routes and merges them with a ranker before reading rows. |
@@ -66,6 +68,7 @@ Create a table with the required properties:
CREATE TABLE my_table (
id INT,
name STRING,
+ tags ARRAY<STRING>,
embedding ARRAY<FLOAT>,
content STRING
) TBLPROPERTIES (
@@ -89,6 +92,7 @@ schema = Schema.from_pyarrow_schema(
pa.schema([
pa.field("id", pa.int32()),
pa.field("name", pa.string()),
+ pa.field("tags", pa.list_(pa.string())),
pa.field("embedding", pa.list_(pa.float32())),
pa.field("content", pa.string()),
]),
@@ -395,12 +399,12 @@ These table options affect global index build and read behavior:
|---|---|---|
| `global-index.enabled` | `true` | Whether scans can use global indexes. |
| `global-index.search-mode` | Not set | Legacy fallback search mode for global-index queries. Family-specific options take precedence. |
-| `scalar-index.search-mode` | `fast` | Search mode for BTree and Bitmap queries. |
+| `scalar-index.search-mode` | `fast` | Search mode for BTree, Bitmap, and Multivalue queries. |
| `vector-index.search-mode` | `fast` | Search mode for vector queries. |
| `full-text-index.search-mode` | `fast` | Search mode for full-text queries. |
| `global-index.external-path` | Not set | Root directory for global index files. If not set, files are stored under the table index directory. |
| `global-index.column-update-action` | `THROW_ERROR` | Action for updates to indexed columns. `THROW_ERROR` rejects the update, `DROP_PARTITION_INDEX` drops affected partition indexes, and `IGNORE` keeps existing index files unchanged during the update and causes the next incremental index build to scan active data manifests and rebuild affected row ranges. Index files created before source metadata was introduced are not refreshed automatically. |
-| `sorted-index.records-per-range` | `10000000` | Expected number of records per sorted global index file for BTree and Bitmap builds. |
+| `sorted-index.records-per-range` | `10000000` | Expected number of records per sorted global index file for BTree, Bitmap, and Multivalue builds. |
| `sorted-index.build.max-parallelism` | `4096` | Maximum Flink or Spark parallelism for building sorted global indexes. |
| `global-index.row-count-per-shard` | `100000` | Target row count per shard for non-sorted global index builds such as vector and full-text indexes. |
| `global-index.build.max-shard` | `32` | Preferred maximum shard count for global index builds. |
@@ -420,6 +424,12 @@ Use Bitmap indexes for enum-like dimensions and tag columns queried by equality,
match, complement, or null predicates. See [Bitmap Index](./global-index/bitmap) for build examples,
options, and file format details.
+## Multivalue Index
+
+Use Multivalue indexes for element-membership predicates on `ARRAY` columns. See
+[Multivalue Index](./global-index/multivalue) for Data Evolution build examples, Core query usage,
+options, and null/empty membership semantics.
+
## Vector Index
Use Vector indexes for approximate nearest neighbor (ANN) search. See [Vector Index](./global-index/vector)
diff --git a/docs/docs/multimodal-table/global-index/bitmap.mdx b/docs/docs/multimodal-table/global-index/bitmap.mdx
index 154110af28..fce2324571 100644
--- a/docs/docs/multimodal-table/global-index/bitmap.mdx
+++ b/docs/docs/multimodal-table/global-index/bitmap.mdx
@@ -171,7 +171,7 @@ print(matched_files)
| Option | Default | Description |
|---|---|---|
-| `sorted-index.records-per-range` | `10000000` | Expected number of records per sorted global index file for BTree and Bitmap builds. |
+| `sorted-index.records-per-range` | `10000000` | Expected number of records per sorted global index file for BTree, Bitmap, and Multivalue builds. |
| `sorted-index.build.max-parallelism` | `4096` | Maximum Flink or Spark parallelism for building sorted global indexes. |
| `bitmap-index.dictionary-block-size` | `16 kb` | Target size of dictionary blocks in bitmap global index files. Smaller blocks reduce dictionary read amplification for high-cardinality columns; larger blocks reduce dictionary block index size. |
| `bitmap-index.compression` | `none` | Compression algorithm for bitmap dictionary blocks and the dictionary block index. Supported values are the same block codecs as BTree index, such as `none`, `lz4`, `lzo`, and `zstd`. |
diff --git a/docs/docs/multimodal-table/global-index/btree.mdx b/docs/docs/multimodal-table/global-index/btree.mdx
index afc953ebe5..cfb01d3375 100644
--- a/docs/docs/multimodal-table/global-index/btree.mdx
+++ b/docs/docs/multimodal-table/global-index/btree.mdx
@@ -147,7 +147,7 @@ print(matched_files)
| Option | Default | Description |
|---|---|---|
-| `sorted-index.records-per-range` | `10000000` | Expected number of records per sorted global index file for BTree and Bitmap builds. |
+| `sorted-index.records-per-range` | `10000000` | Expected number of records per sorted global index file for BTree, Bitmap, and Multivalue builds. |
| `sorted-index.build.max-parallelism` | `4096` | Maximum Flink or Spark parallelism for building sorted global indexes. |
| `btree-index.block-size` | `64 kb` | Block size used by BTree index files. |
| `btree-index.cache-size` | `128 mb` | Cache size used by BTree index readers. |
diff --git a/docs/docs/multimodal-table/global-index/multivalue.mdx b/docs/docs/multimodal-table/global-index/multivalue.mdx
new file mode 100644
index 0000000000..484da0dcdf
--- /dev/null
+++ b/docs/docs/multimodal-table/global-index/multivalue.mdx
@@ -0,0 +1,118 @@
+---
+title: "Multivalue Index"
+sidebar_position: 3
+---
+
+<!--
+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.
+-->
+
+# Multivalue Index
+
+A Multivalue index maps every distinct non-null element of an `ARRAY` column to a compressed
+64-bit row-id bitmap. It accelerates element-membership predicates without treating the complete
+array as one scalar key.
+
+For membership lookup, the index has these semantics:
+
+- Duplicate elements in one array contribute the row ID only once.
+- Null arrays, empty arrays, and null elements contribute no posting. The index does not
+ distinguish those cases and does not accelerate array `IS NULL` or `IS NOT NULL` predicates.
+
+The indexed column must be an `ARRAY` whose element type is supported by the global-index key
+serializer. A Multivalue index is single-column.
+
+## Build on a Data Evolution Table
+
+Create a table with the regular global-index prerequisites:
+
+```sql
+CREATE TABLE my_table (
+ id INT,
+ tags ARRAY<STRING>
+) TBLPROPERTIES (
+ 'bucket' = '-1',
+ 'row-tracking.enabled' = 'true',
+ 'data-evolution.enabled' = 'true',
+ 'global-index.enabled' = 'true'
+);
+```
+
+Flink and Spark can build the index with `create_global_index`:
+
+```sql
+CALL sys.create_global_index(
+ table => 'db.my_table',
+ index_column => 'tags',
+ index_type => 'multivalue',
+ options => 'sorted-index.records-per-range=10000000'
+);
+```
+
+Arrays do not need to be ordered. Flink and Spark first assign source rows to stable row-id ranges,
+expand every non-null element into an `(element, row-id)` entry, and sort those entries with their
+spillable sort paths. The writer then streams the sorted entries into bitmap posting lists while
+retaining the original source-row count for index coverage, including ranges containing only null
+or empty arrays.
+
+Build the procedure again after appending rows to cover newly added row-id ranges. The
+`scalar-index.search-mode` option controls whether an indexed scan includes uncovered rows:
+`fast` returns indexed coverage only, while `full` and `detail` retain their regular global-index
+coverage behavior.
+
+## Query Through Paimon Core
+
+The Core predicate API exposes array membership directly:
+
+```java
+int tagsFieldIndex = table.rowType().getFieldIndex("tags");
+Predicate predicate = new PredicateBuilder(table.rowType())
+ .arrayContains(tagsFieldIndex, BinaryString.fromString("blue"));
+
+ReadBuilder readBuilder = table.newReadBuilder().withFilter(predicate);
+TableScan.Plan plan = readBuilder.newScan().plan();
+```
+
+Paimon automatically uses matching Multivalue global index files and maps their row IDs to Data
+Evolution splits. SQL engines use the index after their connector translates an array-membership
+expression to Paimon's `ARRAY_CONTAINS` predicate; that connector translation is separate from the
+Core index implementation.
+
+## Drop the Index
+
+```sql
+CALL sys.drop_global_index(
+ table => 'db.my_table',
+ index_column => 'tags',
+ index_type => 'multivalue'
+);
+```
+
+## Options
+
+| Option | Default | Description |
+|---|---|---|
+| `sorted-index.records-per-range` | `10000000` | Expected number of source records per generated Multivalue index file. |
+| `sorted-index.build.max-parallelism` | `4096` | Maximum Flink or Spark parallelism for building the index. |
+| `multivalue-index.dictionary-block-size` | `16 kb` | Target size of the block-indexed element dictionary. |
+| `multivalue-index.compression` | `none` | Compression algorithm for dictionary blocks. |
+| `multivalue-index.compression-level` | `1` | Compression level used by codecs which support levels. |
+
+The expanded entries use the engine's spillable sort, and the writer keeps only the current
+element's posting list in memory. `sorted-index.records-per-range` bounds the source-row range—and
+therefore the largest posting list—owned by one Data Evolution index file.
diff --git a/docs/docs/multimodal-table/index.mdx b/docs/docs/multimodal-table/index.mdx
index 58363d500e..2f26c41bb3 100644
--- a/docs/docs/multimodal-table/index.mdx
+++ b/docs/docs/multimodal-table/index.mdx
@@ -38,7 +38,7 @@ Key capabilities:
- **[Variant Storage](./variant)**: Store and query schema-flexible semi-structured data with optional typed sub-column shredding.
- **[Blob Storage](./blob)**: Store large binary objects (images, videos, audio) in dedicated `.blob` files with efficient column projection.
- **[Vector Storage](./vector)**: Store and manage vector embeddings in dedicated Vortex-format files optimized for vector workloads.
-- **[Global Index](./global-index)**: Build BTree, Bitmap, vector, and full-text indexes for efficient lookups and similarity search.
+- **[Global Index](./global-index)**: Build BTree, Bitmap, Multivalue, vector, and full-text indexes for efficient lookups and similarity search.
Data Evolution, Blob Storage, Vector Storage, and Global Index require the following table properties.
Variant Storage can also be used in a regular Paimon table without these properties:
diff --git a/docs/docs/primary-key-table/global-index.mdx b/docs/docs/primary-key-table/global-index.mdx
index b96341e7b1..2bd2f3886e 100644
--- a/docs/docs/primary-key-table/global-index.mdx
+++ b/docs/docs/primary-key-table/global-index.mdx
@@ -27,10 +27,10 @@ under the License.
# Primary-Key Indexes
-Primary-key tables can maintain Vector, Full Text, 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 tables can maintain Vector, Full Text, BTree, Bitmap, and Multivalue 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
@@ -62,6 +62,22 @@ See [Bitmap Index](../multimodal-table/global-index/bitmap) for predicate and fo
</TabItem>
+<TabItem value="multivalue" label="Multivalue">
+
+Use Multivalue for element-membership predicates on an `ARRAY` column. It stores one compressed
+bitmap posting list per distinct non-null element and accelerates Paimon Core `ARRAY_CONTAINS`
+predicates. Null arrays, empty arrays, and null elements produce no membership posting; array null
+checks fall back to the ordinary data path.
+
+The Core predicate and index pushdown are available independently of SQL connectors. A connector
+uses this index after it translates its array-membership expression to Paimon's `ARRAY_CONTAINS`
+predicate.
+
+For an append-only or Data Evolution table whose Multivalue index is built independently, see
+[Global Multivalue Index](../multimodal-table/global-index/multivalue).
+
+</TabItem>
+
<TabItem value="vector" label="Vector">
Use Vector for approximate nearest-neighbor (ANN) Top-K search on embeddings that are updated with
@@ -87,8 +103,8 @@ For an append-only or Data Evolution table whose full-text index is built indepe
</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-full-text.index.columns`, `pk-btree.index.columns`, and
-`pk-bitmap.index.columns`.
+one of `pk-vector.index.columns`, `pk-full-text.index.columns`, `pk-btree.index.columns`,
+`pk-bitmap.index.columns`, and `pk-multivalue.index.columns`.
## Requirements
@@ -104,16 +120,17 @@ supported by this configuration.
<Tabs groupId="primary-key-index-requirements">
-<TabItem value="scalar" label="BTree and Bitmap">
+<TabItem value="sorted" label="BTree, Bitmap, and Multivalue">
-BTree and Bitmap additionally require:
+BTree, Bitmap, and Multivalue additionally require:
- `deletion-vectors.enabled = true`.
- `deletion-vectors.merge-on-read = false`.
-- A supported scalar column type.
+- A supported scalar column type for BTree and Bitmap, or an `ARRAY` of supported scalar elements
+ for Multivalue.
-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.
+Each entry creates an independent single-column index. Multiple columns are supported as long as a
+column is not listed by another primary-key index family.
</TabItem>
@@ -149,8 +166,8 @@ Exactly one full-text column is currently supported per table.
## Create a Table
-The following table uses all four families on different columns: Vector for `embedding`, Full
-Text for `content`, BTree for `amount`, and Bitmap for `status`.
+The following table uses all five families on different columns: Vector for `embedding`, Full Text
+for `content`, BTree for `amount`, Bitmap for `status`, and Multivalue for `tags`.
<Tabs groupId="primary-key-index-create-table">
@@ -161,6 +178,7 @@ CREATE TABLE items (
id BIGINT,
status STRING,
amount DECIMAL(12, 2),
+ tags ARRAY<STRING>,
content STRING,
embedding ARRAY<FLOAT> COMMENT '__VECTOR_FIELD;3',
PRIMARY KEY (id) NOT ENFORCED
@@ -176,7 +194,9 @@ CREATE TABLE items (
'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"}'
+ 'fields.status.pk-bitmap.index.options' = '{"dictionary-block-size":"16 kb"}',
+ 'pk-multivalue.index.columns' = 'tags',
+ 'fields.tags.pk-multivalue.index.options' = '{"dictionary-block-size":"16 kb"}'
);
```
@@ -193,6 +213,7 @@ CREATE TABLE items (
id BIGINT,
status STRING,
amount DECIMAL(12, 2),
+ tags ARRAY<STRING>,
content STRING,
embedding ARRAY<FLOAT> COMMENT '__VECTOR_FIELD;3'
) USING paimon
@@ -209,7 +230,9 @@ TBLPROPERTIES (
'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"}'
+ 'fields.status.pk-bitmap.index.options' = '{"dictionary-block-size":"16 kb"}',
+ 'pk-multivalue.index.columns' = 'tags',
+ 'fields.tags.pk-multivalue.index.options' = '{"dictionary-block-size":"16 kb"}'
);
```
@@ -234,6 +257,11 @@ schema validation.
| `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`. |
+| `pk-multivalue.index.columns` | Not set | Comma-separated `ARRAY` columns which own independent Multivalue indexes. |
+| `fields.<column>.pk-multivalue.index.options` | Not set | JSON object containing Multivalue build options. Unqualified keys are scoped to `multivalue-index`. |
+| `multivalue-index.dictionary-block-size` | `16 kb` | Target dictionary block size. |
+| `multivalue-index.compression` | `none` | Dictionary-block compression algorithm. |
+| `multivalue-index.compression-level` | `1` | Dictionary-block compression level. |
| `global-index.search-mode` | Not set | Legacy fallback search mode. Family-specific options take precedence. |
| `vector-index.search-mode` | `fast` | Search mode for primary-key Vector queries. `full` and `detail` search uncovered files exactly. |
| `full-text-index.search-mode` | `fast` | Search mode for primary-key Full Text queries. Only `fast` is currently supported. |
@@ -279,8 +307,8 @@ Coverage behavior depends on the index family and search mode. `vector-index.sea
queries search only data covered by an active index group. Uncovered files are not searched, so
partial coverage can make results incomplete.
-- BTree and Bitmap scans always read uncovered files through the ordinary data path. Partial
- coverage affects their acceleration, not result completeness. The original scalar predicate and
+- BTree, Bitmap, and Multivalue scans always read uncovered files through the ordinary data path.
+ Partial coverage affects their acceleration, not result completeness. The original predicate and
deletion vectors are applied after index pruning.
- Vector search in `full` or `detail` mode evaluates files without an active ANN group exactly and
merges those results with ANN candidates. In the default `fast` mode, those files are omitted.
@@ -290,13 +318,26 @@ persistent archives and ignores uncovered files; `full` and `detail` are rejecte
merge-aware logical-row fallback is not implemented. Consequently, newly appended Level-0 rows
become full-text searchable only after compaction publishes an eligible data file and archive.
-## BTree and Bitmap Queries
+## BTree, Bitmap, and Multivalue Queries
+
+BTree, Bitmap, and Multivalue indexes are applied automatically to snapshot-scoped Core batch
+scans after a supported predicate reaches Paimon. BTree and Bitmap can accelerate equality, `IN`,
+null checks, comparisons and ranges, complement predicates, and supported string predicates.
+Multivalue accelerates `ARRAY_CONTAINS(array_column, element)`.
+
+The Java/Core predicate can be constructed directly:
+
+```java
+Predicate predicate = new PredicateBuilder(rowType)
+ .arrayContains(tagsFieldIndex, BinaryString.fromString("blue"));
+ReadBuilder readBuilder = table.newReadBuilder().withFilter(predicate);
+TableScan.Plan plan = readBuilder.newScan().plan();
+```
-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.
+SQL engines need a connector-specific expression conversion before their `array_contains` or
+equivalent function reaches this Core predicate.
-Paimon combines indexed scalar results as follows:
+Paimon combines indexed predicate results as follows:
- For `AND`, any usable indexed child can narrow a source file; unindexed children remain residual
filters.
@@ -513,7 +554,10 @@ create a table with the desired definition and migrate the data.
## Limitations
-- BTree and Bitmap definitions are single-column indexes.
+- BTree, Bitmap, and Multivalue definitions are single-column indexes.
+- Multivalue supports arrays of scalar element types accepted by the global-index key serializer.
+ Null elements are not indexed. Higher array and element cardinality increases the expanded-sort
+ I/O and index-file size.
- Exactly one vector index column is currently supported per table.
- Exactly one full-text index column is currently supported per table.
- Only `FLOAT` vectors are supported.
diff --git a/docs/generated/core_configuration.html b/docs/generated/core_configuration.html
index 1d40881dae..3fb25ebce1 100644
--- a/docs/generated/core_configuration.html
+++ b/docs/generated/core_configuration.html
@@ -1290,6 +1290,12 @@ For an internal format table in a REST catalog, it also makes the catalog own th
<td>String</td>
<td>Comma-separated character columns indexed by primary-key full-text indexes. The first release supports exactly one column.</td>
</tr>
+ <tr>
+ <td><h5>pk-multivalue.index.columns</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>Comma-separated ARRAY columns indexed by primary-key Multivalue indexes.</td>
+ </tr>
<tr>
<td><h5>pk-vector.index.columns</h5></td>
<td style="word-wrap: break-word;">(none)</td>
diff --git a/docs/sidebars.js b/docs/sidebars.js
index 606457575c..cb70d4ad19 100644
--- a/docs/sidebars.js
+++ b/docs/sidebars.js
@@ -128,6 +128,7 @@ const sidebars = {
"items": [
"multimodal-table/global-index/btree",
"multimodal-table/global-index/bitmap",
+ "multimodal-table/global-index/multivalue",
"multimodal-table/global-index/vector",
"multimodal-table/global-index/full-text",
"multimodal-table/global-index/hybrid-search"
diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
index 8e5e8c1300..9a6fd514df 100644
--- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
+++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
@@ -2952,6 +2952,13 @@ public class CoreOptions implements Serializable {
.withDescription(
"Comma-separated columns indexed by primary-key Bitmap indexes.");
+ public static final ConfigOption<String> PK_MULTIVALUE_INDEX_COLUMNS =
+ key("pk-multivalue.index.columns")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "Comma-separated ARRAY columns indexed by primary-key Multivalue indexes.");
+
public static final ConfigOption<String> PK_FULL_TEXT_INDEX_COLUMNS =
key("pk-full-text.index.columns")
.stringType()
@@ -4605,6 +4612,10 @@ public class CoreOptions implements Serializable {
return primaryKeyIndexColumns(PK_BITMAP_INDEX_COLUMNS);
}
+ public List<String> primaryKeyMultiValueIndexColumns() {
+ return primaryKeyIndexColumns(PK_MULTIVALUE_INDEX_COLUMNS);
+ }
+
public List<String> primaryKeyFullTextIndexColumns() {
return primaryKeyIndexColumns(PK_FULL_TEXT_INDEX_COLUMNS);
}
@@ -4625,6 +4636,10 @@ public class CoreOptions implements Serializable {
return primaryKeySortedIndexOptions(column, "pk-bitmap", "bitmap-index.");
}
+ public Options primaryKeyMultiValueIndexOptions(String column) {
+ return primaryKeySortedIndexOptions(column, "pk-multivalue", "multivalue-index.");
+ }
+
public Options primaryKeyFullTextIndexOptions(String column) {
String optionKey = "fields." + column + ".pk-full-text.index.options";
TreeMap<String, String> resolved = new TreeMap<>();
diff --git a/paimon-common/src/main/resources/META-INF/services/org.apache.paimon.globalindex.GlobalIndexerFactory b/paimon-common/src/main/resources/META-INF/services/org.apache.paimon.globalindex.GlobalIndexerFactory
index 06722ccd2d..b0cd8e0108 100644
--- a/paimon-common/src/main/resources/META-INF/services/org.apache.paimon.globalindex.GlobalIndexerFactory
+++ b/paimon-common/src/main/resources/META-INF/services/org.apache.paimon.globalindex.GlobalIndexerFactory
@@ -15,3 +15,4 @@
org.apache.paimon.globalindex.btree.BTreeGlobalIndexerFactory
org.apache.paimon.globalindex.bitmap.BitmapGlobalIndexerFactory
+org.apache.paimon.globalindex.bitmap.MultiValueGlobalIndexerFactory
diff --git a/paimon-common/src/test/java/org/apache/paimon/globalindex/bitmap/MultiValueBitmapIndexReaderTest.java b/paimon-common/src/test/java/org/apache/paimon/globalindex/bitmap/MultiValueBitmapIndexReaderTest.java
index 98a5cba9f6..a268c549f8 100644
--- a/paimon-common/src/test/java/org/apache/paimon/globalindex/bitmap/MultiValueBitmapIndexReaderTest.java
+++ b/paimon-common/src/test/java/org/apache/paimon/globalindex/bitmap/MultiValueBitmapIndexReaderTest.java
@@ -30,6 +30,7 @@ import org.apache.paimon.globalindex.GlobalIndexReader;
import org.apache.paimon.globalindex.GlobalIndexResult;
import org.apache.paimon.globalindex.GlobalIndexSingleColumnWriter;
import org.apache.paimon.globalindex.GlobalIndexer;
+import org.apache.paimon.globalindex.GlobalIndexerFactoryUtils;
import org.apache.paimon.globalindex.ResultEntry;
import org.apache.paimon.globalindex.SortedGlobalIndexer;
import org.apache.paimon.globalindex.SortedIndexFileMeta;
@@ -206,10 +207,8 @@ class MultiValueBitmapIndexReaderTest {
@Test
void testFactoryAndTypeValidation() throws Exception {
- MultiValueGlobalIndexerFactory factory = new MultiValueGlobalIndexerFactory();
- assertThat(factory.identifier()).isEqualTo(MultiValueGlobalIndexerFactory.IDENTIFIER);
- assertThat(factory.create(dataField, new Options()))
- .isInstanceOf(MultiValueGlobalIndexer.class);
+ assertThat(GlobalIndexerFactoryUtils.load(MultiValueGlobalIndexerFactory.IDENTIFIER))
+ .isInstanceOf(MultiValueGlobalIndexerFactory.class);
assertThat(globalIndexer).isInstanceOf(SortedGlobalIndexer.class);
GlobalIndexKeyExtractor extractor = ((SortedGlobalIndexer) globalIndexer).keyExtractor();
assertThat(extractor.keyType()).isEqualTo(DataTypes.STRING());
diff --git a/paimon-core/src/main/java/org/apache/paimon/KeyValueFileStore.java b/paimon-core/src/main/java/org/apache/paimon/KeyValueFileStore.java
index 8e072a0736..d5ded947ad 100644
--- a/paimon-core/src/main/java/org/apache/paimon/KeyValueFileStore.java
+++ b/paimon-core/src/main/java/org/apache/paimon/KeyValueFileStore.java
@@ -187,7 +187,8 @@ public class KeyValueFileStore extends AbstractFileStore<KeyValue> {
if (writeOptions.primaryKeyVectorIndexEnabled()
|| writeOptions.primaryKeyFullTextIndexEnabled()
|| !writeOptions.primaryKeyBTreeIndexColumns().isEmpty()
- || !writeOptions.primaryKeyBitmapIndexColumns().isEmpty()) {
+ || !writeOptions.primaryKeyBitmapIndexColumns().isEmpty()
+ || !writeOptions.primaryKeyMultiValueIndexColumns().isEmpty()) {
primaryKeyIndexMaintainerFactory =
BucketedPrimaryKeyIndexMaintainer.Factory.create(
newIndexFileHandler(), newReaderFactoryBuilder(), schema);
diff --git a/paimon-core/src/main/java/org/apache/paimon/globalindex/GlobalIndexBuilderUtils.java b/paimon-core/src/main/java/org/apache/paimon/globalindex/GlobalIndexBuilderUtils.java
index 5e838e204b..657ba0098c 100644
--- a/paimon-core/src/main/java/org/apache/paimon/globalindex/GlobalIndexBuilderUtils.java
+++ b/paimon-core/src/main/java/org/apache/paimon/globalindex/GlobalIndexBuilderUtils.java
@@ -275,6 +275,55 @@ public class GlobalIndexBuilderUtils {
return result;
}
+ /** Splits contiguous build ranges into globally aligned, source-row bounded shards. */
+ public static Map<Range, List<Split>> shardSplitsByRowRange(
+ Map<Range, List<Split>> rangeSplits, long rowsPerShard) {
+ checkArgument(rowsPerShard > 0, "Rows per sorted-index shard must be positive.");
+ Map<Range, List<Split>> result = new LinkedHashMap<>();
+ for (Map.Entry<Range, List<Split>> entry : rangeSplits.entrySet()) {
+ Range buildRange = entry.getKey();
+ checkArgument(
+ buildRange.from >= 0,
+ "Sorted-index row IDs must be non-negative, but range starts at %s.",
+ buildRange.from);
+ long shardStart = (buildRange.from / rowsPerShard) * rowsPerShard;
+ while (shardStart <= buildRange.to) {
+ long unboundedShardEnd = shardStart + rowsPerShard - 1;
+ long shardEnd = unboundedShardEnd < shardStart ? Long.MAX_VALUE : unboundedShardEnd;
+ Range shardRange =
+ new Range(
+ Math.max(buildRange.from, shardStart),
+ Math.min(buildRange.to, shardEnd));
+ List<Split> shardSplits = new ArrayList<>();
+ for (Split split : entry.getValue()) {
+ DataSplit dataSplit;
+ List<Range> splitRanges;
+ if (split instanceof IndexedSplit) {
+ IndexedSplit indexedSplit = (IndexedSplit) split;
+ dataSplit = indexedSplit.dataSplit();
+ splitRanges = indexedSplit.rowRanges();
+ } else {
+ dataSplit = (DataSplit) split;
+ splitRanges = calcRowRanges(Collections.singletonList(dataSplit));
+ }
+ List<Range> intersections =
+ Range.and(splitRanges, Collections.singletonList(shardRange));
+ if (!intersections.isEmpty()) {
+ shardSplits.add(new IndexedSplit(dataSplit, intersections, null));
+ }
+ }
+ if (!shardSplits.isEmpty()) {
+ result.put(shardRange, shardSplits);
+ }
+ if (shardEnd == Long.MAX_VALUE) {
+ break;
+ }
+ shardStart = shardEnd + 1;
+ }
+ }
+ return result;
+ }
+
private static List<DataSplit> splitByContiguousRowRange(DataSplit split) {
List<DataFileMeta> input = split.dataFiles();
RangeHelper<DataFileMeta> rangeHelper = new RangeHelper<>(DataFileMeta::nonNullRowIdRange);
diff --git a/paimon-core/src/main/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexWriter.java b/paimon-core/src/main/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexWriter.java
index 52e6d12ee8..52cfd6479f 100644
--- a/paimon-core/src/main/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexWriter.java
+++ b/paimon-core/src/main/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexWriter.java
@@ -21,9 +21,12 @@ package org.apache.paimon.globalindex.sorted;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.data.InternalRow.FieldGetter;
+import org.apache.paimon.globalindex.GlobalIndexKeyExtractor;
import org.apache.paimon.globalindex.GlobalIndexSingleColumnWriter;
import org.apache.paimon.globalindex.GlobalIndexWriter;
+import org.apache.paimon.globalindex.GlobalIndexer;
import org.apache.paimon.globalindex.ResultEntry;
+import org.apache.paimon.globalindex.SortedGlobalIndexer;
import org.apache.paimon.index.DataEvolutionIndexSourceMeta;
import org.apache.paimon.index.IndexFileMeta;
import org.apache.paimon.io.CompactIncrement;
@@ -61,6 +64,7 @@ public class SortedGlobalIndexWriter implements Serializable {
private final long recordsPerRange;
private DataField indexField;
+ private GlobalIndexKeyExtractor keyExtractor;
public SortedGlobalIndexWriter(Table table, String indexType) {
this(table, indexType, ((FileStoreTable) table).coreOptions().toConfiguration());
@@ -82,9 +86,19 @@ public class SortedGlobalIndexWriter implements Serializable {
indexField,
table.fullName());
this.indexField = rowType.getField(indexField);
+ GlobalIndexer indexer = GlobalIndexer.create(indexType, this.indexField, options);
+ checkArgument(
+ indexer instanceof SortedGlobalIndexer,
+ "Index algorithm %s does not expose sorted index keys.",
+ indexType);
+ this.keyExtractor = ((SortedGlobalIndexer) indexer).keyExtractor();
return this;
}
+ public GlobalIndexKeyExtractor keyExtractor() {
+ return keyExtractor;
+ }
+
public long recordsPerRange() {
return recordsPerRange;
}
@@ -92,9 +106,8 @@ public class SortedGlobalIndexWriter implements Serializable {
public List<CommitMessage> buildForSinglePartition(
Range rowRange, BinaryRow partition, Iterator<InternalRow> data, long scanSnapshotId)
throws IOException {
- FieldGetter indexFieldGetter = InternalRow.createFieldGetter(indexField.type(), 0);
- try (SortedSingleColumnIndexWriter writer =
- new SortedSingleColumnIndexWriter(recordsPerRange, this::createWriter)) {
+ FieldGetter indexFieldGetter = InternalRow.createFieldGetter(keyExtractor.keyType(), 0);
+ try (SortedSingleColumnIndexWriter writer = createTaskWriter(rowRange)) {
while (data.hasNext()) {
InternalRow row = data.next();
long localRowId = row.getLong(1) - rowRange.from;
@@ -109,6 +122,14 @@ public class SortedGlobalIndexWriter implements Serializable {
}
}
+ /** Creates a task writer which owns file rotation and source-row coverage semantics. */
+ public SortedSingleColumnIndexWriter createTaskWriter(Range rowRange) throws IOException {
+ if (keyExtractor.isIdentity()) {
+ return new SortedSingleColumnIndexWriter(recordsPerRange, this::createWriter);
+ }
+ return SortedSingleColumnIndexWriter.forSourceRowCount(rowRange.count(), createWriter());
+ }
+
public GlobalIndexSingleColumnWriter createWriter() throws IOException {
GlobalIndexWriter indexWriter = createIndexWriter(table, indexType, indexField, options);
if (!(indexWriter instanceof GlobalIndexSingleColumnWriter)) {
diff --git a/paimon-core/src/main/java/org/apache/paimon/globalindex/sorted/SortedSingleColumnIndexWriter.java b/paimon-core/src/main/java/org/apache/paimon/globalindex/sorted/SortedSingleColumnIndexWriter.java
index 80a8a17ff4..95529ee710 100644
--- a/paimon-core/src/main/java/org/apache/paimon/globalindex/sorted/SortedSingleColumnIndexWriter.java
+++ b/paimon-core/src/main/java/org/apache/paimon/globalindex/sorted/SortedSingleColumnIndexWriter.java
@@ -32,11 +32,12 @@ import java.util.List;
import static org.apache.paimon.utils.Preconditions.checkArgument;
import static org.apache.paimon.utils.Preconditions.checkState;
-/** Rotates single-column index writers after a configured number of logical records. */
+/** Writes one sorted index task, optionally rotating by source record count. */
public final class SortedSingleColumnIndexWriter implements Closeable {
private final long recordsPerRange;
- private final Factory factory;
+ @Nullable private final Factory factory;
+ @Nullable private final Long sourceRowCount;
private final List<List<ResultEntry>> resultGroups = new ArrayList<>();
@Nullable private GlobalIndexSingleColumnWriter currentWriter;
@@ -47,14 +48,33 @@ public final class SortedSingleColumnIndexWriter implements Closeable {
checkArgument(recordsPerRange > 0, "Records per range must be positive.");
this.recordsPerRange = recordsPerRange;
this.factory = factory;
+ this.sourceRowCount = null;
+ }
+
+ private SortedSingleColumnIndexWriter(
+ long sourceRowCount, GlobalIndexSingleColumnWriter writer) {
+ checkArgument(sourceRowCount >= 0, "Source row count must be non-negative.");
+ this.recordsPerRange = Long.MAX_VALUE;
+ this.factory = null;
+ this.sourceRowCount = sourceRowCount;
+ this.currentWriter = writer;
+ }
+
+ /** Creates a non-rotating task writer with an explicit source-row coverage count. */
+ public static SortedSingleColumnIndexWriter forSourceRowCount(
+ long sourceRowCount, GlobalIndexSingleColumnWriter writer) {
+ return new SortedSingleColumnIndexWriter(sourceRowCount, writer);
}
public void write(@Nullable Object value, long localRowId) throws IOException {
checkState(!finished, "Cannot write after the sorted index writer is finished.");
- if (currentWriter != null && currentRecordCount >= recordsPerRange) {
+ if (sourceRowCount == null
+ && currentWriter != null
+ && currentRecordCount >= recordsPerRange) {
finishCurrentWriter();
}
if (currentWriter == null) {
+ checkState(factory != null, "Cannot create another source-range index writer.");
currentWriter = factory.create();
}
currentWriter.write(value, localRowId);
@@ -92,7 +112,8 @@ public final class SortedSingleColumnIndexWriter implements Closeable {
}
private void finishCurrentWriter() {
- resultGroups.add(currentWriter.finish());
+ resultGroups.add(
+ currentWriter.finish(sourceRowCount == null ? currentRecordCount : sourceRowCount));
currentWriter = null;
currentRecordCount = 0;
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/index/pk/BucketedPrimaryKeyIndexMaintainer.java b/paimon-core/src/main/java/org/apache/paimon/index/pk/BucketedPrimaryKeyIndexMaintainer.java
index b9a9223c11..128491b9e1 100644
--- a/paimon-core/src/main/java/org/apache/paimon/index/pk/BucketedPrimaryKeyIndexMaintainer.java
+++ b/paimon-core/src/main/java/org/apache/paimon/index/pk/BucketedPrimaryKeyIndexMaintainer.java
@@ -376,6 +376,7 @@ public final class BucketedPrimaryKeyIndexMaintainer {
break;
case BTREE:
case BITMAP:
+ case MULTI_VALUE:
sortedFactories.add(
new SortedDefinitionFactory(
readerFactoryBuilder,
diff --git a/paimon-core/src/main/java/org/apache/paimon/index/pk/PrimaryKeyIndexDefinition.java b/paimon-core/src/main/java/org/apache/paimon/index/pk/PrimaryKeyIndexDefinition.java
index f04ea4389a..fbec65af5c 100644
--- a/paimon-core/src/main/java/org/apache/paimon/index/pk/PrimaryKeyIndexDefinition.java
+++ b/paimon-core/src/main/java/org/apache/paimon/index/pk/PrimaryKeyIndexDefinition.java
@@ -28,6 +28,7 @@ public class PrimaryKeyIndexDefinition {
VECTOR,
BTREE,
BITMAP,
+ MULTI_VALUE,
FULL_TEXT
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/index/pk/PrimaryKeyIndexDefinitions.java b/paimon-core/src/main/java/org/apache/paimon/index/pk/PrimaryKeyIndexDefinitions.java
index 9b7bdcdfed..fd7f216fff 100644
--- a/paimon-core/src/main/java/org/apache/paimon/index/pk/PrimaryKeyIndexDefinitions.java
+++ b/paimon-core/src/main/java/org/apache/paimon/index/pk/PrimaryKeyIndexDefinitions.java
@@ -20,6 +20,7 @@ package org.apache.paimon.index.pk;
import org.apache.paimon.CoreOptions;
import org.apache.paimon.globalindex.bitmap.BitmapGlobalIndexerFactory;
+import org.apache.paimon.globalindex.bitmap.MultiValueGlobalIndexerFactory;
import org.apache.paimon.globalindex.btree.BTreeGlobalIndexerFactory;
import org.apache.paimon.schema.TableSchema;
import org.apache.paimon.types.DataField;
@@ -46,12 +47,15 @@ public class PrimaryKeyIndexDefinitions {
List<String> vectorColumns = options.primaryKeyVectorIndexColumns();
List<String> btreeColumns = options.primaryKeyBTreeIndexColumns();
List<String> bitmapColumns = options.primaryKeyBitmapIndexColumns();
+ List<String> multiValueColumns = options.primaryKeyMultiValueIndexColumns();
List<String> fullTextColumns = options.primaryKeyFullTextIndexColumns();
validateNoDuplicates(vectorColumns, CoreOptions.PK_VECTOR_INDEX_COLUMNS.key());
validateNoDuplicates(btreeColumns, CoreOptions.PK_BTREE_INDEX_COLUMNS.key());
validateNoDuplicates(bitmapColumns, CoreOptions.PK_BITMAP_INDEX_COLUMNS.key());
+ validateNoDuplicates(multiValueColumns, CoreOptions.PK_MULTIVALUE_INDEX_COLUMNS.key());
validateNoDuplicates(fullTextColumns, CoreOptions.PK_FULL_TEXT_INDEX_COLUMNS.key());
- validateOneIndexPerColumn(vectorColumns, btreeColumns, bitmapColumns, fullTextColumns);
+ validateOneIndexPerColumn(
+ vectorColumns, btreeColumns, bitmapColumns, multiValueColumns, fullTextColumns);
List<PrimaryKeyIndexDefinition> definitions = new ArrayList<>();
for (DataField field : schema.fields()) {
@@ -72,6 +76,14 @@ public class PrimaryKeyIndexDefinitions {
BitmapGlobalIndexerFactory.IDENTIFIER,
options.primaryKeyBitmapIndexOptions(column),
PrimaryKeyIndexDefinition.Family.BITMAP));
+ } else if (multiValueColumns.contains(column)) {
+ definitions.add(
+ new PrimaryKeyIndexDefinition(
+ column,
+ field.id(),
+ MultiValueGlobalIndexerFactory.IDENTIFIER,
+ options.primaryKeyMultiValueIndexOptions(column),
+ PrimaryKeyIndexDefinition.Family.MULTI_VALUE));
} else if (vectorColumns.contains(column)) {
definitions.add(
new PrimaryKeyIndexDefinition(
@@ -109,11 +121,13 @@ public class PrimaryKeyIndexDefinitions {
List<String> vectorColumns,
List<String> btreeColumns,
List<String> bitmapColumns,
+ List<String> multiValueColumns,
List<String> fullTextColumns) {
Set<String> indexedColumns = new HashSet<>();
validateUniqueColumns(indexedColumns, vectorColumns);
validateUniqueColumns(indexedColumns, btreeColumns);
validateUniqueColumns(indexedColumns, bitmapColumns);
+ validateUniqueColumns(indexedColumns, multiValueColumns);
validateUniqueColumns(indexedColumns, fullTextColumns);
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/index/pksorted/BucketedSortedIndexMaintainer.java b/paimon-core/src/main/java/org/apache/paimon/index/pksorted/BucketedSortedIndexMaintainer.java
index 7c56779b06..e3ac81a1d9 100644
--- a/paimon-core/src/main/java/org/apache/paimon/index/pksorted/BucketedSortedIndexMaintainer.java
+++ b/paimon-core/src/main/java/org/apache/paimon/index/pksorted/BucketedSortedIndexMaintainer.java
@@ -45,7 +45,7 @@ import java.util.concurrent.Future;
import static org.apache.paimon.utils.Preconditions.checkArgument;
-/** Maintains one bucket-local source-backed BTree or Bitmap definition. */
+/** Maintains one bucket-local source-backed sorted-index definition. */
public class BucketedSortedIndexMaintainer {
private static final Logger LOG = LoggerFactory.getLogger(BucketedSortedIndexMaintainer.class);
diff --git a/paimon-core/src/main/java/org/apache/paimon/index/pksorted/PkSortedIndexBuilder.java b/paimon-core/src/main/java/org/apache/paimon/index/pksorted/PkSortedIndexBuilder.java
index ef3c210e21..b5412ad490 100644
--- a/paimon-core/src/main/java/org/apache/paimon/index/pksorted/PkSortedIndexBuilder.java
+++ b/paimon-core/src/main/java/org/apache/paimon/index/pksorted/PkSortedIndexBuilder.java
@@ -23,6 +23,9 @@ import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.disk.IOManager;
+import org.apache.paimon.globalindex.GlobalIndexKeyExtractor;
+import org.apache.paimon.globalindex.GlobalIndexer;
+import org.apache.paimon.globalindex.SortedGlobalIndexer;
import org.apache.paimon.index.IndexFileMeta;
import org.apache.paimon.index.pk.PrimaryKeyIndexSourceFile;
import org.apache.paimon.io.DataFileMeta;
@@ -45,7 +48,7 @@ import java.util.List;
import static org.apache.paimon.utils.Preconditions.checkArgument;
-/** Sorts physical data files and builds their source-backed scalar index payloads. */
+/** Builds source-backed index payloads by extracting and spill-sorting normalized index keys. */
public class PkSortedIndexBuilder {
private static final int ROW_ID_FIELD_ID = Integer.MAX_VALUE;
@@ -96,6 +99,18 @@ public class PkSortedIndexBuilder {
dataFile.level());
}
+ List<PrimaryKeyIndexSourceFile> sourceFiles = new ArrayList<>();
+ for (DataFileMeta dataFile : orderedDataFiles) {
+ sourceFiles.add(
+ new PrimaryKeyIndexSourceFile(dataFile.fileName(), dataFile.rowCount()));
+ }
+ GlobalIndexer indexer = GlobalIndexer.create(indexType, indexField, options);
+ checkArgument(
+ indexer instanceof SortedGlobalIndexer,
+ "Index algorithm %s does not expose sorted index keys.",
+ indexType);
+ GlobalIndexKeyExtractor keyExtractor = ((SortedGlobalIndexer) indexer).keyExtractor();
+
IOManager actualIOManager = ioManager;
boolean ownsIOManager = false;
if (actualIOManager == null) {
@@ -106,27 +121,26 @@ public class PkSortedIndexBuilder {
BinaryExternalSortBuffer sortBuffer = null;
try {
CoreOptions coreOptions = new CoreOptions(options);
+ DataField keyField =
+ new DataField(indexField.id(), indexField.name(), keyExtractor.keyType());
RowType sortRowType =
RowType.of(
- indexField,
+ keyField,
new DataField(
ROW_ID_FIELD_ID, "_ROW_ID", DataTypes.BIGINT().notNull()));
sortBuffer =
BinaryExternalSortBuffer.create(
actualIOManager,
sortRowType,
- new int[] {0},
+ new int[] {0, 1},
coreOptions.writeBufferSize(),
coreOptions.pageSize(),
coreOptions.localSortMaxNumFileHandles(),
coreOptions.spillCompressOptions(),
coreOptions.writeBufferSpillDiskSize());
- List<PrimaryKeyIndexSourceFile> sourceFiles = new ArrayList<>();
long sourceOffset = 0;
for (DataFileMeta dataFile : orderedDataFiles) {
- sourceFiles.add(
- new PrimaryKeyIndexSourceFile(dataFile.fileName(), dataFile.rowCount()));
try (Reader reader = readerFactory.create(dataFile)) {
checkArgument(
reader.rowCount() == dataFile.rowCount(),
@@ -134,6 +148,7 @@ public class PkSortedIndexBuilder {
reader.rowCount(),
dataFile.fileName(),
dataFile.rowCount());
+ long readRows = 0;
PkSortedDataFileReader.Entry entry;
while ((entry = reader.readNext()) != null) {
checkArgument(
@@ -143,11 +158,19 @@ public class PkSortedIndexBuilder {
entry.rowPosition(),
dataFile.fileName(),
dataFile.rowCount());
- sortBuffer.write(
- GenericRow.of(
- entry.value(),
- Math.addExact(sourceOffset, entry.rowPosition())));
+ long rowId = Math.addExact(sourceOffset, entry.rowPosition());
+ BinaryExternalSortBuffer currentSortBuffer = sortBuffer;
+ keyExtractor.extract(
+ entry.value(),
+ key -> currentSortBuffer.write(GenericRow.of(key, rowId)));
+ readRows++;
}
+ checkArgument(
+ readRows == dataFile.rowCount(),
+ "Sorted reader returned %s rows for data file %s, expected %s.",
+ readRows,
+ dataFile.fileName(),
+ dataFile.rowCount());
}
sourceOffset = Math.addExact(sourceOffset, dataFile.rowCount());
}
@@ -157,7 +180,7 @@ public class PkSortedIndexBuilder {
sortBuffer.sortedIterator(),
new BinaryRow(sortRowType.getFieldCount()));
InternalRow.FieldGetter valueGetter =
- InternalRow.createFieldGetter(indexField.type(), 0);
+ InternalRow.createFieldGetter(keyExtractor.keyType(), 0);
Iterator<PkSortedIndexFile.Entry> sortedEntries =
new Iterator<PkSortedIndexFile.Entry>() {
@Override
@@ -188,7 +211,7 @@ public class PkSortedIndexBuilder {
return IOManager.create(System.getProperty("java.io.tmpdir"));
}
- /** Physical source reader used by the sorter. */
+ /** Physical source reader used by the index builder. */
interface Reader extends Closeable {
long rowCount();
diff --git a/paimon-core/src/main/java/org/apache/paimon/index/pksorted/PkSortedIndexFile.java b/paimon-core/src/main/java/org/apache/paimon/index/pksorted/PkSortedIndexFile.java
index 944cd4ef32..e3b4a1d1c3 100644
--- a/paimon-core/src/main/java/org/apache/paimon/index/pksorted/PkSortedIndexFile.java
+++ b/paimon-core/src/main/java/org/apache/paimon/index/pksorted/PkSortedIndexFile.java
@@ -46,7 +46,7 @@ import java.util.Map;
import static org.apache.paimon.utils.Preconditions.checkArgument;
-/** Builds source-backed BTree or Bitmap payloads for ordered physical data files. */
+/** Builds source-backed sorted-index payloads for ordered physical data files. */
public class PkSortedIndexFile extends IndexFile {
public PkSortedIndexFile(FileIO fileIO, IndexPathFactory pathFactory) {
@@ -74,7 +74,6 @@ public class PkSortedIndexFile extends IndexFile {
try {
writer = createWriter(indexType, indexField, indexOptions, fileWriter);
- long writtenRows = 0;
while (sortedEntries.hasNext()) {
Entry entry = sortedEntries.next();
checkArgument(
@@ -83,15 +82,9 @@ public class PkSortedIndexFile extends IndexFile {
entry.rowId,
sourceRowCount);
writer.write(entry.value, entry.rowId);
- writtenRows++;
}
- checkArgument(
- writtenRows == sourceRowCount,
- "Sorted index input row count %s does not match source row count %s.",
- writtenRows,
- sourceRowCount);
- List<ResultEntry> results = writer.finish();
+ List<ResultEntry> results = writer.finish(sourceRowCount);
checkArgument(
results.size() == 1,
"Sorted index build must produce exactly one payload file, but produced %s.",
@@ -145,7 +138,7 @@ public class PkSortedIndexFile extends IndexFile {
return (GlobalIndexSingleColumnWriter) writer;
}
- /** One sorted scalar value and its zero-based ordinal in the ordered source group. */
+ /** One sorted normalized key and its zero-based source-row ordinal. */
public static final class Entry {
@Nullable private final Object value;
diff --git a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java
index 2c83cbcb57..b55bff318b 100644
--- a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java
+++ b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java
@@ -1015,6 +1015,7 @@ public class SchemaManager implements Serializable {
if (options.primaryKeyVectorIndexColumns().contains(fieldName)
|| options.primaryKeyBTreeIndexColumns().contains(fieldName)
|| options.primaryKeyBitmapIndexColumns().contains(fieldName)
+ || options.primaryKeyMultiValueIndexColumns().contains(fieldName)
|| options.primaryKeyFullTextIndexColumns().contains(fieldName)) {
throw new UnsupportedOperationException(
String.format(
diff --git a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
index 1fe54ea479..5d55a73cba 100644
--- a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
+++ b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
@@ -31,6 +31,7 @@ import org.apache.paimon.fileindex.FileIndexerFactoryUtils;
import org.apache.paimon.format.FileFormat;
import org.apache.paimon.globalindex.GlobalIndexer;
import org.apache.paimon.globalindex.bitmap.BitmapGlobalIndexerFactory;
+import org.apache.paimon.globalindex.bitmap.MultiValueGlobalIndexerFactory;
import org.apache.paimon.globalindex.btree.BTreeGlobalIndexerFactory;
import org.apache.paimon.iceberg.IcebergOptions;
import org.apache.paimon.mergetree.compact.aggregate.FieldAggregator;
@@ -1217,6 +1218,7 @@ public class SchemaValidation {
List<String> vectorColumns = options.primaryKeyVectorIndexColumns();
List<String> btreeColumns = options.primaryKeyBTreeIndexColumns();
List<String> bitmapColumns = options.primaryKeyBitmapIndexColumns();
+ List<String> multiValueColumns = options.primaryKeyMultiValueIndexColumns();
List<String> fullTextColumns = options.primaryKeyFullTextIndexColumns();
validateNoDuplicatePrimaryKeyIndexColumns(
vectorColumns, CoreOptions.PK_VECTOR_INDEX_COLUMNS.key());
@@ -1224,6 +1226,8 @@ public class SchemaValidation {
btreeColumns, CoreOptions.PK_BTREE_INDEX_COLUMNS.key());
validateNoDuplicatePrimaryKeyIndexColumns(
bitmapColumns, CoreOptions.PK_BITMAP_INDEX_COLUMNS.key());
+ validateNoDuplicatePrimaryKeyIndexColumns(
+ multiValueColumns, CoreOptions.PK_MULTIVALUE_INDEX_COLUMNS.key());
validateNoDuplicatePrimaryKeyIndexColumns(
fullTextColumns, CoreOptions.PK_FULL_TEXT_INDEX_COLUMNS.key());
@@ -1231,6 +1235,7 @@ public class SchemaValidation {
validateUniquePrimaryKeyIndexColumns(indexedColumns, vectorColumns);
validateUniquePrimaryKeyIndexColumns(indexedColumns, btreeColumns);
validateUniquePrimaryKeyIndexColumns(indexedColumns, bitmapColumns);
+ validateUniquePrimaryKeyIndexColumns(indexedColumns, multiValueColumns);
validateUniquePrimaryKeyIndexColumns(indexedColumns, fullTextColumns);
}
@@ -1255,27 +1260,28 @@ public class SchemaValidation {
private static void validatePrimaryKeySortedIndexes(TableSchema schema, CoreOptions options) {
if (options.primaryKeyBTreeIndexColumns().isEmpty()
- && options.primaryKeyBitmapIndexColumns().isEmpty()) {
+ && options.primaryKeyBitmapIndexColumns().isEmpty()
+ && options.primaryKeyMultiValueIndexColumns().isEmpty()) {
return;
}
checkArgument(
options.deletionVectorsEnabled(),
- "Primary-key BTree and Bitmap indexes require deletion-vectors.enabled = true.");
+ "Primary-key BTree, Bitmap, and Multivalue indexes require deletion-vectors.enabled = true.");
checkArgument(
!schema.primaryKeys().isEmpty(),
- "Primary-key BTree and Bitmap indexes require a primary-key table.");
+ "Primary-key BTree, Bitmap, and Multivalue indexes require a primary-key table.");
checkArgument(
options.bucket() > 0 || options.bucket() == BucketMode.POSTPONE_BUCKET,
- "Primary-key BTree and Bitmap indexes require fixed or postpone bucket mode "
+ "Primary-key BTree, Bitmap, and Multivalue indexes require fixed or postpone bucket mode "
+ "(bucket > 0 or bucket = -2), but bucket is %s.",
options.bucket());
checkArgument(
!options.deletionVectorsMergeOnRead(),
- "Primary-key BTree and Bitmap indexes require deletion-vectors.merge-on-read = false.");
+ "Primary-key BTree, Bitmap, and Multivalue indexes require deletion-vectors.merge-on-read = false.");
checkArgument(
!options.pkClusteringOverride(),
- "Primary-key BTree and Bitmap indexes do not support pk-clustering-override.");
+ "Primary-key BTree, Bitmap, and Multivalue indexes do not support pk-clustering-override.");
validatePrimaryKeySortedIndexColumns(
schema,
@@ -1285,6 +1291,10 @@ public class SchemaValidation {
schema,
options.primaryKeyBitmapIndexColumns(),
CoreOptions.PK_BITMAP_INDEX_COLUMNS.key());
+ validatePrimaryKeySortedIndexColumns(
+ schema,
+ options.primaryKeyMultiValueIndexColumns(),
+ CoreOptions.PK_MULTIVALUE_INDEX_COLUMNS.key());
Map<String, DataField> fields = schema.nameToFieldMap();
for (String column : options.primaryKeyBTreeIndexColumns()) {
@@ -1299,6 +1309,12 @@ public class SchemaValidation {
fields.get(column),
options.primaryKeyBitmapIndexOptions(column));
}
+ for (String column : options.primaryKeyMultiValueIndexColumns()) {
+ GlobalIndexer.create(
+ MultiValueGlobalIndexerFactory.IDENTIFIER,
+ fields.get(column),
+ options.primaryKeyMultiValueIndexOptions(column));
+ }
}
private static void validatePrimaryKeySortedIndexColumns(
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 7fc98c959e..82f658f88d 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
@@ -71,7 +71,8 @@ public class PrimaryKeyBatchScan extends AbstractBatchTableScan {
for (PrimaryKeyIndexDefinition definition :
PrimaryKeyIndexDefinitions.create(table.schema()).definitions()) {
if (definition.family() == PrimaryKeyIndexDefinition.Family.BTREE
- || definition.family() == PrimaryKeyIndexDefinition.Family.BITMAP) {
+ || definition.family() == PrimaryKeyIndexDefinition.Family.BITMAP
+ || definition.family() == PrimaryKeyIndexDefinition.Family.MULTI_VALUE) {
definitions.add(definition);
definitionFieldIds.add(definition.fieldId());
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeySortedIndexScan.java b/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeySortedIndexScan.java
index b1458eadbc..d919cb85ac 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeySortedIndexScan.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeySortedIndexScan.java
@@ -74,7 +74,7 @@ import static org.apache.paimon.CoreOptions.GLOBAL_INDEX_THREAD_NUM;
import static org.apache.paimon.utils.Preconditions.checkArgument;
import static org.apache.paimon.utils.Preconditions.checkNotNull;
-/** Plans source-backed scalar index groups in file-local row-position space. */
+/** Plans source-backed sorted-index groups in file-local row-position space. */
public final class PrimaryKeySortedIndexScan {
private static final Logger LOG = LoggerFactory.getLogger(PrimaryKeySortedIndexScan.class);
@@ -138,7 +138,8 @@ public final class PrimaryKeySortedIndexScan {
List<PrimaryKeyIndexDefinition> scalarDefinitions = new ArrayList<>();
for (PrimaryKeyIndexDefinition definition : definitions) {
if (definition.family() == PrimaryKeyIndexDefinition.Family.BTREE
- || definition.family() == PrimaryKeyIndexDefinition.Family.BITMAP) {
+ || definition.family() == PrimaryKeyIndexDefinition.Family.BITMAP
+ || definition.family() == PrimaryKeyIndexDefinition.Family.MULTI_VALUE) {
scalarDefinitions.add(definition);
}
}
@@ -242,7 +243,8 @@ public final class PrimaryKeySortedIndexScan {
Map<Integer, PrimaryKeyIndexDefinition> definitionsByField = new LinkedHashMap<>();
for (PrimaryKeyIndexDefinition definition : definitions) {
if (definition.family() == PrimaryKeyIndexDefinition.Family.BTREE
- || definition.family() == PrimaryKeyIndexDefinition.Family.BITMAP) {
+ || definition.family() == PrimaryKeyIndexDefinition.Family.BITMAP
+ || definition.family() == PrimaryKeyIndexDefinition.Family.MULTI_VALUE) {
definitionsByField.put(definition.fieldId(), definition);
}
}
@@ -447,6 +449,14 @@ public final class PrimaryKeySortedIndexScan {
() -> reader().visitContains(fieldRef, literal));
}
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitArrayContains(
+ FieldRef fieldRef, Object literal) {
+ return query(
+ QueryKey.of(QueryOperation.ARRAY_CONTAINS, fieldRef, literal),
+ () -> reader().visitArrayContains(fieldRef, literal));
+ }
+
@Override
public CompletableFuture<Optional<GlobalIndexResult>> visitLike(
FieldRef fieldRef, Object literal) {
@@ -628,6 +638,7 @@ public final class PrimaryKeySortedIndexScan {
STARTS_WITH,
ENDS_WITH,
CONTAINS,
+ ARRAY_CONTAINS,
LIKE,
LESS_THAN,
GREATER_OR_EQUAL,
@@ -722,6 +733,12 @@ public final class PrimaryKeySortedIndexScan {
return localize(wrapped.visitContains(fieldRef, literal));
}
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitArrayContains(
+ FieldRef fieldRef, Object literal) {
+ return localize(wrapped.visitArrayContains(fieldRef, literal));
+ }
+
@Override
public CompletableFuture<Optional<GlobalIndexResult>> visitLike(
FieldRef fieldRef, Object literal) {
diff --git a/paimon-core/src/test/java/org/apache/paimon/globalindex/GlobalIndexBuilderUtilsTest.java b/paimon-core/src/test/java/org/apache/paimon/globalindex/GlobalIndexBuilderUtilsTest.java
index cf0e6820ad..549814fafe 100644
--- a/paimon-core/src/test/java/org/apache/paimon/globalindex/GlobalIndexBuilderUtilsTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/globalindex/GlobalIndexBuilderUtilsTest.java
@@ -49,6 +49,7 @@ import org.junit.jupiter.api.io.TempDir;
import java.io.IOException;
import java.util.Arrays;
import java.util.Collections;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;
@@ -279,6 +280,28 @@ class GlobalIndexBuilderUtilsTest {
assertIndexedSplitRowRanges(ranges.get(new Range(5938, 7599)), new Range(5938, 7599));
}
+ @Test
+ void testShardSplitsUsesStableGlobalRowIdWindows() {
+ DataSplit split =
+ DataSplit.builder()
+ .withPartition(BinaryRow.EMPTY_ROW)
+ .withBucket(0)
+ .withBucketPath("bucket-0")
+ .withDataFiles(Collections.singletonList(createDataFileMeta(5L, 10L)))
+ .rawConvertible(false)
+ .build();
+ Map<Range, List<Split>> input = new LinkedHashMap<>();
+ input.put(new Range(5, 14), Collections.singletonList(split));
+
+ Map<Range, List<Split>> shards = GlobalIndexBuilderUtils.shardSplitsByRowRange(input, 6);
+
+ assertThat(shards.keySet())
+ .containsExactly(new Range(5, 5), new Range(6, 11), new Range(12, 14));
+ for (Map.Entry<Range, List<Split>> shard : shards.entrySet()) {
+ assertIndexedSplitRowRanges(shard.getValue(), shard.getKey());
+ }
+ }
+
private List<ResultEntry> createDummyResultEntries() throws IOException {
String fileName = "test-index-" + UUID.randomUUID();
Path filePath = indexPathFactory.toPath(fileName);
diff --git a/paimon-core/src/test/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexScannerTest.java b/paimon-core/src/test/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexScannerTest.java
index ab2b87d93b..4770726e4b 100644
--- a/paimon-core/src/test/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexScannerTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexScannerTest.java
@@ -191,12 +191,11 @@ public class SortedGlobalIndexScannerTest extends TableTestBase {
assertThat(entries).isNotEmpty();
assertThat(entries)
.allSatisfy(
- entry ->
- assertThat(
- DataEvolutionIndexSourceMeta.fromIndexFile(
- entry.indexFile())
- .scanSnapshotId())
- .isEqualTo(scanSnapshotId));
+ entry -> {
+ DataEvolutionIndexSourceMeta sourceMeta =
+ DataEvolutionIndexSourceMeta.fromIndexFile(entry.indexFile());
+ assertThat(sourceMeta.scanSnapshotId()).isEqualTo(scanSnapshotId);
+ });
}
@Test
diff --git a/paimon-core/src/test/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexTestUtils.java b/paimon-core/src/test/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexTestUtils.java
index ed9238068f..d25f68cda6 100644
--- a/paimon-core/src/test/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexTestUtils.java
+++ b/paimon-core/src/test/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexTestUtils.java
@@ -20,6 +20,7 @@ package org.apache.paimon.globalindex.sorted;
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.globalindex.GlobalIndexKeyExtractor;
import org.apache.paimon.globalindex.KeySerializer;
import org.apache.paimon.reader.RecordReader;
import org.apache.paimon.table.FileStoreTable;
@@ -60,6 +61,7 @@ public final class SortedGlobalIndexTestUtils {
SpecialFields.rowTypeWithRowId(table.rowType())
.project(Arrays.asList(indexFieldName, SpecialFields.ROW_ID.name()));
InternalRow.FieldGetter fieldGetter = InternalRow.createFieldGetter(indexField.type(), 0);
+ GlobalIndexKeyExtractor keyExtractor = writer.keyExtractor();
List<Pair<Object, Long>> rows = new ArrayList<>();
try (RecordReader<InternalRow> reader =
table.newReadBuilder()
@@ -70,19 +72,33 @@ public final class SortedGlobalIndexTestUtils {
while (iterator.hasNext()) {
InternalRow row = iterator.next();
Object value = fieldGetter.getFieldOrNull(row);
- rows.add(Pair.of(InternalRowUtils.copy(value, indexField.type()), row.getLong(1)));
+ long rowId = row.getLong(1);
+ keyExtractor.extract(
+ value,
+ key ->
+ rows.add(
+ Pair.of(
+ InternalRowUtils.copy(key, keyExtractor.keyType()),
+ rowId)));
}
}
- Comparator<Object> comparator = KeySerializer.create(indexField.type()).createComparator();
+ Comparator<Object> comparator =
+ KeySerializer.create(keyExtractor.keyType()).createComparator();
rows.sort(
(left, right) -> {
if (left.getKey() == null) {
- return right.getKey() == null ? 0 : -1;
+ return right.getKey() == null
+ ? Long.compare(left.getValue(), right.getValue())
+ : -1;
}
- return right.getKey() == null
- ? 1
- : comparator.compare(left.getKey(), right.getKey());
+ int comparison =
+ right.getKey() == null
+ ? 1
+ : comparator.compare(left.getKey(), right.getKey());
+ return comparison == 0
+ ? Long.compare(left.getValue(), right.getValue())
+ : comparison;
});
List<InternalRow> sortedRows = new ArrayList<>(rows.size());
diff --git a/paimon-core/src/test/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexWriterTest.java b/paimon-core/src/test/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexWriterTest.java
index e449d63211..8d7a443ec4 100644
--- a/paimon-core/src/test/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexWriterTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexWriterTest.java
@@ -49,9 +49,9 @@ public class SortedGlobalIndexWriterTest extends TableTestBase {
public void testSingleColumnWriterRotationPreservesResultGroups() throws Exception {
GlobalIndexSingleColumnWriter first = mock(GlobalIndexSingleColumnWriter.class);
GlobalIndexSingleColumnWriter second = mock(GlobalIndexSingleColumnWriter.class);
- when(first.finish())
+ when(first.finish(2))
.thenReturn(Collections.singletonList(new ResultEntry("index-1", 2, null)));
- when(second.finish())
+ when(second.finish(1))
.thenReturn(Collections.singletonList(new ResultEntry("index-2", 1, null)));
Queue<GlobalIndexSingleColumnWriter> writers =
new ArrayDeque<>(Arrays.asList(first, second));
@@ -71,6 +71,38 @@ public class SortedGlobalIndexWriterTest extends TableTestBase {
assertThat(results.get(1)).extracting(ResultEntry::fileName).containsExactly("index-2");
}
+ @Test
+ public void testSingleColumnWriterPreservesSourceRowCount() throws Exception {
+ GlobalIndexSingleColumnWriter writer = mock(GlobalIndexSingleColumnWriter.class);
+ when(writer.finish(5))
+ .thenReturn(Collections.singletonList(new ResultEntry("index", 5, null)));
+ SortedSingleColumnIndexWriter taskWriter =
+ SortedSingleColumnIndexWriter.forSourceRowCount(5, writer);
+
+ taskWriter.write(10, 0);
+ taskWriter.write(20, 0);
+ List<List<ResultEntry>> results = taskWriter.finish();
+
+ verify(writer).finish(5);
+ assertThat(results).hasSize(1);
+ assertThat(results.get(0)).extracting(ResultEntry::rowCount).containsExactly(5L);
+ }
+
+ @Test
+ public void testSingleColumnWriterPreservesSourceRangeWithoutEntries() {
+ GlobalIndexSingleColumnWriter writer = mock(GlobalIndexSingleColumnWriter.class);
+ when(writer.finish(3))
+ .thenReturn(Collections.singletonList(new ResultEntry("index", 3, null)));
+ SortedSingleColumnIndexWriter taskWriter =
+ SortedSingleColumnIndexWriter.forSourceRowCount(3, writer);
+
+ List<List<ResultEntry>> results = taskWriter.finish();
+
+ verify(writer).finish(3);
+ assertThat(results).hasSize(1);
+ assertThat(results.get(0)).extracting(ResultEntry::rowCount).containsExactly(3L);
+ }
+
@Test
public void testSingleColumnWriterClosesActiveWriter() throws Exception {
GlobalIndexSingleColumnWriter activeWriter =
diff --git a/paimon-core/src/test/java/org/apache/paimon/index/pk/PrimaryKeyIndexDefinitionsTest.java b/paimon-core/src/test/java/org/apache/paimon/index/pk/PrimaryKeyIndexDefinitionsTest.java
index 882c714e50..ec9be827b3 100644
--- a/paimon-core/src/test/java/org/apache/paimon/index/pk/PrimaryKeyIndexDefinitionsTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/index/pk/PrimaryKeyIndexDefinitionsTest.java
@@ -44,25 +44,27 @@ class PrimaryKeyIndexDefinitionsTest {
options.put("fields.embedding.pk-vector.index.type", "ivf-pq");
options.put(CoreOptions.PK_BTREE_INDEX_COLUMNS.key(), "name");
options.put(CoreOptions.PK_BITMAP_INDEX_COLUMNS.key(), "status");
+ options.put(CoreOptions.PK_MULTIVALUE_INDEX_COLUMNS.key(), "tags");
List<PrimaryKeyIndexDefinition> definitions =
PrimaryKeyIndexDefinitions.create(schema(options)).definitions();
assertThat(definitions)
.extracting(PrimaryKeyIndexDefinition::column)
- .containsExactly("name", "status", "embedding");
+ .containsExactly("name", "status", "embedding", "tags");
assertThat(definitions)
.extracting(PrimaryKeyIndexDefinition::family)
.containsExactly(
PrimaryKeyIndexDefinition.Family.BTREE,
PrimaryKeyIndexDefinition.Family.BITMAP,
- PrimaryKeyIndexDefinition.Family.VECTOR);
+ PrimaryKeyIndexDefinition.Family.VECTOR,
+ PrimaryKeyIndexDefinition.Family.MULTI_VALUE);
assertThat(definitions)
.extracting(PrimaryKeyIndexDefinition::fieldId)
- .containsExactly(1, 2, 3);
+ .containsExactly(1, 2, 3, 4);
assertThat(definitions)
.extracting(PrimaryKeyIndexDefinition::indexType)
- .containsExactly("btree", "bitmap", "ivf-pq");
+ .containsExactly("btree", "bitmap", "ivf-pq", "multivalue");
}
@Test
@@ -126,7 +128,8 @@ class PrimaryKeyIndexDefinitionsTest {
new DataField(0, "id", DataTypes.INT().notNull()),
new DataField(1, "name", DataTypes.STRING()),
new DataField(2, "status", DataTypes.INT()),
- new DataField(3, "embedding", DataTypes.VECTOR(3, DataTypes.FLOAT()))),
+ new DataField(3, "embedding", DataTypes.VECTOR(3, DataTypes.FLOAT())),
+ new DataField(4, "tags", DataTypes.ARRAY(DataTypes.STRING()))),
3,
Collections.emptyList(),
Collections.singletonList("id"),
diff --git a/paimon-core/src/test/java/org/apache/paimon/index/pksorted/PkSortedIndexBuilderTest.java b/paimon-core/src/test/java/org/apache/paimon/index/pksorted/PkSortedIndexBuilderTest.java
index a57363e4b1..85f7c04344 100644
--- a/paimon-core/src/test/java/org/apache/paimon/index/pksorted/PkSortedIndexBuilderTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/index/pksorted/PkSortedIndexBuilderTest.java
@@ -19,6 +19,7 @@
package org.apache.paimon.index.pksorted;
import org.apache.paimon.CoreOptions;
+import org.apache.paimon.data.GenericArray;
import org.apache.paimon.disk.BufferFileWriter;
import org.apache.paimon.disk.FileIOChannel;
import org.apache.paimon.disk.IOManager;
@@ -110,6 +111,157 @@ class PkSortedIndexBuilderTest {
}
}
+ @Test
+ void testBuildsQueryableMultiValueFromUnsortedArrayRows() throws Exception {
+ DataField tags = new DataField(8, "tags", DataTypes.ARRAY(DataTypes.INT()));
+ List<PkSortedDataFileReader.Entry> entries =
+ Arrays.asList(
+ entry(new GenericArray(new Integer[] {2, 1}), 0),
+ entry(null, 1),
+ entry(new GenericArray(new Integer[0]), 2),
+ entry(new GenericArray(new Integer[] {3, 2}), 3));
+ DataFileMeta source = dataFile("data-file", entries.size());
+ LocalFileIO fileIO = LocalFileIO.create();
+ IndexPathFactory pathFactory = pathFactory(tempPath);
+ Options options = options();
+ IndexFileMeta payload =
+ new PkSortedIndexBuilder(
+ ignored -> new ArrayReader(entries),
+ new PkSortedIndexFile(fileIO, pathFactory),
+ tags,
+ "multivalue",
+ options,
+ null)
+ .build(Collections.singletonList(source));
+
+ List<GlobalIndexIOMeta> ioMetas =
+ Collections.singletonList(
+ new GlobalIndexIOMeta(
+ pathFactory.toPath(payload.fileName()),
+ payload.fileSize(),
+ payload.globalIndexMeta().indexMeta()));
+ ExecutorService executor = newDirectExecutorService();
+ try (GlobalIndexReader reader =
+ GlobalIndexer.create("multivalue", tags, options)
+ .createReader(
+ new GlobalIndexFileReadWrite(fileIO, pathFactory),
+ ioMetas,
+ payload.rowCount(),
+ executor)) {
+ FieldRef fieldRef = new FieldRef(8, "tags", tags.type());
+ assertThat(reader.visitArrayContains(fieldRef, 2).join().get().results())
+ .containsExactlyInAnyOrder(0L, 3L);
+ assertThat(reader.visitArrayContains(fieldRef, 4).join().get().results()).isEmpty();
+ } finally {
+ executor.shutdownNow();
+ }
+ }
+
+ @Test
+ void testExpandsAndSortsMultivalueInputInSourceOrdinalOrder() throws Exception {
+ DataField tags = new DataField(8, "tags", DataTypes.ARRAY(DataTypes.INT()));
+ DataFileMeta sourceB = dataFile("data-b", 2);
+ DataFileMeta sourceA = dataFile("data-a", 2);
+ List<PrimaryKeyIndexSourceFile> capturedSources = new ArrayList<>();
+ List<PkSortedIndexFile.Entry> capturedEntries = new ArrayList<>();
+ List<ArrayReader> openedReaders = new ArrayList<>();
+ List<PkSortedDataFileReader.Entry> entriesA =
+ Arrays.asList(
+ entry(new GenericArray(new Integer[] {3}), 0),
+ entry(new GenericArray(new Integer[] {1}), 1));
+ List<PkSortedDataFileReader.Entry> entriesB =
+ Arrays.asList(
+ entry(new GenericArray(new Integer[] {4}), 0),
+ entry(new GenericArray(new Integer[] {2}), 1));
+ PkSortedIndexFile capturingFile =
+ new PkSortedIndexFile(LocalFileIO.create(), pathFactory(tempPath)) {
+ @Override
+ public IndexFileMeta build(
+ int dataLevel,
+ List<PrimaryKeyIndexSourceFile> sourceFiles,
+ DataField indexField,
+ String indexType,
+ Options indexOptions,
+ Iterator<Entry> entries) {
+ capturedSources.addAll(sourceFiles);
+ entries.forEachRemaining(capturedEntries::add);
+ return ignoredPayload();
+ }
+ };
+
+ new PkSortedIndexBuilder(
+ dataFile -> {
+ ArrayReader reader =
+ new ArrayReader(
+ dataFile.fileName().equals("data-a")
+ ? entriesA
+ : entriesB);
+ openedReaders.add(reader);
+ return reader;
+ },
+ capturingFile,
+ tags,
+ "multivalue",
+ options(),
+ null)
+ .build(Arrays.asList(sourceB, sourceA));
+
+ assertThat(capturedSources)
+ .extracting(PrimaryKeyIndexSourceFile::fileName)
+ .containsExactly("data-a", "data-b");
+ assertThat(capturedEntries)
+ .extracting(PkSortedIndexFile.Entry::value)
+ .containsExactly(1, 2, 3, 4);
+ assertThat(capturedEntries)
+ .extracting(PkSortedIndexFile.Entry::rowId)
+ .containsExactly(1L, 3L, 0L, 2L);
+ assertThat(openedReaders).allMatch(ArrayReader::isClosed);
+ }
+
+ @Test
+ void testAllEmptyMultivalueRowsStillBuildSourceCoverage() throws Exception {
+ DataField tags = new DataField(8, "tags", DataTypes.ARRAY(DataTypes.INT()));
+ List<PkSortedIndexFile.Entry> capturedEntries = new ArrayList<>();
+ List<PrimaryKeyIndexSourceFile> capturedSources = new ArrayList<>();
+ PkSortedIndexFile capturingFile =
+ new PkSortedIndexFile(LocalFileIO.create(), pathFactory(tempPath)) {
+ @Override
+ public IndexFileMeta build(
+ int dataLevel,
+ List<PrimaryKeyIndexSourceFile> sourceFiles,
+ DataField indexField,
+ String indexType,
+ Options indexOptions,
+ Iterator<Entry> entries) {
+ capturedSources.addAll(sourceFiles);
+ entries.forEachRemaining(capturedEntries::add);
+ return ignoredPayload();
+ }
+ };
+
+ new PkSortedIndexBuilder(
+ ignored ->
+ new ArrayReader(
+ Arrays.asList(
+ entry(null, 0),
+ entry(new GenericArray(new Integer[0]), 1),
+ entry(
+ new GenericArray(
+ new Integer[] {null, null}),
+ 2))),
+ capturingFile,
+ tags,
+ "multivalue",
+ options(),
+ null)
+ .build(Collections.singletonList(dataFile("empty-data", 3)));
+
+ assertThat(capturedEntries).isEmpty();
+ assertThat(capturedSources)
+ .extracting(PrimaryKeyIndexSourceFile::rowCount)
+ .containsOnly(3L);
+ }
+
@Test
void testBuildsSeveralSourcesInDeterministicOrdinalOrder() throws Exception {
DataFileMeta sourceB = dataFile("data-b", 3);
@@ -163,13 +315,17 @@ class PkSortedIndexBuilderTest {
}
@Test
- void testForcedSpillSortsRowsAndClosesTaskOwnedIoManager() throws Exception {
+ void testForcedSpillSortsExpandedMultivalueKeysAndClosesTaskOwnedIoManager() throws Exception {
int rowCount = 20_000;
List<PkSortedDataFileReader.Entry> entries = new ArrayList<>(rowCount);
for (int position = 0; position < rowCount; position++) {
- entries.add(entry(rowCount - position - 1, position));
+ entries.add(
+ entry(
+ new GenericArray(
+ new Integer[] {rowCount - position - 1, rowCount + position}),
+ position));
}
- List<Integer> sortedValues = new ArrayList<>(rowCount);
+ List<Integer> sortedValues = new ArrayList<>(rowCount * 2);
TrackingIOManager ioManager =
new TrackingIOManager(tempPath.resolve("owned-spill").toString());
PkSortedIndexFile capturingFile =
@@ -197,8 +353,8 @@ class PkSortedIndexBuilderTest {
new PkSortedIndexBuilder(
ignored -> new ArrayReader(entries),
capturingFile,
- field(),
- "btree",
+ new DataField(8, "tags", DataTypes.ARRAY(DataTypes.INT())),
+ "multivalue",
options,
null) {
@Override
@@ -209,7 +365,7 @@ class PkSortedIndexBuilderTest {
assertThat(sortedValues)
.containsExactlyElementsOf(
- LongStream.range(0, rowCount)
+ LongStream.range(0, rowCount * 2L)
.mapToObj(value -> (int) value)
.collect(Collectors.toList()));
assertThat(ioManager.createdSpillWriters).isGreaterThan(0);
@@ -310,6 +466,7 @@ class PkSortedIndexBuilderTest {
private final List<PkSortedDataFileReader.Entry> entries;
private int position;
+ private boolean closed;
private ArrayReader(List<PkSortedDataFileReader.Entry> entries) {
this.entries = entries;
@@ -326,7 +483,13 @@ class PkSortedIndexBuilderTest {
}
@Override
- public void close() {}
+ public void close() {
+ closed = true;
+ }
+
+ private boolean isClosed() {
+ return closed;
+ }
}
private static final class TrackingIOManager extends IOManagerImpl {
diff --git a/paimon-core/src/test/java/org/apache/paimon/index/pksorted/PkSortedIndexFileTest.java b/paimon-core/src/test/java/org/apache/paimon/index/pksorted/PkSortedIndexFileTest.java
index e8e8775e1e..ca2ea01984 100644
--- a/paimon-core/src/test/java/org/apache/paimon/index/pksorted/PkSortedIndexFileTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/index/pksorted/PkSortedIndexFileTest.java
@@ -125,6 +125,33 @@ class PkSortedIndexFileTest {
}
}
+ @Test
+ void testBuildsMultiValuePayloadFromArrayRows() throws Exception {
+ PkSortedIndexFile indexFile =
+ new PkSortedIndexFile(LocalFileIO.create(), pathFactory(tempPath));
+ PrimaryKeyIndexSourceFile source = new PrimaryKeyIndexSourceFile("data-file", 4);
+ DataField tags = new DataField(8, "tags", DataTypes.ARRAY(DataTypes.INT()));
+
+ IndexFileMeta payload =
+ indexFile.build(
+ 1,
+ Collections.singletonList(source),
+ tags,
+ "multivalue",
+ options(),
+ Arrays.asList(
+ new PkSortedIndexFile.Entry(1, 0),
+ new PkSortedIndexFile.Entry(2, 0),
+ new PkSortedIndexFile.Entry(2, 3))
+ .iterator());
+
+ assertThat(payload.indexType()).isEqualTo("multivalue");
+ assertThat(payload.rowCount()).isEqualTo(4L);
+ assertThat(payload.globalIndexMeta().indexFieldId()).isEqualTo(8);
+ assertThat(PrimaryKeyIndexSourceMeta.fromIndexFile(payload).sourceFile()).isEqualTo(source);
+ assertThat(indexFile.exists(payload)).isTrue();
+ }
+
@Test
void testBuildsMultiSourcePayloadsInOneOrdinalDomain() throws Exception {
PkSortedIndexFile indexFile =
diff --git a/paimon-core/src/test/java/org/apache/paimon/index/pksorted/PrimaryKeySortedIndexMaintenanceTest.java b/paimon-core/src/test/java/org/apache/paimon/index/pksorted/PrimaryKeySortedIndexMaintenanceTest.java
index 473ffe7963..6d0718ac48 100644
--- a/paimon-core/src/test/java/org/apache/paimon/index/pksorted/PrimaryKeySortedIndexMaintenanceTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/index/pksorted/PrimaryKeySortedIndexMaintenanceTest.java
@@ -73,7 +73,7 @@ import java.util.stream.Collectors;
import static org.assertj.core.api.Assertions.assertThat;
-/** End-to-end maintenance tests for source-backed BTree and Bitmap indexes. */
+/** End-to-end maintenance tests for source-backed sorted indexes. */
class PrimaryKeySortedIndexMaintenanceTest {
@TempDir java.nio.file.Path tempDir;
diff --git a/paimon-core/src/test/java/org/apache/paimon/index/pksorted/PrimaryKeySortedIndexOptionsTest.java b/paimon-core/src/test/java/org/apache/paimon/index/pksorted/PrimaryKeySortedIndexOptionsTest.java
index 7bb2018e0e..0f7ad773a9 100644
--- a/paimon-core/src/test/java/org/apache/paimon/index/pksorted/PrimaryKeySortedIndexOptionsTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/index/pksorted/PrimaryKeySortedIndexOptionsTest.java
@@ -49,6 +49,17 @@ class PrimaryKeySortedIndexOptionsTest {
assertThat(options.primaryKeyBitmapIndexColumns()).containsExactly("status", "region");
}
+ @Test
+ void testResolvesMultiValueIndexColumns() {
+ CoreOptions options =
+ new CoreOptions(
+ Collections.singletonMap(
+ "pk-multivalue.index.columns", " tags, categories "));
+
+ assertThat(options.primaryKeyMultiValueIndexColumns())
+ .containsExactly("tags", "categories");
+ }
+
@Test
void testResolvesBTreeIndexAndSortOptions() {
Map<String, String> values = new HashMap<>();
@@ -80,4 +91,15 @@ class PrimaryKeySortedIndexOptionsTest {
assertThat(options.get("bitmap-index.dictionary-block-size")).isEqualTo("8 kb");
}
+
+ @Test
+ void testResolvesMultiValueIndexOptions() {
+ Map<String, String> values = new HashMap<>();
+ values.put(
+ "fields.tags.pk-multivalue.index.options", "{\"dictionary-block-size\":\"8 kb\"}");
+
+ Options options = new CoreOptions(values).primaryKeyMultiValueIndexOptions("tags");
+
+ assertThat(options.get("multivalue-index.dictionary-block-size")).isEqualTo("8 kb");
+ }
}
diff --git a/paimon-core/src/test/java/org/apache/paimon/operation/PrimaryKeyIndexWriteTest.java b/paimon-core/src/test/java/org/apache/paimon/operation/PrimaryKeyIndexWriteTest.java
index 4d21109310..5198bce33d 100644
--- a/paimon-core/src/test/java/org/apache/paimon/operation/PrimaryKeyIndexWriteTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/operation/PrimaryKeyIndexWriteTest.java
@@ -52,7 +52,7 @@ import java.util.Map;
import static org.assertj.core.api.Assertions.assertThat;
-/** Tests BTree and Bitmap primary-key index wiring through the file-store writer. */
+/** Tests source-backed primary-key index wiring through the file-store writer. */
class PrimaryKeyIndexWriteTest {
@TempDir java.nio.file.Path tempDir;
@@ -89,6 +89,30 @@ class PrimaryKeyIndexWriteTest {
write.close();
}
+ @Test
+ void testCreatesCoordinatorForMultiValueDefinition() throws Exception {
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.BUCKET.key(), "10");
+ options.put(CoreOptions.DELETION_VECTORS_ENABLED.key(), "true");
+ options.put(CoreOptions.PK_MULTIVALUE_INDEX_COLUMNS.key(), "tags");
+ List<DataField> fields =
+ new ArrayList<>(TestKeyValueGenerator.DEFAULT_ROW_TYPE.getFields());
+ fields.add(new DataField(7, "tags", DataTypes.ARRAY(DataTypes.STRING())));
+ TestFileStore store = createStore(options, new RowType(fields));
+ KeyValueFileStoreWrite write = (KeyValueFileStoreWrite) store.newWrite();
+ write.withIOManager(ioManager);
+ TestKeyValueGenerator generator = new TestKeyValueGenerator();
+ KeyValue record = generator.next();
+
+ AbstractFileStoreWrite.WriterContainer<KeyValue> container =
+ write.createWriterContainer(generator.getPartition(record), 1);
+
+ assertThat(container.primaryKeyIndexMaintainer).isNotNull();
+ assertThat((List<?>) readField(container.primaryKeyIndexMaintainer, "sortedMaintainers"))
+ .hasSize(1);
+ write.close();
+ }
+
@Test
void testCreatesCoordinatorForMultipleScalarDefinitionsAndVector() throws Exception {
Map<String, String> options = new HashMap<>();
diff --git a/paimon-core/src/test/java/org/apache/paimon/schema/PrimaryKeySortedIndexValidationTest.java b/paimon-core/src/test/java/org/apache/paimon/schema/PrimaryKeySortedIndexValidationTest.java
index d55dd836ef..ff00caed39 100644
--- a/paimon-core/src/test/java/org/apache/paimon/schema/PrimaryKeySortedIndexValidationTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/schema/PrimaryKeySortedIndexValidationTest.java
@@ -33,7 +33,7 @@ import static org.apache.paimon.schema.SchemaValidation.validateTableSchema;
import static org.assertj.core.api.Assertions.assertThatCode;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
-/** Tests for primary-key BTree and Bitmap index validation. */
+/** Tests for source-backed primary-key sorted index validation. */
class PrimaryKeySortedIndexValidationTest {
@Test
@@ -67,6 +67,34 @@ class PrimaryKeySortedIndexValidationTest {
.hasMessageContaining("duplicate");
}
+ @Test
+ void testSupportsMultiValueIndexOnArrayColumn() {
+ Map<String, String> options = enabledOptions();
+ options.put(CoreOptions.PK_MULTIVALUE_INDEX_COLUMNS.key(), "tags");
+
+ assertThatCode(() -> validateTableSchema(arraySchema(options))).doesNotThrowAnyException();
+ }
+
+ @Test
+ void testRejectsDuplicateMultiValueIndexColumns() {
+ Map<String, String> options = enabledOptions();
+ options.put(CoreOptions.PK_MULTIVALUE_INDEX_COLUMNS.key(), "tags,tags");
+
+ assertThatThrownBy(() -> validateTableSchema(arraySchema(options)))
+ .hasMessageContaining(CoreOptions.PK_MULTIVALUE_INDEX_COLUMNS.key())
+ .hasMessageContaining("duplicate");
+ }
+
+ @Test
+ void testRejectsMultiValueIndexOnScalarColumn() {
+ Map<String, String> options = enabledOptions();
+ options.put(CoreOptions.PK_MULTIVALUE_INDEX_COLUMNS.key(), "payload");
+
+ assertThatThrownBy(() -> validateTableSchema(schema(options)))
+ .hasMessageContaining("Multivalue index requires an ARRAY column")
+ .hasMessageContaining("payload");
+ }
+
@Test
void testRequiresDeletionVectors() {
Map<String, String> options = enabledOptions();
@@ -75,7 +103,7 @@ class PrimaryKeySortedIndexValidationTest {
assertThatThrownBy(() -> validateTableSchema(schema(options)))
.hasMessageContaining(
- "Primary-key BTree and Bitmap indexes require deletion-vectors.enabled = true");
+ "Primary-key BTree, Bitmap, and Multivalue indexes require deletion-vectors.enabled = true");
}
@Test
@@ -86,7 +114,7 @@ class PrimaryKeySortedIndexValidationTest {
assertThatThrownBy(() -> validateTableSchema(schema(options, Collections.emptyList())))
.hasMessageContaining(
- "Primary-key BTree and Bitmap indexes require a primary-key table");
+ "Primary-key BTree, Bitmap, and Multivalue indexes require a primary-key table");
}
@Test
@@ -97,7 +125,7 @@ class PrimaryKeySortedIndexValidationTest {
assertThatThrownBy(() -> validateTableSchema(schema(options)))
.hasMessageContaining(
- "Primary-key BTree and Bitmap indexes require fixed or postpone bucket mode");
+ "Primary-key BTree, Bitmap, and Multivalue indexes require fixed or postpone bucket mode");
}
@Test
@@ -117,7 +145,7 @@ class PrimaryKeySortedIndexValidationTest {
assertThatThrownBy(() -> validateTableSchema(schema(options)))
.hasMessageContaining(
- "Primary-key BTree and Bitmap indexes require deletion-vectors.merge-on-read = false");
+ "Primary-key BTree, Bitmap, and Multivalue indexes require deletion-vectors.merge-on-read = false");
}
@Test
@@ -129,7 +157,7 @@ class PrimaryKeySortedIndexValidationTest {
assertThatThrownBy(() -> validateTableSchema(schema(options)))
.hasMessageContaining(
- "Primary-key BTree and Bitmap indexes do not support pk-clustering-override");
+ "Primary-key BTree, Bitmap, and Multivalue indexes do not support pk-clustering-override");
}
@Test
@@ -175,6 +203,19 @@ class PrimaryKeySortedIndexValidationTest {
return schema(options, Collections.singletonList("id"));
}
+ private static TableSchema arraySchema(Map<String, String> options) {
+ return new TableSchema(
+ 0,
+ Arrays.asList(
+ new DataField(0, "id", DataTypes.INT().notNull()),
+ new DataField(1, "tags", DataTypes.ARRAY(DataTypes.STRING()))),
+ 0,
+ Collections.emptyList(),
+ Collections.singletonList("id"),
+ options,
+ "");
+ }
+
private static TableSchema schema(
Map<String, String> options, java.util.List<String> primaryKeys) {
return new TableSchema(
diff --git a/paimon-core/src/test/java/org/apache/paimon/schema/SchemaManagerTest.java b/paimon-core/src/test/java/org/apache/paimon/schema/SchemaManagerTest.java
index 044513251c..1529f73ff4 100644
--- a/paimon-core/src/test/java/org/apache/paimon/schema/SchemaManagerTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/schema/SchemaManagerTest.java
@@ -485,6 +485,36 @@ public class SchemaManagerTest {
.hasMessage("Cannot update type of primary-key index column: [status]");
}
+ @Test
+ public void testRejectChangeOfPrimaryKeyMultiValueIndexColumn() throws Exception {
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.BUCKET.key(), "1");
+ options.put(CoreOptions.DELETION_VECTORS_ENABLED.key(), "true");
+ options.put(CoreOptions.PK_MULTIVALUE_INDEX_COLUMNS.key(), "tags");
+ Schema schema =
+ new Schema(
+ Arrays.asList(
+ new DataField(0, "id", DataTypes.INT().notNull()),
+ new DataField(1, "tags", DataTypes.ARRAY(DataTypes.STRING()))),
+ Collections.emptyList(),
+ Collections.singletonList("id"),
+ options,
+ "");
+ SchemaManager manager = new SchemaManager(LocalFileIO.create(), path);
+ manager.createTable(schema);
+
+ assertThatThrownBy(() -> manager.commitChanges(SchemaChange.dropColumn("tags")))
+ .isInstanceOf(UnsupportedOperationException.class)
+ .hasMessage("Cannot drop primary-key index column: [tags]");
+ assertThatThrownBy(
+ () ->
+ manager.commitChanges(
+ SchemaChange.updateColumnType(
+ "tags", DataTypes.ARRAY(DataTypes.BIGINT()))))
+ .isInstanceOf(UnsupportedOperationException.class)
+ .hasMessage("Cannot update type of primary-key index column: [tags]");
+ }
+
@Test
public void testResetSequenceGroupForAggregateFunction() throws Exception {
Map<String, String> options = new HashMap<>();
diff --git a/paimon-core/src/test/java/org/apache/paimon/table/MultiValueGlobalIndexTableTest.java b/paimon-core/src/test/java/org/apache/paimon/table/MultiValueGlobalIndexTableTest.java
new file mode 100644
index 0000000000..ed52c8c727
--- /dev/null
+++ b/paimon-core/src/test/java/org/apache/paimon/table/MultiValueGlobalIndexTableTest.java
@@ -0,0 +1,164 @@
+/*
+ * 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.table;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.data.GenericArray;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.globalindex.IndexedSplit;
+import org.apache.paimon.globalindex.ScanResult;
+import org.apache.paimon.globalindex.sorted.SortedGlobalIndexScanner;
+import org.apache.paimon.globalindex.sorted.SortedGlobalIndexTestUtils;
+import org.apache.paimon.io.CompactIncrement;
+import org.apache.paimon.io.DataIncrement;
+import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.predicate.PredicateBuilder;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.table.sink.BatchTableCommit;
+import org.apache.paimon.table.sink.CommitMessage;
+import org.apache.paimon.table.sink.CommitMessageImpl;
+import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.table.source.ReadBuilder;
+import org.apache.paimon.table.source.TableScan;
+import org.apache.paimon.types.DataTypes;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests the multivalue global index on Data Evolution tables. */
+public class MultiValueGlobalIndexTableTest extends TableTestBase {
+
+ private static final Integer RED = 1;
+ private static final Integer BLUE = 2;
+
+ @Override
+ protected Schema schemaDefault() {
+ return Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column("tags", DataTypes.ARRAY(DataTypes.INT()))
+ .option("bucket", "-1")
+ .option(CoreOptions.ROW_TRACKING_ENABLED.key(), "true")
+ .option(CoreOptions.DATA_EVOLUTION_ENABLED.key(), "true")
+ .option(CoreOptions.GLOBAL_INDEX_ENABLED.key(), "true")
+ .build();
+ }
+
+ @Test
+ public void testCoreScanUsesMultiValueIndexAndPreservesCoverage() throws Exception {
+ createTableDefault();
+ FileStoreTable table = getTableDefault();
+ write(
+ table,
+ GenericRow.of(1, array(RED, BLUE)),
+ GenericRow.of(2, array(BLUE)),
+ GenericRow.of(3, null),
+ GenericRow.of(4, array()),
+ GenericRow.of(5, array(null, RED, RED)));
+
+ buildIndex(table);
+ table = getTableDefault();
+
+ PredicateBuilder builder = new PredicateBuilder(table.rowType());
+ Predicate containsRed = builder.arrayContains(1, RED);
+ assertThat(readIds(table, containsRed)).containsExactlyInAnyOrder(1, 5);
+ assertThat(readIdsWithFallback(table, builder.isNull(1))).containsExactly(3);
+
+ write(table, GenericRow.of(6, array(RED)));
+ table = getTableDefault();
+
+ // Fast search only returns covered rows. Full search retains the uncovered append.
+ assertThat(readIds(table, containsRed)).containsExactlyInAnyOrder(1, 5);
+ FileStoreTable fullSearchTable =
+ table.copy(
+ Collections.singletonMap(
+ CoreOptions.GLOBAL_INDEX_SEARCH_MODE.key(), "full"));
+ assertThat(readIds(fullSearchTable, containsRed)).containsExactlyInAnyOrder(1, 5, 6);
+ }
+
+ private void buildIndex(FileStoreTable table) throws Exception {
+ SortedGlobalIndexScanner scanner =
+ new SortedGlobalIndexScanner(table, "multivalue").withIndexField("tags");
+ ScanResult<DataSplit> scanResult =
+ scanner.incrementalScan()
+ .orElseThrow(
+ () ->
+ new IllegalStateException(
+ "Expected data to build multivalue index."));
+ List<CommitMessage> messages = new ArrayList<>();
+ for (DataSplit split : scanResult.entries()) {
+ messages.addAll(
+ SortedGlobalIndexTestUtils.buildIndex(
+ table, "multivalue", "tags", split, scanResult.scanSnapshotId()));
+ }
+ scanResult
+ .deletedIndexEntries()
+ .forEach(
+ entry ->
+ messages.add(
+ new CommitMessageImpl(
+ entry.partition(),
+ entry.bucket(),
+ null,
+ DataIncrement.deleteIndexIncrement(
+ Collections.singletonList(
+ entry.indexFile())),
+ CompactIncrement.emptyIncrement())));
+ try (BatchTableCommit commit = table.newBatchWriteBuilder().newCommit()) {
+ commit.commit(messages);
+ }
+ }
+
+ private List<Integer> readIds(FileStoreTable table, Predicate predicate) throws Exception {
+ return readIds(table, predicate, true);
+ }
+
+ private List<Integer> readIdsWithFallback(FileStoreTable table, Predicate predicate)
+ throws Exception {
+ return readIds(table, predicate, false);
+ }
+
+ private List<Integer> readIds(
+ FileStoreTable table, Predicate predicate, boolean expectIndexedSplits)
+ throws Exception {
+ ReadBuilder readBuilder = table.newReadBuilder().withFilter(predicate);
+ TableScan.Plan plan = readBuilder.newScan().plan();
+ if (expectIndexedSplits) {
+ assertThat(plan.splits()).allMatch(IndexedSplit.class::isInstance);
+ } else {
+ assertThat(plan.splits()).noneMatch(IndexedSplit.class::isInstance);
+ }
+
+ List<Integer> ids = new ArrayList<>();
+ readBuilder
+ .newRead()
+ .executeFilter()
+ .createReader(plan)
+ .forEachRemaining(row -> ids.add(row.getInt(0)));
+ return ids;
+ }
+
+ private GenericArray array(Object... elements) {
+ return new GenericArray(elements);
+ }
+}
diff --git a/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeySortedIndexReadTest.java b/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeySortedIndexReadTest.java
index 14b476c368..19f568152f 100644
--- a/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeySortedIndexReadTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeySortedIndexReadTest.java
@@ -22,6 +22,7 @@ import org.apache.paimon.CoreOptions;
import org.apache.paimon.Snapshot;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.GenericArray;
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.globalindex.IndexedSplit;
@@ -43,7 +44,7 @@ import java.util.List;
import static org.assertj.core.api.Assertions.assertThat;
-/** End-to-end reads for source-backed scalar indexes, residual filters, and deletion vectors. */
+/** End-to-end reads for source-backed sorted indexes, residual filters, and deletion vectors. */
class PrimaryKeySortedIndexReadTest extends TableTestBase {
@Override
@@ -200,4 +201,54 @@ class PrimaryKeySortedIndexReadTest extends TableTestBase {
}
assertThat(ids).containsExactlyInAnyOrderElementsOf(expectedIds);
}
+
+ @Test
+ void testReadWithMultiValueIndex() throws Exception {
+ Schema schema =
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column("tags", DataTypes.ARRAY(DataTypes.STRING()))
+ .primaryKey("id")
+ .option(CoreOptions.BUCKET.key(), "1")
+ .option(CoreOptions.DELETION_VECTORS_ENABLED.key(), "true")
+ .option(CoreOptions.PK_MULTIVALUE_INDEX_COLUMNS.key(), "tags")
+ .build();
+ catalog.createTable(identifier(), schema, false);
+ FileStoreTable table = getTableDefault();
+ BinaryString red = BinaryString.fromString("red");
+ BinaryString blue = BinaryString.fromString("blue");
+ write(
+ table,
+ ioManager,
+ GenericRow.of(1, new GenericArray(new BinaryString[] {red, blue})),
+ GenericRow.of(2, new GenericArray(new BinaryString[] {blue})),
+ GenericRow.of(3, null));
+ write(
+ table,
+ ioManager,
+ GenericRow.of(4, new GenericArray(new BinaryString[0])),
+ GenericRow.of(5, new GenericArray(new BinaryString[] {null, red})));
+ compact(table, BinaryRow.EMPTY_ROW, 0, ioManager, true);
+
+ Snapshot snapshot = table.store().snapshotManager().latestSnapshot();
+ assertThat(
+ table.store()
+ .newIndexFileHandler()
+ .scanSourceIndexes(snapshot, BinaryRow.EMPTY_ROW, 0))
+ .singleElement()
+ .extracting(IndexFileMeta::indexType)
+ .isEqualTo("multivalue");
+
+ Predicate predicate = new PredicateBuilder(table.rowType()).arrayContains(1, red);
+ ReadBuilder readBuilder = table.newReadBuilder().withFilter(predicate);
+ TableScan.Plan plan = readBuilder.newScan().plan();
+ assertThat(plan.splits()).isNotEmpty().allMatch(IndexedSplit.class::isInstance);
+
+ List<Integer> ids = new ArrayList<>();
+ try (RecordReader<InternalRow> reader =
+ readBuilder.newRead().executeFilter().createReader(plan)) {
+ reader.forEachRemaining(row -> ids.add(row.getInt(0)));
+ }
+ assertThat(ids).containsExactlyInAnyOrder(1, 5);
+ }
}
diff --git a/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeySortedIndexScanTest.java b/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeySortedIndexScanTest.java
index 9281ec395a..d4b6d73ed9 100644
--- a/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeySortedIndexScanTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeySortedIndexScanTest.java
@@ -23,6 +23,7 @@ import org.apache.paimon.globalindex.GlobalIndexReader;
import org.apache.paimon.globalindex.GlobalIndexResult;
import org.apache.paimon.globalindex.IndexedSplit;
import org.apache.paimon.globalindex.bitmap.BitmapGlobalIndexerFactory;
+import org.apache.paimon.globalindex.bitmap.MultiValueGlobalIndexerFactory;
import org.apache.paimon.globalindex.btree.BTreeGlobalIndexerFactory;
import org.apache.paimon.index.GlobalIndexMeta;
import org.apache.paimon.index.IndexFileMeta;
@@ -71,7 +72,7 @@ import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
-/** Tests source-backed BTree and Bitmap planning in file-local row-position space. */
+/** Tests source-backed sorted-index planning in file-local row-position space. */
class PrimaryKeySortedIndexScanTest {
@Test
@@ -362,6 +363,63 @@ class PrimaryKeySortedIndexScanTest {
assertThat(secondSplit.rowRanges()).containsExactly(new Range(0, 0), new Range(2, 2));
}
+ @Test
+ void testArrayContainsIsCachedAndLocalized() throws IOException {
+ DataFileMeta first = dataFile("data-1", 2);
+ DataFileMeta second = dataFile("data-2", 3);
+ DataSplit split = dataSplit(11, 0, first, second);
+ PrimaryKeyIndexDefinition definition =
+ definition(
+ 7,
+ MultiValueGlobalIndexerFactory.IDENTIFIER,
+ PrimaryKeyIndexDefinition.Family.MULTI_VALUE);
+ IndexFileMeta mergedPayload =
+ payload(
+ "multivalue-merged",
+ Arrays.asList(
+ new PrimaryKeyIndexSourceFile("data-1", 2),
+ new PrimaryKeyIndexSourceFile("data-2", 3)),
+ "multivalue",
+ 7,
+ 5);
+ PrimaryKeySortedIndexScan.Plan plan =
+ PrimaryKeySortedIndexScan.plan(
+ 11,
+ Collections.singletonList(split),
+ Collections.singletonList(definition),
+ Collections.singletonList(payloadEntry(0, mergedPayload)));
+ RowType rowType = RowType.of(new DataField(7, "tags", DataTypes.ARRAY(DataTypes.INT())));
+ Predicate predicate = new PredicateBuilder(rowType).arrayContains(0, 42);
+ AtomicInteger queries = new AtomicInteger();
+ GlobalIndexReader reader = mock(GlobalIndexReader.class);
+ when(reader.visitArrayContains(any(), eq(42)))
+ .thenAnswer(
+ ignored -> {
+ queries.incrementAndGet();
+ return completedResult(1, 2, 4);
+ });
+
+ PrimaryKeySortedIndexScan.EvaluatedPlan evaluated =
+ PrimaryKeySortedIndexScan.evaluate(
+ plan,
+ rowType,
+ predicate,
+ Collections.singletonList(definition),
+ (ignoredFile, ignoredDefinition, payloads, totalRowCount) -> {
+ assertThat(payloads).containsExactly(mergedPayload);
+ assertThat(totalRowCount).isEqualTo(5);
+ return reader;
+ });
+
+ assertThat(queries).hasValue(1);
+ assertThat(evaluated.files()).hasSize(2);
+ assertThat(evaluated.files().get(0).result()).isPresent();
+ assertThat(evaluated.files().get(0).result().get().results()).containsExactly(1L);
+ assertThat(evaluated.files().get(1).result()).isPresent();
+ assertThat(evaluated.files().get(1).result().get().results()).containsExactly(0L, 2L);
+ verify(reader).close();
+ }
+
@Test
void testPerFileBooleanFallbackSemantics() {
DataSplit split = dataSplit(11, 0, dataFile("data-1", 4));
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/DataEvolutionMergeIntoAction.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/DataEvolutionMergeIntoAction.java
index 5c5cce7491..d0babc5b6f 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/DataEvolutionMergeIntoAction.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/DataEvolutionMergeIntoAction.java
@@ -46,10 +46,13 @@ import org.apache.paimon.types.DataTypeRoot;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.Preconditions;
+import org.apache.flink.api.common.functions.OpenContext;
+import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.dag.Transformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.typeutils.TupleTypeInfo;
+import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.functions.sink.v2.DiscardingSink;
import org.apache.flink.streaming.api.operators.StreamFlatMap;
@@ -62,6 +65,7 @@ import org.apache.flink.table.data.RowData;
import org.apache.flink.table.data.conversion.DataStructureConverter;
import org.apache.flink.table.data.conversion.DataStructureConverters;
import org.apache.flink.table.types.logical.LogicalTypeRoot;
+import org.apache.flink.types.Row;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -440,14 +444,36 @@ public class DataEvolutionMergeIntoAction extends TableActionBase {
return batchTEnv
.toDataStream(source)
.map(
- row -> {
- int arity = row.getArity();
- GenericRowData rowData = new GenericRowData(row.getKind(), arity);
- for (int i = 0; i < arity; i++) {
- rowData.setField(
- i, converters.get(i).toInternalOrNull(row.getField(i)));
+ new RichMapFunction<Row, RowData>() {
+
+ /**
+ * Do not annotate with <code>@override</code> here to maintain
+ * compatibility with Flink 1.18-.
+ */
+ public void open(OpenContext openContext) {
+ open(new Configuration());
+ }
+
+ /**
+ * Do not annotate with <code>@override</code> here to maintain
+ * compatibility with Flink 2.0+.
+ */
+ public void open(Configuration parameters) {
+ ClassLoader classLoader =
+ getRuntimeContext().getUserCodeClassLoader();
+ converters.forEach(converter -> converter.open(classLoader));
+ }
+
+ @Override
+ public RowData map(Row row) {
+ int arity = row.getArity();
+ GenericRowData rowData = new GenericRowData(row.getKind(), arity);
+ for (int i = 0; i < arity; i++) {
+ rowData.setField(
+ i, converters.get(i).toInternalOrNull(row.getField(i)));
+ }
+ return rowData;
}
- return rowData;
});
}
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/globalindex/SortedIndexTopoBuilder.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/globalindex/SortedIndexTopoBuilder.java
index 7e3154bdbc..cda2275903 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/globalindex/SortedIndexTopoBuilder.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/globalindex/SortedIndexTopoBuilder.java
@@ -22,7 +22,6 @@ import org.apache.paimon.CoreOptions;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.data.InternalRow;
-import org.apache.paimon.data.JoinedRow;
import org.apache.paimon.data.serializer.BinaryRowSerializer;
import org.apache.paimon.flink.FlinkRowData;
import org.apache.paimon.flink.FlinkRowWrapper;
@@ -32,16 +31,22 @@ import org.apache.paimon.flink.sink.CommittableTypeInfo;
import org.apache.paimon.flink.sink.CommitterOperatorFactory;
import org.apache.paimon.flink.sink.NoopCommittableStateManager;
import org.apache.paimon.flink.sink.StoreCommitter;
+import org.apache.paimon.flink.sorter.SortOperator;
import org.apache.paimon.flink.sorter.TableSortInfo;
import org.apache.paimon.flink.sorter.TableSorter;
import org.apache.paimon.flink.utils.BoundedOneInputOperator;
+import org.apache.paimon.flink.utils.InternalTypeInfo;
import org.apache.paimon.flink.utils.JavaTypeInfo;
import org.apache.paimon.flink.utils.StreamExecutionEnvironmentUtils;
-import org.apache.paimon.globalindex.GlobalIndexSingleColumnWriter;
+import org.apache.paimon.globalindex.GlobalIndexKeyExtractor;
+import org.apache.paimon.globalindex.GlobalIndexer;
+import org.apache.paimon.globalindex.ResultEntry;
import org.apache.paimon.globalindex.ScanResult;
+import org.apache.paimon.globalindex.SortedGlobalIndexer;
import org.apache.paimon.globalindex.sorted.SortedGlobalIndexScanner;
import org.apache.paimon.globalindex.sorted.SortedGlobalIndexWriter;
import org.apache.paimon.globalindex.sorted.SortedIndexOptions;
+import org.apache.paimon.globalindex.sorted.SortedSingleColumnIndexWriter;
import org.apache.paimon.manifest.IndexManifestEntry;
import org.apache.paimon.options.Options;
import org.apache.paimon.partition.PartitionPredicate;
@@ -61,12 +66,12 @@ import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.Range;
+import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.operators.OneInputStreamOperatorFactory;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.table.data.RowData;
-import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
import java.io.IOException;
import java.io.Serializable;
@@ -81,6 +86,7 @@ import java.util.Optional;
import java.util.function.Supplier;
import static org.apache.paimon.globalindex.GlobalIndexBuilderUtils.groupSplitsByRange;
+import static org.apache.paimon.globalindex.GlobalIndexBuilderUtils.shardSplitsByRowRange;
import static org.apache.paimon.globalindex.GlobalIndexBuilderUtils.splitByContiguousRowRange;
import static org.apache.paimon.io.CompactIncrement.emptyIncrement;
import static org.apache.paimon.io.DataIncrement.deleteIndexIncrement;
@@ -91,7 +97,7 @@ public class SortedIndexTopoBuilder {
private static final String BUILD_TASK_ID_FIELD = "_SORTED_INDEX_BUILD_TASK_ID";
private static final int BUILD_TASK_ID_FIELD_ID = -1;
private static final HashSet<String> SUPPORTED_INDEX_TYPES =
- new HashSet<>(Arrays.asList("btree", "bitmap"));
+ new HashSet<>(Arrays.asList("btree", "bitmap", "multivalue"));
public static boolean supports(String indexType) {
return SUPPORTED_INDEX_TYPES.contains(indexType);
@@ -169,11 +175,22 @@ public class SortedIndexTopoBuilder {
RowType dataReadType =
SpecialFields.rowTypeWithRowId(table.rowType().project(selectedColumns));
String buildTaskIdField = buildTaskIdFieldName(dataReadType);
- RowType sortReadType = withBuildTaskId(dataReadType, buildTaskIdField);
+ GlobalIndexer indexer =
+ GlobalIndexer.create(
+ indexType, table.rowType().getField(indexColumn), userOptions);
+ if (!(indexer instanceof SortedGlobalIndexer)) {
+ throw new IllegalArgumentException(
+ "Index algorithm " + indexType + " does not expose sorted index keys.");
+ }
+ GlobalIndexKeyExtractor keyExtractor = ((SortedGlobalIndexer) indexer).keyExtractor();
+ RowType sortReadType =
+ normalizedReadType(
+ dataReadType, buildTaskIdField, indexColumn, keyExtractor.keyType());
int taskIdPos = sortReadType.getFieldIndex(buildTaskIdField);
int indexFieldPos = sortReadType.getFieldIndex(indexColumn);
int rowIdPos = sortReadType.getFieldIndex(SpecialFields.ROW_ID.name());
DataType indexFieldType = sortReadType.getTypeAt(indexFieldPos);
+ DataType sourceFieldType = table.rowType().getField(indexColumn).type();
// 3. Calculate maximum parallelism bound
long recordsPerRange =
@@ -184,7 +201,6 @@ public class SortedIndexTopoBuilder {
// 4. Build one topology for all contiguous row ranges
CoreOptions coreOptions = table.coreOptions();
ReadBuilder readBuilder = table.newReadBuilder().withReadType(dataReadType);
- List<String> sortColumns = createSortColumns(buildTaskIdField, indexColumn);
int partitionFieldSize = table.partitionKeys().size();
BinaryRowSerializer binaryRowSerializer = new BinaryRowSerializer(partitionFieldSize);
List<SortedBuildTask> buildTasks = new ArrayList<>();
@@ -193,7 +209,11 @@ public class SortedIndexTopoBuilder {
partitionRangeSplits.entrySet()) {
BinaryRow partition = partitionEntry.getKey();
byte[] partitionBytes = binaryRowSerializer.serializeToBytes(partition);
- for (Map.Entry<Range, List<Split>> entry : partitionEntry.getValue().entrySet()) {
+ Map<Range, List<Split>> ranges =
+ keyExtractor.isIdentity()
+ ? partitionEntry.getValue()
+ : shardSplitsByRowRange(partitionEntry.getValue(), recordsPerRange);
+ for (Map.Entry<Range, List<Split>> entry : ranges.entrySet()) {
Range range = entry.getKey();
List<Split> rangeSplits = entry.getValue();
if (rangeSplits.isEmpty()) {
@@ -226,7 +246,8 @@ public class SortedIndexTopoBuilder {
indexFieldPos,
rowIdPos,
indexFieldType,
- sortColumns,
+ sourceFieldType,
+ keyExtractor,
coreOptions,
sortReadType,
recordsPerRange,
@@ -304,7 +325,8 @@ public class SortedIndexTopoBuilder {
int indexFieldPos,
int rowIdPos,
DataType indexFieldType,
- List<String> sortColumns,
+ DataType sourceFieldType,
+ GlobalIndexKeyExtractor keyExtractor,
CoreOptions coreOptions,
RowType readType,
long recordsPerRange,
@@ -317,27 +339,24 @@ public class SortedIndexTopoBuilder {
.name("Global Index Source")
.setParallelism(1);
- DataStream<RowData> rowDataStream =
+ DataStream<InternalRow> rowDataStream =
sourceStream
.transform(
"Read Data",
- InternalTypeInfo.of(LogicalTypeConversion.toLogicalType(readType)),
- new ReadDataOperator(readBuilder))
+ InternalTypeInfo.fromRowType(readType),
+ new ReadDataOperator(readBuilder, keyExtractor, sourceFieldType))
.setParallelism(parallelism);
- TableSortInfo sortInfo =
- new TableSortInfo.Builder()
- .setSortColumns(sortColumns)
- .setSortStrategy(CoreOptions.OrderType.ORDER)
- .setSinkParallelism(parallelism)
- .setLocalSampleSize(parallelism * coreOptions.getLocalSampleMagnification())
- .setGlobalSampleSize(parallelism * 1000)
- .setRangeNumber(parallelism * 10)
- .build();
-
- TableSorter sorter =
- TableSorter.getSorter(env, rowDataStream, coreOptions, readType, sortInfo);
- DataStream<RowData> sortedStream = sorter.sort();
+ DataStream<InternalRow> sortedStream =
+ sortRows(
+ env,
+ rowDataStream,
+ keyExtractor.isIdentity(),
+ taskIdPos,
+ indexFieldPos,
+ coreOptions,
+ readType,
+ parallelism);
return sortedStream
.transform(
@@ -355,6 +374,56 @@ public class SortedIndexTopoBuilder {
.setParallelism(parallelism);
}
+ private static DataStream<InternalRow> sortRows(
+ StreamExecutionEnvironment env,
+ DataStream<InternalRow> input,
+ boolean identity,
+ int taskIdPos,
+ int indexFieldPos,
+ CoreOptions coreOptions,
+ RowType readType,
+ int parallelism) {
+ if (!identity) {
+ return input.keyBy((KeySelector<InternalRow, Integer>) row -> row.getInt(taskIdPos))
+ .transform(
+ "Sort Normalized Index Keys",
+ InternalTypeInfo.fromRowType(readType),
+ new SortOperator(
+ readType,
+ readType,
+ coreOptions.writeBufferSize(),
+ coreOptions.pageSize(),
+ coreOptions.localSortMaxNumFileHandles(),
+ coreOptions.spillCompressOptions(),
+ parallelism,
+ coreOptions.writeBufferSpillDiskSize()))
+ .setParallelism(parallelism);
+ }
+
+ DataStream<RowData> rowData =
+ input.map(
+ row -> (RowData) new FlinkRowData(row),
+ org.apache.flink.table.runtime.typeutils.InternalTypeInfo.of(
+ LogicalTypeConversion.toLogicalType(readType)))
+ .setParallelism(parallelism);
+ TableSortInfo sortInfo =
+ new TableSortInfo.Builder()
+ .setSortColumns(
+ createSortColumns(
+ readType.getFieldNames().get(taskIdPos),
+ readType.getFieldNames().get(indexFieldPos)))
+ .setSortStrategy(CoreOptions.OrderType.ORDER)
+ .setSinkParallelism(parallelism)
+ .setLocalSampleSize(parallelism * coreOptions.getLocalSampleMagnification())
+ .setGlobalSampleSize(parallelism * 1000)
+ .setRangeNumber(parallelism * 10)
+ .build();
+ return TableSorter.getSorter(env, rowData, coreOptions, readType, sortInfo)
+ .sort()
+ .map(FlinkRowWrapper::new, InternalTypeInfo.fromRowType(readType))
+ .setParallelism(parallelism);
+ }
+
static int calculateParallelism(
List<SortedBuildTask> buildTasks, long recordsPerRange, int maxParallelism) {
long totalRecords = 0;
@@ -372,8 +441,6 @@ public class SortedIndexTopoBuilder {
}
static List<String> createSortColumns(String buildTaskIdField, String indexColumn) {
- // Range shuffle may spread duplicate boundary keys randomly. ROW_ID makes the full sort key
- // unique while keeping equal index keys adjacent, so output file key ranges stay ordered.
return Arrays.asList(buildTaskIdField, indexColumn, SpecialFields.ROW_ID.name());
}
@@ -385,11 +452,14 @@ public class SortedIndexTopoBuilder {
return fieldName;
}
- private static RowType withBuildTaskId(RowType readType, String buildTaskIdField) {
+ private static RowType normalizedReadType(
+ RowType readType, String buildTaskIdField, String indexColumn, DataType keyType) {
List<DataField> fields = new ArrayList<>();
fields.add(
new DataField(BUILD_TASK_ID_FIELD_ID, buildTaskIdField, DataTypes.INT().notNull()));
- fields.addAll(readType.getFields());
+ DataField sourceField = readType.getField(indexColumn);
+ fields.add(new DataField(sourceField.id(), sourceField.name(), keyType));
+ fields.add(readType.getField(SpecialFields.ROW_ID.name()));
return new RowType(readType.isNullable(), fields);
}
@@ -412,41 +482,71 @@ public class SortedIndexTopoBuilder {
/** Operator to read data from splits. */
private static class ReadDataOperator
- extends org.apache.flink.table.runtime.operators.TableStreamOperator<RowData>
+ extends org.apache.flink.table.runtime.operators.TableStreamOperator<InternalRow>
implements org.apache.flink.streaming.api.operators.OneInputStreamOperator<
- SortedSplitTask, RowData> {
+ SortedSplitTask, InternalRow> {
private static final long serialVersionUID = 1L;
private final ReadBuilder readBuilder;
+ private final GlobalIndexKeyExtractor keyExtractor;
+ private final DataType sourceFieldType;
private transient TableRead tableRead;
+ private transient InternalRow.FieldGetter sourceFieldGetter;
- public ReadDataOperator(ReadBuilder readBuilder) {
+ public ReadDataOperator(
+ ReadBuilder readBuilder,
+ GlobalIndexKeyExtractor keyExtractor,
+ DataType sourceFieldType) {
this.readBuilder = readBuilder;
+ this.keyExtractor = keyExtractor;
+ this.sourceFieldType = sourceFieldType;
}
@Override
public void open() throws Exception {
super.open();
this.tableRead = readBuilder.newRead();
+ this.sourceFieldGetter = InternalRow.createFieldGetter(sourceFieldType, 0);
}
@Override
public void processElement(StreamRecord<SortedSplitTask> element) throws Exception {
SortedSplitTask buildTask = element.getValue();
- GenericRow taskId = GenericRow.of(buildTask.taskId);
try (RecordReader<InternalRow> reader = tableRead.createReader(buildTask.split)) {
- reader.forEachRemaining(
- row ->
+ RecordReader.RecordIterator<InternalRow> batch;
+ while ((batch = reader.readBatch()) != null) {
+ try {
+ InternalRow row;
+ while ((row = batch.next()) != null) {
+ long rowId = row.getLong(1);
+ boolean[] emitted = new boolean[1];
+ keyExtractor.extract(
+ sourceFieldGetter.getFieldOrNull(row),
+ key -> {
+ emitted[0] = true;
+ output.collect(
+ new StreamRecord<>(
+ GenericRow.of(
+ buildTask.taskId, key, rowId)));
+ });
+ if (!emitted[0]) {
output.collect(
new StreamRecord<>(
- new FlinkRowData(new JoinedRow(taskId, row)))));
+ GenericRow.of(buildTask.taskId, null, rowId)));
+ }
+ }
+ } finally {
+ batch.releaseBatch();
+ }
+ }
}
}
}
- private static class WriteIndexOperator extends BoundedOneInputOperator<RowData, Committable> {
+ private static class WriteIndexOperator
+ extends BoundedOneInputOperator<InternalRow, Committable> {
private final List<SortedBuildTask> buildTasks;
private final int partitionFieldSize;
@@ -457,10 +557,9 @@ public class SortedIndexTopoBuilder {
private final int rowIdPos;
private final DataType indexFieldType;
- private transient long counter;
private transient SortedBuildTask currentTask;
private transient BinaryRow currentPartition;
- private transient GlobalIndexSingleColumnWriter currentWriter;
+ private transient SortedSingleColumnIndexWriter currentWriter;
private transient List<CommitMessage> commitMessages;
private transient Map<Integer, SortedBuildTask> buildTasksById;
private transient InternalRow.FieldGetter indexFieldGetter;
@@ -498,8 +597,8 @@ public class SortedIndexTopoBuilder {
}
@Override
- public void processElement(StreamRecord<RowData> element) throws IOException {
- InternalRow row = new FlinkRowWrapper(element.getValue());
+ public void processElement(StreamRecord<InternalRow> element) throws IOException {
+ InternalRow row = element.getValue();
int taskId = row.getInt(taskIdPos);
SortedBuildTask task = buildTasksById.get(taskId);
if (task == null) {
@@ -512,14 +611,8 @@ public class SortedIndexTopoBuilder {
currentPartition = binaryRowSerializer.deserializeFromBytes(task.partition);
}
- if (currentWriter != null && counter >= writer.recordsPerRange()) {
- flushCurrentWriter();
- }
-
- counter++;
-
if (currentWriter == null) {
- currentWriter = writer.createWriter();
+ currentWriter = writer.createTaskWriter(currentTask.rowRange);
}
long localRowId = row.getLong(rowIdPos) - currentTask.rowRange.from;
@@ -540,11 +633,10 @@ public class SortedIndexTopoBuilder {
@Override
public void close() throws Exception {
try {
- GlobalIndexSingleColumnWriter writer = currentWriter;
+ SortedSingleColumnIndexWriter writer = currentWriter;
currentWriter = null;
- counter = 0;
- if (writer instanceof AutoCloseable) {
- ((AutoCloseable) writer).close();
+ if (writer != null) {
+ writer.close();
}
} finally {
super.close();
@@ -552,16 +644,17 @@ public class SortedIndexTopoBuilder {
}
private void flushCurrentWriter() throws IOException {
- if (counter > 0 && currentWriter != null) {
- commitMessages.add(
- writer.flushIndex(
- currentTask.rowRange,
- currentWriter.finish(),
- currentPartition,
- scanSnapshotId));
+ if (currentWriter != null) {
+ for (List<ResultEntry> resultEntries : currentWriter.finish()) {
+ commitMessages.add(
+ writer.flushIndex(
+ currentTask.rowRange,
+ resultEntries,
+ currentPartition,
+ scanSnapshotId));
+ }
}
currentWriter = null;
- counter = 0;
}
}
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/CreateGlobalIndexProcedure.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/CreateGlobalIndexProcedure.java
index 71cb07e398..8fb6fac49e 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/CreateGlobalIndexProcedure.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/CreateGlobalIndexProcedure.java
@@ -191,6 +191,9 @@ public class CreateGlobalIndexProcedure extends ProcedureBase {
if ("bitmap".equals(indexType)) {
return "Bitmap";
}
+ if ("multivalue".equals(indexType)) {
+ return "Multivalue";
+ }
return indexType;
}
}
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/SortedGlobalIndexITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/SortedGlobalIndexITCase.java
index 14f768c1f7..70be19ff6d 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/SortedGlobalIndexITCase.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/SortedGlobalIndexITCase.java
@@ -19,10 +19,16 @@
package org.apache.paimon.flink;
import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.globalindex.IndexedSplit;
import org.apache.paimon.index.DataEvolutionIndexSourceMeta;
import org.apache.paimon.index.IndexFileMeta;
import org.apache.paimon.manifest.IndexManifestEntry;
+import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.predicate.PredicateBuilder;
import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.source.ReadBuilder;
+import org.apache.paimon.table.source.TableScan;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
@@ -107,6 +113,100 @@ public class SortedGlobalIndexITCase extends CatalogITCaseBase {
.containsOnly(Row.of(100, "name_100"));
}
+ @Test
+ public void testMultiValueIndex() throws Exception {
+ tEnv.getConfig().set(TableConfigOptions.TABLE_DML_SYNC, true);
+ sql(
+ "CREATE TABLE T_MULTIVALUE (id INT, tags ARRAY<STRING>) WITH ("
+ + "'bucket' = '-1', "
+ + "'global-index.enabled' = 'true', "
+ + "'row-tracking.enabled' = 'true', "
+ + "'data-evolution.enabled' = 'true', "
+ + "'global-index.column-update-action' = 'IGNORE'"
+ + ")");
+ sql(
+ "INSERT INTO T_MULTIVALUE VALUES "
+ + "(1, ARRAY['red', 'blue']), "
+ + "(2, ARRAY['blue']), "
+ + "(3, ARRAY['green']), "
+ + "(4, ARRAY['red', 'red']), "
+ + "(5, CAST(NULL AS ARRAY<STRING>)), "
+ + "(6, ARRAY[CAST(NULL AS STRING)]), "
+ + "(7, ARRAY[CAST(NULL AS STRING)]), "
+ + "(8, ARRAY['red', CAST(NULL AS STRING)])");
+ sql(
+ "CALL sys.create_global_index(`table` => 'default.T_MULTIVALUE', "
+ + "index_column => 'tags', index_type => 'multivalue', "
+ + "options => 'sorted-index.records-per-range=2')");
+
+ FileStoreTable table = paimonTable("T_MULTIVALUE");
+ List<IndexFileMeta> entries =
+ table.store().newIndexFileHandler().scanEntries().stream()
+ .map(IndexManifestEntry::indexFile)
+ .filter(file -> "multivalue".equals(file.indexType()))
+ .collect(Collectors.toList());
+ assertThat(entries).hasSizeGreaterThan(1);
+ assertThat(entries).allSatisfy(entry -> assertThat(entry.globalIndexMeta()).isNotNull());
+ assertThat(entries.stream().mapToLong(IndexFileMeta::rowCount).sum()).isEqualTo(8L);
+ Set<String> initialFiles =
+ entries.stream().map(IndexFileMeta::fileName).collect(Collectors.toSet());
+
+ Predicate predicate =
+ new PredicateBuilder(table.rowType())
+ .arrayContains(1, BinaryString.fromString("red"));
+ ReadBuilder readBuilder = table.newReadBuilder().withFilter(predicate);
+ TableScan.Plan plan = readBuilder.newScan().plan();
+ assertThat(plan.splits()).allMatch(IndexedSplit.class::isInstance);
+ List<Integer> ids = new java.util.ArrayList<>();
+ readBuilder
+ .newRead()
+ .executeFilter()
+ .createReader(plan)
+ .forEachRemaining(row -> ids.add(row.getInt(0)));
+ assertThat(ids).containsExactlyInAnyOrder(1, 4, 8);
+
+ sql("CREATE TABLE S_MULTIVALUE (id INT, tags ARRAY<STRING>)");
+ sql("INSERT INTO S_MULTIVALUE VALUES (2, ARRAY['red'])");
+ sql(
+ "CALL sys.data_evolution_merge_into("
+ + "'default.T_MULTIVALUE', '', '', 'S_MULTIVALUE', "
+ + "'T_MULTIVALUE.id=S_MULTIVALUE.id', 'tags=S_MULTIVALUE.tags', 2)");
+ long updateSnapshotId = paimonTable("T_MULTIVALUE").snapshotManager().latestSnapshot().id();
+
+ sql(
+ "CALL sys.create_global_index(`table` => 'default.T_MULTIVALUE', "
+ + "index_column => 'tags', index_type => 'multivalue', "
+ + "options => 'sorted-index.records-per-range=2')");
+ table = paimonTable("T_MULTIVALUE");
+ List<IndexFileMeta> refreshed =
+ table.store().newIndexFileHandler().scanEntries().stream()
+ .map(IndexManifestEntry::indexFile)
+ .filter(file -> "multivalue".equals(file.indexType()))
+ .collect(Collectors.toList());
+ assertThat(refreshed.stream().map(IndexFileMeta::fileName).collect(Collectors.toSet()))
+ .doesNotContainAnyElementsOf(initialFiles);
+ assertThat(refreshed)
+ .allSatisfy(
+ entry ->
+ assertThat(
+ DataEvolutionIndexSourceMeta.fromIndexFile(entry)
+ .scanSnapshotId())
+ .isEqualTo(updateSnapshotId));
+
+ predicate =
+ new PredicateBuilder(table.rowType())
+ .arrayContains(1, BinaryString.fromString("red"));
+ readBuilder = table.newReadBuilder().withFilter(predicate);
+ plan = readBuilder.newScan().plan();
+ ids.clear();
+ readBuilder
+ .newRead()
+ .executeFilter()
+ .createReader(plan)
+ .forEachRemaining(row -> ids.add(row.getInt(0)));
+ assertThat(ids).containsExactlyInAnyOrder(1, 2, 4, 8);
+ }
+
@Test
public void testBitmapRefreshesUpdatedDataEvolutionRange() throws Exception {
tEnv.getConfig().set(TableConfigOptions.TABLE_DML_SYNC, true);
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/globalindex/SortedIndexTopoBuilderTest.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/globalindex/SortedIndexTopoBuilderTest.java
index 4fc485bfd9..a29eb9e7be 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/globalindex/SortedIndexTopoBuilderTest.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/globalindex/SortedIndexTopoBuilderTest.java
@@ -22,6 +22,7 @@ import org.apache.paimon.flink.globalindex.SortedIndexTopoBuilder.SortedBuildTas
import org.apache.paimon.globalindex.GlobalIndexSingleColumnWriter;
import org.apache.paimon.globalindex.sorted.SortedGlobalIndexScanner;
import org.apache.paimon.globalindex.sorted.SortedGlobalIndexWriter;
+import org.apache.paimon.globalindex.sorted.SortedSingleColumnIndexWriter;
import org.apache.paimon.options.Options;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.SpecialFields;
@@ -53,6 +54,7 @@ public class SortedIndexTopoBuilderTest {
public void testSupportsBitmapAndBTree() {
assertThat(SortedIndexTopoBuilder.supports("bitmap")).isTrue();
assertThat(SortedIndexTopoBuilder.supports("btree")).isTrue();
+ assertThat(SortedIndexTopoBuilder.supports("multivalue")).isTrue();
assertThat(SortedIndexTopoBuilder.supports("inverted")).isFalse();
}
@@ -91,9 +93,11 @@ public class SortedIndexTopoBuilderTest {
mock(
GlobalIndexSingleColumnWriter.class,
org.mockito.Mockito.withSettings().extraInterfaces(Closeable.class));
+ SortedSingleColumnIndexWriter taskWriter =
+ SortedSingleColumnIndexWriter.forSourceRowCount(1, activeWriter);
Field currentWriter = operatorClass.getDeclaredField("currentWriter");
currentWriter.setAccessible(true);
- currentWriter.set(operator, activeWriter);
+ currentWriter.set(operator, taskWriter);
Method close = operatorClass.getMethod("close");
close.invoke(operator);
@@ -172,7 +176,7 @@ public class SortedIndexTopoBuilderTest {
}
@Test
- public void testSortColumnsUseRowIdAsTieBreaker() {
+ public void testNormalizedSortColumnsUseRowIdAsTieBreaker() {
assertThat(SortedIndexTopoBuilder.createSortColumns("task-id", "index-key"))
.containsExactly("task-id", "index-key", SpecialFields.ROW_ID.name());
}
diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/sorted/SortedIndexTopoBuilder.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/sorted/SortedIndexTopoBuilder.java
index 57fa28a47a..62bfb157d3 100644
--- a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/sorted/SortedIndexTopoBuilder.java
+++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/sorted/SortedIndexTopoBuilder.java
@@ -19,9 +19,13 @@
package org.apache.paimon.spark.globalindex.sorted;
import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.GenericRow;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.data.serializer.BinaryRowSerializer;
+import org.apache.paimon.globalindex.GlobalIndexKeyExtractor;
+import org.apache.paimon.globalindex.GlobalIndexer;
import org.apache.paimon.globalindex.ScanResult;
+import org.apache.paimon.globalindex.SortedGlobalIndexer;
import org.apache.paimon.globalindex.sorted.SortedGlobalIndexScanner;
import org.apache.paimon.globalindex.sorted.SortedGlobalIndexWriter;
import org.apache.paimon.globalindex.sorted.SortedIndexOptions;
@@ -34,12 +38,14 @@ import org.apache.paimon.spark.SparkRow;
import org.apache.paimon.spark.globalindex.GlobalIndexTopologyBuilder;
import org.apache.paimon.spark.util.ScanPlanHelper$;
import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.SpecialFields;
import org.apache.paimon.table.sink.CommitMessage;
import org.apache.paimon.table.sink.CommitMessageImpl;
import org.apache.paimon.table.sink.CommitMessageSerializer;
import org.apache.paimon.table.source.DataSplit;
import org.apache.paimon.table.source.Split;
import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.InstantiationUtil;
import org.apache.paimon.utils.Range;
@@ -58,20 +64,26 @@ import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
+import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
+import java.util.NoSuchElementException;
import java.util.Optional;
import static org.apache.paimon.globalindex.GlobalIndexBuilderUtils.groupSplitsByRange;
+import static org.apache.paimon.globalindex.GlobalIndexBuilderUtils.shardSplitsByRowRange;
import static org.apache.paimon.globalindex.GlobalIndexBuilderUtils.splitByContiguousRowRange;
/** The {@link GlobalIndexTopologyBuilder} for sorted indexes. */
public class SortedIndexTopoBuilder implements GlobalIndexTopologyBuilder {
+ private static final String BUILD_TASK_ID_FIELD = "_SORTED_INDEX_BUILD_TASK_ID";
+ private static final int BUILD_TASK_ID_FIELD_ID = -1;
+
private static final HashSet<String> SUPPORTED_INDEX_TYPES =
- new HashSet<>(Arrays.asList("btree", "bitmap"));
+ new HashSet<>(Arrays.asList("btree", "bitmap", "multivalue"));
public static boolean supports(String indexType) {
return SUPPORTED_INDEX_TYPES.contains(indexType);
@@ -113,15 +125,20 @@ public class SortedIndexTopoBuilder implements GlobalIndexTopologyBuilder {
return Collections.emptyList();
}
- List<String> selectedColumns = new ArrayList<>(readType.getFieldNames());
-
// Calculate maximum parallelism bound
long recordsPerRange = options.get(SortedIndexOptions.SORTED_INDEX_RECORDS_PER_RANGE);
int maxParallelism = options.get(SortedIndexOptions.SORTED_INDEX_BUILD_MAX_PARALLELISM);
List<CommitMessage> allMessages = new ArrayList<>();
- List<String> sortColumns = new ArrayList<>();
- sortColumns.add(indexField.name());
+ GlobalIndexer indexer = GlobalIndexer.create(indexType, indexField, options);
+ if (!(indexer instanceof SortedGlobalIndexer)) {
+ throw new IllegalArgumentException(
+ "Index algorithm " + indexType + " does not expose sorted index keys.");
+ }
+ GlobalIndexKeyExtractor keyExtractor = ((SortedGlobalIndexer) indexer).keyExtractor();
+ String taskIdField = buildTaskIdFieldName(readType);
+ RowType normalizedReadType =
+ normalizedReadType(readType, taskIdField, indexField, keyExtractor.keyType());
final int partitionKeyNum = table.partitionKeys().size();
BinaryRowSerializer binaryRowSerializer = new BinaryRowSerializer(partitionKeyNum);
SortedGlobalIndexWriter indexWriter =
@@ -135,42 +152,114 @@ public class SortedIndexTopoBuilder implements GlobalIndexTopologyBuilder {
if (rangeSplits.isEmpty()) {
continue;
}
- int partitionNum = Math.max((int) (range.count() / recordsPerRange), 1);
- partitionNum = Math.min(partitionNum, maxParallelism);
+
+ final byte[] serializedWriter = InstantiationUtil.serializeObject(indexWriter);
+ final byte[] partitionBytes =
+ binaryRowSerializer.serializeToBytes(partitionEntry.getKey());
+ if (keyExtractor.isIdentity()) {
+ int partitionNum = Math.max((int) (range.count() / recordsPerRange), 1);
+ partitionNum = Math.min(partitionNum, maxParallelism);
+
+ Dataset<Row> source =
+ PaimonUtils.createDataset(
+ spark,
+ ScanPlanHelper$.MODULE$.createNewScanPlan(
+ rangeSplits.toArray(new Split[0]), relation));
+ Dataset<Row> selected =
+ source.select(
+ readType.getFieldNames().stream()
+ .map(functions::col)
+ .toArray(Column[]::new));
+ Column[] sortFields =
+ new Column[] {
+ functions.col(indexField.name()),
+ functions.col(SpecialFields.ROW_ID.name())
+ };
+ Dataset<Row> partitioned =
+ selected.repartitionByRange(partitionNum, sortFields)
+ .sortWithinPartitions(sortFields);
+ JavaRDD<byte[]> written =
+ partitioned
+ .javaRDD()
+ .map(row -> (InternalRow) (new SparkRow(readType, row)))
+ .mapPartitions(
+ (FlatMapFunction<Iterator<InternalRow>, byte[]>)
+ iter ->
+ buildSortedIndex(
+ iter,
+ serializedWriter,
+ range,
+ partitionKeyNum,
+ partitionBytes,
+ scanSnapshotId));
+ allMessages.addAll(CommitMessageSerializer.deserializeAll(written.collect()));
+ continue;
+ }
+
+ Map<Range, List<Split>> shardedSplits =
+ shardSplitsByRowRange(
+ Collections.singletonMap(range, rangeSplits), recordsPerRange);
+ if (shardedSplits.isEmpty()) {
+ continue;
+ }
+ int partitionNum = Math.min(shardedSplits.size(), maxParallelism);
+ List<Split> splitsToRead = new ArrayList<>();
+ Map<Long, Range> taskRanges = new HashMap<>();
+ for (Map.Entry<Range, List<Split>> shard : shardedSplits.entrySet()) {
+ splitsToRead.addAll(shard.getValue());
+ taskRanges.put(shard.getKey().from / recordsPerRange, shard.getKey());
+ }
Dataset<Row> source =
PaimonUtils.createDataset(
spark,
ScanPlanHelper$.MODULE$.createNewScanPlan(
- rangeSplits.toArray(new Split[0]), relation));
+ splitsToRead.toArray(new Split[0]), relation));
Dataset<Row> selected =
source.select(
- selectedColumns.stream()
- .map(functions::col)
- .toArray(Column[]::new));
+ functions.col(indexField.name()),
+ functions.col(SpecialFields.ROW_ID.name()))
+ .withColumn(
+ taskIdField,
+ functions
+ .expr(
+ "`"
+ + SpecialFields.ROW_ID.name()
+ + "` DIV "
+ + recordsPerRange)
+ .cast("long"));
+ Dataset<Row> normalized =
+ selected.select(
+ functions.col(taskIdField),
+ functions
+ .explode_outer(functions.col(indexField.name()))
+ .alias(indexField.name()),
+ functions.col(SpecialFields.ROW_ID.name()));
Column[] sortFields =
- sortColumns.stream().map(functions::col).toArray(Column[]::new);
+ new Column[] {
+ functions.col(taskIdField),
+ functions.col(indexField.name()),
+ functions.col(SpecialFields.ROW_ID.name())
+ };
Dataset<Row> partitioned =
- selected.repartitionByRange(partitionNum, sortFields)
+ normalized
+ .repartition(partitionNum, functions.col(taskIdField))
.sortWithinPartitions(sortFields);
- final byte[] serializedWriter = InstantiationUtil.serializeObject(indexWriter);
- final byte[] partitionBytes =
- binaryRowSerializer.serializeToBytes(partitionEntry.getKey());
JavaRDD<byte[]> written =
partitioned
.javaRDD()
- .map(row -> (InternalRow) (new SparkRow(readType, row)))
+ .map(row -> (InternalRow) (new SparkRow(normalizedReadType, row)))
.mapPartitions(
(FlatMapFunction<Iterator<InternalRow>, byte[]>)
iter ->
- buildSortedIndex(
+ buildSortedIndexes(
iter,
serializedWriter,
- range,
+ taskRanges,
partitionKeyNum,
partitionBytes,
scanSnapshotId));
@@ -208,4 +297,107 @@ public class SortedIndexTopoBuilder implements GlobalIndexTopologyBuilder {
writer.buildForSinglePartition(range, partition, input, scanSnapshotId))
.iterator();
}
+
+ private static Iterator<byte[]> buildSortedIndexes(
+ Iterator<InternalRow> input,
+ byte[] serializedWriter,
+ Map<Long, Range> taskRanges,
+ int partitionKeyNum,
+ byte[] partitionBytes,
+ long scanSnapshotId)
+ throws IOException, ClassNotFoundException {
+ final BinaryRowSerializer binaryRowSerializer = new BinaryRowSerializer(partitionKeyNum);
+ BinaryRow partition = binaryRowSerializer.deserializeFromBytes(partitionBytes);
+ SortedGlobalIndexWriter writer =
+ InstantiationUtil.deserializeObject(
+ serializedWriter, SortedGlobalIndexWriter.class.getClassLoader());
+ SortedTaskInput taskInput = new SortedTaskInput(input, writer.keyExtractor().keyType());
+ List<byte[]> results = new ArrayList<>();
+ while (taskInput.hasTask()) {
+ long taskId = taskInput.taskId();
+ Range range = taskRanges.get(taskId);
+ if (range == null) {
+ throw new IllegalArgumentException("Unknown sorted index build task id: " + taskId);
+ }
+ results.addAll(
+ CommitMessageSerializer.serializeAll(
+ writer.buildForSinglePartition(
+ range,
+ partition,
+ taskInput.consumeTask(taskId),
+ scanSnapshotId)));
+ }
+ return results.iterator();
+ }
+
+ private static String buildTaskIdFieldName(RowType readType) {
+ String fieldName = BUILD_TASK_ID_FIELD;
+ while (readType.containsField(fieldName)) {
+ fieldName = "_" + fieldName;
+ }
+ return fieldName;
+ }
+
+ private static RowType normalizedReadType(
+ RowType readType,
+ String taskIdField,
+ DataField sourceField,
+ org.apache.paimon.types.DataType keyType) {
+ return RowType.of(
+ new DataField(BUILD_TASK_ID_FIELD_ID, taskIdField, DataTypes.BIGINT().notNull()),
+ new DataField(sourceField.id(), sourceField.name(), keyType),
+ readType.getField(SpecialFields.ROW_ID.name()));
+ }
+
+ /** Groups a partition already sorted by task id, normalized key and row id. */
+ private static class SortedTaskInput {
+
+ private final Iterator<InternalRow> input;
+ private final InternalRow.FieldGetter keyGetter;
+
+ private InternalRow next;
+
+ private SortedTaskInput(
+ Iterator<InternalRow> input, org.apache.paimon.types.DataType keyType) {
+ this.input = input;
+ this.keyGetter = InternalRow.createFieldGetter(keyType, 1);
+ advance();
+ }
+
+ private boolean hasTask() {
+ return next != null;
+ }
+
+ private long taskId() {
+ if (next == null) {
+ throw new NoSuchElementException();
+ }
+ return next.getLong(0);
+ }
+
+ private Iterator<InternalRow> consumeTask(long taskId) {
+ return new Iterator<InternalRow>() {
+ @Override
+ public boolean hasNext() {
+ return next != null && next.getLong(0) == taskId;
+ }
+
+ @Override
+ public InternalRow next() {
+ if (!hasNext()) {
+ throw new NoSuchElementException();
+ }
+ InternalRow row = next;
+ InternalRow result =
+ GenericRow.of(keyGetter.getFieldOrNull(row), row.getLong(2));
+ advance();
+ return result;
+ }
+ };
+ }
+
+ private void advance() {
+ next = input.hasNext() ? input.next() : null;
+ }
+ }
}
diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/CreateGlobalIndexProcedureTest.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/CreateGlobalIndexProcedureTest.scala
index 9f4910cbe0..94d836a613 100644
--- a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/CreateGlobalIndexProcedureTest.scala
+++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/CreateGlobalIndexProcedureTest.scala
@@ -244,6 +244,50 @@ class CreateGlobalIndexProcedureTest extends PaimonSparkTestBase with StreamTest
}
}
+ test("create multivalue global index") {
+ withTable("T") {
+ spark.sql("""
+ |CREATE TABLE T (id INT, tags ARRAY<STRING>)
+ |TBLPROPERTIES (
+ | 'bucket' = '-1',
+ | 'global-index.enabled' = 'true',
+ | 'row-tracking.enabled' = 'true',
+ | 'data-evolution.enabled' = 'true')
+ |""".stripMargin)
+
+ spark.sql(
+ "INSERT INTO T VALUES " +
+ "(1, array('red', 'blue')), " +
+ "(2, array('blue')), " +
+ "(3, array('green')), " +
+ "(4, array('red', 'red')), " +
+ "(5, CAST(NULL AS ARRAY<STRING>)), " +
+ "(6, CAST(array() AS ARRAY<STRING>)), " +
+ "(7, array(CAST(NULL AS STRING))), " +
+ "(8, array('red', CAST(NULL AS STRING)))"
+ )
+
+ val output =
+ spark
+ .sql("CALL sys.create_global_index(table => 'test.T', index_column => 'tags', " +
+ "index_type => 'multivalue', options => 'sorted-index.records-per-range=2')")
+ .collect()
+ .head
+
+ assert(output.getBoolean(0))
+ val entries = loadTable("T")
+ .store()
+ .newIndexFileHandler()
+ .scanEntries()
+ .asScala
+ .map(_.indexFile())
+ .filter(_.indexType() == "multivalue")
+ assert(entries.size > 1)
+ assert(entries.map(_.rowCount()).sum == 8L)
+ entries.foreach(entry => assert(entry.globalIndexMeta() != null))
+ }
+ }
+
test("create btree global index with multiple partitions") {
withTable("T") {
spark.sql("""