Skip to content

[spark][flink] Support primary-key vector search#8581

Open
JingsongLi wants to merge 7 commits into
masterfrom
codex/pk-vector-07-spark-search
Open

[spark][flink] Support primary-key vector search#8581
JingsongLi wants to merge 7 commits into
masterfrom
codex/pk-vector-07-spark-search

Conversation

@JingsongLi

@JingsongLi JingsongLi commented Jul 12, 2026

Copy link
Copy Markdown
Contributor

Summary

Add end-to-end Spark SQL and Flink procedure support for primary-key vector search backed by the bucket-local indexes introduced in #8579. The PR supports every primary-key merge engine, distributes sufficiently large Spark searches by bucket group, and documents table creation, index maintenance, compaction freshness, query APIs, and current limitations.

Changes

  • Route configured primary-key vector fields to SparkPrimaryKeyVectorRead, while Data Evolution tables use the renamed SparkDataEvolutionVectorRead.
  • Reuse Core bucket search in Spark executors by keeping only serializable query state in PrimaryKeyVectorRead and creating file readers, ANN searchers, and executors per task invocation.
  • Explicitly serialize bucket splits and task-local candidates so execution is independent of Spark's configured Java/Kryo serializer.
  • Introduce ScoreRecordReader so Spark metadata handling depends on a shared score-reader contract instead of a concrete reader implementation.
  • Expose primary-key physical-position scores through the same contract and skip Spark's metadata-only local scan for split-based primary-key vector results.
  • Support deduplicate, partial-update, aggregation, and first-row merge engines by indexing their complete Level-1 compact output.
  • Require deletion vectors for deduplicate, partial-update, and aggregation, while first-row requires deletion vectors to be disabled.
  • Add Flink vector_search IT coverage and a primary-key vector index guide under docs/docs/primary-key-table/.

Behavior

  • Spark vector_search returns physical rows and __paimon_search_score from one captured snapshot.
  • With spark.paimon.vector-search.distribute.enabled=true, sufficiently large bucket plans run as Spark task groups. Each task returns only its local Top-K and the driver performs the final deterministic global Top-K merge; small plans keep the Core local path to avoid Spark startup overhead.
  • Flink CALL sys.vector_search uses the same Core primary-key vector search path and returns projected rows as JSON.
  • Partition predicates are applied before bucket-level Top-K; non-partition predicates remain unsupported for primary-key vector search.
  • deduplicate indexes the retained physical row, and partial-update indexes the lookup-completed row.
  • aggregation indexes the aggregated Level-1 row; first-row indexes the retained first row and ignores later rows with the same primary key.
  • Deletion vectors filter stale ANN candidates for merge engines that replace old rows. They are rejected for first-row, whose later duplicates are ignored instead of replacing the retained row.
  • Level-0 rows follow compaction freshness: streaming read-after-write search waits for Level-1 output and its synchronously built ANN segment.

Testing

  • Spark PrimaryKeyVectorSearchTest: 7 tests covering distributed reader routing, a real Spark ANN job, global Top-K across four buckets/two task groups, score projection, partition pruning, DEDUPLICATE updates/deletes, and PARTIAL_UPDATE completion.
  • Spark SparkDataEvolutionVectorReadTest: 3 existing distributed index/raw/refine tests after the reader rename.
  • Spark VectorSearchOptionsTest: existing Data Evolution query-option and metadata-only path.
  • Flink VectorSearchProcedureITCase: 7 tests with both flink1 and flink2 profiles.
  • Core PrimaryKeyVectorIndexValidationTest and PrimaryKeyVectorSearchTest: all merge engines, first-row with deletion vectors disabled, first-row plus deletion-vector rejection, and compact-output ANN semantics (22 tests).
  • Core PrimaryKeyVectorReadTest and PrimaryKeyVectorSearchTest: local deterministic Top-K and bucket search regression coverage.
  • Core PrimaryKeyVectorPositionReaderTest and IndexedSplitRecordReaderTest: 12 score-reader tests.
  • Java Spotless and git diff --check.
  • Docusaurus production build. It reports the repository's existing site-wide /concepts/overview broken-link warning, but the build and the new page succeed.

Notes

Builds on #8579. This PR adds no new configuration keys or storage-format changes. Hybrid search and lateral/batch vector search remain outside its scope.

@JingsongLi JingsongLi changed the title [spark] Support primary-key vector search [spark][flink] Support primary-key vector search Jul 12, 2026
@JingsongLi JingsongLi marked this pull request as ready for review July 12, 2026 14:48

@leaves12138 leaves12138 left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Reviewed the latest head. The Spark primary-key vector path now uses the Core primary-key read flow, score propagation is preserved through the generic score reader, and the metadata-only shortcut is correctly avoided for physical-position results. The merge-engine and deletion-vector validation changes are consistent with the corresponding read semantics.\n\nFocused verification passed:\n- Core primary-key vector and score-reader tests: 35 tests\n- Spark PrimaryKeyVectorSearchTest: 5 tests\n- Flink VectorSearchProcedureITCase: 7 tests

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants