From 5cb2e4763bb016ca065bf4641a288918de9b8410 Mon Sep 17 00:00:00 2001 From: "xiaohongbo.xhb" Date: Thu, 30 Jul 2026 07:13:38 -0700 Subject: [PATCH] [core][flink][lumina] Allow building Lumina global index on DV-enabled tables GenericGlobalIndexBuilder rejected every generic (non-btree/bitmap) global index on deletion-vector tables. Lumina vector search already filters DV-deleted rows at read time via the live-row pre-filter, so building the index over all physical rows (a superset) is safe. Add an opt-in GlobalIndexerFactory.supportsDeletionVectors() capability (default false), override it to true in Lumina, and relax the guard to allow only opt-in index types on DV tables while still rejecting the rest. Wires the resolved index type through the builder. --- .../globalindex/GlobalIndexerFactory.java | 7 ++ .../GenericGlobalIndexBuilder.java | 19 +++- .../globalindex/GenericIndexTopoBuilder.java | 1 + ...cGlobalIndexBuilderDeletionVectorTest.java | 87 +++++++++++++++++++ .../LuminaVectorGlobalIndexerFactory.java | 6 ++ 5 files changed, 118 insertions(+), 2 deletions(-) create mode 100644 paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/globalindex/GenericGlobalIndexBuilderDeletionVectorTest.java diff --git a/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexerFactory.java b/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexerFactory.java index 995d19760767..ef3ac3c4e231 100644 --- a/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexerFactory.java +++ b/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexerFactory.java @@ -33,6 +33,13 @@ default boolean supportsFullTextSearch() { return false; } + /** + * Whether the read path filters DV-deleted rows, so building over all physical rows is safe. + */ + default boolean supportsDeletionVectors() { + return false; + } + GlobalIndexer create(DataField indexField, Options options); /** diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/globalindex/GenericGlobalIndexBuilder.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/globalindex/GenericGlobalIndexBuilder.java index b20c513cd0c3..de57b2a3a83b 100644 --- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/globalindex/GenericGlobalIndexBuilder.java +++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/globalindex/GenericGlobalIndexBuilder.java @@ -19,6 +19,7 @@ package org.apache.paimon.flink.globalindex; import org.apache.paimon.Snapshot; +import org.apache.paimon.globalindex.GlobalIndexerFactoryUtils; import org.apache.paimon.manifest.IndexManifestEntry; import org.apache.paimon.manifest.ManifestEntry; import org.apache.paimon.partition.PartitionPredicate; @@ -41,6 +42,7 @@ public class GenericGlobalIndexBuilder implements Serializable { @Nullable protected PartitionPredicate partitionPredicate; @Nullable private Snapshot scanSnapshot; + @Nullable private String indexType; public GenericGlobalIndexBuilder(FileStoreTable table) { this.table = table; @@ -51,6 +53,11 @@ public GenericGlobalIndexBuilder withPartitionPredicate(PartitionPredicate parti return this; } + public GenericGlobalIndexBuilder withIndexType(String indexType) { + this.indexType = indexType; + return this; + } + public FileStoreTable table() { return table; } @@ -67,11 +74,14 @@ public List scan() { + "but table '%s' has bucket = %d.", table.name(), table.coreOptions().bucket()); + // Allow DV only for index types that opt in (read path filters DV-deleted + // rows via the live-row pre-filter, e.g. Lumina); reject all others. checkArgument( - !table.coreOptions().deletionVectorsEnabled(), - "Generic global index does not support tables with deletion vectors enabled. " + !table.coreOptions().deletionVectorsEnabled() || supportsDeletionVectors(), + "Global index type '%s' does not support tables with deletion vectors enabled. " + "Table '%s' has 'deletion-vectors.enabled' = true, which may cause " + "deleted rows to be indexed.", + indexType, table.name()); scanSnapshot = table.snapshotManager().latestSnapshot(); @@ -87,6 +97,11 @@ public List scan() { .files(); } + private boolean supportsDeletionVectors() { + return indexType != null + && GlobalIndexerFactoryUtils.load(indexType).supportsDeletionVectors(); + } + @Nullable public Snapshot scanSnapshot() { return scanSnapshot; diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/globalindex/GenericIndexTopoBuilder.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/globalindex/GenericIndexTopoBuilder.java index 34461d7f7595..1f53ae1de977 100644 --- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/globalindex/GenericIndexTopoBuilder.java +++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/globalindex/GenericIndexTopoBuilder.java @@ -298,6 +298,7 @@ private static boolean buildIndexInternal( boolean autoIncremental) throws Exception { GenericGlobalIndexBuilder indexBuilder = indexBuilderSupplier.get(); + indexBuilder.withIndexType(indexType); if (partitionPredicate != null) { indexBuilder.withPartitionPredicate(partitionPredicate); } diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/globalindex/GenericGlobalIndexBuilderDeletionVectorTest.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/globalindex/GenericGlobalIndexBuilderDeletionVectorTest.java new file mode 100644 index 000000000000..da4812b1f383 --- /dev/null +++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/globalindex/GenericGlobalIndexBuilderDeletionVectorTest.java @@ -0,0 +1,87 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.paimon.flink.globalindex; + +import org.apache.paimon.CoreOptions; +import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.utils.SnapshotManager; + +import org.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThatCode; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +/** Tests the deletion-vector guard in {@link GenericGlobalIndexBuilder}. */ +class GenericGlobalIndexBuilderDeletionVectorTest { + + private static FileStoreTable table(boolean deletionVectorsEnabled) { + FileStoreTable table = mock(FileStoreTable.class); + CoreOptions options = mock(CoreOptions.class); + when(options.bucket()).thenReturn(-1); + when(options.deletionVectorsEnabled()).thenReturn(deletionVectorsEnabled); + when(table.coreOptions()).thenReturn(options); + when(table.name()).thenReturn("T"); + SnapshotManager snapshotManager = mock(SnapshotManager.class); + when(snapshotManager.latestSnapshot()).thenReturn(null); + when(table.snapshotManager()).thenReturn(snapshotManager); + return table; + } + + @Test + void testLuminaAllowedOnDeletionVectorTable() { + assertThatCode( + () -> + new GenericGlobalIndexBuilder(table(true)) + .withIndexType("lumina") + .scan()) + .doesNotThrowAnyException(); + } + + @Test + void testLegacyLuminaIdentifierAllowedOnDeletionVectorTable() { + assertThatCode( + () -> + new GenericGlobalIndexBuilder(table(true)) + .withIndexType("lumina-vector-ann") + .scan()) + .doesNotThrowAnyException(); + } + + @Test + void testNonLuminaRejectedOnDeletionVectorTable() { + assertThatThrownBy( + () -> + new GenericGlobalIndexBuilder(table(true)) + .withIndexType("bitmap") + .scan()) + .hasMessageContaining("deletion vectors"); + } + + @Test + void testNonLuminaAllowedWithoutDeletionVectors() { + assertThatCode( + () -> + new GenericGlobalIndexBuilder(table(false)) + .withIndexType("bitmap") + .scan()) + .doesNotThrowAnyException(); + } +} diff --git a/paimon-lumina/src/main/java/org/apache/paimon/lumina/index/LuminaVectorGlobalIndexerFactory.java b/paimon-lumina/src/main/java/org/apache/paimon/lumina/index/LuminaVectorGlobalIndexerFactory.java index 7d9c062feb61..89c8c14dbf60 100644 --- a/paimon-lumina/src/main/java/org/apache/paimon/lumina/index/LuminaVectorGlobalIndexerFactory.java +++ b/paimon-lumina/src/main/java/org/apache/paimon/lumina/index/LuminaVectorGlobalIndexerFactory.java @@ -33,6 +33,12 @@ public String identifier() { return IDENTIFIER; } + @Override + public boolean supportsDeletionVectors() { + // Vector search filters DV-deleted rows at read time (live-row pre-filter). + return true; + } + @Override public GlobalIndexer create(DataField field, Options options) { Options fieldOptions = LuminaVectorIndexOptions.resolveFieldOptions(field.name(), options);