pzhdfy commented on code in PR #68988:
URL: https://github.com/apache/doris/pull/68988#discussion_r4321190058
##########
be/src/format_v2/table/paimon_rust_table_reader.cpp:
##########
@@ -509,6 +626,313 @@ 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());
+ if (rdr_res.error != nullptr) {
+ return Status::InternalError("paimon-rust vector search read failed:
{}",
+ consume_error(rdr_res.error));
+ }
+ _handles->reader.reset(rdr_res.reader);
+ return Status::OK();
+}
+
+std::string PaimonRustTableReader::arrow_field_to_block_column(const
std::string& arrow_name) {
+ std::string name = to_lower(arrow_name);
+ if (name == kPaimonRustScoreField) {
+ return kPaimonSearchDistanceColumn;
+ }
+ return name;
+}
+
+Status PaimonRustTableReader::parse_score_transform(TVectorMetric::type metric,
+ ScoreTransform* transform)
{
+ switch (metric) {
+ case TVectorMetric::L2:
+ *transform = ScoreTransform::L2;
+ return Status::OK();
+ case TVectorMetric::DOT_PRODUCT:
+ *transform = ScoreTransform::InnerProduct;
+ return Status::OK();
+ default:
+ // Includes COSINE: paimon-rust does define a score for it
+ // (score = 1 - distance), but v1 does not push cosine down, so
reaching here
+ // means FE and BE disagree. Fail instead of inventing a transform.
+ return Status::InternalError(
+ "paimon-rust vector search unsupported distance metric: {} "
+ "(expected L2 or DOT_PRODUCT)",
+ static_cast<int>(metric));
+ }
+}
+
+void PaimonRustTableReader::_convert_score_to_distance(Block* block, uint32_t
block_idx,
+ size_t row_offset,
size_t num_rows,
+ ScoreTransform
transform) {
+ // Undo what paimon-rust's convert_distance_to_score did, so the column
holds
+ // what the Doris distance function is defined to return.
+ switch (transform) {
+ case ScoreTransform::InnerProduct:
+ // score IS the inner product. Nothing to undo, and no clamping to do
+ // either: the value is unconstrained by definition (it may be
negative).
+ return;
+ case ScoreTransform::L2:
+ break;
+ }
+
+ auto& column_with_type = block->get_by_position(block_idx);
+ auto column = IColumn::mutate(std::move(column_with_type.column));
+ PaddedPODArray<float>* data = nullptr;
+ const NullMap* null_map = nullptr;
+ if (auto* nullable = check_and_get_column<ColumnNullable>(column.get())) {
+ data =
&assert_cast<ColumnFloat32&>(nullable->get_nested_column()).get_data();
+ null_map = &nullable->get_null_map_data();
+ } else {
+ data = &assert_cast<ColumnFloat32&>(*column).get_data();
+ }
+
+ // score = 1/(1 + d) where d is the SQUARED L2 distance, and
+ // l2_distance_approximate returns the true Euclidean one, so invert both
steps:
+ // d_euclidean = sqrt(1/score - 1).
+ const size_t end = std::min(row_offset + num_rows, data->size());
+ for (size_t i = row_offset; i < end; ++i) {
+ if (null_map != nullptr && (*null_map)[i] != 0) {
+ continue;
+ }
+ const float score = (*data)[i];
+ // score is in (0, 1] whenever d is finite and non-negative, and
within that
+ // range the radicand is provably non-negative: a real value above 1
cannot
+ // round down to below 1.0F in f32, so 1/score >= 1. The useful range
+ // therefore never reaches either guard below -- they exist because the
+ // distances come straight out of the closed-source lumina kernel with
no
+ // clamp, and computing the inverse on an illegal score would produce a
+ // non-finite distance (score <= 0 -> 1/score is inf; score > 1 ->
negative
+ // radicand -> NaN), which makes the coordinator's Top-N comparisons
+ // ill-defined. Report a distinct negative sentinel per case instead:
both
+ // are unreachable for a genuine Euclidean distance, so they surface as
+ // visibly bogus rows that also say WHICH way the index is broken.
+ //
+ // A score slightly above 1 is NOT corruption: the kernel computes
squared
+ // L2 as ||a||^2 + ||b||^2 - 2<a,b>, which can yield a small negative
+ // residual (catastrophic cancellation) when a ~= b -- a near-exact
hit.
+ // Clamp those to distance 0; only scores past kScoreAboveOneTolerance
are
+ // treated as genuinely corrupt.
+ if (UNLIKELY(score <= 0.0F)) {
+ LOG(WARNING) << "paimon-rust L2 score_to_distance: corrupt score "
<< score << " (row "
+ << i << ", expect score in (0, 1]), "
+ << "sentinel " << kCorruptScoreNotPositive;
+ (*data)[i] = kCorruptScoreNotPositive;
Review Comment:
▎ Confirmed. A zero score is reachable from a genuine huge distance — the
exact fallback in vector-index.search-mode=full computes squared L2 in f32, so
norms above ~1.8e19 overflow to +inf and 1/(1+d²) rounds to 0 — and the
negative sentinel then sorted ahead of every finite distance, inverting the
coordinator's TopN. Both out-of-range branches now emit +inf
(kCorruptScoreDistance): +inf sorts last in an ASC Top-K, so a corrupt or
infinite row can never displace a valid closer row; the distinct WARNING per
case stays. The design comment in the header now documents the
reachable-zero-score path. UT updated: three out-of-range rows map to +inf with
a legal score between them still converting exactly.
--
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]
