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 7f4134c2a7 [global-index] Reuse sorted build for bitmap indexes (#8283)
7f4134c2a7 is described below
commit 7f4134c2a77e9ba4e61a483676f9b8b1bd3606c5
Author: Jingsong Lee <[email protected]>
AuthorDate: Fri Jun 19 11:19:39 2026 +0800
[global-index] Reuse sorted build for bitmap indexes (#8283)
This PR generalizes the BTree global-index build path into a
sorted-index build path and reuses it for Bitmap global indexes. It also
exposes Bitmap global-index creation through Flink and Spark procedures
and updates the related documentation.
---
docs/docs/flink/procedures.md | 8 +-
docs/docs/multimodal-table/global-index.mdx | 6 +-
docs/docs/multimodal-table/global-index/bitmap.mdx | 8 ++
docs/docs/multimodal-table/global-index/btree.mdx | 7 +-
docs/docs/multimodal-table/index.mdx | 2 +-
docs/docs/spark/procedures.md | 3 +-
.../globalindex/sorted/SortedIndexOptions.java | 41 ++++++++
.../globalindex/sorted/SortedIndexOptionsTest.java | 45 +++++++++
.../SortedGlobalIndexBuilder.java} | 36 ++++---
.../test/java/org/apache/paimon/JavaPyE2ETest.java | 11 ++-
.../DataEvolutionRowIdReassignerTest.java | 7 +-
.../SortedGlobalIndexBuilderSplitTest.java} | 15 +--
.../SortedGlobalIndexBuilderTest.java} | 22 +++--
.../metastore/VisibilityWaitCallbackTest.java | 13 +--
.../paimon/table/BtreeGlobalIndexTableTest.java | 6 +-
.../SortedIndexTopoBuilder.java} | 104 +++++++++++----------
.../procedure/CreateGlobalIndexProcedure.java | 26 ++++--
...dexITCase.java => SortedGlobalIndexITCase.java} | 37 +++++++-
.../SortedIndexTopoBuilderTest.java} | 32 +++----
.../procedure/CreateGlobalIndexProcedureTest.java | 5 +-
.../globalindex/DefaultGlobalIndexTopoBuilder.java | 5 -
.../globalindex/GlobalIndexTopologyBuilder.java | 6 +-
.../GlobalIndexTopologyBuilderUtils.java | 40 ++------
.../SortedIndexTopoBuilder.java} | 42 +++++----
...on.spark.globalindex.GlobalIndexTopologyBuilder | 16 ----
.../procedure/CreateGlobalIndexProcedureTest.java | 5 +-
.../procedure/CreateGlobalIndexProcedureTest.scala | 38 ++++++++
27 files changed, 378 insertions(+), 208 deletions(-)
diff --git a/docs/docs/flink/procedures.md b/docs/docs/flink/procedures.md
index 0dbf8f0c1b..a59306df18 100644
--- a/docs/docs/flink/procedures.md
+++ b/docs/docs/flink/procedures.md
@@ -1004,7 +1004,7 @@ All available procedures are listed below.
To create a global index on a table for accelerating queries.
Arguments:
<li>table(required): the target table identifier.</li>
<li>index_column(required): the column name to build index on.</li>
- <li>index_type(required): the type of global index, supported
types include 'btree', 'ivf-flat', 'ivf-pq', 'ivf-hnsw-flat', 'ivf-hnsw-sq',
'tantivy-fulltext'.</li>
+ <li>index_type(required): the type of global index, supported
types include 'btree', 'bitmap', 'ivf-flat', 'ivf-pq', 'ivf-hnsw-flat',
'ivf-hnsw-sq', 'tantivy-fulltext'.</li>
<li>partitions(optional): partition filter for selective index
creation.</li>
<li>options(optional): additional dynamic options for index
creation.</li>
</td>
@@ -1014,6 +1014,12 @@ All available procedures are listed below.
`table` => 'default.T',<br/>
`index_column` => 'name',<br/>
`index_type` => 'btree')<br/><br/>
+ -- Create bitmap index<br/>
+ CALL sys.create_global_index(<br/>
+ `table` => 'default.T',<br/>
+ `index_column` => 'tag',<br/>
+ `index_type` => 'bitmap',<br/>
+ `options` => 'sorted-index.records-per-range=1000000')<br/><br/>
-- Create index for specific partitions<br/>
CALL sys.create_global_index(<br/>
`table` => 'default.T',<br/>
diff --git a/docs/docs/multimodal-table/global-index.mdx
b/docs/docs/multimodal-table/global-index.mdx
index d7b0041b71..90a8ac33bc 100644
--- a/docs/docs/multimodal-table/global-index.mdx
+++ b/docs/docs/multimodal-table/global-index.mdx
@@ -148,9 +148,11 @@ These table options affect global index build and read
behavior:
|---|---|---|
| `global-index.enabled` | `true` | Whether scans can use global indexes. |
| `global-index.external-path` | Not set | Root directory for global index
files. If not set, files are stored under the table index directory. |
-| `global-index.row-count-per-shard` | `100000` | Target row count per shard
for non-BTree global index builds. |
+| `sorted-index.records-per-range` | `10000000` | Expected number of records
per sorted global index file for BTree and Bitmap builds. |
+| `sorted-index.build.max-parallelism` | `4096` | Maximum Flink or Spark
parallelism for building sorted global indexes. |
+| `global-index.row-count-per-shard` | `100000` | Target row count per shard
for non-sorted global index builds such as vector and full-text indexes. |
| `global-index.build.max-shard` | `32` | Preferred maximum shard count for
global index builds. |
-| `global-index.build.max-parallelism` | `4096` | Maximum Flink or Spark
parallelism for building global indexes. |
+| `global-index.build.max-parallelism` | `4096` | Maximum Flink or Spark
parallelism for building non-sorted global indexes. |
| `global-index.thread-num` | `32` | Maximum number of concurrent threads for
global index I/O. |
| `visibility-callback.enabled` | `false` | Whether batch or bounded-stream
commits wait until existing global indexes cover newly added files. |
| `visibility-callback.timeout` | `30 min` | Maximum wait time for visibility
callback. |
diff --git a/docs/docs/multimodal-table/global-index/bitmap.mdx
b/docs/docs/multimodal-table/global-index/bitmap.mdx
index b9715687d5..453d8dbf00 100644
--- a/docs/docs/multimodal-table/global-index/bitmap.mdx
+++ b/docs/docs/multimodal-table/global-index/bitmap.mdx
@@ -76,10 +76,18 @@ CALL sys.create_global_index(
);
```
+Bitmap indexes share the sorted index build path with BTree indexes. Use
+`sorted-index.records-per-range` to control the expected records per generated
index
+file, and `sorted-index.build.max-parallelism` to cap Flink or Spark build
+parallelism. The legacy `btree-index.records-per-range` and
+`btree-index.build.max-parallelism` keys are still recognized as fallback keys.
+
## Bitmap Options
| Option | Default | Description |
|---|---|---|
+| `sorted-index.records-per-range` | `10000000` | Expected number of records
per sorted global index file for BTree and Bitmap builds. |
+| `sorted-index.build.max-parallelism` | `4096` | Maximum Flink or Spark
parallelism for building sorted global indexes. |
| `bitmap-index.dictionary-block-size` | `16 kb` | Target size of dictionary
blocks in bitmap global index files. Smaller blocks reduce dictionary read
amplification for high-cardinality columns; larger blocks reduce dictionary
block index size. |
| `bitmap-index.compression` | `none` | Compression algorithm for bitmap
dictionary blocks and the dictionary block index. Supported values are the same
block codecs as BTree index, such as `none`, `lz4`, `lzo`, and `zstd`. |
| `bitmap-index.compression-level` | `1` | Compression level used by codecs
that support levels, such as `zstd`. |
diff --git a/docs/docs/multimodal-table/global-index/btree.mdx
b/docs/docs/multimodal-table/global-index/btree.mdx
index 236447c987..ee50db0aed 100644
--- a/docs/docs/multimodal-table/global-index/btree.mdx
+++ b/docs/docs/multimodal-table/global-index/btree.mdx
@@ -72,13 +72,16 @@ CALL sys.create_global_index(
| Option | Default | Description |
|---|---|---|
-| `btree-index.records-per-range` | `10000000` | Expected number of records
per BTree index file. |
-| `btree-index.build.max-parallelism` | `4096` | Maximum Flink or Spark
parallelism for building BTree indexes. |
+| `sorted-index.records-per-range` | `10000000` | Expected number of records
per sorted global index file for BTree and Bitmap builds. |
+| `sorted-index.build.max-parallelism` | `4096` | Maximum Flink or Spark
parallelism for building sorted global indexes. |
| `btree-index.block-size` | `64 kb` | Block size used by BTree index files. |
| `btree-index.cache-size` | `128 mb` | Cache size used by BTree index
readers. |
| `btree-index.fallback-scan-max-size` | `256 mb` | Maximum total size of
candidate BTree global index files to allow fallback index scans. Set to `0 b`
to disable fallback scans. |
| `btree-index.compression` | `none` | Compression algorithm used by BTree
index blocks. |
+The legacy `btree-index.records-per-range` and
+`btree-index.build.max-parallelism` keys are still recognized as fallback keys.
+
## Query with BTree Index
Once a BTree index is built, it is automatically used during scan when a
filter predicate matches the indexed column.
diff --git a/docs/docs/multimodal-table/index.mdx
b/docs/docs/multimodal-table/index.mdx
index f0a4b29750..01a0e65f25 100644
--- a/docs/docs/multimodal-table/index.mdx
+++ b/docs/docs/multimodal-table/index.mdx
@@ -37,7 +37,7 @@ Key capabilities:
- **[Data Evolution](./data-evolution)**: Update partial columns without
rewriting entire files, enabling efficient schema evolution.
- **[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, vector, and full-text
(Tantivy) indexes for efficient lookups and similarity search.
+- **[Global Index](./global-index)**: Build BTree, Bitmap, vector, and
full-text (Tantivy) indexes for efficient lookups and similarity search.
All multimodal features require the following table properties:
diff --git a/docs/docs/spark/procedures.md b/docs/docs/spark/procedures.md
index d693346e00..0fd5f733b6 100644
--- a/docs/docs/spark/procedures.md
+++ b/docs/docs/spark/procedures.md
@@ -517,12 +517,13 @@ This section introduce all available spark procedures
about paimon.
To create global index files for a given column. The table must have
<code>row-tracking.enabled=true</code>. Arguments:
<li>table: the target table identifier. Cannot be empty.</li>
<li>index_column: the name of the column to index. Cannot be
empty.</li>
- <li>index_type: type of the index to build, e.g. 'btree'. Cannot
be empty.</li>
+ <li>index_type: type of the index to build, e.g. 'btree' or
'bitmap'. Cannot be empty.</li>
<li>partitions: partition filter to limit the partitions on which
to build the index. The comma (",") represents "AND", the semicolon (";")
represents "OR". Left empty for all partitions.</li>
<li>options: additional dynamic options of the table. It
prioritizes higher than original `tableProp` and lower than `procedureArg`.</li>
</td>
<td>
CALL sys.create_global_index(table => 'default.T', index_column =>
'name', index_type => 'btree')<br/><br/>
+ CALL sys.create_global_index(table => 'default.T', index_column =>
'tag', index_type => 'bitmap', options =>
'sorted-index.records-per-range=1000000')<br/><br/>
CALL sys.create_global_index(table => 'default.T', index_column =>
'name', index_type => 'btree', partitions => 'pt=p1;pt=p2')<br/><br/>
CALL sys.create_global_index(table => 'default.T', index_column =>
'content', index_type => 'tantivy-fulltext', options =>
'tantivy.tokenizer=ngram,tantivy.ngram.min-gram=2,tantivy.ngram.max-gram=2')<br/><br/>
CALL sys.create_global_index(table => 'default.T', index_column =>
'content', index_type => 'tantivy-fulltext', options =>
'tantivy.tokenizer=jieba')<br/><br/>
diff --git
a/paimon-common/src/main/java/org/apache/paimon/globalindex/sorted/SortedIndexOptions.java
b/paimon-common/src/main/java/org/apache/paimon/globalindex/sorted/SortedIndexOptions.java
new file mode 100644
index 0000000000..4c437d890a
--- /dev/null
+++
b/paimon-common/src/main/java/org/apache/paimon/globalindex/sorted/SortedIndexOptions.java
@@ -0,0 +1,41 @@
+/*
+ * 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.globalindex.sorted;
+
+import org.apache.paimon.options.ConfigOption;
+import org.apache.paimon.options.ConfigOptions;
+
+/** Options for sorted global index build. */
+public class SortedIndexOptions {
+
+ public static final ConfigOption<Long> SORTED_INDEX_RECORDS_PER_RANGE =
+ ConfigOptions.key("sorted-index.records-per-range")
+ .longType()
+ .defaultValue(10_000_000L)
+ .withFallbackKeys("btree-index.records-per-range")
+ .withDescription("The expected number of records per
sorted index file.");
+
+ public static final ConfigOption<Integer>
SORTED_INDEX_BUILD_MAX_PARALLELISM =
+ ConfigOptions.key("sorted-index.build.max-parallelism")
+ .intType()
+ .defaultValue(4096)
+ .withFallbackKeys("btree-index.build.max-parallelism")
+ .withDescription(
+ "The max parallelism of Flink/Spark for building
sorted indexes.");
+}
diff --git
a/paimon-common/src/test/java/org/apache/paimon/globalindex/sorted/SortedIndexOptionsTest.java
b/paimon-common/src/test/java/org/apache/paimon/globalindex/sorted/SortedIndexOptionsTest.java
new file mode 100644
index 0000000000..8ad1c55793
--- /dev/null
+++
b/paimon-common/src/test/java/org/apache/paimon/globalindex/sorted/SortedIndexOptionsTest.java
@@ -0,0 +1,45 @@
+/*
+ * 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.globalindex.sorted;
+
+import org.apache.paimon.options.Options;
+
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link SortedIndexOptions}. */
+class SortedIndexOptionsTest {
+
+ @Test
+ void testDefaultRecordsPerRange() {
+
assertThat(SortedIndexOptions.SORTED_INDEX_RECORDS_PER_RANGE.defaultValue())
+ .isEqualTo(10_000_000L);
+ }
+
+ @Test
+ void testBTreeBuildOptionFallbacks() {
+ Options options = new Options();
+ options.setString("btree-index.records-per-range", "100");
+ options.setString("btree-index.build.max-parallelism", "8");
+
+
assertThat(options.get(SortedIndexOptions.SORTED_INDEX_RECORDS_PER_RANGE)).isEqualTo(100L);
+
assertThat(options.get(SortedIndexOptions.SORTED_INDEX_BUILD_MAX_PARALLELISM)).isEqualTo(8);
+ }
+}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/globalindex/btree/BTreeGlobalIndexBuilder.java
b/paimon-core/src/main/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexBuilder.java
similarity index 94%
rename from
paimon-core/src/main/java/org/apache/paimon/globalindex/btree/BTreeGlobalIndexBuilder.java
rename to
paimon-core/src/main/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexBuilder.java
index 0fcbbcbb43..e8ba93ffc8 100644
---
a/paimon-core/src/main/java/org/apache/paimon/globalindex/btree/BTreeGlobalIndexBuilder.java
+++
b/paimon-core/src/main/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexBuilder.java
@@ -16,7 +16,7 @@
* limitations under the License.
*/
-package org.apache.paimon.globalindex.btree;
+package org.apache.paimon.globalindex.sorted;
import org.apache.paimon.CoreOptions;
import org.apache.paimon.Snapshot;
@@ -80,13 +80,12 @@ import static
org.apache.paimon.globalindex.GlobalIndexBuilderUtils.toIndexFileM
import static org.apache.paimon.types.VectorType.isVectorStoreFile;
import static org.apache.paimon.utils.Preconditions.checkArgument;
-/** Builder to build btree global index. */
-public class BTreeGlobalIndexBuilder implements Serializable {
+/** Builder to build sorted global index. */
+public class SortedGlobalIndexBuilder implements Serializable {
private static final long serialVersionUID = 1L;
private static final double FLOATING = 1.2;
- private static final String INDEX_TYPE = "btree";
-
+ private final String indexType;
private final FileStoreTable table;
private final RowType rowType;
private final Options options;
@@ -100,15 +99,20 @@ public class BTreeGlobalIndexBuilder implements
Serializable {
@Nullable private PartitionPredicate partitionPredicate;
- public BTreeGlobalIndexBuilder(Table table) {
+ public SortedGlobalIndexBuilder(Table table, String indexType) {
+ this(table, indexType, ((FileStoreTable)
table).coreOptions().toConfiguration());
+ }
+
+ public SortedGlobalIndexBuilder(Table table, String indexType, Options
options) {
+ this.indexType = indexType;
this.table = (FileStoreTable) table;
this.rowType = this.table.rowType();
- this.options = this.table.coreOptions().toConfiguration();
+ this.options = options;
this.recordsPerRange =
- (long)
(options.get(BTreeIndexOptions.BTREE_INDEX_RECORDS_PER_RANGE) * FLOATING);
+ (long)
(options.get(SortedIndexOptions.SORTED_INDEX_RECORDS_PER_RANGE) * FLOATING);
}
- public BTreeGlobalIndexBuilder withIndexField(String indexField) {
+ public SortedGlobalIndexBuilder withIndexField(String indexField) {
checkArgument(
rowType.containsField(indexField),
"Column '%s' does not exist in table '%s'.",
@@ -123,12 +127,12 @@ public class BTreeGlobalIndexBuilder implements
Serializable {
return this;
}
- public BTreeGlobalIndexBuilder withPartitionPredicate(PartitionPredicate
partitionPredicate) {
+ public SortedGlobalIndexBuilder withPartitionPredicate(PartitionPredicate
partitionPredicate) {
this.partitionPredicate = partitionPredicate;
return this;
}
- public BTreeGlobalIndexBuilder withSnapshot(Snapshot snapshot) {
+ public SortedGlobalIndexBuilder withSnapshot(Snapshot snapshot) {
this.snapshot = snapshot;
return this;
}
@@ -192,7 +196,7 @@ public class BTreeGlobalIndexBuilder implements
Serializable {
private List<Range> indexedRowRanges(Snapshot snapshot) {
List<Range> ranges = new ArrayList<>();
for (IndexManifestEntry entry :
- table.store().newIndexFileHandler().scan(snapshot, "btree")) {
+ table.store().newIndexFileHandler().scan(snapshot, indexType))
{
if (partitionPredicate != null &&
!partitionPredicate.test(entry.partition())) {
continue;
}
@@ -286,10 +290,12 @@ public class BTreeGlobalIndexBuilder implements
Serializable {
public GlobalIndexSingleColumnWriter createWriter() throws IOException {
GlobalIndexSingleColumnWriter currentWriter;
- GlobalIndexWriter indexWriter = createIndexWriter(table, INDEX_TYPE,
indexField, options);
+ GlobalIndexWriter indexWriter = createIndexWriter(table, indexType,
indexField, options);
if (!(indexWriter instanceof GlobalIndexSingleColumnWriter)) {
throw new RuntimeException(
- "Unexpected implementation, the index writer of BTree
should be an instance of GlobalIndexSingleColumnWriter, but found: "
+ "Unexpected implementation, the index writer of "
+ + indexType
+ + " should be an instance of
GlobalIndexSingleColumnWriter, but found: "
+ indexWriter.getClass().getName());
}
currentWriter = (GlobalIndexSingleColumnWriter) indexWriter;
@@ -306,7 +312,7 @@ public class BTreeGlobalIndexBuilder implements
Serializable {
table.coreOptions(),
rowRange,
indexField.id(),
- INDEX_TYPE,
+ indexType,
resultEntries);
DataIncrement dataIncrement =
DataIncrement.indexIncrement(indexFileMetas);
return new CommitMessageImpl(
diff --git a/paimon-core/src/test/java/org/apache/paimon/JavaPyE2ETest.java
b/paimon-core/src/test/java/org/apache/paimon/JavaPyE2ETest.java
index 543ca8eab6..1d2a4f3b85 100644
--- a/paimon-core/src/test/java/org/apache/paimon/JavaPyE2ETest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/JavaPyE2ETest.java
@@ -34,7 +34,7 @@ import org.apache.paimon.disk.IOManager;
import org.apache.paimon.fs.FileIOFinder;
import org.apache.paimon.fs.Path;
import org.apache.paimon.fs.local.LocalFileIO;
-import org.apache.paimon.globalindex.btree.BTreeGlobalIndexBuilder;
+import org.apache.paimon.globalindex.sorted.SortedGlobalIndexBuilder;
import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.manifest.IndexManifestEntry;
import org.apache.paimon.options.MemorySize;
@@ -572,7 +572,8 @@ public class JavaPyE2ETest {
}
// build index
- BTreeGlobalIndexBuilder builder = new
BTreeGlobalIndexBuilder(table).withIndexField("k");
+ SortedGlobalIndexBuilder builder =
+ new SortedGlobalIndexBuilder(table,
"btree").withIndexField("k");
try (BatchTableCommit commit = writeBuilder.newCommit()) {
commit.commit(
builder.build(
@@ -648,7 +649,8 @@ public class JavaPyE2ETest {
}
// build index
- BTreeGlobalIndexBuilder builder = new
BTreeGlobalIndexBuilder(table).withIndexField("k");
+ SortedGlobalIndexBuilder builder =
+ new SortedGlobalIndexBuilder(table,
"btree").withIndexField("k");
try (BatchTableCommit commit = writeBuilder.newCommit()) {
commit.commit(
builder.build(
@@ -726,7 +728,8 @@ public class JavaPyE2ETest {
}
// build index
- BTreeGlobalIndexBuilder builder = new
BTreeGlobalIndexBuilder(table).withIndexField("k");
+ SortedGlobalIndexBuilder builder =
+ new SortedGlobalIndexBuilder(table,
"btree").withIndexField("k");
try (BatchTableCommit commit = writeBuilder.newCommit()) {
commit.commit(
builder.build(
diff --git
a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdReassignerTest.java
b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdReassignerTest.java
index ffaade20cb..b17c912aaf 100644
---
a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdReassignerTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdReassignerTest.java
@@ -26,8 +26,8 @@ import org.apache.paimon.data.BinaryString;
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.data.serializer.InternalRowSerializer;
-import org.apache.paimon.globalindex.btree.BTreeGlobalIndexBuilder;
import org.apache.paimon.globalindex.btree.BTreeIndexOptions;
+import org.apache.paimon.globalindex.sorted.SortedGlobalIndexBuilder;
import org.apache.paimon.index.GlobalIndexMeta;
import org.apache.paimon.index.IndexFileMeta;
import org.apache.paimon.io.DataFileMeta;
@@ -910,7 +910,8 @@ public class DataEvolutionRowIdReassignerTest extends
TableTestBase {
}
private void createBTreeIndex(FileStoreTable table) throws Exception {
- BTreeGlobalIndexBuilder builder = new
BTreeGlobalIndexBuilder(table).withIndexField("id");
+ SortedGlobalIndexBuilder builder =
+ new SortedGlobalIndexBuilder(table,
"btree").withIndexField("id");
List<DataSplit> dataSplits =
builder.scan()
.map(Pair::getRight)
@@ -919,7 +920,7 @@ public class DataEvolutionRowIdReassignerTest extends
TableTestBase {
new IllegalStateException(
"Expected scan result when
building index."));
List<CommitMessage> commitMessages = new ArrayList<>();
- for (DataSplit dataSplit :
BTreeGlobalIndexBuilder.splitByContiguousRowRange(dataSplits)) {
+ for (DataSplit dataSplit :
SortedGlobalIndexBuilder.splitByContiguousRowRange(dataSplits)) {
commitMessages.addAll(builder.build(dataSplit, ioManager));
}
try (BatchTableCommit commit =
table.newBatchWriteBuilder().newCommit()) {
diff --git
a/paimon-core/src/test/java/org/apache/paimon/globalindex/btree/BTreeGlobalIndexBuilderSplitTest.java
b/paimon-core/src/test/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexBuilderSplitTest.java
similarity index 90%
rename from
paimon-core/src/test/java/org/apache/paimon/globalindex/btree/BTreeGlobalIndexBuilderSplitTest.java
rename to
paimon-core/src/test/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexBuilderSplitTest.java
index 60431faa12..2d64ce9db5 100644
---
a/paimon-core/src/test/java/org/apache/paimon/globalindex/btree/BTreeGlobalIndexBuilderSplitTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexBuilderSplitTest.java
@@ -16,7 +16,7 @@
* limitations under the License.
*/
-package org.apache.paimon.globalindex.btree;
+package org.apache.paimon.globalindex.sorted;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.globalindex.IndexedSplit;
@@ -38,8 +38,8 @@ import java.util.UUID;
import static org.assertj.core.api.Assertions.assertThat;
-/** Tests for split regrouping in {@link BTreeGlobalIndexBuilder}. */
-public class BTreeGlobalIndexBuilderSplitTest {
+/** Tests for split regrouping in {@link SortedGlobalIndexBuilder}. */
+public class SortedGlobalIndexBuilderSplitTest {
@Test
public void testSplitByContiguousRowRangeFromDataFiles() {
@@ -58,14 +58,15 @@ public class BTreeGlobalIndexBuilderSplitTest {
.build();
List<DataSplit> rebuilt =
-
BTreeGlobalIndexBuilder.splitByContiguousRowRange(Collections.singletonList(split));
+ SortedGlobalIndexBuilder.splitByContiguousRowRange(
+ Collections.singletonList(split));
assertThat(rebuilt).hasSize(2);
assertThat(rebuilt.get(0).dataFiles()).containsExactly(file1, file3);
assertThat(rebuilt.get(1).dataFiles()).containsExactly(file2);
- assertThat(BTreeGlobalIndexBuilder.calcRowRange(rebuilt.get(0)))
+ assertThat(SortedGlobalIndexBuilder.calcRowRange(rebuilt.get(0)))
.isEqualTo(new Range(0, 199));
- assertThat(BTreeGlobalIndexBuilder.calcRowRange(rebuilt.get(1)))
+ assertThat(SortedGlobalIndexBuilder.calcRowRange(rebuilt.get(1)))
.isEqualTo(new Range(300, 399));
}
@@ -86,7 +87,7 @@ public class BTreeGlobalIndexBuilderSplitTest {
.build();
Map<BinaryRow, Map<Range, List<Split>>> result =
- BTreeGlobalIndexBuilder.groupSplitsByRange(
+ SortedGlobalIndexBuilder.groupSplitsByRange(
RowRangeIndex.create(
Arrays.asList(new Range(4750, 4900), new
Range(5938, 7599))),
Collections.singletonList(split));
diff --git
a/paimon-core/src/test/java/org/apache/paimon/globalindex/btree/BTreeGlobalIndexBuilderTest.java
b/paimon-core/src/test/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexBuilderTest.java
similarity index 95%
rename from
paimon-core/src/test/java/org/apache/paimon/globalindex/btree/BTreeGlobalIndexBuilderTest.java
rename to
paimon-core/src/test/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexBuilderTest.java
index 5e8c5cb82e..7e0d18c2c9 100644
---
a/paimon-core/src/test/java/org/apache/paimon/globalindex/btree/BTreeGlobalIndexBuilderTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexBuilderTest.java
@@ -16,7 +16,7 @@
* limitations under the License.
*/
-package org.apache.paimon.globalindex.btree;
+package org.apache.paimon.globalindex.sorted;
import org.apache.paimon.CoreOptions;
import org.apache.paimon.Snapshot;
@@ -25,6 +25,7 @@ import org.apache.paimon.data.BinaryString;
import org.apache.paimon.data.BlobData;
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.globalindex.KeySerializer;
+import org.apache.paimon.globalindex.btree.BTreeIndexOptions;
import org.apache.paimon.index.GlobalIndexMeta;
import org.apache.paimon.index.IndexFileHandler;
import org.apache.paimon.index.IndexFileMeta;
@@ -57,8 +58,8 @@ import java.util.List;
import java.util.Map;
import java.util.Optional;
-/** Test class for {@link BTreeGlobalIndexBuilder}. */
-public class BTreeGlobalIndexBuilderTest extends TableTestBase {
+/** Test class for {@link SortedGlobalIndexBuilder}. */
+public class SortedGlobalIndexBuilderTest extends TableTestBase {
private static final long PART_ROW_NUM = 1000L;
private static final KeySerializer KEY_SERIALIZER =
KeySerializer.create(DataTypes.INT());
@@ -109,7 +110,7 @@ public class BTreeGlobalIndexBuilderTest extends
TableTestBase {
private void createIndex(PartitionPredicate partitionPredicate) throws
Exception {
FileStoreTable table = getTableDefault();
- BTreeGlobalIndexBuilder builder = new BTreeGlobalIndexBuilder(table);
+ SortedGlobalIndexBuilder builder = new SortedGlobalIndexBuilder(table,
"btree");
builder.withIndexField("f0");
builder.withPartitionPredicate(partitionPredicate);
List<DataSplit> dataSplits =
@@ -168,7 +169,7 @@ public class BTreeGlobalIndexBuilderTest extends
TableTestBase {
createTableDefault();
FileStoreTable table = getTableDefault();
- BTreeGlobalIndexBuilder builder = new BTreeGlobalIndexBuilder(table);
+ SortedGlobalIndexBuilder builder = new SortedGlobalIndexBuilder(table,
"btree");
builder.withIndexField("f0");
Assertions.assertFalse(
@@ -182,7 +183,7 @@ public class BTreeGlobalIndexBuilderTest extends
TableTestBase {
createIndex(null);
FileStoreTable table = getTableDefault();
- BTreeGlobalIndexBuilder builder = new BTreeGlobalIndexBuilder(table);
+ SortedGlobalIndexBuilder builder = new SortedGlobalIndexBuilder(table,
"btree");
builder.withIndexField("f0");
Assertions.assertFalse(
@@ -212,7 +213,7 @@ public class BTreeGlobalIndexBuilderTest extends
TableTestBase {
}
table = getTableDefault();
- BTreeGlobalIndexBuilder builder = new BTreeGlobalIndexBuilder(table);
+ SortedGlobalIndexBuilder builder = new SortedGlobalIndexBuilder(table,
"btree");
builder.withIndexField("f0");
Optional<Pair<org.apache.paimon.utils.RowRangeIndex, List<DataSplit>>>
incrementalScan =
@@ -270,7 +271,7 @@ public class BTreeGlobalIndexBuilderTest extends
TableTestBase {
predicate =
PartitionPredicate.createPartitionPredicate(
partType, Collections.singletonMap("dt",
BinaryString.fromString("p0")));
- BTreeGlobalIndexBuilder builder = new BTreeGlobalIndexBuilder(table);
+ SortedGlobalIndexBuilder builder = new SortedGlobalIndexBuilder(table,
"btree");
builder.withIndexField("f0");
builder.withPartitionPredicate(PartitionPredicate.fromPredicate(partType,
predicate));
@@ -330,7 +331,8 @@ public class BTreeGlobalIndexBuilderTest extends
TableTestBase {
containsBlobFile(table.store().newScan().plan().files()),
"Test table should contain blob manifest entries.");
- BTreeGlobalIndexBuilder builder = new
BTreeGlobalIndexBuilder(table).withIndexField("f0");
+ SortedGlobalIndexBuilder builder =
+ new SortedGlobalIndexBuilder(table,
"btree").withIndexField("f0");
assertNoBlobFiles(
builder.scan()
.map(Pair::getRight)
@@ -385,7 +387,7 @@ public class BTreeGlobalIndexBuilderTest extends
TableTestBase {
Assertions.assertNotEquals(
"blob",
file.fileFormat(),
- "BTree global index scan should not include blob
files.");
+ "Sorted global index scan should not include blob
files.");
}
}
}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/metastore/VisibilityWaitCallbackTest.java
b/paimon-core/src/test/java/org/apache/paimon/metastore/VisibilityWaitCallbackTest.java
index 030b2af41c..9440988e88 100644
---
a/paimon-core/src/test/java/org/apache/paimon/metastore/VisibilityWaitCallbackTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/metastore/VisibilityWaitCallbackTest.java
@@ -22,7 +22,7 @@ import org.apache.paimon.CoreOptions;
import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.data.BinaryString;
import org.apache.paimon.data.GenericRow;
-import org.apache.paimon.globalindex.btree.BTreeGlobalIndexBuilder;
+import org.apache.paimon.globalindex.sorted.SortedGlobalIndexBuilder;
import org.apache.paimon.partition.PartitionPredicate;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.schema.Schema;
@@ -97,8 +97,8 @@ public class VisibilityWaitCallbackTest extends TableTestBase
{
buildIndex(getTableDefault(), true);
writeFuture.get(10, TimeUnit.SECONDS);
- BTreeGlobalIndexBuilder builder =
- new
BTreeGlobalIndexBuilder(getTableDefault()).withIndexField("f1");
+ SortedGlobalIndexBuilder builder =
+ new SortedGlobalIndexBuilder(getTableDefault(),
"btree").withIndexField("f1");
assertThat(builder.incrementalScan()).isNotPresent();
} finally {
executor.shutdownNow();
@@ -159,7 +159,8 @@ public class VisibilityWaitCallbackTest extends
TableTestBase {
}
private void buildIndex(FileStoreTable table, boolean incremental) throws
Exception {
- BTreeGlobalIndexBuilder builder = new
BTreeGlobalIndexBuilder(table).withIndexField("f1");
+ SortedGlobalIndexBuilder builder =
+ new SortedGlobalIndexBuilder(table,
"btree").withIndexField("f1");
Optional<Pair<RowRangeIndex, List<DataSplit>>> scan =
incremental ? builder.incrementalScan() : builder.scan();
assertThat(scan).isPresent();
@@ -175,8 +176,8 @@ public class VisibilityWaitCallbackTest extends
TableTestBase {
}
private void buildPartitionIndex(FileStoreTable table, String partition)
throws Exception {
- BTreeGlobalIndexBuilder builder =
- new BTreeGlobalIndexBuilder(table)
+ SortedGlobalIndexBuilder builder =
+ new SortedGlobalIndexBuilder(table, "btree")
.withIndexField("f1")
.withPartitionPredicate(partitionPredicate(table,
partition));
Optional<Pair<RowRangeIndex, List<DataSplit>>> scan = builder.scan();
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/BtreeGlobalIndexTableTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/BtreeGlobalIndexTableTest.java
index 80c82fd626..3fcd8f2ae4 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/BtreeGlobalIndexTableTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/BtreeGlobalIndexTableTest.java
@@ -24,7 +24,7 @@ import org.apache.paimon.globalindex.DataEvolutionBatchScan;
import org.apache.paimon.globalindex.GlobalIndexResult;
import org.apache.paimon.globalindex.GlobalIndexScanner;
import org.apache.paimon.globalindex.IndexedSplit;
-import org.apache.paimon.globalindex.btree.BTreeGlobalIndexBuilder;
+import org.apache.paimon.globalindex.sorted.SortedGlobalIndexBuilder;
import org.apache.paimon.partition.PartitionPredicate;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.predicate.PredicateBuilder;
@@ -206,8 +206,8 @@ public class BtreeGlobalIndexTableTest extends
DataEvolutionTestBase {
private void createIndex(String fieldName, List<Range> rowRanges) throws
Exception {
FileStoreTable table = (FileStoreTable) catalog.getTable(identifier());
- BTreeGlobalIndexBuilder builder =
- new BTreeGlobalIndexBuilder(table).withIndexField(fieldName);
+ SortedGlobalIndexBuilder builder =
+ new SortedGlobalIndexBuilder(table,
"btree").withIndexField(fieldName);
List<DataSplit> dataSplits =
builder.scan()
.map(org.apache.paimon.utils.Pair::getRight)
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/btree/BTreeIndexTopoBuilder.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/globalindex/SortedIndexTopoBuilder.java
similarity index 84%
rename from
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/btree/BTreeIndexTopoBuilder.java
rename to
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/globalindex/SortedIndexTopoBuilder.java
index 98697fae73..4a17bb68f1 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/btree/BTreeIndexTopoBuilder.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/globalindex/SortedIndexTopoBuilder.java
@@ -16,7 +16,7 @@
* limitations under the License.
*/
-package org.apache.paimon.flink.btree;
+package org.apache.paimon.flink.globalindex;
import org.apache.paimon.CoreOptions;
import org.apache.paimon.data.BinaryRow;
@@ -38,8 +38,8 @@ import org.apache.paimon.flink.utils.BoundedOneInputOperator;
import org.apache.paimon.flink.utils.JavaTypeInfo;
import org.apache.paimon.flink.utils.StreamExecutionEnvironmentUtils;
import org.apache.paimon.globalindex.GlobalIndexSingleColumnWriter;
-import org.apache.paimon.globalindex.btree.BTreeGlobalIndexBuilder;
-import org.apache.paimon.globalindex.btree.BTreeIndexOptions;
+import org.apache.paimon.globalindex.sorted.SortedGlobalIndexBuilder;
+import org.apache.paimon.globalindex.sorted.SortedIndexOptions;
import org.apache.paimon.options.Options;
import org.apache.paimon.partition.PartitionPredicate;
import org.apache.paimon.reader.RecordReader;
@@ -69,26 +69,34 @@ import
org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
import java.io.IOException;
import java.io.Serializable;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.UUID;
import java.util.function.Supplier;
-import static
org.apache.paimon.globalindex.btree.BTreeGlobalIndexBuilder.groupSplitsByRange;
-import static
org.apache.paimon.globalindex.btree.BTreeGlobalIndexBuilder.splitByContiguousRowRange;
+import static
org.apache.paimon.globalindex.sorted.SortedGlobalIndexBuilder.groupSplitsByRange;
+import static
org.apache.paimon.globalindex.sorted.SortedGlobalIndexBuilder.splitByContiguousRowRange;
-/** The {@link BTreeIndexTopoBuilder} for BTree index in Flink. */
-public class BTreeIndexTopoBuilder {
+/** The topology builder for sorted indexes in Flink. */
+public class SortedIndexTopoBuilder {
- private static final String BUILD_TASK_ID_FIELD = "_BTREE_BUILD_TASK_ID";
+ private static final String BUILD_TASK_ID_FIELD =
"_SORTED_INDEX_BUILD_TASK_ID";
private static final int BUILD_TASK_ID_FIELD_ID = -1;
+ private static final HashSet<String> SUPPORTED_INDEX_TYPES =
+ new HashSet<>(Arrays.asList("btree", "bitmap"));
+
+ public static boolean supports(String indexType) {
+ return SUPPORTED_INDEX_TYPES.contains(indexType);
+ }
public static boolean buildIndex(
StreamExecutionEnvironment env,
- Supplier<BTreeGlobalIndexBuilder> indexBuilderSupplier,
+ Supplier<SortedGlobalIndexBuilder> indexBuilderSupplier,
FileStoreTable table,
List<String> indexColumns,
PartitionPredicate partitionPredicate,
@@ -96,7 +104,7 @@ public class BTreeIndexTopoBuilder {
throws Exception {
List<DataStream<Committable>> allStreams = new ArrayList<>();
for (String indexColumn : indexColumns) {
- BTreeGlobalIndexBuilder indexBuilder =
+ SortedGlobalIndexBuilder indexBuilder =
indexBuilderSupplier.get().withIndexField(indexColumn);
if (partitionPredicate != null) {
indexBuilder =
indexBuilder.withPartitionPredicate(partitionPredicate);
@@ -133,9 +141,10 @@ public class BTreeIndexTopoBuilder {
DataType indexFieldType = sortReadType.getTypeAt(indexFieldPos);
// 3. Calculate maximum parallelism bound
- long recordsPerRange =
userOptions.get(BTreeIndexOptions.BTREE_INDEX_RECORDS_PER_RANGE);
+ long recordsPerRange =
+
userOptions.get(SortedIndexOptions.SORTED_INDEX_RECORDS_PER_RANGE);
int maxParallelism =
-
userOptions.get(BTreeIndexOptions.BTREE_INDEX_BUILD_MAX_PARALLELISM);
+
userOptions.get(SortedIndexOptions.SORTED_INDEX_BUILD_MAX_PARALLELISM);
// 4. Build one topology for all contiguous row ranges
CoreOptions coreOptions = table.coreOptions();
@@ -145,8 +154,8 @@ public class BTreeIndexTopoBuilder {
sortColumns.add(indexColumn);
int partitionFieldSize = table.partitionKeys().size();
BinaryRowSerializer binaryRowSerializer = new
BinaryRowSerializer(partitionFieldSize);
- List<BTreeBuildTask> buildTasks = new ArrayList<>();
- List<BTreeSplitTask> splitTasks = new ArrayList<>();
+ List<SortedBuildTask> buildTasks = new ArrayList<>();
+ List<SortedSplitTask> splitTasks = new ArrayList<>();
for (Map.Entry<BinaryRow, Map<Range, List<Split>>> partitionEntry :
partitionRangeSplits.entrySet()) {
BinaryRow partition = partitionEntry.getKey();
@@ -159,9 +168,9 @@ public class BTreeIndexTopoBuilder {
}
int taskId = buildTasks.size();
- buildTasks.add(new BTreeBuildTask(taskId, range,
partitionBytes));
+ buildTasks.add(new SortedBuildTask(taskId, range,
partitionBytes));
for (Split split : rangeSplits) {
- splitTasks.add(new BTreeSplitTask(taskId, split));
+ splitTasks.add(new SortedSplitTask(taskId, split));
}
}
}
@@ -205,26 +214,27 @@ public class BTreeIndexTopoBuilder {
StreamExecutionEnvironment env,
FileStoreTable table,
String indexColumn,
+ String indexType,
PartitionPredicate partitionPredicate,
Options userOptions)
throws Exception {
if (buildIndex(
env,
- () -> new BTreeGlobalIndexBuilder(table),
+ () -> new SortedGlobalIndexBuilder(table, indexType,
userOptions),
table,
Collections.singletonList(indexColumn),
partitionPredicate,
userOptions)) {
- env.execute("Create btree global index for table: " +
table.name());
+ env.execute("Create " + indexType + " global index for table: " +
table.name());
}
}
protected static DataStream<Committable> executeForBuildTasks(
StreamExecutionEnvironment env,
- List<BTreeBuildTask> buildTasks,
- List<BTreeSplitTask> splitTasks,
+ List<SortedBuildTask> buildTasks,
+ List<SortedSplitTask> splitTasks,
ReadBuilder readBuilder,
- BTreeGlobalIndexBuilder indexBuilder,
+ SortedGlobalIndexBuilder indexBuilder,
int partitionFieldSize,
int taskIdPos,
int indexFieldPos,
@@ -237,9 +247,9 @@ public class BTreeIndexTopoBuilder {
int maxParallelism) {
int parallelism = calculateParallelism(buildTasks, recordsPerRange,
maxParallelism);
- DataStream<BTreeSplitTask> sourceStream =
+ DataStream<SortedSplitTask> sourceStream =
StreamExecutionEnvironmentUtils.fromData(
- env, splitTasks, new
JavaTypeInfo<>(BTreeSplitTask.class))
+ env, splitTasks, new
JavaTypeInfo<>(SortedSplitTask.class))
.name("Global Index Source")
.setParallelism(1);
@@ -267,7 +277,7 @@ public class BTreeIndexTopoBuilder {
return sortedStream
.transform(
- "write-btree-index",
+ "write-sorted-index",
new CommittableTypeInfo(),
new WriteIndexOperator(
buildTasks,
@@ -281,9 +291,9 @@ public class BTreeIndexTopoBuilder {
}
static int calculateParallelism(
- List<BTreeBuildTask> buildTasks, long recordsPerRange, int
maxParallelism) {
+ List<SortedBuildTask> buildTasks, long recordsPerRange, int
maxParallelism) {
long totalRecords = 0;
- for (BTreeBuildTask task : buildTasks) {
+ for (SortedBuildTask task : buildTasks) {
long count = task.rowRange.count();
if (Long.MAX_VALUE - totalRecords < count) {
totalRecords = Long.MAX_VALUE;
@@ -317,7 +327,7 @@ public class BTreeIndexTopoBuilder {
new CommitterOperatorFactory<>(
false,
true,
- "BTreeIndexCommitter-" + UUID.randomUUID(),
+ "SortedIndexCommitter-" + UUID.randomUUID(),
context ->
new StoreCommitter(
table,
table.newCommit(context.commitUser()), context),
@@ -332,7 +342,7 @@ public class BTreeIndexTopoBuilder {
private static class ReadDataOperator
extends
org.apache.flink.table.runtime.operators.TableStreamOperator<RowData>
implements
org.apache.flink.streaming.api.operators.OneInputStreamOperator<
- BTreeSplitTask, RowData> {
+ SortedSplitTask, RowData> {
private static final long serialVersionUID = 1L;
@@ -351,8 +361,8 @@ public class BTreeIndexTopoBuilder {
}
@Override
- public void processElement(StreamRecord<BTreeSplitTask> element)
throws Exception {
- BTreeSplitTask buildTask = element.getValue();
+ public void processElement(StreamRecord<SortedSplitTask> element)
throws Exception {
+ SortedSplitTask buildTask = element.getValue();
GenericRow taskId = GenericRow.of(buildTask.taskId);
try (RecordReader<InternalRow> reader =
tableRead.createReader(buildTask.split)) {
reader.forEachRemaining(
@@ -366,27 +376,27 @@ public class BTreeIndexTopoBuilder {
private static class WriteIndexOperator extends
BoundedOneInputOperator<RowData, Committable> {
- private final List<BTreeBuildTask> buildTasks;
+ private final List<SortedBuildTask> buildTasks;
private final int partitionFieldSize;
- private final BTreeGlobalIndexBuilder builder;
+ private final SortedGlobalIndexBuilder builder;
private final int taskIdPos;
private final int indexFieldPos;
private final int rowIdPos;
private final DataType indexFieldType;
private transient long counter;
- private transient BTreeBuildTask currentTask;
+ private transient SortedBuildTask currentTask;
private transient BinaryRow currentPartition;
private transient GlobalIndexSingleColumnWriter currentWriter;
private transient List<CommitMessage> commitMessages;
- private transient Map<Integer, BTreeBuildTask> buildTasksById;
+ private transient Map<Integer, SortedBuildTask> buildTasksById;
private transient InternalRow.FieldGetter indexFieldGetter;
private transient BinaryRowSerializer binaryRowSerializer;
public WriteIndexOperator(
- List<BTreeBuildTask> buildTasks,
+ List<SortedBuildTask> buildTasks,
int partitionFieldSize,
- BTreeGlobalIndexBuilder builder,
+ SortedGlobalIndexBuilder builder,
int taskIdPos,
int indexFieldPos,
int rowIdPos,
@@ -405,7 +415,7 @@ public class BTreeIndexTopoBuilder {
super.open();
commitMessages = new ArrayList<>();
buildTasksById = new HashMap<>();
- for (BTreeBuildTask task : buildTasks) {
+ for (SortedBuildTask task : buildTasks) {
buildTasksById.put(task.taskId, task);
}
indexFieldGetter = InternalRow.createFieldGetter(indexFieldType,
indexFieldPos);
@@ -416,9 +426,9 @@ public class BTreeIndexTopoBuilder {
public void processElement(StreamRecord<RowData> element) throws
IOException {
InternalRow row = new FlinkRowWrapper(element.getValue());
int taskId = row.getInt(taskIdPos);
- BTreeBuildTask task = buildTasksById.get(taskId);
+ SortedBuildTask task = buildTasksById.get(taskId);
if (task == null) {
- throw new IllegalArgumentException("Unknown BTree build task
id: " + taskId);
+ throw new IllegalArgumentException("Unknown sorted index build
task id: " + taskId);
}
if (currentTask == null || currentTask.taskId != taskId) {
@@ -463,8 +473,8 @@ public class BTreeIndexTopoBuilder {
}
}
- /** Metadata for one BTree index build range. */
- public static class BTreeBuildTask implements Serializable {
+ /** Metadata for one sorted index build range. */
+ public static class SortedBuildTask implements Serializable {
private static final long serialVersionUID = 1L;
@@ -472,26 +482,26 @@ public class BTreeIndexTopoBuilder {
private Range rowRange;
private byte[] partition;
- public BTreeBuildTask() {}
+ public SortedBuildTask() {}
- BTreeBuildTask(int taskId, Range rowRange, byte[] partition) {
+ public SortedBuildTask(int taskId, Range rowRange, byte[] partition) {
this.taskId = taskId;
this.rowRange = rowRange;
this.partition = partition;
}
}
- /** Split assigned to one BTree index build task. */
- public static class BTreeSplitTask implements Serializable {
+ /** Split assigned to one sorted index build task. */
+ public static class SortedSplitTask implements Serializable {
private static final long serialVersionUID = 1L;
private int taskId;
private Split split;
- public BTreeSplitTask() {}
+ public SortedSplitTask() {}
- BTreeSplitTask(int taskId, Split split) {
+ public SortedSplitTask(int taskId, Split split) {
this.taskId = taskId;
this.split = split;
}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/CreateGlobalIndexProcedure.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/CreateGlobalIndexProcedure.java
index 1979547777..71cb07e398 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/CreateGlobalIndexProcedure.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/CreateGlobalIndexProcedure.java
@@ -18,8 +18,8 @@
package org.apache.paimon.flink.procedure;
-import org.apache.paimon.flink.btree.BTreeIndexTopoBuilder;
import org.apache.paimon.flink.globalindex.GenericIndexTopoBuilder;
+import org.apache.paimon.flink.globalindex.SortedIndexTopoBuilder;
import org.apache.paimon.globalindex.GlobalIndexer;
import org.apache.paimon.options.Options;
import org.apache.paimon.partition.PartitionPredicate;
@@ -114,7 +114,6 @@ public class CreateGlobalIndexProcedure extends
ProcedureBase {
// Parse options
Options userOptions = createUserOptions(table, options);
- // Build global index based on index type
indexType = indexType.toLowerCase().trim();
if (indexColumns.size() > 1) {
// Fail fast before submitting the job: index types that do not
support multi-column
@@ -134,15 +133,18 @@ public class CreateGlobalIndexProcedure extends
ProcedureBase {
}
}
try {
- if ("btree".equals(indexType)) {
- BTreeIndexTopoBuilder.buildIndexAndExecute(
+ if (SortedIndexTopoBuilder.supports(indexType)) {
+ SortedIndexTopoBuilder.buildIndexAndExecute(
procedureContext.getExecutionEnvironment(),
table,
indexColumns.get(0),
+ indexType,
partitionPredicate,
userOptions);
return new String[] {
- "BTree global index created successfully for table: " +
table.name()
+ displayIndexType(indexType)
+ + " global index created successfully for table: "
+ + table.name()
};
} else {
GenericIndexTopoBuilder.buildIndexAndExecute(
@@ -162,7 +164,9 @@ public class CreateGlobalIndexProcedure extends
ProcedureBase {
e);
}
return new String[] {
- indexType + " global index created successfully for table: " +
table.name()
+ displayIndexType(indexType)
+ + " global index created successfully for table: "
+ + table.name()
};
}
@@ -179,4 +183,14 @@ public class CreateGlobalIndexProcedure extends
ProcedureBase {
table.coreOptions().partitionDefaultName());
return
PartitionPredicate.fromPredicate(table.schema().logicalPartitionType(),
predicate);
}
+
+ private static String displayIndexType(String indexType) {
+ if ("btree".equals(indexType)) {
+ return "BTree";
+ }
+ if ("bitmap".equals(indexType)) {
+ return "Bitmap";
+ }
+ return indexType;
+ }
}
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/BTreeGlobalIndexITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/SortedGlobalIndexITCase.java
similarity index 87%
rename from
paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/BTreeGlobalIndexITCase.java
rename to
paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/SortedGlobalIndexITCase.java
index 9433034d39..8b730819bb 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/BTreeGlobalIndexITCase.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/SortedGlobalIndexITCase.java
@@ -36,8 +36,8 @@ import java.util.stream.IntStream;
import static org.assertj.core.api.Assertions.assertThat;
-/** Test case for btree global index. */
-public class BTreeGlobalIndexITCase extends CatalogITCaseBase {
+/** Test case for sorted global indexes. */
+public class SortedGlobalIndexITCase extends CatalogITCaseBase {
@Test
public void testBTreeIndex() throws Catalog.TableNotExistException {
@@ -71,6 +71,39 @@ public class BTreeGlobalIndexITCase extends
CatalogITCaseBase {
assertThat(sql("SELECT * FROM T WHERE id =
100")).containsOnly(Row.of(100, "name_100"));
}
+ @Test
+ public void testBitmapIndex() throws Catalog.TableNotExistException {
+ sql(
+ "CREATE TABLE T_BITMAP (id INT, name STRING) WITH ("
+ + "'global-index.enabled' = 'true', "
+ + "'row-tracking.enabled' = 'true', "
+ + "'data-evolution.enabled' = 'true'"
+ + ")");
+ String values =
+ IntStream.range(0, 1_000)
+ .mapToObj(i -> String.format("(%s, %s)", i, "'name_" +
i + "'"))
+ .collect(Collectors.joining(","));
+ sql("INSERT INTO T_BITMAP VALUES " + values);
+ sql(
+ "CALL sys.create_global_index(`table` => 'default.T_BITMAP', "
+ + "index_column => 'id', index_type => 'bitmap', "
+ + "options => 'sorted-index.records-per-range=200')");
+
+ FileStoreTable table = paimonTable("T_BITMAP");
+ List<IndexFileMeta> bitmapEntries =
+ table.store().newIndexFileHandler().scanEntries().stream()
+ .map(IndexManifestEntry::indexFile)
+ .filter(f -> "bitmap".equals(f.indexType()))
+ .collect(Collectors.toList());
+
+ long totalRowCount =
bitmapEntries.stream().mapToLong(IndexFileMeta::rowCount).sum();
+ assertThat(bitmapEntries).hasSizeGreaterThan(1);
+ assertThat(totalRowCount).isEqualTo(1000L);
+
+ assertThat(sql("SELECT * FROM T_BITMAP WHERE id = 100"))
+ .containsOnly(Row.of(100, "name_100"));
+ }
+
@Test
public void testBTreeIndexWithMultiPartition() throws
Catalog.TableNotExistException {
sql(
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/btree/BTreeIndexTopoBuilderTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/globalindex/SortedIndexTopoBuilderTest.java
similarity index 67%
rename from
paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/btree/BTreeIndexTopoBuilderTest.java
rename to
paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/globalindex/SortedIndexTopoBuilderTest.java
index 25260ee5c4..89a15bef31 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/btree/BTreeIndexTopoBuilderTest.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/globalindex/SortedIndexTopoBuilderTest.java
@@ -16,10 +16,10 @@
* limitations under the License.
*/
-package org.apache.paimon.flink.btree;
+package org.apache.paimon.flink.globalindex;
-import org.apache.paimon.flink.btree.BTreeIndexTopoBuilder.BTreeBuildTask;
-import org.apache.paimon.globalindex.btree.BTreeGlobalIndexBuilder;
+import
org.apache.paimon.flink.globalindex.SortedIndexTopoBuilder.SortedBuildTask;
+import org.apache.paimon.globalindex.sorted.SortedGlobalIndexBuilder;
import org.apache.paimon.options.Options;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.utils.Range;
@@ -37,18 +37,18 @@ import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verifyNoInteractions;
import static org.mockito.Mockito.when;
-/** Tests for {@link BTreeIndexTopoBuilder}. */
-public class BTreeIndexTopoBuilderTest {
+/** Tests for {@link SortedIndexTopoBuilder}. */
+public class SortedIndexTopoBuilderTest {
@Test
public void testBuildIndexReturnsFalseWhenNoBuildTask() throws Exception {
- BTreeGlobalIndexBuilder indexBuilder =
mock(BTreeGlobalIndexBuilder.class);
+ SortedGlobalIndexBuilder indexBuilder =
mock(SortedGlobalIndexBuilder.class);
when(indexBuilder.withIndexField("id")).thenReturn(indexBuilder);
when(indexBuilder.scan()).thenReturn(Optional.empty());
StreamExecutionEnvironment env =
mock(StreamExecutionEnvironment.class);
assertThat(
- BTreeIndexTopoBuilder.buildIndex(
+ SortedIndexTopoBuilder.buildIndex(
env,
() -> indexBuilder,
mock(FileStoreTable.class),
@@ -61,29 +61,29 @@ public class BTreeIndexTopoBuilderTest {
@Test
public void testCalculateParallelismByTotalRowsInsteadOfRangeCount() {
- List<BTreeBuildTask> tasks = new ArrayList<>();
+ List<SortedBuildTask> tasks = new ArrayList<>();
for (int i = 0; i < 100; i++) {
- tasks.add(new BTreeBuildTask(i, new Range(i * 10L, i * 10L + 9),
new byte[0]));
+ tasks.add(new SortedBuildTask(i, new Range(i * 10L, i * 10L + 9),
new byte[0]));
}
- assertThat(BTreeIndexTopoBuilder.calculateParallelism(tasks, 1000L,
4096)).isEqualTo(1);
+ assertThat(SortedIndexTopoBuilder.calculateParallelism(tasks, 1000L,
4096)).isEqualTo(1);
}
@Test
public void testCalculateParallelismHonorsMaxParallelism() {
- List<BTreeBuildTask> tasks = new ArrayList<>();
+ List<SortedBuildTask> tasks = new ArrayList<>();
for (int i = 0; i < 100; i++) {
- tasks.add(new BTreeBuildTask(i, new Range(i * 1000L, i * 1000L +
999), new byte[0]));
+ tasks.add(new SortedBuildTask(i, new Range(i * 1000L, i * 1000L +
999), new byte[0]));
}
- assertThat(BTreeIndexTopoBuilder.calculateParallelism(tasks, 1000L,
16)).isEqualTo(16);
+ assertThat(SortedIndexTopoBuilder.calculateParallelism(tasks, 1000L,
16)).isEqualTo(16);
}
@Test
public void testCalculateParallelismKeepsSingleRangeBehavior() {
- List<BTreeBuildTask> tasks = new ArrayList<>();
- tasks.add(new BTreeBuildTask(0, new Range(0, 1499), new byte[0]));
+ List<SortedBuildTask> tasks = new ArrayList<>();
+ tasks.add(new SortedBuildTask(0, new Range(0, 1499), new byte[0]));
- assertThat(BTreeIndexTopoBuilder.calculateParallelism(tasks, 1000L,
16)).isEqualTo(1);
+ assertThat(SortedIndexTopoBuilder.calculateParallelism(tasks, 1000L,
16)).isEqualTo(1);
}
}
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/CreateGlobalIndexProcedureTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/CreateGlobalIndexProcedureTest.java
index 5b879628b2..025623bcec 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/CreateGlobalIndexProcedureTest.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/CreateGlobalIndexProcedureTest.java
@@ -19,6 +19,7 @@
package org.apache.paimon.flink.procedure;
import org.apache.paimon.globalindex.btree.BTreeIndexOptions;
+import org.apache.paimon.globalindex.sorted.SortedIndexOptions;
import org.apache.paimon.options.Options;
import org.junit.jupiter.api.Test;
@@ -41,11 +42,11 @@ public class CreateGlobalIndexProcedureTest {
Options userOptions =
CreateGlobalIndexProcedure.createUserOptions(
tableOptions,
- BTreeIndexOptions.BTREE_INDEX_RECORDS_PER_RANGE.key()
+ SortedIndexOptions.SORTED_INDEX_RECORDS_PER_RANGE.key()
+ "=200;procedure-only=procedure-value");
assertThat(userOptions.get(BTreeIndexOptions.BTREE_INDEX_COMPRESSION)).isEqualTo("zstd");
-
assertThat(userOptions.get(BTreeIndexOptions.BTREE_INDEX_RECORDS_PER_RANGE))
+
assertThat(userOptions.get(SortedIndexOptions.SORTED_INDEX_RECORDS_PER_RANGE))
.isEqualTo(200L);
assertThat(userOptions.get("unrelated-table-option")).isEqualTo("table-value");
assertThat(userOptions.get("procedure-only")).isEqualTo("procedure-value");
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/DefaultGlobalIndexTopoBuilder.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/DefaultGlobalIndexTopoBuilder.java
index 4671006929..b2112149a1 100644
---
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/DefaultGlobalIndexTopoBuilder.java
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/DefaultGlobalIndexTopoBuilder.java
@@ -63,11 +63,6 @@ import static
org.apache.paimon.utils.Preconditions.checkArgument;
/** Default topology builder. */
public class DefaultGlobalIndexTopoBuilder implements
GlobalIndexTopologyBuilder {
- @Override
- public String identifier() {
- return "default";
- }
-
@Override
public List<CommitMessage> buildIndex(
SparkSession spark,
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/GlobalIndexTopologyBuilder.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/GlobalIndexTopologyBuilder.java
index d7a47cfdc9..74fcec67f6 100644
---
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/GlobalIndexTopologyBuilder.java
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/GlobalIndexTopologyBuilder.java
@@ -31,11 +31,9 @@ import
org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation;
import java.io.IOException;
import java.util.List;
-/** User defined topology builder. */
+/** Topology builder for Spark global index creation. */
public interface GlobalIndexTopologyBuilder {
- String identifier();
-
List<CommitMessage> buildIndex(
SparkSession spark,
DataSourceV2Relation relation,
@@ -62,7 +60,7 @@ public interface GlobalIndexTopologyBuilder {
throw new UnsupportedOperationException(
String.format(
"Topology builder '%s' does not support
multi-column index, got extra columns: %s",
- identifier(), extraFields));
+ getClass().getSimpleName(), extraFields));
}
return buildIndex(
spark,
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/GlobalIndexTopologyBuilderUtils.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/GlobalIndexTopologyBuilderUtils.java
index 022e6607eb..405438c593 100644
---
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/GlobalIndexTopologyBuilderUtils.java
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/GlobalIndexTopologyBuilderUtils.java
@@ -18,43 +18,15 @@
package org.apache.paimon.spark.globalindex;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import org.apache.paimon.spark.globalindex.sorted.SortedIndexTopoBuilder;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.ServiceLoader;
-
-/**
- * Utility class for loading {@link GlobalIndexTopologyBuilder}
implementations via Java's {@link
- * ServiceLoader} mechanism.
- *
- * <p>Factories are loaded once during class initialization and cached for
subsequent lookups.
- */
+/** Utility class for creating {@link GlobalIndexTopologyBuilder}
implementations. */
public class GlobalIndexTopologyBuilderUtils {
- private static final Logger LOG =
- LoggerFactory.getLogger(GlobalIndexTopologyBuilderUtils.class);
-
- private static final Map<String, GlobalIndexTopologyBuilder> FACTORIES =
new HashMap<>();
-
- static {
- ServiceLoader<GlobalIndexTopologyBuilder> serviceLoader =
- ServiceLoader.load(GlobalIndexTopologyBuilder.class);
-
- for (GlobalIndexTopologyBuilder builder : serviceLoader) {
- String identifier = builder.identifier();
- if (FACTORIES.put(identifier, builder) != null) {
- LOG.warn(
- "Found multiple GlobalIndexBuilderFactory
implementations for type '{}'. "
- + "Using the last one loaded.",
- identifier);
- }
- }
- }
-
public static GlobalIndexTopologyBuilder createTopoBuilder(String
indexType) {
- GlobalIndexTopologyBuilder builder = FACTORIES.get(indexType);
- return builder == null ? new DefaultGlobalIndexTopoBuilder() : builder;
+ if (SortedIndexTopoBuilder.supports(indexType)) {
+ return new SortedIndexTopoBuilder();
+ }
+ return new DefaultGlobalIndexTopoBuilder();
}
}
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/btree/BTreeIndexTopoBuilder.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/sorted/SortedIndexTopoBuilder.java
similarity index 83%
rename from
paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/btree/BTreeIndexTopoBuilder.java
rename to
paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/sorted/SortedIndexTopoBuilder.java
index 5f7bfa453e..2174418009 100644
---
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/btree/BTreeIndexTopoBuilder.java
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/sorted/SortedIndexTopoBuilder.java
@@ -16,13 +16,13 @@
* limitations under the License.
*/
-package org.apache.paimon.spark.globalindex.btree;
+package org.apache.paimon.spark.globalindex.sorted;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.data.serializer.BinaryRowSerializer;
-import org.apache.paimon.globalindex.btree.BTreeGlobalIndexBuilder;
-import org.apache.paimon.globalindex.btree.BTreeIndexOptions;
+import org.apache.paimon.globalindex.sorted.SortedGlobalIndexBuilder;
+import org.apache.paimon.globalindex.sorted.SortedIndexOptions;
import org.apache.paimon.options.Options;
import org.apache.paimon.partition.PartitionPredicate;
import org.apache.paimon.spark.SparkRow;
@@ -52,21 +52,25 @@ import org.apache.spark.sql.functions;
import java.io.IOException;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collections;
+import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Optional;
-import static
org.apache.paimon.globalindex.btree.BTreeGlobalIndexBuilder.groupSplitsByRange;
-import static
org.apache.paimon.globalindex.btree.BTreeGlobalIndexBuilder.splitByContiguousRowRange;
+import static
org.apache.paimon.globalindex.sorted.SortedGlobalIndexBuilder.groupSplitsByRange;
+import static
org.apache.paimon.globalindex.sorted.SortedGlobalIndexBuilder.splitByContiguousRowRange;
-/** The {@link GlobalIndexTopologyBuilder} for BTree index. */
-public class BTreeIndexTopoBuilder implements GlobalIndexTopologyBuilder {
+/** The {@link GlobalIndexTopologyBuilder} for sorted indexes. */
+public class SortedIndexTopoBuilder implements GlobalIndexTopologyBuilder {
- @Override
- public String identifier() {
- return "btree";
+ private static final HashSet<String> SUPPORTED_INDEX_TYPES =
+ new HashSet<>(Arrays.asList("btree", "bitmap"));
+
+ public static boolean supports(String indexType) {
+ return SUPPORTED_INDEX_TYPES.contains(indexType);
}
@Override
@@ -80,9 +84,9 @@ public class BTreeIndexTopoBuilder implements
GlobalIndexTopologyBuilder {
DataField indexField,
Options options)
throws IOException {
- // 1. read the whole dataset of target partitions
- BTreeGlobalIndexBuilder indexBuilder =
- new
BTreeGlobalIndexBuilder(table).withIndexField(indexField.name());
+ SortedGlobalIndexBuilder indexBuilder =
+ new SortedGlobalIndexBuilder(table, indexType, options)
+ .withIndexField(indexField.name());
if (partitionPredicate != null) {
indexBuilder =
indexBuilder.withPartitionPredicate(partitionPredicate);
}
@@ -107,8 +111,8 @@ public class BTreeIndexTopoBuilder implements
GlobalIndexTopologyBuilder {
List<String> selectedColumns = new
ArrayList<>(readType.getFieldNames());
// Calculate maximum parallelism bound
- long recordsPerRange =
options.get(BTreeIndexOptions.BTREE_INDEX_RECORDS_PER_RANGE);
- int maxParallelism =
options.get(BTreeIndexOptions.BTREE_INDEX_BUILD_MAX_PARALLELISM);
+ long recordsPerRange =
options.get(SortedIndexOptions.SORTED_INDEX_RECORDS_PER_RANGE);
+ int maxParallelism =
options.get(SortedIndexOptions.SORTED_INDEX_BUILD_MAX_PARALLELISM);
List<CommitMessage> allMessages = new ArrayList<>();
List<String> sortColumns = new ArrayList<>();
@@ -155,7 +159,7 @@ public class BTreeIndexTopoBuilder implements
GlobalIndexTopologyBuilder {
.mapPartitions(
(FlatMapFunction<Iterator<InternalRow>, byte[]>)
iter ->
- buildBTreeIndex(
+ buildSortedIndex(
iter,
serializedBuilder,
range,
@@ -168,7 +172,7 @@ public class BTreeIndexTopoBuilder implements
GlobalIndexTopologyBuilder {
return allMessages;
}
- private static Iterator<byte[]> buildBTreeIndex(
+ private static Iterator<byte[]> buildSortedIndex(
Iterator<InternalRow> input,
byte[] serializedBuilder,
Range range,
@@ -177,9 +181,9 @@ public class BTreeIndexTopoBuilder implements
GlobalIndexTopologyBuilder {
throws IOException, ClassNotFoundException {
final BinaryRowSerializer binaryRowSerializer = new
BinaryRowSerializer(partitionKeyNum);
BinaryRow partition =
binaryRowSerializer.deserializeFromBytes(partitionBytes);
- BTreeGlobalIndexBuilder builder =
+ SortedGlobalIndexBuilder builder =
InstantiationUtil.deserializeObject(
- serializedBuilder,
BTreeGlobalIndexBuilder.class.getClassLoader());
+ serializedBuilder,
SortedGlobalIndexBuilder.class.getClassLoader());
return CommitMessageSerializer.serializeAll(
builder.buildForSinglePartition(range, partition,
input))
.iterator();
diff --git
a/paimon-spark/paimon-spark-common/src/main/resources/META-INF/services/org.apache.paimon.spark.globalindex.GlobalIndexTopologyBuilder
b/paimon-spark/paimon-spark-common/src/main/resources/META-INF/services/org.apache.paimon.spark.globalindex.GlobalIndexTopologyBuilder
deleted file mode 100644
index 07a41cb47a..0000000000
---
a/paimon-spark/paimon-spark-common/src/main/resources/META-INF/services/org.apache.paimon.spark.globalindex.GlobalIndexTopologyBuilder
+++ /dev/null
@@ -1,16 +0,0 @@
-# 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.
-
-org.apache.paimon.spark.globalindex.btree.BTreeIndexTopoBuilder
diff --git
a/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/procedure/CreateGlobalIndexProcedureTest.java
b/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/procedure/CreateGlobalIndexProcedureTest.java
index 0a13f06197..991f5d8627 100644
---
a/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/procedure/CreateGlobalIndexProcedureTest.java
+++
b/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/procedure/CreateGlobalIndexProcedureTest.java
@@ -23,6 +23,7 @@ import org.apache.paimon.data.BinaryRowWriter;
import org.apache.paimon.fs.Path;
import org.apache.paimon.globalindex.IndexedSplit;
import org.apache.paimon.globalindex.btree.BTreeIndexOptions;
+import org.apache.paimon.globalindex.sorted.SortedIndexOptions;
import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.io.PojoDataFileMeta;
import org.apache.paimon.manifest.FileKind;
@@ -63,11 +64,11 @@ public class CreateGlobalIndexProcedureTest {
Options userOptions =
CreateGlobalIndexProcedure.createUserOptions(
tableOptions,
- BTreeIndexOptions.BTREE_INDEX_RECORDS_PER_RANGE.key()
+ SortedIndexOptions.SORTED_INDEX_RECORDS_PER_RANGE.key()
+ "=200, procedure-only=procedure-value");
assertThat(userOptions.get(BTreeIndexOptions.BTREE_INDEX_COMPRESSION)).isEqualTo("zstd");
-
assertThat(userOptions.get(BTreeIndexOptions.BTREE_INDEX_RECORDS_PER_RANGE))
+
assertThat(userOptions.get(SortedIndexOptions.SORTED_INDEX_RECORDS_PER_RANGE))
.isEqualTo(200L);
assertThat(userOptions.get("unrelated-table-option")).isEqualTo("table-value");
assertThat(userOptions.get("procedure-only")).isEqualTo("procedure-value");
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/CreateGlobalIndexProcedureTest.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/CreateGlobalIndexProcedureTest.scala
index 9a28029f1c..9182ff70d5 100644
---
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/CreateGlobalIndexProcedureTest.scala
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/CreateGlobalIndexProcedureTest.scala
@@ -110,6 +110,44 @@ class CreateGlobalIndexProcedureTest extends
PaimonSparkTestBase with StreamTest
}
}
+ test("create bitmap global index") {
+ withTable("T") {
+ spark.sql("""
+ |CREATE TABLE T (id INT, name STRING)
+ |TBLPROPERTIES (
+ | 'bucket' = '-1',
+ | 'global-index.row-count-per-shard' = '10000',
+ | 'row-tracking.enabled' = 'true',
+ | 'data-evolution.enabled' = 'true')
+ |""".stripMargin)
+
+ val values =
+ (0 until 10000).map(i => s"($i, 'name_$i')").mkString(",")
+ spark.sql(s"INSERT INTO T VALUES $values")
+
+ val output =
+ spark
+ .sql(
+ "CALL sys.create_global_index(table => 'test.T', index_column =>
'name', index_type => 'bitmap'," +
+ " options => 'sorted-index.records-per-range=1000')")
+ .collect()
+ .head
+
+ assert(output.getBoolean(0))
+ val table = loadTable("T")
+ val bitmapEntries = table
+ .store()
+ .newIndexFileHandler()
+ .scanEntries()
+ .asScala
+ .filter(_.indexFile().indexType() == "bitmap")
+ .map(_.indexFile())
+ assert(bitmapEntries.nonEmpty)
+ assert(bitmapEntries.map(_.rowCount()).sum == 10000L)
+ bitmapEntries.foreach(e => assert(e.globalIndexMeta() != null))
+ }
+ }
+
test("create btree global index with multiple partitions") {
withTable("T") {
spark.sql("""
