[core][flink][lumina] Allow building Lumina global index on DV-enabled tables

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);
 
     /**

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<ManifestEntry> 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<ManifestEntry> scan() {
                 .files();
     }
 
+    private boolean supportsDeletionVectors() {
+        return indexType != null
+                && GlobalIndexerFactoryUtils.load(indexType).supportsDeletionVectors();
+    }
+
     @Nullable
     public Snapshot scanSnapshot() {
         return scanSnapshot;

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);
         }

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();
+    }
+}

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);