[flink] Support distributed vector search

[liangjie3138]

• ### Purpose

Close #8825

This PR adds distributed Vector Search support for Flink.

Key changes:

• Support local and distributed search for Data Evolution and primary-key tables.
• Search Indexed and Raw data in one Flink job with global TopK merging and optional refinement.
• Extend sys.vector_search with filters, partition pruning, score projection, snapshot consistency, and parameter validation.
• Support multi-field vector indexes and reduce Raw prefilter serialization overhead.
• Update the related Flink documentation.

Tests

Added unit and integration tests covering:

• Local and distributed search for Data Evolution and primary-key tables.
• Indexed, Raw, and mixed Indexed + Raw search with filtering and global TopK.
• Multi-field indexes, refinement, partition pruning, score projection, and snapshot consistency.
• Empty tables, schema evolution, parameter validation, and unsupported configurations.

[steFaiz]

Thanks for your contrib! Left some comments

[steFaiz]

Thanks, can u explain why hard-code '2' as the factor of indexSplits size and row count num here? Can we make it configurable? Or maybe we can merge some small splits of low cardianlity to a single task

[liangjie3138]

Thanks. The factor 2 follows Spark’s heuristic to amortize distributed execution overhead. Small index splits are already grouped into at most parallelism tasks, while Raw ranges are balanced by row count. Keeping it fixed avoids another configuration option.

[steFaiz]

This is a dangerous action. Add one byte to the serialized bytes without any code-level specification, then strip the first byte on read. I think we could add a new column, or just
introduce a format for this field, then deal with the read & write in a single format class, to make sure the read/write format is aligned.

[liangjie3138]

Good point. Although this payload is only used within one bounded Flink job, the byte prefix is implicit and fragile. I’ll replace it with Tuple2<Byte, byte[]> to represent the result type and payload as separate fields, and remove the manual tag/untag logic.

[steFaiz]

Why we need to do this copy? If we want to pin a snapshot, we could just pass the target snapshot id, not clear all other options and copy the table again.
For example, you can check: org.apache.paimon.flink.dataevolution.DataEvolutionPartialWriteOperator#DataEvolutionPartialWriteOperator

[liangjie3138]

DataEvolutionPartialWriteOperator uses a single low-level FileStoreScan, which supports withSnapshot(snapshotId). Vector Search uses several high-level scan paths: DataEvolutionVectorScan, Raw/refine reads through ReadBuilder, and the final row lookup. These APIs do not expose a common snapshot parameter.
We could remove the copy after adding explicit snapshot support to both VectorSearchBuilder and ReadBuilder, but that would require a broader Core API change.