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 f31f199ebc [core] Define scalar primary-key global indexes (#8599)
f31f199ebc is described below
commit f31f199ebc25585a25079bf53708a5c2aa0e8a5b
Author: Jingsong Lee <[email protected]>
AuthorDate: Mon Jul 13 19:46:06 2026 +0800
[core] Define scalar primary-key global indexes (#8599)
Define the configuration and schema contract for source-backed scalar
primary-key indexes, preparing BTree and Bitmap indexes for the later
build, write, and scan changes.
---
docs/generated/core_configuration.html | 12 ++
.../main/java/org/apache/paimon/CoreOptions.java | 78 ++++++++-
.../paimon/index/pk/PrimaryKeyIndexDefinition.java | 67 ++++++++
.../index/pk/PrimaryKeyIndexDefinitions.java | 118 +++++++++++++
.../org/apache/paimon/schema/SchemaManager.java | 17 +-
.../org/apache/paimon/schema/SchemaValidation.java | 100 +++++++++++
.../index/pk/PrimaryKeyIndexDefinitionsTest.java | 104 +++++++++++
.../pksorted/PrimaryKeySortedIndexOptionsTest.java | 78 +++++++++
.../PrimaryKeySortedIndexValidationTest.java | 191 +++++++++++++++++++++
.../apache/paimon/schema/SchemaManagerTest.java | 79 ++++++++-
10 files changed, 837 insertions(+), 7 deletions(-)
diff --git a/docs/generated/core_configuration.html
b/docs/generated/core_configuration.html
index 5d46fda800..cc8ceaaec3 100644
--- a/docs/generated/core_configuration.html
+++ b/docs/generated/core_configuration.html
@@ -1199,6 +1199,18 @@ This config option does not affect the default
filesystem metastore.</td>
<td>String</td>
<td>You can specify a pattern to get a timestamp from partitions.
The formatter pattern is defined by 'partition.timestamp-formatter'.<ul><li>By
default, read from the first field.</li><li>If the timestamp in the partition
is a single field called 'dt', you can use '$dt'.</li><li>If it is spread
across multiple fields for year, month, day, and hour, you can use
'$year-$month-$day $hour:00:00'.</li><li>If the timestamp is in fields dt and
hour, you can use '$dt $hour:00:00'.</ [...]
</tr>
+ <tr>
+ <td><h5>pk-bitmap.index.columns</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>Comma-separated columns indexed by primary-key Bitmap
indexes.</td>
+ </tr>
+ <tr>
+ <td><h5>pk-btree.index.columns</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>Comma-separated columns indexed by primary-key BTree
indexes.</td>
+ </tr>
<tr>
<td><h5>pk-clustering-override</h5></td>
<td style="word-wrap: break-word;">false</td>
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 e4f43582a1..9130e4bc03 100644
--- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
+++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
@@ -2741,6 +2741,20 @@ public class CoreOptions implements Serializable {
+ "metric are also field-scoped. The first
release supports exactly "
+ "one column.");
+ public static final ConfigOption<String> PK_BTREE_INDEX_COLUMNS =
+ key("pk-btree.index.columns")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "Comma-separated columns indexed by primary-key
BTree indexes.");
+
+ public static final ConfigOption<String> PK_BITMAP_INDEX_COLUMNS =
+ key("pk-bitmap.index.columns")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "Comma-separated columns indexed by primary-key
Bitmap indexes.");
+
public static final ConfigOption<Integer>
PK_VECTOR_INDEX_COMPACTION_LEVEL_FANOUT =
key("pk-vector.index.compaction.level-fanout")
.intType()
@@ -4296,13 +4310,75 @@ public class CoreOptions implements Serializable {
}
public List<String> primaryKeyVectorIndexColumns() {
- String columns = options.get(PK_VECTOR_INDEX_COLUMNS);
+ return primaryKeyIndexColumns(PK_VECTOR_INDEX_COLUMNS);
+ }
+
+ public List<String> primaryKeyBTreeIndexColumns() {
+ return primaryKeyIndexColumns(PK_BTREE_INDEX_COLUMNS);
+ }
+
+ public List<String> primaryKeyBitmapIndexColumns() {
+ return primaryKeyIndexColumns(PK_BITMAP_INDEX_COLUMNS);
+ }
+
+ private List<String> primaryKeyIndexColumns(ConfigOption<String> option) {
+ String columns = options.get(option);
if (columns == null) {
return Collections.emptyList();
}
return Arrays.stream(columns.split(",",
-1)).map(String::trim).collect(Collectors.toList());
}
+ public Options primaryKeyBTreeIndexOptions(String column) {
+ return primaryKeySortedIndexOptions(column, "pk-btree",
"btree-index.");
+ }
+
+ public Options primaryKeyBitmapIndexOptions(String column) {
+ return primaryKeySortedIndexOptions(column, "pk-bitmap",
"bitmap-index.");
+ }
+
+ private Options primaryKeySortedIndexOptions(
+ String column, String optionFamily, String algorithmPrefix) {
+ Options resolved = new Options(toConfiguration().toMap());
+ String optionKey = "fields." + column + "." + optionFamily +
".index.options";
+ String serialized = options.get(optionKey);
+ if (serialized == null || serialized.trim().isEmpty()) {
+ return resolved;
+ }
+
+ LinkedHashMap<String, String> parsed;
+ try {
+ parsed = JsonSerdeUtil.parseJsonMap(serialized, String.class);
+ } catch (RuntimeException e) {
+ throw new IllegalArgumentException(
+ optionKey + " must be a JSON object of option key-value
pairs.", e);
+ }
+
+ for (Map.Entry<String, String> entry : parsed.entrySet()) {
+ String key = entry.getKey();
+ String value = entry.getValue();
+ checkArgument(
+ key != null && !key.trim().isEmpty(),
+ "%s contains an empty option key.",
+ optionKey);
+ checkArgument(value != null, "%s value for key %s must not be
null.", optionKey, key);
+ String qualifiedKey =
+ key.startsWith(algorithmPrefix)
+ || key.startsWith("sorted-index.")
+ || key.startsWith("fields.")
+ ? key
+ : algorithmPrefix + key;
+ String previous = resolved.get(qualifiedKey);
+ checkArgument(
+ previous == null || previous.equals(value),
+ "%s defines conflicting values for %s.",
+ optionKey,
+ qualifiedKey);
+ resolved.setString(qualifiedKey, value);
+ }
+ return resolved;
+ }
+
public String primaryKeyVectorIndexColumn() {
List<String> columns = primaryKeyVectorIndexColumns();
checkArgument(
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
new file mode 100644
index 0000000000..76f7b3a2ef
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/index/pk/PrimaryKeyIndexDefinition.java
@@ -0,0 +1,67 @@
+/*
+ * 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.pk;
+
+import org.apache.paimon.options.Options;
+
+/** Resolved definition of one source-backed primary-key index. */
+public class PrimaryKeyIndexDefinition {
+
+ /** Built-in primary-key index families. */
+ public enum Family {
+ VECTOR,
+ BTREE,
+ BITMAP
+ }
+
+ private final String column;
+ private final int fieldId;
+ private final String indexType;
+ private final Options options;
+ private final Family family;
+
+ public PrimaryKeyIndexDefinition(
+ String column, int fieldId, String indexType, Options options,
Family family) {
+ this.column = column;
+ this.fieldId = fieldId;
+ this.indexType = indexType;
+ this.options = options;
+ this.family = family;
+ }
+
+ public String column() {
+ return column;
+ }
+
+ public int fieldId() {
+ return fieldId;
+ }
+
+ public String indexType() {
+ return indexType;
+ }
+
+ public Options options() {
+ return options;
+ }
+
+ public Family family() {
+ return family;
+ }
+}
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
new file mode 100644
index 0000000000..16f2fdaa3a
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/index/pk/PrimaryKeyIndexDefinitions.java
@@ -0,0 +1,118 @@
+/*
+ * 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.pk;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.globalindex.bitmap.BitmapGlobalIndexerFactory;
+import org.apache.paimon.globalindex.btree.BTreeGlobalIndexerFactory;
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.types.DataField;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/** Resolves all configured source-backed primary-key indexes. */
+public class PrimaryKeyIndexDefinitions {
+
+ private final List<PrimaryKeyIndexDefinition> definitions;
+
+ private PrimaryKeyIndexDefinitions(List<PrimaryKeyIndexDefinition>
definitions) {
+ this.definitions = Collections.unmodifiableList(definitions);
+ }
+
+ public static PrimaryKeyIndexDefinitions create(TableSchema schema) {
+ CoreOptions options = new CoreOptions(schema.options());
+ List<String> vectorColumns = options.primaryKeyVectorIndexColumns();
+ List<String> btreeColumns = options.primaryKeyBTreeIndexColumns();
+ List<String> bitmapColumns = options.primaryKeyBitmapIndexColumns();
+ validateNoDuplicates(vectorColumns,
CoreOptions.PK_VECTOR_INDEX_COLUMNS.key());
+ validateNoDuplicates(btreeColumns,
CoreOptions.PK_BTREE_INDEX_COLUMNS.key());
+ validateNoDuplicates(bitmapColumns,
CoreOptions.PK_BITMAP_INDEX_COLUMNS.key());
+ validateOneIndexPerColumn(vectorColumns, btreeColumns, bitmapColumns);
+ List<PrimaryKeyIndexDefinition> definitions = new ArrayList<>();
+
+ for (DataField field : schema.fields()) {
+ String column = field.name();
+ if (btreeColumns.contains(column)) {
+ definitions.add(
+ new PrimaryKeyIndexDefinition(
+ column,
+ field.id(),
+ BTreeGlobalIndexerFactory.IDENTIFIER,
+ options.primaryKeyBTreeIndexOptions(column),
+ PrimaryKeyIndexDefinition.Family.BTREE));
+ } else if (bitmapColumns.contains(column)) {
+ definitions.add(
+ new PrimaryKeyIndexDefinition(
+ column,
+ field.id(),
+ BitmapGlobalIndexerFactory.IDENTIFIER,
+ options.primaryKeyBitmapIndexOptions(column),
+ PrimaryKeyIndexDefinition.Family.BITMAP));
+ } else if (vectorColumns.contains(column)) {
+ definitions.add(
+ new PrimaryKeyIndexDefinition(
+ column,
+ field.id(),
+ options.primaryKeyVectorIndexType(column),
+ options.primaryKeyVectorIndexOptions(column),
+ PrimaryKeyIndexDefinition.Family.VECTOR));
+ }
+ }
+
+ return new PrimaryKeyIndexDefinitions(definitions);
+ }
+
+ private static void validateNoDuplicates(List<String> columns, String
optionKey) {
+ Set<String> uniqueColumns = new HashSet<>();
+ for (String column : columns) {
+ checkArgument(
+ uniqueColumns.add(column),
+ "%s contains duplicate column '%s'.",
+ optionKey,
+ column);
+ }
+ }
+
+ private static void validateOneIndexPerColumn(
+ List<String> vectorColumns, List<String> btreeColumns,
List<String> bitmapColumns) {
+ Set<String> indexedColumns = new HashSet<>();
+ validateUniqueColumns(indexedColumns, vectorColumns);
+ validateUniqueColumns(indexedColumns, btreeColumns);
+ validateUniqueColumns(indexedColumns, bitmapColumns);
+ }
+
+ private static void validateUniqueColumns(Set<String> indexedColumns,
List<String> columns) {
+ for (String column : columns) {
+ checkArgument(
+ indexedColumns.add(column),
+ "Column '%s' can own at most one primary-key index.",
+ column);
+ }
+ }
+
+ public List<PrimaryKeyIndexDefinition> definitions() {
+ return definitions;
+ }
+}
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 6db4136f1b..10d73eeb9d 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
@@ -439,7 +439,8 @@ public class SchemaManager implements Serializable {
} else if (change instanceof RenameColumn) {
RenameColumn rename = (RenameColumn) change;
assertNotUpdatingPartitionKeys(oldTableSchema,
rename.fieldNames(), "rename");
- assertNotRenamingPrimaryKeyVectorIndexColumn(oldTableSchema,
rename.fieldNames());
+ assertNotUpdatingPrimaryKeyIndexColumn(
+ oldTableSchema, rename.fieldNames(), "rename");
assertNotRenamingBlobColumn(newFields, rename.fieldNames());
new NestedColumnModifier(rename.fieldNames(), lazyIdentifier) {
@Override
@@ -496,6 +497,8 @@ public class SchemaManager implements Serializable {
UpdateColumnType update = (UpdateColumnType) change;
assertNotUpdatingPartitionKeys(oldTableSchema,
update.fieldNames(), "update");
assertNotUpdatingPrimaryKeys(oldTableSchema,
update.fieldNames(), "update");
+ assertNotUpdatingPrimaryKeyIndexColumn(
+ oldTableSchema, update.fieldNames(), "update type of");
assertNotChangingBlobColumnType(
newFields, update.fieldNames(), update.newDataType());
updateNestedColumn(
@@ -943,6 +946,7 @@ public class SchemaManager implements Serializable {
throw new UnsupportedOperationException(
String.format("Cannot drop partition key or primary key:
[%s]", columnToDrop));
}
+ assertNotUpdatingPrimaryKeyIndexColumn(schema, change.fieldNames(),
"drop");
}
private static void assertNotUpdatingPartitionKeys(
@@ -984,16 +988,19 @@ public class SchemaManager implements Serializable {
}
}
- private static void assertNotRenamingPrimaryKeyVectorIndexColumn(
- TableSchema schema, String[] fieldNames) {
+ private static void assertNotUpdatingPrimaryKeyIndexColumn(
+ TableSchema schema, String[] fieldNames, String operation) {
if (fieldNames.length > 1) {
return;
}
String fieldName = fieldNames[0];
- if (new
CoreOptions(schema.options()).primaryKeyVectorIndexColumns().contains(fieldName))
{
+ CoreOptions options = new CoreOptions(schema.options());
+ if (options.primaryKeyVectorIndexColumns().contains(fieldName)
+ || options.primaryKeyBTreeIndexColumns().contains(fieldName)
+ || options.primaryKeyBitmapIndexColumns().contains(fieldName))
{
throw new UnsupportedOperationException(
String.format(
- "Cannot rename primary-key vector index column:
[%s]", fieldName));
+ "Cannot %s primary-key index column: [%s]",
operation, fieldName));
}
}
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 746338314d..49dfcb9b06 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
@@ -29,6 +29,9 @@ import org.apache.paimon.fileindex.FileIndexOptions;
import org.apache.paimon.fileindex.FileIndexerFactory;
import org.apache.paimon.fileindex.FileIndexerFactoryUtils;
import org.apache.paimon.format.FileFormat;
+import org.apache.paimon.globalindex.GlobalIndexer;
+import org.apache.paimon.globalindex.bitmap.BitmapGlobalIndexerFactory;
+import org.apache.paimon.globalindex.btree.BTreeGlobalIndexerFactory;
import org.apache.paimon.mergetree.compact.aggregate.FieldAggregator;
import
org.apache.paimon.mergetree.compact.aggregate.factory.FieldAggregatorFactory;
import org.apache.paimon.options.ConfigOption;
@@ -350,6 +353,8 @@ public class SchemaValidation {
fieldNamesSpecifiedAsVector.isEmpty(),
"Some of the columns specified as vector-field are unknown.");
+ validatePrimaryKeyIndexColumns(options);
+ validatePrimaryKeySortedIndexes(schema, options);
validatePrimaryKeyVectorIndex(schema, options);
validateMergeFunctionFactory(schema);
@@ -960,6 +965,101 @@ public class SchemaValidation {
options.primaryKeyVectorDistanceMetric(indexColumn));
}
+ private static void validatePrimaryKeyIndexColumns(CoreOptions options) {
+ List<String> vectorColumns = options.primaryKeyVectorIndexColumns();
+ List<String> btreeColumns = options.primaryKeyBTreeIndexColumns();
+ List<String> bitmapColumns = options.primaryKeyBitmapIndexColumns();
+ validateNoDuplicatePrimaryKeyIndexColumns(
+ vectorColumns, CoreOptions.PK_VECTOR_INDEX_COLUMNS.key());
+ validateNoDuplicatePrimaryKeyIndexColumns(
+ btreeColumns, CoreOptions.PK_BTREE_INDEX_COLUMNS.key());
+ validateNoDuplicatePrimaryKeyIndexColumns(
+ bitmapColumns, CoreOptions.PK_BITMAP_INDEX_COLUMNS.key());
+
+ Set<String> indexedColumns = new HashSet<>();
+ validateUniquePrimaryKeyIndexColumns(indexedColumns, vectorColumns);
+ validateUniquePrimaryKeyIndexColumns(indexedColumns, btreeColumns);
+ validateUniquePrimaryKeyIndexColumns(indexedColumns, bitmapColumns);
+ }
+
+ private static void validateNoDuplicatePrimaryKeyIndexColumns(
+ List<String> columns, String optionKey) {
+ checkArgument(
+ new HashSet<>(columns).size() == columns.size(),
+ "%s must not contain duplicate columns, but is %s.",
+ optionKey,
+ columns);
+ }
+
+ private static void validateUniquePrimaryKeyIndexColumns(
+ Set<String> indexedColumns, List<String> columns) {
+ for (String column : columns) {
+ checkArgument(
+ indexedColumns.add(column),
+ "Column '%s' can own at most one primary-key index.",
+ column);
+ }
+ }
+
+ private static void validatePrimaryKeySortedIndexes(TableSchema schema,
CoreOptions options) {
+ if (options.primaryKeyBTreeIndexColumns().isEmpty()
+ && options.primaryKeyBitmapIndexColumns().isEmpty()) {
+ return;
+ }
+
+ checkArgument(
+ options.deletionVectorsEnabled(),
+ "Primary-key BTree and Bitmap indexes require
deletion-vectors.enabled = true.");
+ checkArgument(
+ !schema.primaryKeys().isEmpty(),
+ "Primary-key BTree and Bitmap indexes require a primary-key
table.");
+ checkArgument(
+ options.bucket() > 0 || options.bucket() ==
BucketMode.POSTPONE_BUCKET,
+ "Primary-key BTree and Bitmap indexes require fixed or
postpone bucket mode "
+ + "(bucket > 0 or bucket = -2), but bucket is %s.",
+ options.bucket());
+ checkArgument(
+ !options.deletionVectorsMergeOnRead(),
+ "Primary-key BTree and Bitmap indexes require
deletion-vectors.merge-on-read = false.");
+ checkArgument(
+ !options.pkClusteringOverride(),
+ "Primary-key BTree and Bitmap indexes do not support
pk-clustering-override.");
+
+ validatePrimaryKeySortedIndexColumns(
+ schema,
+ options.primaryKeyBTreeIndexColumns(),
+ CoreOptions.PK_BTREE_INDEX_COLUMNS.key());
+ validatePrimaryKeySortedIndexColumns(
+ schema,
+ options.primaryKeyBitmapIndexColumns(),
+ CoreOptions.PK_BITMAP_INDEX_COLUMNS.key());
+
+ Map<String, DataField> fields = schema.nameToFieldMap();
+ for (String column : options.primaryKeyBTreeIndexColumns()) {
+ GlobalIndexer.create(
+ BTreeGlobalIndexerFactory.IDENTIFIER,
+ fields.get(column),
+ options.primaryKeyBTreeIndexOptions(column));
+ }
+ for (String column : options.primaryKeyBitmapIndexColumns()) {
+ GlobalIndexer.create(
+ BitmapGlobalIndexerFactory.IDENTIFIER,
+ fields.get(column),
+ options.primaryKeyBitmapIndexOptions(column));
+ }
+ }
+
+ private static void validatePrimaryKeySortedIndexColumns(
+ TableSchema schema, List<String> columns, String optionKey) {
+ for (String column : columns) {
+ checkArgument(
+ schema.fieldNames().contains(column),
+ "%s entry '%s' must reference an existing column.",
+ optionKey,
+ column);
+ }
+ }
+
private static void validateSequenceField(TableSchema schema, CoreOptions
options) {
List<String> sequenceField = options.sequenceField();
if (!sequenceField.isEmpty()) {
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
new file mode 100644
index 0000000000..b5d7875adc
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/index/pk/PrimaryKeyIndexDefinitionsTest.java
@@ -0,0 +1,104 @@
+/*
+ * 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.pk;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.schema.TableSchema;
+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.List;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link PrimaryKeyIndexDefinitions}. */
+class PrimaryKeyIndexDefinitionsTest {
+
+ @Test
+ void testCreatesMixedDefinitionsInSchemaFieldOrder() {
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.PK_VECTOR_INDEX_COLUMNS.key(), "embedding");
+ options.put("fields.embedding.pk-vector.index.type", "ivf-pq");
+ options.put(CoreOptions.PK_BTREE_INDEX_COLUMNS.key(), "name");
+ options.put(CoreOptions.PK_BITMAP_INDEX_COLUMNS.key(), "status");
+
+ List<PrimaryKeyIndexDefinition> definitions =
+
PrimaryKeyIndexDefinitions.create(schema(options)).definitions();
+
+ assertThat(definitions)
+ .extracting(PrimaryKeyIndexDefinition::column)
+ .containsExactly("name", "status", "embedding");
+ assertThat(definitions)
+ .extracting(PrimaryKeyIndexDefinition::family)
+ .containsExactly(
+ PrimaryKeyIndexDefinition.Family.BTREE,
+ PrimaryKeyIndexDefinition.Family.BITMAP,
+ PrimaryKeyIndexDefinition.Family.VECTOR);
+ assertThat(definitions)
+ .extracting(PrimaryKeyIndexDefinition::fieldId)
+ .containsExactly(1, 2, 3);
+ assertThat(definitions)
+ .extracting(PrimaryKeyIndexDefinition::indexType)
+ .containsExactly("btree", "bitmap", "ivf-pq");
+ }
+
+ @Test
+ void testRejectsDuplicateColumnWithinFamily() {
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.PK_BTREE_INDEX_COLUMNS.key(), "name,name");
+
+ assertThatThrownBy(() ->
PrimaryKeyIndexDefinitions.create(schema(options)))
+ .hasMessageContaining("pk-btree.index.columns")
+ .hasMessageContaining("duplicate")
+ .hasMessageContaining("name");
+ }
+
+ @Test
+ void testRejectsColumnConfiguredByMultipleFamilies() {
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.PK_BTREE_INDEX_COLUMNS.key(), "name");
+ options.put(CoreOptions.PK_BITMAP_INDEX_COLUMNS.key(), "name");
+
+ assertThatThrownBy(() ->
PrimaryKeyIndexDefinitions.create(schema(options)))
+ .hasMessageContaining("name")
+ .hasMessageContaining("at most one primary-key index");
+ }
+
+ private static TableSchema schema(Map<String, String> options) {
+ return new TableSchema(
+ 0,
+ Arrays.asList(
+ new DataField(0, "id", DataTypes.INT().notNull()),
+ new DataField(1, "name", DataTypes.STRING()),
+ new DataField(2, "status", DataTypes.INT()),
+ new DataField(3, "embedding", DataTypes.VECTOR(3,
DataTypes.FLOAT()))),
+ 3,
+ Collections.emptyList(),
+ Collections.singletonList("id"),
+ options,
+ "");
+ }
+}
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
new file mode 100644
index 0000000000..1822598773
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/index/pksorted/PrimaryKeySortedIndexOptionsTest.java
@@ -0,0 +1,78 @@
+/*
+ * 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.CoreOptions;
+import org.apache.paimon.options.Options;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for primary-key sorted index options in {@link CoreOptions}. */
+class PrimaryKeySortedIndexOptionsTest {
+
+ @Test
+ void testResolvesBTreeIndexColumns() {
+ CoreOptions options =
+ new
CoreOptions(Collections.singletonMap("pk-btree.index.columns", " name, age "));
+
+
assertThat(options.primaryKeyBTreeIndexColumns()).containsExactly("name",
"age");
+ }
+
+ @Test
+ void testResolvesBitmapIndexColumns() {
+ CoreOptions options =
+ new CoreOptions(
+ Collections.singletonMap("pk-bitmap.index.columns", "
status, region "));
+
+
assertThat(options.primaryKeyBitmapIndexColumns()).containsExactly("status",
"region");
+ }
+
+ @Test
+ void testResolvesBTreeIndexOptions() {
+ Map<String, String> values = new HashMap<>();
+ values.put(
+ "fields.name.pk-btree.index.options",
+ "{\"block-size\":\"4
kb\",\"sorted-index.records-per-range\":\"10\"}");
+
+ Options options = new
CoreOptions(values).primaryKeyBTreeIndexOptions("name");
+
+ assertThat(options.get("btree-index.block-size")).isEqualTo("4 kb");
+
assertThat(options.get("sorted-index.records-per-range")).isEqualTo("10");
+ }
+
+ @Test
+ void testResolvesBitmapIndexOptions() {
+ Map<String, String> values = new HashMap<>();
+ values.put(
+ "fields.status.pk-bitmap.index.options",
+ "{\"dictionary-block-size\":\"8 kb\","
+ + "\"sorted-index.records-per-range\":\"20\"}");
+
+ Options options = new
CoreOptions(values).primaryKeyBitmapIndexOptions("status");
+
+
assertThat(options.get("bitmap-index.dictionary-block-size")).isEqualTo("8 kb");
+
assertThat(options.get("sorted-index.records-per-range")).isEqualTo("20");
+ }
+}
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
new file mode 100644
index 0000000000..d55dd836ef
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/schema/PrimaryKeySortedIndexValidationTest.java
@@ -0,0 +1,191 @@
+/*
+ * 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 BTree and Bitmap index validation. */
+class PrimaryKeySortedIndexValidationTest {
+
+ @Test
+ void testRejectsColumnConfiguredForMultipleIndexFamilies() {
+ Map<String, String> options = enabledOptions();
+ options.put(CoreOptions.PK_BTREE_INDEX_COLUMNS.key(), "payload");
+ options.put(CoreOptions.PK_BITMAP_INDEX_COLUMNS.key(), "payload");
+
+ assertThatThrownBy(() -> validateTableSchema(schema(options)))
+ .hasMessageContaining("payload")
+ .hasMessageContaining("at most one primary-key index");
+ }
+
+ @Test
+ void testRejectsDuplicateBTreeIndexColumns() {
+ Map<String, String> options = enabledOptions();
+ options.put(CoreOptions.PK_BTREE_INDEX_COLUMNS.key(),
"payload,payload");
+
+ assertThatThrownBy(() -> validateTableSchema(schema(options)))
+ .hasMessageContaining(CoreOptions.PK_BTREE_INDEX_COLUMNS.key())
+ .hasMessageContaining("duplicate");
+ }
+
+ @Test
+ void testRejectsDuplicateBitmapIndexColumns() {
+ Map<String, String> options = enabledOptions();
+ options.put(CoreOptions.PK_BITMAP_INDEX_COLUMNS.key(),
"payload,payload");
+
+ assertThatThrownBy(() -> validateTableSchema(schema(options)))
+
.hasMessageContaining(CoreOptions.PK_BITMAP_INDEX_COLUMNS.key())
+ .hasMessageContaining("duplicate");
+ }
+
+ @Test
+ void testRequiresDeletionVectors() {
+ Map<String, String> options = enabledOptions();
+ options.put(CoreOptions.DELETION_VECTORS_ENABLED.key(), "false");
+ options.put(CoreOptions.PK_BTREE_INDEX_COLUMNS.key(), "payload");
+
+ assertThatThrownBy(() -> validateTableSchema(schema(options)))
+ .hasMessageContaining(
+ "Primary-key BTree and Bitmap indexes require
deletion-vectors.enabled = true");
+ }
+
+ @Test
+ void testRequiresPrimaryKeyTable() {
+ Map<String, String> options = enabledOptions();
+ options.put(CoreOptions.BUCKET_KEY.key(), "id");
+ options.put(CoreOptions.PK_BITMAP_INDEX_COLUMNS.key(), "payload");
+
+ assertThatThrownBy(() -> validateTableSchema(schema(options,
Collections.emptyList())))
+ .hasMessageContaining(
+ "Primary-key BTree and Bitmap indexes require a
primary-key table");
+ }
+
+ @Test
+ void testRequiresFixedBucketMode() {
+ Map<String, String> options = enabledOptions();
+ options.put(CoreOptions.BUCKET.key(), "-1");
+ options.put(CoreOptions.PK_BTREE_INDEX_COLUMNS.key(), "payload");
+
+ assertThatThrownBy(() -> validateTableSchema(schema(options)))
+ .hasMessageContaining(
+ "Primary-key BTree and Bitmap indexes require fixed or
postpone bucket mode");
+ }
+
+ @Test
+ void testSupportsPostponeBucket() {
+ Map<String, String> options = enabledOptions();
+ options.put(CoreOptions.BUCKET.key(), "-2");
+ options.put(CoreOptions.PK_BTREE_INDEX_COLUMNS.key(), "payload");
+
+ assertThatCode(() ->
validateTableSchema(schema(options))).doesNotThrowAnyException();
+ }
+
+ @Test
+ void testRejectsDeletionVectorMergeOnRead() {
+ Map<String, String> options = enabledOptions();
+ options.put(CoreOptions.DELETION_VECTORS_MERGE_ON_READ.key(), "true");
+ options.put(CoreOptions.PK_BITMAP_INDEX_COLUMNS.key(), "payload");
+
+ assertThatThrownBy(() -> validateTableSchema(schema(options)))
+ .hasMessageContaining(
+ "Primary-key BTree and Bitmap indexes require
deletion-vectors.merge-on-read = false");
+ }
+
+ @Test
+ void testRejectsPkClusteringOverride() {
+ Map<String, String> options = enabledOptions();
+ options.put(CoreOptions.PK_CLUSTERING_OVERRIDE.key(), "true");
+ options.put(CoreOptions.CLUSTERING_COLUMNS.key(), "payload");
+ options.put(CoreOptions.PK_BTREE_INDEX_COLUMNS.key(), "payload");
+
+ assertThatThrownBy(() -> validateTableSchema(schema(options)))
+ .hasMessageContaining(
+ "Primary-key BTree and Bitmap indexes do not support
pk-clustering-override");
+ }
+
+ @Test
+ void testRejectsUnknownIndexColumn() {
+ Map<String, String> options = enabledOptions();
+ options.put(CoreOptions.PK_BITMAP_INDEX_COLUMNS.key(), "unknown");
+
+ assertThatThrownBy(() -> validateTableSchema(schema(options)))
+ .hasMessageContaining("pk-bitmap.index.columns entry
'unknown'")
+ .hasMessageContaining("must reference an existing column");
+ }
+
+ @Test
+ void testRejectsUnsupportedIndexColumnType() {
+ Map<String, String> options = enabledOptions();
+ options.put(CoreOptions.PK_BTREE_INDEX_COLUMNS.key(), "payload");
+
+ TableSchema schema =
+ new TableSchema(
+ 0,
+ Arrays.asList(
+ new DataField(0, "id",
DataTypes.INT().notNull()),
+ new DataField(1, "payload",
DataTypes.ARRAY(DataTypes.INT()))),
+ 0,
+ Collections.emptyList(),
+ Collections.singletonList("id"),
+ options,
+ "");
+
+ assertThatThrownBy(() -> validateTableSchema(schema))
+ .hasMessageContaining("ARRAY<INT>")
+ .hasMessageContaining("not supported by global index");
+ }
+
+ 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");
+ return options;
+ }
+
+ private static TableSchema schema(Map<String, String> options) {
+ return schema(options, Collections.singletonList("id"));
+ }
+
+ private static TableSchema schema(
+ Map<String, String> options, java.util.List<String> primaryKeys) {
+ return new TableSchema(
+ 0,
+ Arrays.asList(
+ new DataField(0, "id", DataTypes.INT().notNull()),
+ new DataField(1, "payload", DataTypes.STRING())),
+ 0,
+ Collections.emptyList(),
+ primaryKeys,
+ options,
+ "");
+ }
+}
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 2c065176d1..9cb401c98e 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
@@ -196,7 +196,84 @@ public class SchemaManagerTest {
SchemaChange.renameColumn(
new String[] {"embedding"},
"renamed_embedding")))
.isInstanceOf(UnsupportedOperationException.class)
- .hasMessage("Cannot rename primary-key vector index column:
[embedding]");
+ .hasMessage("Cannot rename primary-key index column:
[embedding]");
+ }
+
+ @Test
+ public void testRejectRenamePrimaryKeyBTreeIndexColumn() 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_BTREE_INDEX_COLUMNS.key(), "name");
+ Schema schema =
+ new Schema(
+ Arrays.asList(
+ new DataField(0, "id",
DataTypes.INT().notNull()),
+ new DataField(1, "name", 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[] {"name"},
"renamed_name")))
+ .isInstanceOf(UnsupportedOperationException.class)
+ .hasMessage("Cannot rename primary-key index column: [name]");
+ }
+
+ @Test
+ public void testRejectDropPrimaryKeyBitmapIndexColumn() 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_BITMAP_INDEX_COLUMNS.key(), "status");
+ Schema schema =
+ new Schema(
+ Arrays.asList(
+ new DataField(0, "id",
DataTypes.INT().notNull()),
+ new DataField(1, "status", DataTypes.INT())),
+ Collections.emptyList(),
+ Collections.singletonList("id"),
+ options,
+ "");
+ SchemaManager manager = new SchemaManager(LocalFileIO.create(), path);
+ manager.createTable(schema);
+
+ assertThatThrownBy(() ->
manager.commitChanges(SchemaChange.dropColumn("status")))
+ .isInstanceOf(UnsupportedOperationException.class)
+ .hasMessage("Cannot drop primary-key index column: [status]");
+ }
+
+ @Test
+ public void testRejectTypeChangeOfPrimaryKeyBitmapIndexColumn() 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_BITMAP_INDEX_COLUMNS.key(), "status");
+ Schema schema =
+ new Schema(
+ Arrays.asList(
+ new DataField(0, "id",
DataTypes.INT().notNull()),
+ new DataField(1, "status", DataTypes.INT())),
+ Collections.emptyList(),
+ Collections.singletonList("id"),
+ options,
+ "");
+ SchemaManager manager = new SchemaManager(LocalFileIO.create(), path);
+ manager.createTable(schema);
+
+ assertThatThrownBy(
+ () ->
+ manager.commitChanges(
+ SchemaChange.updateColumnType(
+ "status", DataTypes.BIGINT())))
+ .isInstanceOf(UnsupportedOperationException.class)
+ .hasMessage("Cannot update type of primary-key index column:
[status]");
}
@Test
