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/28 10:49:37 UTC
(paimon) branch master updated: [core] Support FM indexes for primary-key tables (#9447)
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 f01570c54f [core] Support FM indexes for primary-key tables (#9447)
f01570c54f is described below
commit f01570c54f4cf84384456843650e9bd3aa8cb432
Author: Jingsong Lee <ji...@gmail.com>
AuthorDate: Fri Aug 28 18:49:31 2026 +0800
[core] Support FM indexes for primary-key tables (#9447)
---
docs/docs/multimodal-table/global-index.mdx | 10 +-
docs/docs/multimodal-table/global-index/fm.mdx | 111 +++++++
docs/docs/multimodal-table/index.mdx | 2 +-
docs/docs/primary-key-table/global-index.mdx | 72 +++--
docs/sidebars.js | 1 +
.../main/java/org/apache/paimon/CoreOptions.java | 19 ++
.../globalindex/fmindex/FMGlobalIndexReader.java | 63 ++--
.../globalindex/fmindex/FMGlobalIndexWriter.java | 232 ++++++++------
.../globalindex/fmindex/FMGlobalIndexer.java | 46 ++-
.../paimon/globalindex/fmindex/FMIndexFile.java | 335 ++++++++++++++++++---
.../globalindex/fmindex/FMGlobalIndexTest.java | 53 ++--
.../src/test/resources/fmindex-v1-golden.base64 | 2 +-
.../java/org/apache/paimon/KeyValueFileStore.java | 1 +
.../pk/BucketedPrimaryKeyIndexMaintainer.java | 42 ++-
.../paimon/index/pk/PrimaryKeyIndexDefinition.java | 7 +-
.../index/pk/PrimaryKeyIndexDefinitions.java | 22 +-
.../pksorted/BucketedSortedIndexMaintainer.java | 6 +-
.../index/pksorted/PkSequentialIndexBuilder.java | 192 ++++++++++++
.../paimon/index/pksorted/PkSortedIndexFile.java | 13 +-
.../org/apache/paimon/schema/SchemaManager.java | 1 +
.../org/apache/paimon/schema/SchemaValidation.java | 25 +-
.../paimon/table/source/PrimaryKeyBatchScan.java | 4 +-
.../table/source/PrimaryKeySortedIndexScan.java | 8 +-
.../index/pk/PrimaryKeyIndexDefinitionsTest.java | 17 ++
.../BucketedSortedIndexMaintainerTest.java | 23 +-
.../pksorted/PkSequentialIndexBuilderTest.java | 226 ++++++++++++++
.../index/pksorted/PkSortedIndexFileTest.java | 9 +-
.../pksorted/PrimaryKeySortedIndexOptionsTest.java | 18 ++
.../schema/PrimaryKeyFMIndexValidationTest.java | 147 +++++++++
.../PrimaryKeySortedIndexValidationTest.java | 10 +-
.../apache/paimon/schema/SchemaManagerTest.java | 36 +++
.../spark/sql/PrimaryKeySortedIndexTest.scala | 56 ++++
32 files changed, 1547 insertions(+), 262 deletions(-)
diff --git a/docs/docs/multimodal-table/global-index.mdx b/docs/docs/multimodal-table/global-index.mdx
index 6de531fcb2..5217089d38 100644
--- a/docs/docs/multimodal-table/global-index.mdx
+++ b/docs/docs/multimodal-table/global-index.mdx
@@ -35,6 +35,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.
+- **[FM Index](./global-index/fm)**: An exact partitioned substring index for `CONTAINS` predicates on character 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.
@@ -44,6 +45,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`, `ARRAYS_OVERLAP`, and `ARRAY_CONTAINS_ALL`. |
+| FM | Exact substring filters on character columns | Supports needles of any byte length and partitions the indexed text for bounded-memory construction and demand-loaded reads. |
| 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. |
@@ -396,7 +398,7 @@ 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, Bitmap, and Multivalue queries. |
+| `scalar-index.search-mode` | `fast` | Search mode for BTree, Bitmap, Multivalue, and FM 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. |
@@ -427,6 +429,12 @@ 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.
+## FM Index
+
+Use FM indexes for exact substring predicates on character columns. See
+[FM Index](./global-index/fm) for build examples, exactness and coverage semantics, tuning options,
+and primary-key table configuration.
+
## 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/fm.mdx b/docs/docs/multimodal-table/global-index/fm.mdx
new file mode 100644
index 0000000000..6ab2bf4a99
--- /dev/null
+++ b/docs/docs/multimodal-table/global-index/fm.mdx
@@ -0,0 +1,111 @@
+---
+title: "FM Index"
+sidebar_position: 4
+---
+
+<!--
+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.
+-->
+
+# FM Index
+
+The FM index is an exact byte-oriented substring index for `CHAR`, `VARCHAR`, and `STRING`
+columns. It supports `CONTAINS` needles of any byte length without a configured gram size. Null
+values do not match; empty needles follow the normal Paimon predicate semantics.
+
+The writer divides source rows into independent partitions and appends them to one checksummed
+container file. Each partition stores a compressed wavelet matrix, sampled suffix-array values,
+row boundaries, null rows, and exact verification pages. Reads demand-load bounded blocks instead
+of downloading the complete container. If locating matches would cost more than exact
+verification, the reader scans the relevant verification pages and still returns an exact result.
+
+## Create a Global FM Index
+
+Create the index on a Data Evolution table with row tracking enabled:
+
+```sql
+CALL sys.create_global_index(
+ table => 'db.documents',
+ index_column => 'content',
+ index_type => 'fmindex',
+ options => 'fm-index.partition-row-count=100000'
+);
+```
+
+Drop it with the same index type:
+
+```sql
+CALL sys.drop_global_index(
+ table => 'db.documents',
+ index_column => 'content',
+ index_type => 'fmindex'
+);
+```
+
+See [Global Index](../global-index) for row-tracking requirements, lifecycle, and partial-coverage
+semantics.
+
+## Query
+
+After a connector converts a literal substring expression to Paimon's `CONTAINS` predicate, the
+normal batch scan uses the FM index automatically. For example, Spark pushes down:
+
+```sql
+SELECT id, content
+FROM documents
+WHERE content LIKE '%needle%';
+```
+
+Results are exact inside every indexed row range. A table-wide global-index query can still omit
+matches in row ranges that have never been indexed; rebuild the index for newly appended ranges or
+use the visibility callback described on the Global Index page.
+
+For a primary-key table, configure `pk-fm.index.columns` instead. Primary-key FM indexes follow
+data compaction and scan uncovered files through the ordinary data path, so partial coverage does
+not make `CONTAINS` results incomplete. See [Primary-Key Indexes](../../primary-key-table/global-index).
+
+## Options
+
+Global options can be passed to `create_global_index`. For a primary-key index, place the same
+keys in `fields.<column>.pk-fm.index.options`; the `fm-index.` prefix may be omitted inside that
+JSON object.
+
+| Option | Default | Description |
+|---|---|---|
+| `fm-index.partition-size` | `16 mb` | Maximum encoded text buffered by one independently readable partition. |
+| `fm-index.partition-row-count` | `100000` | Maximum source rows in one partition. |
+| `fm-index.sa-sample-rate` | `32` | Suffix-array sample rate. A smaller power of two speeds locate at the cost of a larger index. |
+| `fm-index.compression` | `lz4` | Compression codec for independently checksummed FM blocks. |
+| `fm-index.compression-level` | `1` | Compression level for codecs which support levels. |
+| `fm-index.read-cache-size` | `64 mb` | Maximum decoded rank and sample block cache per indexer. |
+| `fm-index.demand-page-size` | `512 kb` | Target contiguous range size when demand-loading blocks. |
+| `fm-index.locate-cost-ratio` | `0.001` | Maximum estimated suffix-array locate work relative to exact stored-value scan bytes before verification fallback. |
+
+`fm-index.partition-size` and `fm-index.partition-row-count` bound build memory and the unit of
+independent reads. Smaller partitions reduce peak construction memory but increase the number of
+partitions searched per query. All partitions produced by one writer are stored in one physical
+index file and represented by one index manifest entry. A lower `fm-index.sa-sample-rate`
+accelerates locating matched rows but stores more suffix-array samples.
+
+## Limitations
+
+- The index is single-column and accepts only character string types.
+- Matching is byte-oriented and case-sensitive; it does not apply a tokenizer, collation, or
+ Unicode normalization.
+- Global FM indexes inherit the partial-coverage behavior of table-wide global indexes.
+- Primary-key FM indexes are built from eligible compact output, not directly from Level-0 appends.
diff --git a/docs/docs/multimodal-table/index.mdx b/docs/docs/multimodal-table/index.mdx
index 2f26c41bb3..15a5e9305f 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, Multivalue, vector, and full-text indexes for efficient lookups and similarity search.
+- **[Global Index](./global-index)**: Build BTree, Bitmap, Multivalue, FM, vector, and full-text indexes for efficient lookups, exact substring filtering, 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 00361af4c7..fb19e7d452 100644
--- a/docs/docs/primary-key-table/global-index.mdx
+++ b/docs/docs/primary-key-table/global-index.mdx
@@ -27,7 +27,7 @@ under the License.
# Primary-Key Indexes
-Primary-key tables can maintain Vector, Full Text, BTree, Bitmap, and Multivalue indexes together
+Primary-key tables can maintain Vector, Full Text, FM, 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.
@@ -100,11 +100,23 @@ For an append-only or Data Evolution table whose full-text index is built indepe
</TabItem>
+<TabItem value="fm" label="FM">
+
+Use FM for exact substring predicates on a `CHAR`, `VARCHAR`, or `STRING` column. Normal batch
+scans apply it automatically after a connector converts a literal substring expression to
+Paimon's `CONTAINS` predicate. Needles of any byte length are supported without choosing a gram
+size.
+
+For an append-only or Data Evolution table whose FM index is built independently, see
+[FM Index](../multimodal-table/global-index/fm).
+
+</TabItem>
+
</Tabs>
Different columns in one table can use different index families. One column can occur in at most
one of `pk-vector.index.columns`, `pk-full-text.index.columns`, `pk-btree.index.columns`,
-`pk-bitmap.index.columns`, and `pk-multivalue.index.columns`.
+`pk-bitmap.index.columns`, `pk-multivalue.index.columns`, and `pk-fm.index.columns`.
## Requirements
@@ -120,14 +132,14 @@ supported by this configuration.
<Tabs groupId="primary-key-index-requirements">
-<TabItem value="sorted" label="BTree, Bitmap, and Multivalue">
+<TabItem value="sorted" label="BTree, Bitmap, Multivalue, and FM">
-BTree, Bitmap, and Multivalue additionally require:
+BTree, Bitmap, Multivalue, and FM additionally require:
- `deletion-vectors.enabled = true`.
- `deletion-vectors.merge-on-read = false`.
- A supported scalar column type for BTree and Bitmap, or an `ARRAY` of supported scalar elements
- for Multivalue.
+ for Multivalue. FM requires a `CHAR`, `VARCHAR`, or `STRING` column.
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.
@@ -166,8 +178,9 @@ Exactly one full-text column is currently supported per table.
## Create a Table
-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`.
+The following table uses all six families on different columns: Vector for `embedding`, Full Text
+for `content`, FM for `raw_text`, BTree for `amount`, Bitmap for `status`, and Multivalue for
+`tags`.
<Tabs groupId="primary-key-index-create-table">
@@ -180,6 +193,7 @@ CREATE TABLE items (
amount DECIMAL(12, 2),
tags ARRAY<STRING>,
content STRING,
+ raw_text STRING,
embedding ARRAY<FLOAT> COMMENT '__VECTOR_FIELD;3',
PRIMARY KEY (id) NOT ENFORCED
) WITH (
@@ -191,6 +205,8 @@ CREATE TABLE items (
'fields.embedding.pk-vector.index.options' = '{"nlist":"256"}',
'pk-full-text.index.columns' = 'content',
'fields.content.pk-full-text.index.options' = '{"tokenizer":"jieba"}',
+ 'pk-fm.index.columns' = 'raw_text',
+ 'fields.raw_text.pk-fm.index.options' = '{"partition-row-count":"100000"}',
'pk-btree.index.columns' = 'amount',
'fields.amount.pk-btree.index.options' = '{"block-size":"64 kb"}',
'pk-bitmap.index.columns' = 'status',
@@ -215,6 +231,7 @@ CREATE TABLE items (
amount DECIMAL(12, 2),
tags ARRAY<STRING>,
content STRING,
+ raw_text STRING,
embedding ARRAY<FLOAT> COMMENT '__VECTOR_FIELD;3'
) USING paimon
TBLPROPERTIES (
@@ -227,6 +244,8 @@ TBLPROPERTIES (
'fields.embedding.pk-vector.index.options' = '{"nlist":"256"}',
'pk-full-text.index.columns' = 'content',
'fields.content.pk-full-text.index.options' = '{"tokenizer":"jieba"}',
+ 'pk-fm.index.columns' = 'raw_text',
+ 'fields.raw_text.pk-fm.index.options' = '{"partition-row-count":"100000"}',
'pk-btree.index.columns' = 'amount',
'fields.amount.pk-btree.index.options' = '{"block-size":"64 kb"}',
'pk-bitmap.index.columns' = 'status',
@@ -253,6 +272,8 @@ schema validation.
| `fields.<column>.pk-vector.index.options` | Not set | JSON object containing build options for the selected ANN implementation. |
| `pk-full-text.index.columns` | Not set | Character column to index. Exactly one full-text column is currently supported. |
| `fields.<column>.pk-full-text.index.options` | Not set | JSON object containing native analyzer options. Unqualified keys are scoped to `full-text`. |
+| `pk-fm.index.columns` | Not set | Comma-separated character columns which own independent exact FM indexes. |
+| `fields.<column>.pk-fm.index.options` | Not set | JSON object containing FM build and read options. Unqualified keys are scoped to `fm-index`. |
| `pk-btree.index.columns` | Not set | Comma-separated columns which own independent BTree indexes. |
| `fields.<column>.pk-btree.index.options` | Not set | JSON object containing BTree build options. Unqualified keys are scoped to `btree-index`. |
| `pk-bitmap.index.columns` | Not set | Comma-separated columns which own independent Bitmap indexes. |
@@ -268,7 +289,9 @@ schema validation.
For algorithm-specific options, see the corresponding
[BTree](../multimodal-table/global-index/btree),
-[Bitmap](../multimodal-table/global-index/bitmap), or
+[Bitmap](../multimodal-table/global-index/bitmap),
+[Multivalue](../multimodal-table/global-index/multivalue),
+[FM](../multimodal-table/global-index/fm), or
[Vector](../multimodal-table/global-index/vector) index page. Full-text analyzer options are listed
in the [`paimon-full-text` README](https://github.com/apache/paimon/blob/master/paimon-full-text/README.md).
@@ -289,13 +312,14 @@ output; simply assigning or upgrading a pending file does not make it an index s
### Data-Level Maintenance
-Each indexed column maintains one immutable index payload for the complete eligible source-file
-set in every non-zero data level. When data compaction changes a level, Paimon rebuilds that whole
-level payload, including files in the target level which were not direct compaction inputs. A
-level payload is used only when its ordered source names and row counts exactly match the current
-data level; partial, duplicate, stale, and cross-level payloads are rejected.
+Each indexed column maintains one immutable index group for the complete eligible source-file set
+in every non-zero data level. Each group writes one payload file; an FM payload can contain several
+independently readable internal partitions. When data compaction changes a level, Paimon rebuilds
+that whole level group, including files in the target level which were not direct compaction
+inputs. A level group is used only when its ordered source names and row counts exactly match the
+current data level; stale payloads and cross-level payloads are rejected.
-A rebuild atomically replaces the old payload after the complete new payload is ready. Unrelated
+A rebuild atomically replaces the old group after the complete new group is ready. Unrelated
data levels retain their existing payloads.
Index construction can execute asynchronously inside the writer. A writer which waits for
@@ -307,7 +331,7 @@ 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, Bitmap, and Multivalue scans always read uncovered files through the ordinary data path.
+- BTree, Bitmap, Multivalue, and FM 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
@@ -318,12 +342,24 @@ 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, Bitmap, and Multivalue Queries
+## BTree, Bitmap, Multivalue, and FM Queries
-BTree, Bitmap, and Multivalue indexes are applied automatically to snapshot-scoped Core batch
+BTree, Bitmap, Multivalue, and FM 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`, `ARRAYS_OVERLAP`, and `ARRAY_CONTAINS_ALL` predicates.
+FM accelerates exact `CONTAINS` predicates. For example, Spark converts the following literal
+substring filter to Paimon's `CONTAINS` predicate:
+
+```sql
+SELECT id, raw_text
+FROM items
+WHERE raw_text LIKE '%needle%';
+```
+
+The FM result is exact for indexed files. Paimon reads uncovered files normally and reapplies the
+original predicate and deletion vectors, so partial index coverage affects performance rather than
+result completeness.
The Java/Core predicate can be constructed directly:
@@ -558,7 +594,7 @@ create a table with the desired definition and migrate the data.
## Limitations
-- BTree, Bitmap, and Multivalue definitions are single-column indexes.
+- BTree, Bitmap, Multivalue, and FM 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.
diff --git a/docs/sidebars.js b/docs/sidebars.js
index cd2b6b2d95..d6f18ae5a8 100644
--- a/docs/sidebars.js
+++ b/docs/sidebars.js
@@ -130,6 +130,7 @@ const sidebars = {
"multimodal-table/global-index/btree",
"multimodal-table/global-index/bitmap",
"multimodal-table/global-index/multivalue",
+ "multimodal-table/global-index/fm",
"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 9d2b2c6b59..0494d9d4cd 100644
--- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
+++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
@@ -2990,6 +2990,13 @@ public class CoreOptions implements Serializable {
"Comma-separated character columns indexed by primary-key full-text indexes. "
+ "The first release supports exactly one column.");
+ public static final ConfigOption<String> PK_FM_INDEX_COLUMNS =
+ key("pk-fm.index.columns")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "Comma-separated character columns indexed by primary-key FM indexes.");
+
@Immutable
public static final ConfigOption<Boolean> PK_CLUSTERING_OVERRIDE =
key("pk-clustering-override")
@@ -4647,6 +4654,10 @@ public class CoreOptions implements Serializable {
return options.getOptional(PK_FULL_TEXT_INDEX_COLUMNS).isPresent();
}
+ public boolean primaryKeyFMIndexEnabled() {
+ return options.getOptional(PK_FM_INDEX_COLUMNS).isPresent();
+ }
+
public List<String> primaryKeyVectorIndexColumns() {
return primaryKeyIndexColumns(PK_VECTOR_INDEX_COLUMNS);
}
@@ -4667,6 +4678,10 @@ public class CoreOptions implements Serializable {
return primaryKeyIndexColumns(PK_FULL_TEXT_INDEX_COLUMNS);
}
+ public List<String> primaryKeyFMIndexColumns() {
+ return primaryKeyIndexColumns(PK_FM_INDEX_COLUMNS);
+ }
+
private List<String> primaryKeyIndexColumns(ConfigOption<String> option) {
String columns = options.get(option);
if (columns == null) {
@@ -4687,6 +4702,10 @@ public class CoreOptions implements Serializable {
return primaryKeySortedIndexOptions(column, "pk-multivalue", "multivalue-index.");
}
+ public Options primaryKeyFMIndexOptions(String column) {
+ return primaryKeySortedIndexOptions(column, "pk-fm", "fm-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/java/org/apache/paimon/globalindex/fmindex/FMGlobalIndexReader.java b/paimon-common/src/main/java/org/apache/paimon/globalindex/fmindex/FMGlobalIndexReader.java
index b1b6380f75..8189e60cee 100644
--- a/paimon-common/src/main/java/org/apache/paimon/globalindex/fmindex/FMGlobalIndexReader.java
+++ b/paimon-common/src/main/java/org/apache/paimon/globalindex/fmindex/FMGlobalIndexReader.java
@@ -56,9 +56,8 @@ final class FMGlobalIndexReader implements ContainsRefiningGlobalIndexReader {
@Nullable private final GlobalIndexIOMeta file;
private final ExecutorService executor;
private final FMIndexReadContext readContext;
- @Nullable private final FileSetRowCountValidator rowCountValidator;
- @Nullable private final FMIndexFile.IndexMeta indexMeta;
- private final int filePosition;
+ @Nullable private final ContainerMetadataLoader containerLoader;
+ @Nullable private final FMIndexFile.PartitionMeta partition;
private final int demandPageSize;
private final double locateCostRatio;
@@ -69,18 +68,16 @@ final class FMGlobalIndexReader implements ContainsRefiningGlobalIndexReader {
GlobalIndexIOMeta file,
ExecutorService executor,
FMIndexReadContext readContext,
- FileSetRowCountValidator rowCountValidator,
- @Nullable FMIndexFile.IndexMeta indexMeta,
- int filePosition,
+ ContainerMetadataLoader containerLoader,
+ FMIndexFile.PartitionMeta partition,
int demandPageSize,
double locateCostRatio) {
this.fileReader = fileReader;
this.file = file;
this.executor = executor;
this.readContext = readContext;
- this.rowCountValidator = rowCountValidator;
- this.indexMeta = indexMeta;
- this.filePosition = filePosition;
+ this.containerLoader = containerLoader;
+ this.partition = partition;
this.demandPageSize = readContext.effectiveDemandPageSize(demandPageSize);
this.locateCostRatio = locateCostRatio;
}
@@ -94,9 +91,8 @@ final class FMGlobalIndexReader implements ContainsRefiningGlobalIndexReader {
this.file = null;
this.executor = executor;
this.readContext = readContext;
- this.rowCountValidator = null;
- this.indexMeta = null;
- this.filePosition = -1;
+ this.containerLoader = null;
+ this.partition = null;
this.demandPageSize = readContext.effectiveDemandPageSize(demandPageSize);
this.locateCostRatio = locateCostRatio;
}
@@ -239,11 +235,11 @@ final class FMGlobalIndexReader implements ContainsRefiningGlobalIndexReader {
private boolean candidatesDisjointFromIndexMeta(@Nullable GlobalIndexResult candidates) {
return candidates != null
- && indexMeta != null
+ && partition != null
&& !candidates
.results()
.intersects(
- indexMeta.firstRowId, indexMeta.firstRowId + indexMeta.rowCount);
+ partition.firstRowId, partition.firstRowId + partition.rowCount);
}
private Optional<GlobalIndexResult> queryNullsUnchecked(boolean nulls) {
@@ -809,16 +805,14 @@ final class FMGlobalIndexReader implements ContainsRefiningGlobalIndexReader {
current = metadata;
if (current == null) {
Preconditions.checkState(file != null, "Missing FM index file.");
- FMIndexFile.Footer footer = FMIndexFile.readFooter(input, file.fileSize());
+ Preconditions.checkState(
+ containerLoader != null && partition != null,
+ "Missing FM index container metadata.");
+ containerLoader.validate(input);
+ FMIndexFile.Footer footer =
+ FMIndexFile.readFooter(input, partition, file.fileSize());
FMIndexFile.Directory directory =
FMIndexFile.readDirectory(input, footer, file.fileSize());
- Preconditions.checkState(
- rowCountValidator != null, "Missing FM index row-count validator.");
- rowCountValidator.validate(
- filePosition,
- footer.rowCount,
- footer.firstRowId,
- footer.firstRowId + footer.rowCount - 1L);
current = new Metadata(footer, directory);
metadata = current;
}
@@ -982,6 +976,31 @@ final class FMGlobalIndexReader implements ContainsRefiningGlobalIndexReader {
}
}
+ static final class ContainerMetadataLoader {
+ private final GlobalIndexIOMeta file;
+ private final FMIndexFile.IndexMeta expected;
+ private boolean validated;
+
+ ContainerMetadataLoader(GlobalIndexIOMeta file, FMIndexFile.IndexMeta expected) {
+ this.file = file;
+ this.expected = expected;
+ }
+
+ synchronized void validate(SeekableInputStream input) throws IOException {
+ if (validated) {
+ return;
+ }
+ FMIndexFile.ContainerFooter footer =
+ FMIndexFile.readContainerFooter(input, file.fileSize());
+ FMIndexFile.IndexMeta actual =
+ FMIndexFile.readContainerDirectory(input, footer, file.fileSize());
+ Preconditions.checkState(
+ expected.sameLayout(actual),
+ "FM index manifest metadata does not match the container directory.");
+ validated = true;
+ }
+ }
+
static final class FileSetRowCountValidator {
private final long expectedTotalRowCount;
private final FileRowRange[] ranges;
diff --git a/paimon-common/src/main/java/org/apache/paimon/globalindex/fmindex/FMGlobalIndexWriter.java b/paimon-common/src/main/java/org/apache/paimon/globalindex/fmindex/FMGlobalIndexWriter.java
index d0f19c1595..01b5ca8449 100644
--- a/paimon-common/src/main/java/org/apache/paimon/globalindex/fmindex/FMGlobalIndexWriter.java
+++ b/paimon-common/src/main/java/org/apache/paimon/globalindex/fmindex/FMGlobalIndexWriter.java
@@ -34,6 +34,7 @@ import java.io.DataOutputStream;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collections;
import java.util.List;
/** Streaming, bounded-partition writer for an exact byte-oriented FM index. */
@@ -47,10 +48,15 @@ public class FMGlobalIndexWriter implements GlobalIndexSingleColumnWriter, Close
private final int maxPartitionRowCount;
private final int sampleRate;
@Nullable private final BlockCompressionFactory compressionFactory;
- private final List<ResultEntry> results = new ArrayList<>();
+ private final List<FMIndexFile.PartitionMeta> partitions = new ArrayList<>();
private CharBuilder text = new CharBuilder();
private boolean[] nullRows = new boolean[128];
+ @Nullable private String fileName;
+ @Nullable private PositionOutputStream stream;
+ @Nullable private DataOutputStream output;
+ private long firstRowId;
+ private long totalRowCount;
private long partitionFirstRowId;
private int partitionRowCount;
private long lastRowId;
@@ -105,6 +111,9 @@ public class FMGlobalIndexWriter implements GlobalIndexSingleColumnWriter, Close
flushPartition();
}
+ if (!hasLastRowId) {
+ firstRowId = relativeRowId;
+ }
if (partitionRowCount == 0) {
partitionFirstRowId = relativeRowId;
}
@@ -117,6 +126,7 @@ public class FMGlobalIndexWriter implements GlobalIndexSingleColumnWriter, Close
}
text.add(FMIndexFile.SEPARATOR);
partitionRowCount++;
+ totalRowCount = Math.addExact(totalRowCount, 1L);
lastRowId = relativeRowId;
hasLastRowId = true;
}
@@ -125,13 +135,35 @@ public class FMGlobalIndexWriter implements GlobalIndexSingleColumnWriter, Close
public List<ResultEntry> finish() {
Preconditions.checkState(!finished, "FM index writer is already finished.");
finished = true;
- flushPartition();
- return new ArrayList<>(results);
+ try {
+ flushPartition();
+ if (partitions.isEmpty()) {
+ return Collections.emptyList();
+ }
+ Preconditions.checkState(
+ stream != null && output != null && fileName != null,
+ "FM index container output is missing.");
+ byte[] indexMeta = FMIndexFile.writeIndexMeta(firstRowId, totalRowCount, partitions);
+ FMIndexFile.BlockInfo directory =
+ FMIndexFile.writeContainerDirectory(
+ stream, output, indexMeta, compressionFactory);
+ FMIndexFile.writeContainerFooter(
+ output, directory, firstRowId, totalRowCount, partitions.size());
+ closeOutput();
+ return Collections.singletonList(new ResultEntry(fileName, totalRowCount, indexMeta));
+ } catch (IOException e) {
+ closeOutputQuietly();
+ throw new RuntimeException("Failed to finish FM global index container.", e);
+ } catch (RuntimeException e) {
+ closeOutputQuietly();
+ throw e;
+ }
}
@Override
public void close() {
finished = true;
+ closeOutputQuietly();
text = new CharBuilder();
partitionRowCount = 0;
}
@@ -141,6 +173,10 @@ public class FMGlobalIndexWriter implements GlobalIndexSingleColumnWriter, Close
return;
}
try {
+ ensureOutput();
+ Preconditions.checkState(
+ stream != null && output != null, "FM index container output is missing.");
+ long partitionStart = stream.getPos();
text.add(FMIndexFile.TERMINATOR);
char[] symbols = text.toArray();
text = new CharBuilder();
@@ -168,96 +204,92 @@ public class FMGlobalIndexWriter implements GlobalIndexSingleColumnWriter, Close
int levelCount = FMIndexFile.levelsForAlphabet(alphabet.alphabetSize);
int[][] digitStarts = new int[levelCount][4];
FMIndexFile.QuadVectorMeta[] wavelets = new FMIndexFile.QuadVectorMeta[levelCount];
- String fileName = fileWriter.newFileName("fmindex");
- try (PositionOutputStream stream = fileWriter.newOutputStream(fileName)) {
- DataOutputStream output = new DataOutputStream(stream);
- short[] current = bwt;
- short[] reordered = new short[bwt.length];
- for (int level = 0; level < levelCount; level++) {
- int shift = (levelCount - level - 1) * 2;
- long[] quads = new long[FMIndexFile.wordsForQuads(current.length)];
- int[] counts = new int[4];
- for (int i = 0; i < current.length; i++) {
- int digit = ((current[i] & 0xFFFF) >>> shift) & 3;
- counts[digit]++;
- quads[i >>> 5] |= (long) digit << ((i & 31) * 2);
- }
- int next = 0;
- for (int digit = 0; digit < 4; digit++) {
- digitStarts[level][digit] = next;
- next += counts[digit];
- }
- int[] positions = java.util.Arrays.copyOf(digitStarts[level], 4);
- for (short encoded : current) {
- int symbol = encoded & 0xFFFF;
- int digit = (symbol >>> shift) & 3;
- reordered[positions[digit]++] = encoded;
- }
- wavelets[level] =
- FMIndexFile.writeQuadVector(
- stream, output, quads, current.length, compressionFactory);
- short[] swap = current;
- current = reordered;
- reordered = swap;
+ short[] current = bwt;
+ short[] reordered = new short[bwt.length];
+ for (int level = 0; level < levelCount; level++) {
+ int shift = (levelCount - level - 1) * 2;
+ long[] quads = new long[FMIndexFile.wordsForQuads(current.length)];
+ int[] counts = new int[4];
+ for (int i = 0; i < current.length; i++) {
+ int digit = ((current[i] & 0xFFFF) >>> shift) & 3;
+ counts[digit]++;
+ quads[i >>> 5] |= (long) digit << ((i & 31) * 2);
}
+ int next = 0;
+ for (int digit = 0; digit < 4; digit++) {
+ digitStarts[level][digit] = next;
+ next += counts[digit];
+ }
+ int[] positions = java.util.Arrays.copyOf(digitStarts[level], 4);
+ for (short encoded : current) {
+ int symbol = encoded & 0xFFFF;
+ int digit = (symbol >>> shift) & 3;
+ reordered[positions[digit]++] = encoded;
+ }
+ wavelets[level] =
+ FMIndexFile.writeQuadVector(
+ stream, output, quads, current.length, compressionFactory);
+ short[] swap = current;
+ current = reordered;
+ reordered = swap;
+ }
- FMIndexFile.BitVectorMeta sampled =
- FMIndexFile.writeBitVector(
- stream, output, sampledWords, symbols.length, compressionFactory);
- FMIndexFile.IntVectorMeta samples =
- FMIndexFile.writeIntVector(
- stream, output, sampleValues, compressionFactory);
- long[] nullWords = new long[wordsForBits(partitionRowCount)];
- for (int i = 0; i < partitionRowCount; i++) {
- if (nullRows[i]) {
- nullWords[i >>> 6] |= 1L << (i & 63);
- }
+ FMIndexFile.BitVectorMeta sampled =
+ FMIndexFile.writeBitVector(
+ stream, output, sampledWords, symbols.length, compressionFactory);
+ FMIndexFile.IntVectorMeta samples =
+ FMIndexFile.writeIntVector(stream, output, sampleValues, compressionFactory);
+ long[] nullWords = new long[wordsForBits(partitionRowCount)];
+ for (int i = 0; i < partitionRowCount; i++) {
+ if (nullRows[i]) {
+ nullWords[i >>> 6] |= 1L << (i & 63);
}
- FMIndexFile.BitVectorMeta nullVector =
- FMIndexFile.writeBitVector(
- stream, output, nullWords, partitionRowCount, compressionFactory);
- long[] boundaryWords = new long[wordsForBits(symbols.length)];
- for (int i = 0; i < symbols.length; i++) {
- if (symbols[i] == FMIndexFile.SEPARATOR) {
- boundaryWords[i >>> 6] |= 1L << (i & 63);
- }
+ }
+ FMIndexFile.BitVectorMeta nullVector =
+ FMIndexFile.writeBitVector(
+ stream, output, nullWords, partitionRowCount, compressionFactory);
+ long[] boundaryWords = new long[wordsForBits(symbols.length)];
+ for (int i = 0; i < symbols.length; i++) {
+ if (symbols[i] == FMIndexFile.SEPARATOR) {
+ boundaryWords[i >>> 6] |= 1L << (i & 63);
}
- FMIndexFile.BitVectorMeta rowBoundaries =
- FMIndexFile.writeBitVector(
- stream, output, boundaryWords, symbols.length, compressionFactory);
- List<FMIndexFile.VerificationPageMeta> verificationPages =
- writeVerificationPages(stream, output, symbols, alphabet.symbolToByte);
- FMIndexFile.Directory directory =
- new FMIndexFile.Directory(
- partitionRowCount,
- symbols.length,
- sampleRate,
- levelCount,
- alphabet.alphabetSize,
- alphabet.byteToSymbol,
- cumulative,
- digitStarts,
- wavelets,
- sampled,
- samples,
- nullVector,
- rowBoundaries,
- verificationPages);
- FMIndexFile.BlockInfo directoryBlock =
- FMIndexFile.writeDirectory(stream, output, directory, compressionFactory);
- FMIndexFile.writeFooter(
- output,
- directoryBlock,
- partitionFirstRowId,
- partitionRowCount,
- symbols.length,
- sampleRate);
}
- results.add(
- new ResultEntry(
- fileName,
+ FMIndexFile.BitVectorMeta rowBoundaries =
+ FMIndexFile.writeBitVector(
+ stream, output, boundaryWords, symbols.length, compressionFactory);
+ List<FMIndexFile.VerificationPageMeta> verificationPages =
+ writeVerificationPages(stream, output, symbols, alphabet.symbolToByte);
+ FMIndexFile.Directory directory =
+ new FMIndexFile.Directory(
partitionRowCount,
- FMIndexFile.writeIndexMeta(partitionFirstRowId, partitionRowCount)));
+ symbols.length,
+ sampleRate,
+ levelCount,
+ alphabet.alphabetSize,
+ alphabet.byteToSymbol,
+ cumulative,
+ digitStarts,
+ wavelets,
+ sampled,
+ samples,
+ nullVector,
+ rowBoundaries,
+ verificationPages);
+ FMIndexFile.BlockInfo directoryBlock =
+ FMIndexFile.writeDirectory(stream, output, directory, compressionFactory);
+ FMIndexFile.writeFooter(
+ output,
+ directoryBlock,
+ partitionFirstRowId,
+ partitionRowCount,
+ symbols.length,
+ sampleRate);
+ partitions.add(
+ new FMIndexFile.PartitionMeta(
+ partitionStart,
+ stream.getPos(),
+ partitionFirstRowId,
+ partitionRowCount));
} catch (IOException e) {
throw new RuntimeException("Failed to write FM global index.", e);
} finally {
@@ -267,6 +299,32 @@ public class FMGlobalIndexWriter implements GlobalIndexSingleColumnWriter, Close
}
}
+ private void ensureOutput() throws IOException {
+ if (output != null) {
+ return;
+ }
+ fileName = fileWriter.newFileName("fmindex");
+ stream = fileWriter.newOutputStream(fileName);
+ output = new DataOutputStream(stream);
+ }
+
+ private void closeOutput() throws IOException {
+ if (output != null) {
+ DataOutputStream current = output;
+ output = null;
+ stream = null;
+ current.close();
+ }
+ }
+
+ private void closeOutputQuietly() {
+ try {
+ closeOutput();
+ } catch (IOException ignored) {
+ // Best effort; the build owner deletes unpublished output on failure.
+ }
+ }
+
private static int[] cumulativeCounts(char[] symbols, int alphabetSize) {
int[] counts = new int[alphabetSize + 1];
for (char symbol : symbols) {
diff --git a/paimon-common/src/main/java/org/apache/paimon/globalindex/fmindex/FMGlobalIndexer.java b/paimon-common/src/main/java/org/apache/paimon/globalindex/fmindex/FMGlobalIndexer.java
index 1a1c1cac47..0ff9582088 100644
--- a/paimon-common/src/main/java/org/apache/paimon/globalindex/fmindex/FMGlobalIndexer.java
+++ b/paimon-common/src/main/java/org/apache/paimon/globalindex/fmindex/FMGlobalIndexer.java
@@ -117,36 +117,32 @@ public class FMGlobalIndexer implements GlobalIndexer {
FMGlobalIndexReader.FileSetRowCountValidator validator =
new FMGlobalIndexReader.FileSetRowCountValidator(files.size(), totalRowCount);
FMIndexFile.IndexMeta[] indexMetas = new FMIndexFile.IndexMeta[files.size()];
- boolean allIndexMetasPresent = true;
for (int i = 0; i < files.size(); i++) {
byte[] metadata = files.get(i).metadata();
- if (metadata == null || metadata.length == 0) {
- allIndexMetasPresent = false;
- break;
- }
+ checkArgument(
+ metadata != null && metadata.length > 0,
+ "FM index container metadata is missing for %s.",
+ files.get(i).filePath());
indexMetas[i] = FMIndexFile.readIndexMeta(metadata);
+ validator.validate(
+ i, indexMetas[i].rowCount, indexMetas[i].firstRowId, indexMetas[i].lastRowId());
}
- if (allIndexMetasPresent) {
- for (int i = 0; i < indexMetas.length; i++) {
- FMIndexFile.IndexMeta metadata = indexMetas[i];
- validator.validate(i, metadata.rowCount, metadata.firstRowId, metadata.lastRowId());
- }
- } else {
- java.util.Arrays.fill(indexMetas, null);
- }
- List<GlobalIndexReader> readers = new ArrayList<>(files.size());
+ List<GlobalIndexReader> readers = new ArrayList<>();
for (int i = 0; i < files.size(); i++) {
- readers.add(
- new FMGlobalIndexReader(
- fileReader,
- files.get(i),
- executor,
- readContext,
- validator,
- indexMetas[i],
- i,
- demandPageSize,
- locateCostRatio));
+ FMGlobalIndexReader.ContainerMetadataLoader container =
+ new FMGlobalIndexReader.ContainerMetadataLoader(files.get(i), indexMetas[i]);
+ for (FMIndexFile.PartitionMeta partition : indexMetas[i].partitions) {
+ readers.add(
+ new FMGlobalIndexReader(
+ fileReader,
+ files.get(i),
+ executor,
+ readContext,
+ container,
+ partition,
+ demandPageSize,
+ locateCostRatio));
+ }
}
return readers.size() == 1 ? readers.get(0) : new UnionGlobalIndexReader(readers);
}
diff --git a/paimon-common/src/main/java/org/apache/paimon/globalindex/fmindex/FMIndexFile.java b/paimon-common/src/main/java/org/apache/paimon/globalindex/fmindex/FMIndexFile.java
index e52695068a..d02b68dc21 100644
--- a/paimon-common/src/main/java/org/apache/paimon/globalindex/fmindex/FMIndexFile.java
+++ b/paimon-common/src/main/java/org/apache/paimon/globalindex/fmindex/FMIndexFile.java
@@ -38,18 +38,21 @@ import java.io.DataOutputStream;
import java.io.EOFException;
import java.io.IOException;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
import static org.apache.paimon.sst.SstFileUtils.crc32c;
/**
- * Portable V1 layout for partitioned, demand-paged FM indexes.
+ * Portable V1 container layout for partitioned, demand-paged FM indexes.
*
- * <p>The payload is canonical and contiguous: dense-alphabet blocked quaternary wavelet levels,
- * sampled-SA mask and values, null mask, row-boundary mask, exact-verification value pages,
- * directory, then a fixed footer. Every independently readable block records its offset, stored and
- * uncompressed lengths, compression ID and CRC32C. The reader validates all physical ranges before
- * allocating decoded buffers and verifies the stored checksum before decompression.
+ * <p>Each physical index file contains one or more canonical, contiguous partitions followed by a
+ * checksummed container directory and fixed footer. A partition contains dense-alphabet blocked
+ * quaternary wavelet levels, sampled-SA mask and values, null mask, row-boundary mask,
+ * exact-verification value pages, its directory, and a fixed footer. Every independently readable
+ * block records its offset, stored and uncompressed lengths, compression ID and CRC32C. The reader
+ * validates all physical ranges before allocating decoded buffers and verifies the stored checksum
+ * before decompression.
*/
final class FMIndexFile {
@@ -58,12 +61,14 @@ final class FMIndexFile {
static final int FIRST_BYTE_SYMBOL = 2;
static final int MAX_ALPHABET_SIZE = 258;
- private static final int MAGIC = 0x464D4958;
+ private static final int PARTITION_MAGIC = 0x464D4950;
+ private static final int CONTAINER_MAGIC = 0x464D4958;
private static final int VERSION = 1;
private static final int INDEX_META_MAGIC = 0x464D4D45;
private static final int INDEX_META_VERSION = 1;
- private static final int INDEX_META_LENGTH = 24;
- private static final int INDEX_META_CHECKSUM_OFFSET = 20;
+ private static final int INDEX_META_HEADER_LENGTH = 28;
+ private static final int INDEX_META_PARTITION_LENGTH = 28;
+ private static final int INDEX_META_CHECKSUM_LENGTH = Integer.BYTES;
private static final int FEATURE_VALUE_SAMPLED_SA = 1;
private static final int FEATURE_DENSE_QUAD_WAVELET = 1 << 1;
private static final int FEATURE_SEPARATOR_ROW_IDS = 1 << 2;
@@ -80,7 +85,8 @@ final class FMIndexFile {
static final int QUAD_BLOCK_VALUES = BLOCK_WORDS * QUAD_VALUES_PER_WORD;
static final int VALUE_BLOCK_INTS = 8192;
static final int BLOCK_INFO_LENGTH = 24;
- static final int FOOTER_LENGTH = 64;
+ static final int PARTITION_FOOTER_LENGTH = 64;
+ static final int CONTAINER_FOOTER_LENGTH = 64;
static final int FOOTER_CHECKSUM_OFFSET = 60;
static final int MAX_DIRECTORY_UNCOMPRESSED_LENGTH = 16 * 1024 * 1024;
static final int MAX_DATA_BLOCK_UNCOMPRESSED_LENGTH = 64 * 1024;
@@ -88,39 +94,99 @@ final class FMIndexFile {
private FMIndexFile() {}
- static byte[] writeIndexMeta(long firstRowId, int rowCount) {
+ static byte[] writeIndexMeta(long firstRowId, long rowCount, List<PartitionMeta> partitions) {
Preconditions.checkArgument(firstRowId >= 0 && rowCount > 0, "Invalid FM index row range.");
Preconditions.checkArgument(
firstRowId <= Long.MAX_VALUE - rowCount,
"FM index row range overflows the supported row ID space.");
- byte[] bytes = new byte[INDEX_META_LENGTH];
+ Preconditions.checkArgument(!partitions.isEmpty(), "FM index must contain partitions.");
+ long encodedLength =
+ INDEX_META_HEADER_LENGTH
+ + (long) partitions.size() * INDEX_META_PARTITION_LENGTH
+ + INDEX_META_CHECKSUM_LENGTH;
+ Preconditions.checkArgument(
+ encodedLength <= MAX_DIRECTORY_UNCOMPRESSED_LENGTH,
+ "FM index partition directory exceeds the supported size.");
+ byte[] bytes = new byte[(int) encodedLength];
writeInt(bytes, 0, INDEX_META_MAGIC);
writeInt(bytes, 4, INDEX_META_VERSION);
writeLong(bytes, 8, firstRowId);
- writeInt(bytes, 16, rowCount);
- writeInt(bytes, INDEX_META_CHECKSUM_OFFSET, indexMetaChecksum(bytes));
+ writeLong(bytes, 16, rowCount);
+ writeInt(bytes, 24, partitions.size());
+ int offset = INDEX_META_HEADER_LENGTH;
+ for (PartitionMeta partition : partitions) {
+ writeLong(bytes, offset, partition.startOffset);
+ writeLong(bytes, offset + 8, partition.endOffset);
+ writeLong(bytes, offset + 16, partition.firstRowId);
+ writeInt(bytes, offset + 24, partition.rowCount);
+ offset += INDEX_META_PARTITION_LENGTH;
+ }
+ writeInt(bytes, offset, indexMetaChecksum(bytes));
+ // Validate writer-produced metadata through the same canonical parser used by readers.
+ readIndexMeta(bytes);
return bytes;
}
static IndexMeta readIndexMeta(byte[] bytes) {
Preconditions.checkState(
- bytes.length == INDEX_META_LENGTH, "Invalid FM index manifest metadata length.");
+ bytes.length >= INDEX_META_HEADER_LENGTH + INDEX_META_CHECKSUM_LENGTH
+ && bytes.length <= MAX_DIRECTORY_UNCOMPRESSED_LENGTH,
+ "Invalid FM index manifest metadata length.");
Preconditions.checkState(
readInt(bytes, 0) == INDEX_META_MAGIC, "Invalid FM index manifest metadata magic.");
Preconditions.checkState(
readInt(bytes, 4) == INDEX_META_VERSION,
"Unsupported FM index manifest metadata version: %s.",
readInt(bytes, 4));
+ int partitionCount = readInt(bytes, 24);
+ Preconditions.checkState(partitionCount > 0, "FM index must contain partitions.");
+ long expectedLength =
+ INDEX_META_HEADER_LENGTH
+ + (long) partitionCount * INDEX_META_PARTITION_LENGTH
+ + INDEX_META_CHECKSUM_LENGTH;
+ Preconditions.checkState(
+ expectedLength == bytes.length, "Invalid FM index manifest metadata length.");
Preconditions.checkState(
- readInt(bytes, INDEX_META_CHECKSUM_OFFSET) == indexMetaChecksum(bytes),
+ readInt(bytes, bytes.length - INDEX_META_CHECKSUM_LENGTH)
+ == indexMetaChecksum(bytes),
"FM index manifest metadata checksum mismatch.");
long firstRowId = readLong(bytes, 8);
- int rowCount = readInt(bytes, 16);
+ long rowCount = readLong(bytes, 16);
Preconditions.checkState(firstRowId >= 0 && rowCount > 0, "Invalid FM index row range.");
Preconditions.checkState(
firstRowId <= Long.MAX_VALUE - rowCount,
"FM index row range overflows the supported row ID space.");
- return new IndexMeta(firstRowId, rowCount);
+ List<PartitionMeta> partitions = new ArrayList<>(partitionCount);
+ long expectedOffset = 0;
+ long expectedRowId = firstRowId;
+ int offset = INDEX_META_HEADER_LENGTH;
+ for (int i = 0; i < partitionCount; i++) {
+ long startOffset = readLong(bytes, offset);
+ long endOffset = readLong(bytes, offset + 8);
+ long partitionFirstRowId = readLong(bytes, offset + 16);
+ int partitionRowCount = readInt(bytes, offset + 24);
+ Preconditions.checkState(
+ startOffset == expectedOffset
+ && endOffset > startOffset
+ && endOffset - startOffset >= PARTITION_FOOTER_LENGTH,
+ "FM index partitions are not canonical and contiguous.");
+ Preconditions.checkState(
+ partitionFirstRowId == expectedRowId && partitionRowCount > 0,
+ "FM index partition row ranges are not canonical and contiguous.");
+ Preconditions.checkState(
+ expectedRowId <= Long.MAX_VALUE - partitionRowCount,
+ "FM index partition row range overflows the supported row ID space.");
+ partitions.add(
+ new PartitionMeta(
+ startOffset, endOffset, partitionFirstRowId, partitionRowCount));
+ expectedOffset = endOffset;
+ expectedRowId += partitionRowCount;
+ offset += INDEX_META_PARTITION_LENGTH;
+ }
+ Preconditions.checkState(
+ expectedRowId == firstRowId + rowCount,
+ "FM index partition row counts do not match the file row count.");
+ return new IndexMeta(firstRowId, rowCount, partitions);
}
static BlockInfo writeBlock(
@@ -299,7 +365,7 @@ final class FMIndexFile {
int textLength,
int sampleRate)
throws IOException {
- byte[] bytes = new byte[FOOTER_LENGTH];
+ byte[] bytes = new byte[PARTITION_FOOTER_LENGTH];
writeBlockInfo(bytes, 0, directory);
writeLong(bytes, 24, firstRowId);
writeInt(bytes, 32, rowCount);
@@ -307,34 +373,136 @@ final class FMIndexFile {
writeInt(bytes, 40, sampleRate);
writeInt(bytes, 44, FEATURE_FLAGS);
writeInt(bytes, 52, VERSION);
- writeInt(bytes, 56, MAGIC);
+ writeInt(bytes, 56, PARTITION_MAGIC);
writeInt(bytes, FOOTER_CHECKSUM_OFFSET, footerChecksum(bytes));
out.write(bytes);
out.flush();
}
- static Footer readFooter(SeekableInputStream input, long fileSize) throws IOException {
+ static BlockInfo writeContainerDirectory(
+ PositionOutputStream stream,
+ DataOutputStream out,
+ byte[] indexMeta,
+ @Nullable BlockCompressionFactory compressionFactory)
+ throws IOException {
+ return writeBlock(stream, out, indexMeta, compressionFactory);
+ }
+
+ static void writeContainerFooter(
+ DataOutputStream out,
+ BlockInfo directory,
+ long firstRowId,
+ long rowCount,
+ int partitionCount)
+ throws IOException {
+ byte[] bytes = new byte[CONTAINER_FOOTER_LENGTH];
+ writeBlockInfo(bytes, 0, directory);
+ writeLong(bytes, 24, firstRowId);
+ writeLong(bytes, 32, rowCount);
+ writeInt(bytes, 40, partitionCount);
+ writeInt(bytes, 44, FEATURE_FLAGS);
+ writeInt(bytes, 52, VERSION);
+ writeInt(bytes, 56, CONTAINER_MAGIC);
+ writeInt(bytes, FOOTER_CHECKSUM_OFFSET, footerChecksum(bytes));
+ out.write(bytes);
+ out.flush();
+ }
+
+ static ContainerFooter readContainerFooter(SeekableInputStream input, long fileSize)
+ throws IOException {
Preconditions.checkState(
- fileSize >= FOOTER_LENGTH, "Invalid FM index file size: %s.", fileSize);
- byte[] bytes = readAt(input, fileSize - FOOTER_LENGTH, FOOTER_LENGTH);
+ fileSize >= CONTAINER_FOOTER_LENGTH,
+ "Invalid FM index container size: %s.",
+ fileSize);
+ byte[] bytes = readAt(input, fileSize - CONTAINER_FOOTER_LENGTH, CONTAINER_FOOTER_LENGTH);
Preconditions.checkState(
- readInt(bytes, 56) == MAGIC, "File is not an FM index (bad footer magic).");
+ readInt(bytes, 56) == CONTAINER_MAGIC,
+ "File is not an FM index container (bad footer magic).");
Preconditions.checkState(
readInt(bytes, 52) == VERSION,
- "Unsupported FM index version: %s.",
+ "Unsupported FM index container version: %s.",
readInt(bytes, 52));
int expectedChecksum = readInt(bytes, FOOTER_CHECKSUM_OFFSET);
int actualChecksum = footerChecksum(bytes);
Preconditions.checkState(
expectedChecksum == actualChecksum,
- "FM index footer checksum mismatch: expected=%s, actual=%s.",
+ "FM index container footer checksum mismatch: expected=%s, actual=%s.",
expectedChecksum,
actualChecksum);
Preconditions.checkState(
readInt(bytes, 44) == FEATURE_FLAGS,
- "Unsupported FM index feature flags: %s.",
+ "Unsupported FM index container feature flags: %s.",
readInt(bytes, 44));
- Preconditions.checkState(readInt(bytes, 48) == 0, "Invalid FM index reserved field.");
+ Preconditions.checkState(
+ readInt(bytes, 48) == 0, "Invalid FM index container reserved field.");
+
+ DataInputStream data = new DataInputStream(new ByteArrayInputStream(bytes));
+ BlockInfo directory = readBlockInfo(data);
+ long firstRowId = data.readLong();
+ long rowCount = data.readLong();
+ int partitionCount = data.readInt();
+ Preconditions.checkState(firstRowId >= 0 && rowCount > 0, "Invalid FM index row range.");
+ Preconditions.checkState(
+ firstRowId <= Long.MAX_VALUE - rowCount,
+ "FM index row range overflows the supported row ID space.");
+ Preconditions.checkState(
+ partitionCount > 0 && partitionCount <= rowCount,
+ "Invalid FM index partition count.");
+ validateBlock(
+ directory,
+ fileSize - CONTAINER_FOOTER_LENGTH,
+ MAX_DIRECTORY_UNCOMPRESSED_LENGTH,
+ false);
+ Preconditions.checkState(
+ directory.offset + directory.storedLength == fileSize - CONTAINER_FOOTER_LENGTH,
+ "FM index container directory is not immediately before the footer.");
+ return new ContainerFooter(directory, firstRowId, rowCount, partitionCount);
+ }
+
+ static IndexMeta readContainerDirectory(
+ SeekableInputStream input, ContainerFooter footer, long fileSize) throws IOException {
+ IndexMeta metadata = readIndexMeta(readBlock(input, footer.directory, fileSize));
+ Preconditions.checkState(
+ metadata.firstRowId == footer.firstRowId
+ && metadata.rowCount == footer.rowCount
+ && metadata.partitions.size() == footer.partitionCount,
+ "FM index container footer and directory metadata do not match.");
+ Preconditions.checkState(
+ metadata.partitions.get(metadata.partitions.size() - 1).endOffset
+ == footer.directory.offset,
+ "FM index partitions do not exactly cover the container payload.");
+ return metadata;
+ }
+
+ static Footer readFooter(SeekableInputStream input, PartitionMeta partition, long fileSize)
+ throws IOException {
+ Preconditions.checkState(
+ partition.startOffset >= 0
+ && partition.endOffset <= fileSize
+ && partition.endOffset - partition.startOffset >= PARTITION_FOOTER_LENGTH,
+ "Invalid FM index partition range.");
+ long footerOffset = partition.endOffset - PARTITION_FOOTER_LENGTH;
+ byte[] bytes = readAt(input, footerOffset, PARTITION_FOOTER_LENGTH);
+ Preconditions.checkState(
+ readInt(bytes, 56) == PARTITION_MAGIC,
+ "File is not an FM index partition (bad footer magic).");
+ Preconditions.checkState(
+ readInt(bytes, 52) == VERSION,
+ "Unsupported FM index partition version: %s.",
+ readInt(bytes, 52));
+ int expectedChecksum = readInt(bytes, FOOTER_CHECKSUM_OFFSET);
+ int actualChecksum = footerChecksum(bytes);
+ Preconditions.checkState(
+ expectedChecksum == actualChecksum,
+ "FM index partition footer checksum mismatch: expected=%s, actual=%s.",
+ expectedChecksum,
+ actualChecksum);
+ Preconditions.checkState(
+ readInt(bytes, 44) == FEATURE_FLAGS,
+ "Unsupported FM index partition feature flags: %s.",
+ readInt(bytes, 44));
+ Preconditions.checkState(
+ readInt(bytes, 48) == 0, "Invalid FM index partition reserved field.");
DataInputStream data = new DataInputStream(new ByteArrayInputStream(bytes));
BlockInfo directory = readBlockInfo(data);
@@ -349,17 +517,38 @@ final class FMIndexFile {
Preconditions.checkState(
textLength >= rowCount + 1L, "Invalid FM index encoded text length.");
validateSampleRate(sampleRate);
- validateBlock(
- directory, fileSize - FOOTER_LENGTH, MAX_DIRECTORY_UNCOMPRESSED_LENGTH, false);
+ validateBlock(directory, footerOffset, MAX_DIRECTORY_UNCOMPRESSED_LENGTH, false);
+ Preconditions.checkState(
+ directory.offset >= partition.startOffset
+ && directory.offset + directory.storedLength == footerOffset,
+ "FM index partition directory is not immediately before its footer.");
Preconditions.checkState(
- directory.offset + directory.storedLength == fileSize - FOOTER_LENGTH,
- "FM index directory is not immediately before the footer.");
- return new Footer(directory, firstRowId, rowCount, textLength, sampleRate);
+ firstRowId == partition.firstRowId && rowCount == partition.rowCount,
+ "FM index partition footer and container directory metadata do not match.");
+ return new Footer(
+ directory,
+ firstRowId,
+ rowCount,
+ textLength,
+ sampleRate,
+ partition.startOffset,
+ partition.endOffset);
+ }
+
+ static Footer readFooter(SeekableInputStream input, long fileSize) throws IOException {
+ ContainerFooter containerFooter = readContainerFooter(input, fileSize);
+ IndexMeta metadata = readContainerDirectory(input, containerFooter, fileSize);
+ return readFooter(input, metadata.partitions.get(0), fileSize);
}
static Directory readDirectory(SeekableInputStream input, Footer footer, long fileSize)
throws IOException {
- byte[] bytes = readBlock(input, footer.directory, fileSize);
+ Preconditions.checkState(
+ footer.partitionStartOffset >= 0
+ && footer.partitionEndOffset <= fileSize
+ && footer.partitionStartOffset < footer.partitionEndOffset,
+ "Invalid FM index partition range.");
+ byte[] bytes = readBlock(input, footer.directory, footer.partitionEndOffset);
DataInputStream data = new DataInputStream(new ByteArrayInputStream(bytes));
int rowCount = data.readInt();
int textLength = data.readInt();
@@ -420,7 +609,7 @@ final class FMIndexFile {
}
}
- long[] expectedOffset = {0L};
+ long[] expectedOffset = {footer.partitionStartOffset};
int[][] digitStarts = new int[levelCount][4];
QuadVectorMeta[] wavelets = new QuadVectorMeta[levelCount];
for (int i = 0; i < wavelets.length; i++) {
@@ -1027,7 +1216,10 @@ final class FMIndexFile {
private static int indexMetaChecksum(byte[] metadata) {
return crc32c(
- new MemorySlice(MemorySegment.wrap(metadata), 0, INDEX_META_CHECKSUM_OFFSET),
+ new MemorySlice(
+ MemorySegment.wrap(metadata),
+ 0,
+ metadata.length - INDEX_META_CHECKSUM_LENGTH),
BlockCompressionType.NONE);
}
@@ -1343,21 +1535,51 @@ final class FMIndexFile {
final int rowCount;
final int textLength;
final int sampleRate;
+ final long partitionStartOffset;
+ final long partitionEndOffset;
- Footer(BlockInfo directory, long firstRowId, int rowCount, int textLength, int sampleRate) {
+ Footer(
+ BlockInfo directory,
+ long firstRowId,
+ int rowCount,
+ int textLength,
+ int sampleRate,
+ long partitionStartOffset,
+ long partitionEndOffset) {
this.directory = directory;
this.firstRowId = firstRowId;
this.rowCount = rowCount;
this.textLength = textLength;
this.sampleRate = sampleRate;
+ this.partitionStartOffset = partitionStartOffset;
+ this.partitionEndOffset = partitionEndOffset;
}
}
- static final class IndexMeta {
+ static final class ContainerFooter {
+ final BlockInfo directory;
+ final long firstRowId;
+ final long rowCount;
+ final int partitionCount;
+
+ private ContainerFooter(
+ BlockInfo directory, long firstRowId, long rowCount, int partitionCount) {
+ this.directory = directory;
+ this.firstRowId = firstRowId;
+ this.rowCount = rowCount;
+ this.partitionCount = partitionCount;
+ }
+ }
+
+ static final class PartitionMeta {
+ final long startOffset;
+ final long endOffset;
final long firstRowId;
final int rowCount;
- private IndexMeta(long firstRowId, int rowCount) {
+ PartitionMeta(long startOffset, long endOffset, long firstRowId, int rowCount) {
+ this.startOffset = startOffset;
+ this.endOffset = endOffset;
this.firstRowId = firstRowId;
this.rowCount = rowCount;
}
@@ -1367,6 +1589,41 @@ final class FMIndexFile {
}
}
+ static final class IndexMeta {
+ final long firstRowId;
+ final long rowCount;
+ final List<PartitionMeta> partitions;
+
+ private IndexMeta(long firstRowId, long rowCount, List<PartitionMeta> partitions) {
+ this.firstRowId = firstRowId;
+ this.rowCount = rowCount;
+ this.partitions = Collections.unmodifiableList(new ArrayList<>(partitions));
+ }
+
+ long lastRowId() {
+ return firstRowId + rowCount - 1L;
+ }
+
+ boolean sameLayout(IndexMeta that) {
+ if (firstRowId != that.firstRowId
+ || rowCount != that.rowCount
+ || partitions.size() != that.partitions.size()) {
+ return false;
+ }
+ for (int i = 0; i < partitions.size(); i++) {
+ PartitionMeta left = partitions.get(i);
+ PartitionMeta right = that.partitions.get(i);
+ if (left.startOffset != right.startOffset
+ || left.endOffset != right.endOffset
+ || left.firstRowId != right.firstRowId
+ || left.rowCount != right.rowCount) {
+ return false;
+ }
+ }
+ return true;
+ }
+ }
+
static final class QuadBlock {
private final long[] words;
private final int[] prefixes;
diff --git a/paimon-common/src/test/java/org/apache/paimon/globalindex/fmindex/FMGlobalIndexTest.java b/paimon-common/src/test/java/org/apache/paimon/globalindex/fmindex/FMGlobalIndexTest.java
index 9c4daa2e1a..de8b576df9 100644
--- a/paimon-common/src/test/java/org/apache/paimon/globalindex/fmindex/FMGlobalIndexTest.java
+++ b/paimon-common/src/test/java/org/apache/paimon/globalindex/fmindex/FMGlobalIndexTest.java
@@ -172,7 +172,10 @@ public class FMGlobalIndexTest {
null,
str("needle-4")),
0);
- assertThat(files).hasSize(3);
+ assertThat(files).hasSize(1);
+ FMIndexFile.IndexMeta metadata = FMIndexFile.readIndexMeta(files.get(0).metadata());
+ assertThat(metadata.rowCount).isEqualTo(5L);
+ assertThat(metadata.partitions).hasSize(3);
try (GlobalIndexReader reader = createReader(files, 5)) {
assertRows(reader.visitContains(fieldRef, str("needle")).join(), 0L, 2L, 4L);
}
@@ -182,7 +185,7 @@ public class FMGlobalIndexTest {
public void testFooterAndRankBlockCorruptionFailClosed() throws Exception {
List<GlobalIndexIOMeta> files = writeData(Collections.singletonList(str("abcdef")), 0);
GlobalIndexIOMeta file = files.get(0);
- corruptByte(file, file.fileSize() - FMIndexFile.FOOTER_LENGTH + 8);
+ corruptByte(file, file.fileSize() - FMIndexFile.CONTAINER_FOOTER_LENGTH + 8);
try (GlobalIndexReader reader = createReader(files, 1)) {
assertThatThrownBy(() -> reader.visitContains(fieldRef, str("abc")).join())
.isInstanceOf(CompletionException.class)
@@ -334,7 +337,10 @@ public class FMGlobalIndexTest {
java.nio.file.Path fixturePath = tempPath.resolve("fmindex-v1-golden.index");
Files.write(fixturePath, fixture);
GlobalIndexIOMeta fixtureMeta =
- new GlobalIndexIOMeta(new Path(fixturePath.toUri()), fixture.length, null);
+ new GlobalIndexIOMeta(
+ new Path(fixturePath.toUri()),
+ fixture.length,
+ actualFiles.get(0).metadata());
try (GlobalIndexReader reader = createReader(Collections.singletonList(fixtureMeta), 4)) {
assertRows(reader.visitContains(fieldRef, str("banana")).join(), 0L, 2L);
assertRows(reader.visitContains(fieldRef, str("")).join(), 0L, 2L, 3L);
@@ -554,16 +560,23 @@ public class FMGlobalIndexTest {
str("other-3"),
str("needle-4")),
0);
- assertThat(files).allMatch(file -> file.metadata() != null);
- AtomicInteger unrelatedPartitionOpens = new AtomicInteger();
- Path unrelatedPartition = files.get(0).filePath();
- fileReader =
- meta -> {
- if (meta.filePath().equals(unrelatedPartition)) {
- unrelatedPartitionOpens.incrementAndGet();
- }
- return fileIO.newInputStream(meta.filePath());
- };
+ GlobalIndexIOMeta file = files.get(0);
+ FMIndexFile.IndexMeta indexMeta = FMIndexFile.readIndexMeta(file.metadata());
+ assertThat(indexMeta.partitions).hasSize(3);
+ long unrelatedWaveletOffset;
+ try (org.apache.paimon.fs.SeekableInputStream input =
+ fileIO.newInputStream(file.filePath())) {
+ FMIndexFile.Footer footer =
+ FMIndexFile.readFooter(input, indexMeta.partitions.get(0), file.fileSize());
+ unrelatedWaveletOffset =
+ FMIndexFile.readDirectory(input, footer, file.fileSize())
+ .wavelets[0]
+ .blocks
+ .get(0)
+ .block
+ .offset;
+ }
+ corruptByte(file, unrelatedWaveletOffset);
org.apache.paimon.utils.RoaringNavigableMap64 candidates =
new org.apache.paimon.utils.RoaringNavigableMap64();
@@ -580,7 +593,6 @@ public class FMGlobalIndexTest {
.join(),
4L);
}
- assertThat(unrelatedPartitionOpens).hasValue(0);
assertThat(executor.submittedTasks).hasValue(1);
}
@@ -590,13 +602,10 @@ public class FMGlobalIndexTest {
options.set(FMGlobalIndexOptions.PARTITION_ROW_COUNT, 2);
options.set(FMGlobalIndexOptions.COMPRESSION, "none");
indexer = new FMGlobalIndexer(dataField, options);
- List<GlobalIndexIOMeta> files =
- writeData(
- Arrays.asList(
- str("needle-0"), str("other-1"), str("needle-2"), str("other-3")),
- 0);
- GlobalIndexIOMeta first = files.get(0);
- GlobalIndexIOMeta second = files.get(1);
+ GlobalIndexIOMeta first =
+ writeData(Arrays.asList(str("needle-0"), str("other-1")), 0).get(0);
+ GlobalIndexIOMeta second =
+ writeData(Arrays.asList(str("needle-2"), str("other-3")), 2).get(0);
List<GlobalIndexIOMeta> swapped =
Arrays.asList(
new GlobalIndexIOMeta(
@@ -618,7 +627,7 @@ public class FMGlobalIndexTest {
GlobalIndexResult.create(candidates))
.join())
.isInstanceOf(CompletionException.class)
- .hasMessageContaining("row range changed");
+ .hasMessageContaining("does not match the container directory");
}
}
diff --git a/paimon-common/src/test/resources/fmindex-v1-golden.base64 b/paimon-common/src/test/resources/fmindex-v1-golden.base64
index a5ad46c535..f7efdf4dca 100644
--- a/paimon-common/src/test/resources/fmindex-v1-golden.base64
+++ b/paimon-common/src/test/resources/fmindex-v1-golden.base64
@@ -1 +1 @@
-AAAAAQAAAAIAAAAAAAAAAAAAAAAAAAAAAAAADAAAAAcAAAAAAAAAAAAAAAABVVAAAAAAAQAAAAIAAAAAAAAAAAAAAAAAAAAAAAAAAwAAAAgAAAACAAAABgAAACBVv8X1AAAAAQAAAAIAAAAAAAAABQAAAAAAAaAkAAAAEAAAAAgAAAAAAAAABAAAAAwAAAABAAAAAgAAAAAAAAABAAAAAAAAAAIAAAABAAAAAgAAAAAAAAAEAAAAAAADAMAAAAAGYmFuYW5h/////wAAAAgA/2JhbmFuYQAAAAAAAAAEAAAAEwAAAAQAAAACAAAABwAAEAAAAAAC//////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////// [...]
+AAAAAQAAAAIAAAAAAAAAAAAAAAAAAAAAAAAADAAAAAcAAAAAAAAAAAAAAAABVVAAAAAAAQAAAAIAAAAAAAAAAAAAAAAAAAAAAAAAAwAAAAgAAAACAAAABgAAACBVv8X1AAAAAQAAAAIAAAAAAAAABQAAAAAAAaAkAAAAEAAAAAgAAAAAAAAABAAAAAwAAAABAAAAAgAAAAAAAAABAAAAAAAAAAIAAAABAAAAAgAAAAAAAAAEAAAAAAADAMAAAAAGYmFuYW5h/////wAAAAgA/2JhbmFuYQAAAAAAAAAEAAAAEwAAAAQAAAACAAAABwAAEAAAAAAC//////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////// [...]
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 d5ded947ad..8f84f5e504 100644
--- a/paimon-core/src/main/java/org/apache/paimon/KeyValueFileStore.java
+++ b/paimon-core/src/main/java/org/apache/paimon/KeyValueFileStore.java
@@ -186,6 +186,7 @@ public class KeyValueFileStore extends AbstractFileStore<KeyValue> {
BucketedPrimaryKeyIndexMaintainer.Factory primaryKeyIndexMaintainerFactory = null;
if (writeOptions.primaryKeyVectorIndexEnabled()
|| writeOptions.primaryKeyFullTextIndexEnabled()
+ || writeOptions.primaryKeyFMIndexEnabled()
|| !writeOptions.primaryKeyBTreeIndexColumns().isEmpty()
|| !writeOptions.primaryKeyBitmapIndexColumns().isEmpty()
|| !writeOptions.primaryKeyMultiValueIndexColumns().isEmpty()) {
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 128491b9e1..585c9a7b6c 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
@@ -29,6 +29,7 @@ import org.apache.paimon.index.pkfulltext.PkFullTextDataFileReader;
import org.apache.paimon.index.pkfulltext.PkFullTextIndexBuilder;
import org.apache.paimon.index.pkfulltext.PkFullTextIndexFile;
import org.apache.paimon.index.pksorted.BucketedSortedIndexMaintainer;
+import org.apache.paimon.index.pksorted.PkSequentialIndexBuilder;
import org.apache.paimon.index.pksorted.PkSortedDataFileReader;
import org.apache.paimon.index.pksorted.PkSortedIndexBuilder;
import org.apache.paimon.index.pksorted.PkSortedIndexFile;
@@ -382,7 +383,17 @@ public final class BucketedPrimaryKeyIndexMaintainer {
readerFactoryBuilder,
field,
definition.indexType(),
- definition.options()));
+ definition.options(),
+ false));
+ break;
+ case FM:
+ sortedFactories.add(
+ new SortedDefinitionFactory(
+ readerFactoryBuilder,
+ field,
+ definition.indexType(),
+ definition.options(),
+ true));
break;
case FULL_TEXT:
checkArgument(
@@ -516,16 +527,19 @@ public final class BucketedPrimaryKeyIndexMaintainer {
private final DataField field;
private final String indexType;
private final org.apache.paimon.options.Options options;
+ private final boolean sequential;
private SortedDefinitionFactory(
KeyValueFileReaderFactory.Builder readerFactoryBuilder,
DataField field,
String indexType,
- org.apache.paimon.options.Options options) {
+ org.apache.paimon.options.Options options,
+ boolean sequential) {
this.readerFactoryBuilder = readerFactoryBuilder;
this.field = field;
this.indexType = indexType;
this.options = options;
+ this.sequential = sequential;
}
private BucketedSortedIndexMaintainer create(
@@ -537,15 +551,25 @@ public final class BucketedPrimaryKeyIndexMaintainer {
ExecutorService executor,
@Nullable IOManager ioManager) {
PkSortedIndexFile indexFile = handler.pkSortedIndex(partition, bucket);
+ PkSortedDataFileReader.Factory readerFactory =
+ new PkSortedDataFileReader.Factory(
+ readerFactoryBuilder, partition, bucket, field);
+ if (sequential) {
+ PkSequentialIndexBuilder builder =
+ new PkSequentialIndexBuilder(
+ readerFactory, indexFile, field, indexType, options);
+ return new BucketedSortedIndexMaintainer(
+ field.id(),
+ indexType,
+ indexFile,
+ builder::build,
+ restoredDataFiles,
+ restoredPayloads,
+ executor);
+ }
PkSortedIndexBuilder builder =
new PkSortedIndexBuilder(
- new PkSortedDataFileReader.Factory(
- readerFactoryBuilder, partition, bucket, field),
- indexFile,
- field,
- indexType,
- options,
- ioManager);
+ readerFactory, indexFile, field, indexType, options, ioManager);
return new BucketedSortedIndexMaintainer(
field.id(),
indexType,
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 fbec65af5c..1548d20ff3 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
@@ -29,7 +29,12 @@ public class PrimaryKeyIndexDefinition {
BTREE,
BITMAP,
MULTI_VALUE,
- FULL_TEXT
+ FM,
+ FULL_TEXT;
+
+ public boolean isScalar() {
+ return this == BTREE || this == BITMAP || this == MULTI_VALUE || this == FM;
+ }
}
private final String column;
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 fd7f216fff..f7fef0e9f1 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
@@ -22,6 +22,7 @@ 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.globalindex.fmindex.FMGlobalIndexerFactory;
import org.apache.paimon.schema.TableSchema;
import org.apache.paimon.types.DataField;
@@ -49,13 +50,20 @@ public class PrimaryKeyIndexDefinitions {
List<String> bitmapColumns = options.primaryKeyBitmapIndexColumns();
List<String> multiValueColumns = options.primaryKeyMultiValueIndexColumns();
List<String> fullTextColumns = options.primaryKeyFullTextIndexColumns();
+ List<String> fmColumns = options.primaryKeyFMIndexColumns();
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());
+ validateNoDuplicates(fmColumns, CoreOptions.PK_FM_INDEX_COLUMNS.key());
validateOneIndexPerColumn(
- vectorColumns, btreeColumns, bitmapColumns, multiValueColumns, fullTextColumns);
+ vectorColumns,
+ btreeColumns,
+ bitmapColumns,
+ multiValueColumns,
+ fullTextColumns,
+ fmColumns);
List<PrimaryKeyIndexDefinition> definitions = new ArrayList<>();
for (DataField field : schema.fields()) {
@@ -100,6 +108,14 @@ public class PrimaryKeyIndexDefinitions {
"full-text",
options.primaryKeyFullTextIndexOptions(column),
PrimaryKeyIndexDefinition.Family.FULL_TEXT));
+ } else if (fmColumns.contains(column)) {
+ definitions.add(
+ new PrimaryKeyIndexDefinition(
+ column,
+ field.id(),
+ FMGlobalIndexerFactory.IDENTIFIER,
+ options.primaryKeyFMIndexOptions(column),
+ PrimaryKeyIndexDefinition.Family.FM));
}
}
@@ -122,13 +138,15 @@ public class PrimaryKeyIndexDefinitions {
List<String> btreeColumns,
List<String> bitmapColumns,
List<String> multiValueColumns,
- List<String> fullTextColumns) {
+ List<String> fullTextColumns,
+ List<String> fmColumns) {
Set<String> indexedColumns = new HashSet<>();
validateUniqueColumns(indexedColumns, vectorColumns);
validateUniqueColumns(indexedColumns, btreeColumns);
validateUniqueColumns(indexedColumns, bitmapColumns);
validateUniqueColumns(indexedColumns, multiValueColumns);
validateUniqueColumns(indexedColumns, fullTextColumns);
+ validateUniqueColumns(indexedColumns, fmColumns);
}
private static void validateUniqueColumns(Set<String> indexedColumns, List<String> columns) {
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 e3ac81a1d9..95b192e3ea 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 sorted-index definition. */
+/** Maintains one bucket-local source-backed scalar-index definition. */
public class BucketedSortedIndexMaintainer {
private static final Logger LOG = LoggerFactory.getLogger(BucketedSortedIndexMaintainer.class);
@@ -367,7 +367,7 @@ public class BucketedSortedIndexMaintainer {
try {
indexFile.delete(payload);
} catch (RuntimeException e) {
- LOG.warn("Failed to delete unpublished primary-key sorted index payload.", e);
+ LOG.warn("Failed to delete unpublished primary-key scalar index payload.", e);
}
}
@@ -531,7 +531,7 @@ public class BucketedSortedIndexMaintainer {
IndexFileMeta build(List<DataFileMeta> sourceFiles) throws Exception;
}
- /** Sorted-index changes for append and compact snapshot routing. */
+ /** Scalar-index changes for append and compact snapshot routing. */
public static final class SortedIndexCommit {
private final Optional<SortedIndexIncrement> appendIncrement;
diff --git a/paimon-core/src/main/java/org/apache/paimon/index/pksorted/PkSequentialIndexBuilder.java b/paimon-core/src/main/java/org/apache/paimon/index/pksorted/PkSequentialIndexBuilder.java
new file mode 100644
index 0000000000..debdc44508
--- /dev/null
+++ b/paimon-core/src/main/java/org/apache/paimon/index/pksorted/PkSequentialIndexBuilder.java
@@ -0,0 +1,192 @@
+/*
+ * 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.index.pksorted;
+
+import org.apache.paimon.index.IndexFileMeta;
+import org.apache.paimon.index.pk.PrimaryKeyIndexSourceFile;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.utils.IOUtils;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.Iterator;
+import java.util.List;
+import java.util.NoSuchElementException;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/** Builds source-backed indexes whose writers require values in physical source-row order. */
+public class PkSequentialIndexBuilder {
+
+ private final PkSortedIndexBuilder.ReaderFactory readerFactory;
+ private final PkSortedIndexFile indexFile;
+ private final DataField indexField;
+ private final String indexType;
+ private final Options options;
+
+ public PkSequentialIndexBuilder(
+ PkSortedDataFileReader.Factory readerFactory,
+ PkSortedIndexFile indexFile,
+ DataField indexField,
+ String indexType,
+ Options options) {
+ this(readerFactory::create, indexFile, indexField, indexType, options);
+ }
+
+ PkSequentialIndexBuilder(
+ PkSortedIndexBuilder.ReaderFactory readerFactory,
+ PkSortedIndexFile indexFile,
+ DataField indexField,
+ String indexType,
+ Options options) {
+ this.readerFactory = readerFactory;
+ this.indexFile = indexFile;
+ this.indexField = indexField;
+ this.indexType = indexType;
+ this.options = options;
+ }
+
+ public IndexFileMeta build(List<DataFileMeta> dataFiles) throws IOException {
+ checkArgument(!dataFiles.isEmpty(), "A sequential index build requires source files.");
+ List<DataFileMeta> orderedDataFiles = new ArrayList<>(dataFiles);
+ orderedDataFiles.sort(Comparator.comparing(DataFileMeta::fileName));
+ int dataLevel = orderedDataFiles.get(0).level();
+ checkArgument(dataLevel > 0, "A sequential index build requires a positive data level.");
+
+ List<PrimaryKeyIndexSourceFile> sourceFiles = new ArrayList<>();
+ for (DataFileMeta dataFile : orderedDataFiles) {
+ checkArgument(
+ dataFile.level() == dataLevel,
+ "A sequential index build cannot mix data levels %s and %s.",
+ dataLevel,
+ dataFile.level());
+ sourceFiles.add(
+ new PrimaryKeyIndexSourceFile(dataFile.fileName(), dataFile.rowCount()));
+ }
+
+ try (SourceEntryIterator entries = new SourceEntryIterator(orderedDataFiles)) {
+ try {
+ return indexFile.build(
+ dataLevel, sourceFiles, indexField, indexType, options, entries);
+ } catch (UncheckedIOException e) {
+ throw e.getCause();
+ }
+ }
+ }
+
+ private final class SourceEntryIterator
+ implements Iterator<PkSortedIndexFile.Entry>, AutoCloseable {
+
+ private final List<DataFileMeta> dataFiles;
+
+ private int fileIndex;
+ private long sourceOffset;
+ private long expectedPosition;
+ @Nullable private PkSortedIndexBuilder.Reader currentReader;
+ @Nullable private PkSortedIndexFile.Entry next;
+ private boolean prepared;
+ private boolean finished;
+
+ private SourceEntryIterator(List<DataFileMeta> dataFiles) {
+ this.dataFiles = dataFiles;
+ }
+
+ @Override
+ public boolean hasNext() {
+ prepare();
+ return next != null;
+ }
+
+ @Override
+ public PkSortedIndexFile.Entry next() {
+ prepare();
+ if (next == null) {
+ throw new NoSuchElementException();
+ }
+ PkSortedIndexFile.Entry result = next;
+ next = null;
+ prepared = false;
+ return result;
+ }
+
+ private void prepare() {
+ if (prepared || finished) {
+ return;
+ }
+ try {
+ while (fileIndex < dataFiles.size()) {
+ DataFileMeta dataFile = dataFiles.get(fileIndex);
+ if (currentReader == null) {
+ currentReader = readerFactory.create(dataFile);
+ checkArgument(
+ currentReader.rowCount() == dataFile.rowCount(),
+ "Sequential reader row count %s does not match data file %s row count %s.",
+ currentReader.rowCount(),
+ dataFile.fileName(),
+ dataFile.rowCount());
+ }
+ PkSortedDataFileReader.Entry entry = currentReader.readNext();
+ if (entry != null) {
+ checkArgument(
+ entry.rowPosition() == expectedPosition,
+ "Sequential reader for data file %s returned row position %s, expected %s.",
+ dataFile.fileName(),
+ entry.rowPosition(),
+ expectedPosition);
+ next =
+ new PkSortedIndexFile.Entry(
+ entry.value(),
+ Math.addExact(sourceOffset, entry.rowPosition()));
+ expectedPosition++;
+ prepared = true;
+ return;
+ }
+
+ checkArgument(
+ expectedPosition == dataFile.rowCount(),
+ "Sequential reader returned %s rows for data file %s, expected %s.",
+ expectedPosition,
+ dataFile.fileName(),
+ dataFile.rowCount());
+ currentReader.close();
+ currentReader = null;
+ sourceOffset = Math.addExact(sourceOffset, dataFile.rowCount());
+ expectedPosition = 0;
+ fileIndex++;
+ }
+ finished = true;
+ prepared = true;
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ }
+
+ @Override
+ public void close() {
+ IOUtils.closeQuietly(currentReader);
+ currentReader = null;
+ }
+ }
+}
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 e3b4a1d1c3..fb05e43ba3 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 sorted-index payloads for ordered physical data files. */
+/** Builds source-backed scalar-index payloads for ordered physical data files. */
public class PkSortedIndexFile extends IndexFile {
public PkSortedIndexFile(FileIO fileIO, IndexPathFactory pathFactory) {
@@ -66,7 +66,8 @@ public class PkSortedIndexFile extends IndexFile {
sourceRowCount = Math.addExact(sourceRowCount, sourceFile.rowCount());
}
checkArgument(
- sourceRowCount > 0, "A sorted index group must reference at least one source row.");
+ sourceRowCount > 0,
+ "A source-backed index group must reference at least one source row.");
TrackingFileWriter fileWriter = new TrackingFileWriter();
GlobalIndexSingleColumnWriter writer = null;
@@ -78,7 +79,7 @@ public class PkSortedIndexFile extends IndexFile {
Entry entry = sortedEntries.next();
checkArgument(
entry.rowId >= 0 && entry.rowId < sourceRowCount,
- "Row id %s is outside sorted index group row range [0, %s).",
+ "Row id %s is outside source-backed index group row range [0, %s).",
entry.rowId,
sourceRowCount);
writer.write(entry.value, entry.rowId);
@@ -87,12 +88,12 @@ public class PkSortedIndexFile extends IndexFile {
List<ResultEntry> results = writer.finish(sourceRowCount);
checkArgument(
results.size() == 1,
- "Sorted index build must produce exactly one payload file, but produced %s.",
+ "Source-backed index build must produce exactly one payload file, but produced %s.",
results.size());
ResultEntry result = results.get(0);
checkArgument(
result.rowCount() == sourceRowCount,
- "Sorted payload row count %s does not match source row count %s.",
+ "Source-backed payload row count %s does not match source row count %s.",
result.rowCount(),
sourceRowCount);
byte[] sourceMeta = new PrimaryKeyIndexSourceMeta(dataLevel, sourceFiles).serialize();
@@ -138,7 +139,7 @@ public class PkSortedIndexFile extends IndexFile {
return (GlobalIndexSingleColumnWriter) writer;
}
- /** One sorted normalized key and its zero-based source-row ordinal. */
+ /** One index value 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 69658fc221..d730f332ae 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
@@ -1028,6 +1028,7 @@ public class SchemaManager implements Serializable {
|| options.primaryKeyBTreeIndexColumns().contains(fieldName)
|| options.primaryKeyBitmapIndexColumns().contains(fieldName)
|| options.primaryKeyMultiValueIndexColumns().contains(fieldName)
+ || options.primaryKeyFMIndexColumns().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 5d55a73cba..cc0ee88ad7 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
@@ -33,6 +33,7 @@ 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.globalindex.fmindex.FMGlobalIndexerFactory;
import org.apache.paimon.iceberg.IcebergOptions;
import org.apache.paimon.mergetree.compact.aggregate.FieldAggregator;
import org.apache.paimon.mergetree.compact.aggregate.factory.FieldAggregatorFactory;
@@ -1220,6 +1221,7 @@ public class SchemaValidation {
List<String> bitmapColumns = options.primaryKeyBitmapIndexColumns();
List<String> multiValueColumns = options.primaryKeyMultiValueIndexColumns();
List<String> fullTextColumns = options.primaryKeyFullTextIndexColumns();
+ List<String> fmColumns = options.primaryKeyFMIndexColumns();
validateNoDuplicatePrimaryKeyIndexColumns(
vectorColumns, CoreOptions.PK_VECTOR_INDEX_COLUMNS.key());
validateNoDuplicatePrimaryKeyIndexColumns(
@@ -1230,6 +1232,7 @@ public class SchemaValidation {
multiValueColumns, CoreOptions.PK_MULTIVALUE_INDEX_COLUMNS.key());
validateNoDuplicatePrimaryKeyIndexColumns(
fullTextColumns, CoreOptions.PK_FULL_TEXT_INDEX_COLUMNS.key());
+ validateNoDuplicatePrimaryKeyIndexColumns(fmColumns, CoreOptions.PK_FM_INDEX_COLUMNS.key());
Set<String> indexedColumns = new HashSet<>();
validateUniquePrimaryKeyIndexColumns(indexedColumns, vectorColumns);
@@ -1237,6 +1240,7 @@ public class SchemaValidation {
validateUniquePrimaryKeyIndexColumns(indexedColumns, bitmapColumns);
validateUniquePrimaryKeyIndexColumns(indexedColumns, multiValueColumns);
validateUniquePrimaryKeyIndexColumns(indexedColumns, fullTextColumns);
+ validateUniquePrimaryKeyIndexColumns(indexedColumns, fmColumns);
}
private static void validateNoDuplicatePrimaryKeyIndexColumns(
@@ -1261,27 +1265,28 @@ public class SchemaValidation {
private static void validatePrimaryKeySortedIndexes(TableSchema schema, CoreOptions options) {
if (options.primaryKeyBTreeIndexColumns().isEmpty()
&& options.primaryKeyBitmapIndexColumns().isEmpty()
- && options.primaryKeyMultiValueIndexColumns().isEmpty()) {
+ && options.primaryKeyMultiValueIndexColumns().isEmpty()
+ && options.primaryKeyFMIndexColumns().isEmpty()) {
return;
}
checkArgument(
options.deletionVectorsEnabled(),
- "Primary-key BTree, Bitmap, and Multivalue indexes require deletion-vectors.enabled = true.");
+ "Primary-key BTree, Bitmap, Multivalue, and FM indexes require deletion-vectors.enabled = true.");
checkArgument(
!schema.primaryKeys().isEmpty(),
- "Primary-key BTree, Bitmap, and Multivalue indexes require a primary-key table.");
+ "Primary-key BTree, Bitmap, Multivalue, and FM indexes require a primary-key table.");
checkArgument(
options.bucket() > 0 || options.bucket() == BucketMode.POSTPONE_BUCKET,
- "Primary-key BTree, Bitmap, and Multivalue indexes require fixed or postpone bucket mode "
+ "Primary-key BTree, Bitmap, Multivalue, and FM indexes require fixed or postpone bucket mode "
+ "(bucket > 0 or bucket = -2), but bucket is %s.",
options.bucket());
checkArgument(
!options.deletionVectorsMergeOnRead(),
- "Primary-key BTree, Bitmap, and Multivalue indexes require deletion-vectors.merge-on-read = false.");
+ "Primary-key BTree, Bitmap, Multivalue, and FM indexes require deletion-vectors.merge-on-read = false.");
checkArgument(
!options.pkClusteringOverride(),
- "Primary-key BTree, Bitmap, and Multivalue indexes do not support pk-clustering-override.");
+ "Primary-key BTree, Bitmap, Multivalue, and FM indexes do not support pk-clustering-override.");
validatePrimaryKeySortedIndexColumns(
schema,
@@ -1295,6 +1300,8 @@ public class SchemaValidation {
schema,
options.primaryKeyMultiValueIndexColumns(),
CoreOptions.PK_MULTIVALUE_INDEX_COLUMNS.key());
+ validatePrimaryKeySortedIndexColumns(
+ schema, options.primaryKeyFMIndexColumns(), CoreOptions.PK_FM_INDEX_COLUMNS.key());
Map<String, DataField> fields = schema.nameToFieldMap();
for (String column : options.primaryKeyBTreeIndexColumns()) {
@@ -1315,6 +1322,12 @@ public class SchemaValidation {
fields.get(column),
options.primaryKeyMultiValueIndexOptions(column));
}
+ for (String column : options.primaryKeyFMIndexColumns()) {
+ GlobalIndexer.create(
+ FMGlobalIndexerFactory.IDENTIFIER,
+ fields.get(column),
+ options.primaryKeyFMIndexOptions(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 82f658f88d..e50ff3772d 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
@@ -70,9 +70,7 @@ public class PrimaryKeyBatchScan extends AbstractBatchTableScan {
Set<Integer> definitionFieldIds = new HashSet<>();
for (PrimaryKeyIndexDefinition definition :
PrimaryKeyIndexDefinitions.create(table.schema()).definitions()) {
- if (definition.family() == PrimaryKeyIndexDefinition.Family.BTREE
- || definition.family() == PrimaryKeyIndexDefinition.Family.BITMAP
- || definition.family() == PrimaryKeyIndexDefinition.Family.MULTI_VALUE) {
+ if (definition.family().isScalar()) {
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 cfe3fcea64..2bda216ce0 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
@@ -137,9 +137,7 @@ 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.MULTI_VALUE) {
+ if (definition.family().isScalar()) {
scalarDefinitions.add(definition);
}
}
@@ -242,9 +240,7 @@ public final class PrimaryKeySortedIndexScan {
ReaderFactory readerFactory) {
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.MULTI_VALUE) {
+ if (definition.family().isScalar()) {
definitionsByField.put(definition.fieldId(), definition);
}
}
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 ec9be827b3..351fb64def 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
@@ -99,6 +99,23 @@ class PrimaryKeyIndexDefinitionsTest {
.doesNotContainKey("fields.name.pk-full-text.index.options");
}
+ @Test
+ void testCreatesFMDefinitionAndResolvesOptions() {
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.PK_FM_INDEX_COLUMNS.key(), "name");
+ options.put("fm-index.sa-sample-rate", "16");
+ options.put("fields.name.pk-fm.index.options", "{\"partition-row-count\":\"2\"}");
+
+ PrimaryKeyIndexDefinition definition =
+ PrimaryKeyIndexDefinitions.create(schema(options)).definitions().get(0);
+
+ assertThat(definition.column()).isEqualTo("name");
+ assertThat(definition.indexType()).isEqualTo("fmindex");
+ assertThat(definition.family()).isEqualTo(PrimaryKeyIndexDefinition.Family.FM);
+ assertThat(definition.options().get("fm-index.sa-sample-rate")).isEqualTo("16");
+ assertThat(definition.options().get("fm-index.partition-row-count")).isEqualTo("2");
+ }
+
@Test
void testRejectsDuplicateColumnWithinFamily() {
Map<String, String> options = new HashMap<>();
diff --git a/paimon-core/src/test/java/org/apache/paimon/index/pksorted/BucketedSortedIndexMaintainerTest.java b/paimon-core/src/test/java/org/apache/paimon/index/pksorted/BucketedSortedIndexMaintainerTest.java
index 2b1fcffffe..1e15994e08 100644
--- a/paimon-core/src/test/java/org/apache/paimon/index/pksorted/BucketedSortedIndexMaintainerTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/index/pksorted/BucketedSortedIndexMaintainerTest.java
@@ -50,7 +50,7 @@ import java.util.concurrent.atomic.AtomicReference;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
-/** Tests bucket-local BTree/Bitmap source maintenance. */
+/** Tests bucket-local source-backed scalar-index maintenance. */
class BucketedSortedIndexMaintainerTest {
@TempDir java.nio.file.Path tempPath;
@@ -754,6 +754,27 @@ class BucketedSortedIndexMaintainerTest {
null);
}
+ private static IndexFileMeta payload(
+ String fileName,
+ String indexType,
+ byte[] sourceMeta,
+ long rowRangeStart,
+ long rowCount) {
+ return new IndexFileMeta(
+ indexType,
+ fileName,
+ 1,
+ rowCount,
+ new GlobalIndexMeta(
+ rowRangeStart,
+ rowRangeStart + rowCount - 1,
+ 7,
+ null,
+ new byte[] {1},
+ sourceMeta),
+ null);
+ }
+
private static DataFileMeta dataFile(String fileName, long rowCount) {
return DataFileMeta.forAppend(
fileName,
diff --git a/paimon-core/src/test/java/org/apache/paimon/index/pksorted/PkSequentialIndexBuilderTest.java b/paimon-core/src/test/java/org/apache/paimon/index/pksorted/PkSequentialIndexBuilderTest.java
new file mode 100644
index 0000000000..fa0a55eb2b
--- /dev/null
+++ b/paimon-core/src/test/java/org/apache/paimon/index/pksorted/PkSequentialIndexBuilderTest.java
@@ -0,0 +1,226 @@
+/*
+ * 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.index.pksorted;
+
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.index.GlobalIndexMeta;
+import org.apache.paimon.index.IndexFileMeta;
+import org.apache.paimon.index.IndexPathFactory;
+import org.apache.paimon.index.pk.PrimaryKeyIndexSourceFile;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.manifest.FileSource;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.stats.SimpleStats;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataTypes;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.UUID;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests source-order streaming for sequential primary-key indexes. */
+class PkSequentialIndexBuilderTest {
+
+ @TempDir java.nio.file.Path tempPath;
+
+ @Test
+ void testStreamsFilesAndRowsInCanonicalSourceOrder() throws Exception {
+ DataFileMeta sourceB = dataFile("data-b", 1);
+ DataFileMeta sourceA = dataFile("data-a", 2);
+ List<PrimaryKeyIndexSourceFile> capturedSources = new ArrayList<>();
+ List<PkSortedIndexFile.Entry> capturedEntries = new ArrayList<>();
+ List<ArrayReader> readers = new ArrayList<>();
+ PkSortedIndexFile capturingFile =
+ new PkSortedIndexFile(LocalFileIO.create(), pathFactory()) {
+ @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 PkSequentialIndexBuilder(
+ dataFile -> {
+ ArrayReader reader =
+ new ArrayReader(
+ dataFile.fileName().equals("data-a")
+ ? Arrays.asList(entry("a", 0), entry(null, 1))
+ : Collections.singletonList(entry("b", 0)));
+ readers.add(reader);
+ return reader;
+ },
+ capturingFile,
+ field(),
+ "fmindex",
+ new Options())
+ .build(Arrays.asList(sourceB, sourceA));
+
+ assertThat(capturedSources)
+ .extracting(PrimaryKeyIndexSourceFile::fileName)
+ .containsExactly("data-a", "data-b");
+ assertThat(capturedEntries)
+ .extracting(PkSortedIndexFile.Entry::value)
+ .containsExactly("a", null, "b");
+ assertThat(capturedEntries)
+ .extracting(PkSortedIndexFile.Entry::rowId)
+ .containsExactly(0L, 1L, 2L);
+ assertThat(readers).allMatch(ArrayReader::isClosed);
+ }
+
+ @Test
+ void testRejectsNonConsecutivePhysicalPositionsAndClosesReader() {
+ ArrayReader reader =
+ new ArrayReader(
+ Arrays.asList(
+ entry(BinaryString.fromString("a"), 0),
+ entry(BinaryString.fromString("b"), 2)));
+ PkSequentialIndexBuilder builder =
+ new PkSequentialIndexBuilder(
+ ignored -> reader,
+ new PkSortedIndexFile(LocalFileIO.create(), pathFactory()),
+ field(),
+ "fmindex",
+ new Options());
+
+ assertThatThrownBy(() -> builder.build(Collections.singletonList(dataFile("data-file", 2))))
+ .hasMessageContaining("returned row position 2, expected 1");
+ assertThat(reader.isClosed()).isTrue();
+ }
+
+ @Test
+ void testPropagatesReaderCreationIOException() {
+ PkSequentialIndexBuilder builder =
+ new PkSequentialIndexBuilder(
+ (PkSortedIndexBuilder.ReaderFactory)
+ ignored -> {
+ throw new IOException("expected reader failure");
+ },
+ new PkSortedIndexFile(LocalFileIO.create(), pathFactory()),
+ field(),
+ "fmindex",
+ new Options());
+
+ assertThatThrownBy(() -> builder.build(Collections.singletonList(dataFile("data-file", 1))))
+ .isInstanceOf(IOException.class)
+ .hasMessage("expected reader failure");
+ }
+
+ private static PkSortedDataFileReader.Entry entry(@Nullable Object value, long position) {
+ return new PkSortedDataFileReader.Entry(value, position);
+ }
+
+ private static DataField field() {
+ return new DataField(7, "content", DataTypes.STRING());
+ }
+
+ private static IndexFileMeta ignoredPayload() {
+ return new IndexFileMeta("test", "test", 0, 0, (GlobalIndexMeta) null, null);
+ }
+
+ private static DataFileMeta dataFile(String name, long rowCount) {
+ return DataFileMeta.forAppend(
+ name,
+ 100,
+ rowCount,
+ SimpleStats.EMPTY_STATS,
+ 0,
+ 0,
+ 1,
+ Collections.emptyList(),
+ null,
+ FileSource.COMPACT,
+ null,
+ null,
+ null,
+ null)
+ .upgrade(1);
+ }
+
+ private IndexPathFactory pathFactory() {
+ Path root = new Path(tempPath.toUri());
+ return new IndexPathFactory() {
+ @Override
+ public Path toPath(String fileName) {
+ return new Path(root, fileName);
+ }
+
+ @Override
+ public Path newPath() {
+ return new Path(root, UUID.randomUUID().toString());
+ }
+
+ @Override
+ public boolean isExternalPath() {
+ return false;
+ }
+ };
+ }
+
+ private static final class ArrayReader implements PkSortedIndexBuilder.Reader {
+
+ private final List<PkSortedDataFileReader.Entry> entries;
+ private int position;
+ private boolean closed;
+
+ private ArrayReader(List<PkSortedDataFileReader.Entry> entries) {
+ this.entries = entries;
+ }
+
+ @Override
+ public long rowCount() {
+ return entries.size();
+ }
+
+ @Nullable
+ @Override
+ public PkSortedDataFileReader.Entry readNext() {
+ return position == entries.size() ? null : entries.get(position++);
+ }
+
+ @Override
+ public void close() {
+ closed = true;
+ }
+
+ private boolean isClosed() {
+ return closed;
+ }
+ }
+}
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 ca2ea01984..0e0baf60b4 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
@@ -51,7 +51,7 @@ import java.util.stream.Stream;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
-/** Tests source-backed sorted index payload creation. */
+/** Tests source-backed scalar-index payload creation. */
class PkSortedIndexFileTest {
@TempDir java.nio.file.Path tempPath;
@@ -184,7 +184,7 @@ class PkSortedIndexFileTest {
}
@Test
- void testRejectsMultiplePayloadsAndDeletesWholeGroup() throws Exception {
+ void testBuildRejectsMultiplePayloadsAndDeletesThem() throws Exception {
LocalFileIO fileIO = LocalFileIO.create();
PkSortedIndexFile indexFile =
new PkSortedIndexFile(fileIO, pathFactory(tempPath)) {
@@ -214,7 +214,8 @@ class PkSortedIndexFileTest {
throw new RuntimeException(e);
}
results.add(
- new ResultEntry(fileName, rowCount, new byte[] {2}));
+ new ResultEntry(
+ fileName, rowCount / 2, new byte[] {2}));
}
return results;
}
@@ -285,7 +286,7 @@ class PkSortedIndexFileTest {
Collections.singletonList(
new PkSortedIndexFile.Entry(10, 1))
.iterator()))
- .hasMessageContaining("outside sorted index group row range");
+ .hasMessageContaining("outside source-backed index group row range");
assertThat(closed).isTrue();
}
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 0f7ad773a9..dc78a087df 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
@@ -60,6 +60,24 @@ class PrimaryKeySortedIndexOptionsTest {
.containsExactly("tags", "categories");
}
+ @Test
+ void testResolvesFMIndexColumnsAndOptions() {
+ Map<String, String> values = new HashMap<>();
+ values.put("pk-fm.index.columns", " content, description ");
+ values.put(
+ "fields.content.pk-fm.index.options",
+ "{\"partition-row-count\":\"2000\",\"fm-index.sa-sample-rate\":\"16\"}");
+
+ CoreOptions coreOptions = new CoreOptions(values);
+ Options options = coreOptions.primaryKeyFMIndexOptions("content");
+
+ assertThat(coreOptions.primaryKeyFMIndexEnabled()).isTrue();
+ assertThat(coreOptions.primaryKeyFMIndexColumns())
+ .containsExactly("content", "description");
+ assertThat(options.get("fm-index.partition-row-count")).isEqualTo("2000");
+ assertThat(options.get("fm-index.sa-sample-rate")).isEqualTo("16");
+ }
+
@Test
void testResolvesBTreeIndexAndSortOptions() {
Map<String, String> values = new HashMap<>();
diff --git a/paimon-core/src/test/java/org/apache/paimon/schema/PrimaryKeyFMIndexValidationTest.java b/paimon-core/src/test/java/org/apache/paimon/schema/PrimaryKeyFMIndexValidationTest.java
new file mode 100644
index 0000000000..a279ce771f
--- /dev/null
+++ b/paimon-core/src/test/java/org/apache/paimon/schema/PrimaryKeyFMIndexValidationTest.java
@@ -0,0 +1,147 @@
+/*
+ * 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.schema;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataTypes;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+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 FM index option validation. */
+class PrimaryKeyFMIndexValidationTest {
+
+ @Test
+ void testValidConfiguration() {
+ Map<String, String> options = enabledOptions();
+ options.put(
+ "fields.content.pk-fm.index.options",
+ "{\"partition-row-count\":\"2\",\"sa-sample-rate\":\"16\"}");
+
+ assertThatCode(() -> validateTableSchema(schema(options))).doesNotThrowAnyException();
+ }
+
+ @Test
+ void testRequiresCharacterColumn() {
+ Map<String, String> options = enabledOptions();
+ options.put(CoreOptions.PK_FM_INDEX_COLUMNS.key(), "id");
+
+ assertThatThrownBy(() -> validateTableSchema(schema(options)))
+ .hasMessageContaining("FM index requires a character string column");
+ }
+
+ @Test
+ void testRequiresDeletionVectors() {
+ Map<String, String> options = enabledOptions();
+ options.put(CoreOptions.DELETION_VECTORS_ENABLED.key(), "false");
+
+ assertThatThrownBy(() -> validateTableSchema(schema(options)))
+ .hasMessageContaining("FM indexes require deletion-vectors.enabled = true");
+ }
+
+ @Test
+ void testRejectsUnknownColumn() {
+ Map<String, String> options = enabledOptions();
+ options.put(CoreOptions.PK_FM_INDEX_COLUMNS.key(), "unknown");
+
+ assertThatThrownBy(() -> validateTableSchema(schema(options)))
+ .hasMessageContaining(CoreOptions.PK_FM_INDEX_COLUMNS.key())
+ .hasMessageContaining("entry 'unknown'");
+ }
+
+ @Test
+ void testRejectsColumnConfiguredForAnotherFamily() {
+ Map<String, String> options = enabledOptions();
+ options.put(CoreOptions.PK_BITMAP_INDEX_COLUMNS.key(), "content");
+
+ assertThatThrownBy(() -> validateTableSchema(schema(options)))
+ .hasMessageContaining("content")
+ .hasMessageContaining("at most one primary-key index");
+ }
+
+ @Test
+ void testRejectsMalformedFieldOptions() {
+ Map<String, String> options = enabledOptions();
+ options.put("fields.content.pk-fm.index.options", "{not-json");
+
+ assertThatThrownBy(() -> validateTableSchema(schema(options)))
+ .hasMessageContaining("fields.content.pk-fm.index.options must be a JSON object");
+ }
+
+ @Test
+ void testRejectsInvalidFMOptions() {
+ Map<String, String> options = enabledOptions();
+ options.put("fields.content.pk-fm.index.options", "{\"partition-row-count\":\"0\"}");
+
+ assertThatThrownBy(() -> validateTableSchema(schema(options)))
+ .hasMessageContaining("FM index partition row count must be positive");
+ }
+
+ @Test
+ void testRequiresPrimaryKeyTable() {
+ Map<String, String> options = enabledOptions();
+ options.put(CoreOptions.BUCKET_KEY.key(), "id");
+ TableSchema appendTable =
+ new TableSchema(
+ 0,
+ fields(),
+ 0,
+ Collections.emptyList(),
+ Collections.emptyList(),
+ options,
+ "");
+
+ assertThatThrownBy(() -> validateTableSchema(appendTable))
+ .hasMessageContaining("FM indexes require a primary-key table");
+ }
+
+ private static Map<String, String> enabledOptions() {
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.BUCKET.key(), "1");
+ options.put(CoreOptions.DELETION_VECTORS_ENABLED.key(), "true");
+ options.put(CoreOptions.PK_FM_INDEX_COLUMNS.key(), "content");
+ return options;
+ }
+
+ private static java.util.List<DataField> fields() {
+ return Arrays.asList(
+ new DataField(0, "id", DataTypes.INT().notNull()),
+ new DataField(1, "content", DataTypes.STRING()));
+ }
+
+ private static TableSchema schema(Map<String, String> options) {
+ return new TableSchema(
+ 0,
+ fields(),
+ 0,
+ Collections.emptyList(),
+ Collections.singletonList("id"),
+ options,
+ "");
+ }
+}
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 ff00caed39..1d1ac2f9c5 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
@@ -103,7 +103,7 @@ class PrimaryKeySortedIndexValidationTest {
assertThatThrownBy(() -> validateTableSchema(schema(options)))
.hasMessageContaining(
- "Primary-key BTree, Bitmap, and Multivalue indexes require deletion-vectors.enabled = true");
+ "Primary-key BTree, Bitmap, Multivalue, and FM indexes require deletion-vectors.enabled = true");
}
@Test
@@ -114,7 +114,7 @@ class PrimaryKeySortedIndexValidationTest {
assertThatThrownBy(() -> validateTableSchema(schema(options, Collections.emptyList())))
.hasMessageContaining(
- "Primary-key BTree, Bitmap, and Multivalue indexes require a primary-key table");
+ "Primary-key BTree, Bitmap, Multivalue, and FM indexes require a primary-key table");
}
@Test
@@ -125,7 +125,7 @@ class PrimaryKeySortedIndexValidationTest {
assertThatThrownBy(() -> validateTableSchema(schema(options)))
.hasMessageContaining(
- "Primary-key BTree, Bitmap, and Multivalue indexes require fixed or postpone bucket mode");
+ "Primary-key BTree, Bitmap, Multivalue, and FM indexes require fixed or postpone bucket mode");
}
@Test
@@ -145,7 +145,7 @@ class PrimaryKeySortedIndexValidationTest {
assertThatThrownBy(() -> validateTableSchema(schema(options)))
.hasMessageContaining(
- "Primary-key BTree, Bitmap, and Multivalue indexes require deletion-vectors.merge-on-read = false");
+ "Primary-key BTree, Bitmap, Multivalue, and FM indexes require deletion-vectors.merge-on-read = false");
}
@Test
@@ -157,7 +157,7 @@ class PrimaryKeySortedIndexValidationTest {
assertThatThrownBy(() -> validateTableSchema(schema(options)))
.hasMessageContaining(
- "Primary-key BTree, Bitmap, and Multivalue indexes do not support pk-clustering-override");
+ "Primary-key BTree, Bitmap, Multivalue, and FM indexes do not support pk-clustering-override");
}
@Test
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 a44bd78f00..cb39ea112d 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
@@ -438,6 +438,42 @@ public class SchemaManagerTest {
.hasMessage("Cannot update type of primary-key index column: [content]");
}
+ @Test
+ public void testRejectDestructivePrimaryKeyFMIndexColumnChanges() 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_FM_INDEX_COLUMNS.key(), "content");
+ Schema schema =
+ new Schema(
+ Arrays.asList(
+ new DataField(0, "id", DataTypes.INT().notNull()),
+ new DataField(1, "content", DataTypes.STRING())),
+ Collections.emptyList(),
+ Collections.singletonList("id"),
+ options,
+ "");
+ SchemaManager manager = new SchemaManager(LocalFileIO.create(), path);
+ manager.createTable(schema);
+
+ assertThatThrownBy(
+ () ->
+ manager.commitChanges(
+ SchemaChange.renameColumn(
+ new String[] {"content"}, "renamed_content")))
+ .isInstanceOf(UnsupportedOperationException.class)
+ .hasMessage("Cannot rename primary-key index column: [content]");
+ assertThatThrownBy(() -> manager.commitChanges(SchemaChange.dropColumn("content")))
+ .isInstanceOf(UnsupportedOperationException.class)
+ .hasMessage("Cannot drop primary-key index column: [content]");
+ assertThatThrownBy(
+ () ->
+ manager.commitChanges(
+ SchemaChange.updateColumnType("content", DataTypes.INT())))
+ .isInstanceOf(UnsupportedOperationException.class)
+ .hasMessage("Cannot update type of primary-key index column: [content]");
+ }
+
@Test
public void testRejectDropPrimaryKeyBitmapIndexColumn() throws Exception {
Map<String, String> options = new HashMap<>();
diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PrimaryKeySortedIndexTest.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PrimaryKeySortedIndexTest.scala
index 823b099582..228a925348 100644
--- a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PrimaryKeySortedIndexTest.scala
+++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PrimaryKeySortedIndexTest.scala
@@ -31,6 +31,62 @@ import scala.collection.JavaConverters._
/** End-to-end Spark SQL tests for source-backed primary-key sorted indexes. */
class PrimaryKeySortedIndexTest extends PaimonSparkTestBase {
+ test("primary-key FM index supports exact contains with partitioned container") {
+ withTable("t") {
+ spark.sql("""
+ |CREATE TABLE t (id INT, content STRING)
+ |TBLPROPERTIES (
+ | 'primary-key' = 'id',
+ | 'bucket' = '1',
+ | 'deletion-vectors.enabled' = 'true',
+ | 'pk-fm.index.columns' = 'content',
+ | 'fields.content.pk-fm.index.options' =
+ | '{"partition-row-count":"2"}'
+ |)
+ |""".stripMargin)
+ spark.sql("""
+ |INSERT INTO t VALUES
+ | (1, 'alpha needle omega'),
+ | (2, 'noise'),
+ | (3, 'needle at start')
+ |""".stripMargin)
+ spark.sql("""
+ |INSERT INTO t VALUES
+ | (4, 'unicode 你好 needle'),
+ | (5, CAST(NULL AS STRING)),
+ | (6, 'short e')
+ |""".stripMargin)
+ spark.sql("CALL sys.compact(table => 't')")
+
+ val sourceIndexes = loadTable("t").store.newIndexFileHandler.scanEntries.asScala
+ .map(_.indexFile)
+ .filter(meta => meta.globalIndexMeta != null && meta.globalIndexMeta.sourceMeta != null)
+ assert(sourceIndexes.map(_.indexType).toSet == Set("fmindex"))
+ assert(sourceIndexes.size == 1)
+
+ val predicateBuilder = new PredicateBuilder(loadTable("t").rowType())
+ val indexedQuery = "SELECT id FROM t WHERE content LIKE '%needle%'"
+ val indexedScan = getPaimonScan(indexedQuery)
+ assert(
+ indexedScan.pushedDataFilters.contains(
+ predicateBuilder.contains(1, BinaryString.fromString("needle"))))
+ assert(indexedScan.inputSplits.exists(_.isInstanceOf[IndexedSplit]))
+ checkAnswer(spark.sql(indexedQuery), Seq(Row(1), Row(3), Row(4)))
+
+ spark.sql("UPDATE t SET content = 'updated needle' WHERE id = 2")
+ spark.sql("DELETE FROM t WHERE id = 3")
+ spark.sql("INSERT INTO t VALUES (7, 'new needle row')")
+
+ val mixedScan = getPaimonScan(indexedQuery)
+ assert(mixedScan.inputSplits.exists(_.isInstanceOf[IndexedSplit]))
+ assert(mixedScan.inputSplits.exists(_.isInstanceOf[DataSplit]))
+ checkAnswer(spark.sql(indexedQuery), Seq(Row(1), Row(2), Row(4), Row(7)))
+ checkAnswer(
+ spark.sql("SELECT id FROM t WHERE content LIKE '%e%'"),
+ Seq(Row(1), Row(2), Row(4), Row(6), Row(7)))
+ }
+ }
+
test("Spark array predicates use multivalue index") {
assume(gteqSpark3_3)