pzhdfy commented on code in PR #69493:
URL: https://github.com/apache/doris/pull/69493#discussion_r4360676519
##########
be/src/format_v2/table/paimon_rust_table_reader.cpp:
##########
@@ -509,6 +626,316 @@ Status PaimonRustTableReader::_open_split_reader(const
TFileRangeDesc& range) {
return Status::OK();
}
+Status PaimonRustTableReader::_open_vector_split_reader(const TFileRangeDesc&
range) {
+ // Validate every __isset / value-range up front (mirrors
+ // LanceTableReader::_validate_external_search_request). The build path
below
+ // then uses the fields unchecked, matching lance's split of validate-once
/
+ // build-without-recheck.
+ RETURN_IF_ERROR(_validate_external_search_request());
+ const auto& request = _scan_params->external_search_request;
+ const auto& vector = request.search_query.vector_search;
+
+ // 0. Distance metric. It decides which score transform the reader has to
undo
+ // (see _convert_score_to_distance), so defaulting would silently return
wrong
+ // distances. __isset was checked by _validate_external_search_request; the
+ // value is rejected here if unsupported.
+ ScoreTransform transform;
+ RETURN_IF_ERROR(parse_score_transform(vector.metric, &transform));
+ _score_transform = transform;
+
+ // 1. The per-split payload is the serialized BucketVectorSearchSplit (raw
+ // thrift binary, NOT base64; magic "PKVSPLIT", v1). Grab it first so we
fail
+ // fast before any filesystem IO.
+ const auto& payload =
_current_range.table_format_params.paimon_params.vector_payload;
+ if (payload.payload_type != TPaimonVectorPayloadType::BUCKET_SPLIT_BYTES) {
+ return Status::InternalError("paimon-rust vector search unsupported
payload type: {}",
+ static_cast<int>(payload.payload_type));
+ }
+ const std::string& split_bytes = payload.bytes;
+ if (split_bytes.empty()) {
+ return Status::InternalError("paimon-rust vector search split payload
is empty");
+ }
+
+ // 2. Resolve and open (or reuse) the table (same as the normal read path);
+ // options carry storage credentials to the FileIO.
+ RETURN_IF_ERROR(_open_paimon_table(range));
+
+ // 3. Build and configure the vector-search builder. No read.batch-size
here,
+ // unlike the batch path (_open_split_reader): paimon-rust's PK vector
readers
+ // construct their DataFileReaders without with_batch_size, so a
+ // read.batch-size option — from the schema, t@options or these search
+ // options — is silently ignored on the vector path, which batches at the
+ // parquet-rust default. Known difference; the vector stages are
+ // single-column streamed into bounded Top-K heaps and the output is
+ // capped by top_k, so no action needed until paimon-rust wires it.
+ paimon_result_vector_search_builder vb_res =
+ paimon_table_new_vector_search_builder(_handles->table.get());
+ if (vb_res.error != nullptr) {
+ return Status::InternalError("paimon-rust new vector search builder
failed: {}",
+ consume_error(vb_res.error));
+ }
+ vector_search_builder_ptr vb(vb_res.builder);
+
+ // 3a. Target vector column.
+ if (paimon_error* err =
+ paimon_vector_search_builder_with_vector_column(vb.get(),
vector.column.c_str())) {
+ return Status::InternalError("paimon-rust set vector column failed:
{}",
+ consume_error(err));
+ }
+ // 3b. Query vector. The wire form is a TSearchVector carrying a FLOAT32,
+ // little-endian binary blob; decode it into the f32 array the C API wants.
+ // v1 only supports FLOAT32 (paimon-rust's index backends operate on f32);
any
+ // other element type was already rejected by validate. Dimension and
+ // byte-size were validated there too, so decode unchecked here.
+ const auto& search_vec = vector.query_vector;
+ const auto dimension = static_cast<size_t>(search_vec.dimension);
+ std::vector<float> query_vector(dimension);
+ const auto* bytes = search_vec.values.data();
+ for (size_t i = 0; i < dimension; ++i) {
+ query_vector[i] = std::bit_cast<float>(LittleEndian::Load32(
+ reinterpret_cast<const uint8_t*>(bytes + i *
sizeof(uint32_t))));
+ }
+ if (paimon_error* err = paimon_vector_search_builder_with_query_vector(
+ vb.get(), query_vector.data(), query_vector.size())) {
+ return Status::InternalError("paimon-rust set query vector failed:
{}", consume_error(err));
+ }
+ // 3c. Retrieval limit (user_limit + user_offset, folded on FE). A negative
+ // limit would wrap when cast to uintptr_t, so guard it.
+ if (vector.top_k < 0) {
+ return Status::InternalError("paimon-rust vector search limit is
negative: {}",
+ vector.top_k);
+ }
+ if (paimon_error* err = paimon_vector_search_builder_with_limit(
+ vb.get(), static_cast<uintptr_t>(vector.top_k))) {
+ return Status::InternalError("paimon-rust set limit failed: {}",
consume_error(err));
+ }
+ // 3d. Index/search options (refine_factor, ivf.nprobe, "<algo>.metric",
...).
+ if (request.__isset.paimon_options &&
request.paimon_options.__isset.options &&
+ !request.paimon_options.options.empty()) {
+ std::vector<paimon_option> c_search_options;
+ c_search_options.reserve(request.paimon_options.options.size());
+ for (const auto& kv : request.paimon_options.options) {
+ c_search_options.emplace_back(kv.first.c_str(), kv.second.c_str());
+ }
+ if (paimon_error* err = paimon_vector_search_builder_with_options(
+ vb.get(), c_search_options.data(),
c_search_options.size())) {
+ return Status::InternalError("paimon-rust set search options
failed: {}",
+ consume_error(err));
+ }
+ }
+ // 3e. Materialized user columns. The score field is always appended by
+ // paimon-rust itself and must NOT be listed here. Empty projection = only
the
+ // score column (index-only scan). Log the user columns: an empty list is
+ // legitimate, so it is logged as such rather than skipped.
+ auto read_columns = _build_read_columns();
+ std::vector<const char*> projection;
+ projection.reserve(read_columns.size() + 1);
+ for (const auto& col : read_columns) {
+ projection.push_back(col.c_str());
+ }
+ projection.push_back(nullptr);
+ LOG(INFO) << "paimon-rust vector search: projection=[" <<
join(read_columns, ", ") << "] ("
+ << read_columns.size() << " column(s), distance column '"
+ << kPaimonSearchDistanceColumn << "' produced by the reader)";
+ if (paimon_error* err =
+ paimon_vector_search_builder_with_projection(vb.get(),
projection.data())) {
+ return Status::InternalError("paimon-rust set vector projection
failed: {}",
+ consume_error(err));
+ }
+
+ // 3f. Optional residual scalar filter from the FE data-half push-down. The
+ // FE-side DV/merge-on-read interception is intentionally not done (see
+ // PushDownVectorTopNIntoPaimonScan): the Rust candidate search stage
applies
+ // the filter together with deletion vectors, and a table shape Rust cannot
+ // handle correctly is Rust's responsibility to reject.
+ if (!_conjuncts.empty()) {
+ std::vector<std::string> names;
+ std::vector<DataTypePtr> types;
+ names.reserve(_projected_columns.size());
+ types.reserve(_projected_columns.size());
+ for (const auto& col : _projected_columns) {
+ if (col.is_partition_key || col.name ==
kPaimonSearchDistanceColumn) {
+ continue;
+ }
+ names.push_back(col.name);
+ types.push_back(col.type);
+ }
+ PaimonRustPredicateConverter converter(names, types,
_handles->table.get());
+ paimon_predicate* predicate = converter.build(_conjuncts);
+ if (predicate != nullptr) {
+ // with_filter consumes the predicate on every path.
+ if (paimon_error* err =
paimon_vector_search_builder_with_filter(vb.get(), predicate)) {
+ return Status::InternalError("paimon-rust set vector filter
failed: {}",
+ consume_error(err));
+ }
+ LOG(INFO) << "paimon-rust vector search: applied residual filter";
+ }
+ }
+
+ // 4. Execute the search over the single FE-planned bucket split, via the
+ // eager Top-K chain (paimon-rust #822). Search, refine, local Top-K and
+ // materialization all happen in paimon_vector_read_read; the returned
arrow
+ // stream is best-first and carries the __paimon_search_score column. The
+ // intermediate handles are locals: the migration contract (see
+ // doris-vector-search-api-migration.md) guarantees the record batch reader
+ // is independent once paimon_vector_read_read returns, so they are
released
+ // when this function exits.
+ //
+ // split: the bytes only need to live until deserialize returns.
+ // scan: only needs to live until the plan is constructed.
+ // builder: only needs to live until scan and read are created.
+ // read/plan: only need to live until paimon_vector_read_read returns.
+ paimon_result_bucket_vector_search_split split_result =
+ paimon_bucket_vector_search_split_deserialize(
+ reinterpret_cast<const uint8_t*>(split_bytes.data()),
split_bytes.size());
+ if (split_result.error != nullptr) {
+ return Status::InternalError("paimon-rust deserialize vector split
failed: {}",
+ consume_error(split_result.error));
+ }
+ bucket_vector_search_split_ptr split(split_result.split);
+
+ paimon_result_vector_scan scan_result =
paimon_vector_search_builder_new_scan(vb.get());
+ if (scan_result.error != nullptr) {
+ return Status::InternalError("paimon-rust new vector scan failed: {}",
+ consume_error(scan_result.error));
+ }
+ vector_scan_ptr scan(scan_result.scan);
+
+ paimon_result_vector_read read_result =
paimon_vector_search_builder_new_read(vb.get());
+ if (read_result.error != nullptr) {
+ return Status::InternalError("paimon-rust new vector read failed: {}",
+ consume_error(read_result.error));
+ }
+ vector_read_ptr read(read_result.read);
+
+ // Must use plan_from_bucket_splits: it plans from the FE-assigned split,
+ // whereas paimon_vector_scan_plan / execute_read replan from the table and
+ // would ignore the split this scanner was given.
+ const paimon_bucket_vector_search_split* splits[] = {split.get()};
+ paimon_result_vector_plan plan_result =
+ paimon_vector_scan_plan_from_bucket_splits(scan.get(), splits, 1);
+ if (plan_result.error != nullptr) {
+ return Status::InternalError("paimon-rust build vector plan failed:
{}",
+ consume_error(plan_result.error));
+ }
+ vector_plan_ptr plan(plan_result.plan);
+
+ paimon_result_record_batch_reader rdr_res =
paimon_vector_read_read(read.get(), plan.get());
Review Comment:
Held as a documented limitation by explicit decision this round, with a
concrete upstream path we can point to:
- The verified chain is as reported: `paimon_vector_read_read` completes
search + materialization before returning; `materialize_positions`
retains every projected Top-K batch and then reorders into one output
batch, inside rust allocations that bypass Doris's tracked allocator, with
no bound from the adaptive batch size. The OFFSET-spreading example
(15+ GiB before the first block) is consistent with our analysis
(`vector-eager-materialization-analysis.md`).
- Upstream paimon-rust already merged the accounting foundation: PR #914
("integrate reader memory reservations with DataFusion", merged Sep 23)
adds ResourceContext / MemoryReservation / ReadBudget and a
`with_resources` entry point. We verified our pinned v0.4.0-rc1 does NOT
contain it (zero hits in the tarball), and #914 itself scopes out both the
vector-search path and the C ABI ("no FFI integration").
- Within this very repository, LanceTableReader demonstrates the shape the
vector path should converge to: a streaming `scanner_next` bounded by the
adaptive batch size, each batch converted into a Doris-Allocator block and
released immediately — an unaccounted window of exactly one bounded batch,
not one unbounded result set.
So the end state needs (a) paimon-rust: wire the vector read into
`with_resources`/reservations (or a loud byte cap) and expose the budget
through the C ABI, and optionally (b) an FE-side decline on
`(limit + offset)` beyond a configured cap as the interim guard. We are
tracking both; if maintainers prefer the FE-side interim guard in this PR we
can add it as a small follow-up commit.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
