[core][flink][spark][python][vector] Support vector indexes on DV tables

[XiaoHongbo-Hope]

Purpose

Allow building vector global indexes on deletion-vector (DV) enabled tables. Generic global-index building previously rejected every index type on DV tables, even though the vector read paths can filter DV-deleted rows and safely build over the physical-row superset.

Changes

• Add an opt-in GlobalIndexerFactory.supportsDeletionVectors() capability, defaulting to false.
• Enable the capability for Lumina and all current paimon-vector native index types: ivf-flat, ivf-pq, ivf-sq, ivf-rq, and diskann.
• Relax the Flink builder guard only for opted-in factories and thread the selected index type into the builder.
• Pin the Spark vector reader to the snapshot used during planning.
• Carry the planned snapshot through PyPaimon vector plans and use it for DV live-row filtering, scalar-index filtering, and raw fallback reads.
• Refresh global-index deletions from the latest snapshot on every materialized-deletion compaction commit attempt. This removes indexes committed after compaction preparation and repeats the refresh after optimistic-commit conflicts.

Related merged fixes

• [core][flink] Pin DV vector-search read path to the plan snapshot #8946 pins the core and Flink DV vector-search paths to the plan snapshot.
• [core] Drop concurrently-built global indexes at materialize-deletion compaction #8947 scans the latest snapshot while preparing global-index deletions for materialized-deletion compaction.
• [core][python] Fix global-index live-row filter re-adding DV-deleted rows across overlapping files #8939 contains the union / two-phase live-row filtering fix.

This PR completes snapshot pinning for Spark and PyPaimon and closes the remaining preparation-to-commit race.

Tests

• Lumina, its legacy identifier, and all native vector index types are allowed on DV-enabled tables.
• Non-opt-in index types remain rejected on DV-enabled tables and continue to work without DV.
• Spark and PyPaimon readers retain and propagate the planned snapshot through live-row, scalar-index, and raw-read paths.
• A global index committed after materialized-compaction preparation is deleted when that compaction commits.

[JingsongLi]

Thanks for working on this. I think this PR needs two changes before merge:

1. Please complete snapshot pinning for every reader before advertising deletion-vector support. supportsDeletionVectors() is a factory-wide capability, but snapshot pinning is currently incomplete. The core and Flink readers assign planSnapshot = plan.snapshot(), while SparkDataEvolutionVectorRead#read overrides the method without doing so. PyPaimon's VectorSearchScanPlan also only carries splits, and its live-row filter creates a new scan against the latest snapshot. If materialized-deletion compaction commits between planning and reading, an old index plan can be combined with the new row-id/DV state, producing missing or incorrect top-K results. Please thread the planned snapshot through Spark and PyPaimon as well, including the live-row/scalar filters, and add regression coverage for this race.

2. This PR should also support paimon-vector. paimon-vector is the more important/current native vector-index implementation; Lumina is only the legacy index. NativeVectorGlobalIndexReader already honors includeRowIds for both single and batch search, so NativeVectorGlobalIndexerFactory should opt in to deletion-vector support after the snapshot-pinning issue above is fixed, with tests covering the native index types on DV tables. Limiting this change to Lumina leaves the primary implementation unsupported and makes the feature incomplete.

I consider both items part of this PR rather than follow-ups. The FileStoreCommitImpl refresh still makes sense because it closes the preparation-to-commit/retry race.

[XiaoHongbo-Hope]

Thanks for working on this. I think this PR needs two changes before merge:

1. Please complete snapshot pinning for every reader before advertising deletion-vector support. supportsDeletionVectors() is a factory-wide capability, but snapshot pinning is currently incomplete. The core and Flink readers assign planSnapshot = plan.snapshot(), while SparkDataEvolutionVectorRead#read overrides the method without doing so. PyPaimon's VectorSearchScanPlan also only carries splits, and its live-row filter creates a new scan against the latest snapshot. If materialized-deletion compaction commits between planning and reading, an old index plan can be combined with the new row-id/DV state, producing missing or incorrect top-K results. Please thread the planned snapshot through Spark and PyPaimon as well, including the live-row/scalar filters, and add regression coverage for this race.
2. This PR should also support paimon-vector. paimon-vector is the more important/current native vector-index implementation; Lumina is only the legacy index. NativeVectorGlobalIndexReader already honors includeRowIds for both single and batch search, so NativeVectorGlobalIndexerFactory should opt in to deletion-vector support after the snapshot-pinning issue above is fixed, with tests covering the native index types on DV tables. Limiting this change to Lumina leaves the primary implementation unsupported and makes the feature incomplete.

I consider both items part of this PR rather than follow-ups. The FileStoreCommitImpl refresh still makes sense because it closes the preparation-to-commit/retry race.

Thanks for the detailed review. Both requested changes are addressed in 6363ee8:

[JingsongLi]

Remove this, we have support all vector index.

[JingsongLi]

I found one PyPaimon snapshot-lifecycle correctness issue that should be addressed before merge.

[JingsongLi]

Storing the plan snapshot on the reader makes the result depend on previous calls. read() remains public and does not clear _plan_snapshot, so reusing a reader after read_plan(plan_at_S1) and then calling read(splits_planned_at_S2) makes the S2 live-row, scalar-index, and raw-fallback paths read from S1. The mutable field is also unsafe for concurrent reads, and the batch path has the same issue. After DV materialization this can return deleted rows or omit valid top-K results. Could we keep the reader stateless and pass the snapshot explicitly for each invocation (for example, _read(splits, snapshot), with read_plan forwarding plan.snapshot()), instead of caching it on self?

[JingsongLi]

The latest update exposes one additional DV snapshot-consistency issue outside the vector readers.

[JingsongLi]

Removing the DV guard here enables every generic index type, including full-text and es-index, but those read paths are not snapshot-consistent yet. FullTextRead.read(Plan) discards the plan context, and DataEvolutionFullTextRead computes live rows from the latest snapshot. If materialized-DV compaction commits between planning at S1 and reading at S2, the S1 index row IDs are intersected with S2 live rows, which can produce missing or empty top-K results. Please keep non-vector indexes gated (or restore an explicit vector-capability check) until full-text plans carry their snapshot and both live-row and raw fallback reads are pinned to it. A plan -> materialized compaction -> read regression test would cover this race.

[JingsongLi]

Let‘s add support to full-text too.

[JingsongLi]

+1