[flink] Support distributed vector search#8834
Conversation
steFaiz
left a comment
There was a problem hiding this comment.
Thanks for your contrib! Left some comments
| if (!indexSplits.isEmpty() && !rawSplits.isEmpty()) { | ||
| int parallelism = flinkParallelism(); | ||
| List<Range> rawRowRanges = rawRowRanges(rawSplits); | ||
| if (indexSplits.size() >= parallelism * 2L |
There was a problem hiding this comment.
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
There was a problem hiding this comment.
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.
| List<byte[]> indexedResults = new ArrayList<>(); | ||
| List<byte[]> rawResults = new ArrayList<>(); | ||
| for (byte[] taggedResult : taggedResults) { | ||
| if (taggedResult.length == 0) { |
There was a problem hiding this comment.
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.
There was a problem hiding this comment.
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.
| snapshotOptions.put(SCAN_TIMESTAMP_MILLIS.key(), null); | ||
| snapshotOptions.put(SCAN_MODE.key(), CoreOptions.StartupMode.FROM_SNAPSHOT.toString()); | ||
| snapshotOptions.put(SCAN_SNAPSHOT_ID.key(), String.valueOf(snapshot.id())); | ||
| return table.copyWithoutTimeTravel(snapshotOptions); |
There was a problem hiding this comment.
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
There was a problem hiding this comment.
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.
• ### Purpose
Close #8825
This PR adds distributed Vector Search support for Flink.
Key changes:
Tests
Added unit and integration tests covering: