branch-4.1:[feat](paimon) port PK-vector ANN search and upgrade Paimon to 2.1

be/src/agent/heartbeat_server.cpp

@@ -97,6 +97,10 @@ void HeartbeatServer::heartbeat(THeartbeatResult& heartbeat_result,
         heartbeat_result.backend_info.__set_be_mem(MemInfo::physical_mem());
         // Reader enum support cannot be inferred from a release version during rolling upgrades.
         heartbeat_result.backend_info.__set_supports_paimon_rust_reader(true);
+        // The PK-vector (ANN) protocol is a separate negotiation: an older
+        // rust-capable BE would reject an empty paimon_split + vector payload as
+        // "missing paimon_split", so only BEs carrying the vector protocol report it.
+        heartbeat_result.backend_info.__set_supports_paimon_rust_vector_reader(true);
     }
     watch.stop();
     if (watch.elapsed_time() > 1000L * 1000L * 1000L) {

be/src/format_v2/table/paimon_rust_predicate_converter.h

@@ -42,6 +42,27 @@ namespace doris {} // namespace doris
 
 namespace doris {
 
+// Reader-produced PK-vector search distance column, as Doris names it. Wire
+// contract shared with FE (PaimonVectorSearch.SEARCH_DISTANCE_COLUMN). It is never
+// a projectable table column and never a pushable predicate column, so both the
+// projection builder and the predicate converter must exclude it.
+//
+// It holds the metric-native DISTANCE, not paimon-rust's score: paimon-rust emits
+// an Arrow field named kPaimonRustScoreField carrying a higher-is-better score, and
+// the rust table reader inverts that transform in place while filling the block so
+// the value matches what l2_distance_approximate / inner_product_approximate are
+// defined to return. Lower-is-better for l2.
+//
+// The two names deliberately differ, since the two values mean different things --
+// and keeping them distinct leaves __paimon_search_score free should a later
+// version want to expose the raw score as its own column.
+constexpr const char* kPaimonSearchDistanceColumn = "__paimon_search_score_to_dis";
+
+// The Arrow field name paimon-rust auto-appends to a vector-search batch (see the
+// bindings' materialize_search_result), so it cannot be changed from here; the
+// reader maps it onto kPaimonSearchDistanceColumn when filling the block.
+constexpr const char* kPaimonRustScoreField = "__paimon_search_score";
+
 // Converts Doris push-down conjuncts into a paimon-rust filter predicate.
 //
 // This mirrors PaimonPredicateConverter (the paimon-cpp variant) but targets the

be/src/format_v2/table/paimon_rust_table_reader.h

@@ -18,6 +18,8 @@
 #pragma once
 
 #include <cstddef>
+#include <cstdint>
+#include <limits>
 #include <map>
 #include <memory>
 #include <optional>
@@ -30,6 +32,7 @@
 #include "cctz/time_zone.h"
 #include "common/status.h"
 #include "format_v2/table_reader.h"
+#include "gen_cpp/PlanNodes_types.h" // TVectorMetric
 #include "runtime/runtime_profile.h"
 
 namespace arrow {
@@ -61,6 +64,75 @@ class PaimonRustTableReader final : public format::TableReader {
     PaimonRustTableReader();
     ~PaimonRustTableReader() override;
 
+    // How paimon-rust encoded the distance into its __paimon_search_score field, and
+    // hence which transform the reader has to undo before the value lands in the
+    // __paimon_search_score_to_dis block column. One enumerator per metric Doris can
+    // push down (v1: l2 / inner_product; cosine is rejected by the FE rule and by
+    // parse_score_transform, so it deliberately has no enumerator here).
+    //
+    // Adding a metric is meant to be compiler-guided: add the enumerator, and the
+    // switch in _convert_score_to_distance stops compiling (-Wswitch is an error in
+    // be/src) until its inverse is written. Do not add a `default:` to that switch
+    // or that guarantee is lost.
+    enum class ScoreTransform {
+        // score = 1/(1 + squaredL2). Inverse: sqrt(1/score - 1).
+        L2,
+        // score = the inner product itself. Inverse: identity.
+        InnerProduct,
+    };
+
+    // Map an Arrow field name onto the block column name that receives its values.
+    //
+    // paimon-rust resolves projections case-insensitively, so the Arrow batch carries
+    // the table's canonical field casing (e.g. `is_QT`) while block columns are
+    // FE-normalized to lowercase (`is_qt`). Fold the Rust-side name down so the two match.
+    //
+    // The vector-search field is additionally renamed: paimon-rust calls it
+    // kPaimonRustScoreField ("__paimon_search_score") and puts a score in it, but the
+    // reader converts that to a distance in place, so FE asks for the column as
+    // kPaimonSearchDistanceColumn ("__paimon_search_score_to_dis"). The fill loop
+    // skips Arrow fields it cannot match (they may legitimately be absent from the
+    // block), so without this alias the rename would not fail loudly -- the distance
+    // column would just come back empty.
+    static std::string arrow_field_to_block_column(const std::string& arrow_name);
+
+    // Map a thrift TVectorMetric onto its score transform. Unknown or unsupported
+    // metrics (including COSINE) are an error rather than a silent fallback: guessing
+    // would return wrong distances instead of failing the query. Takes the thrift enum
+    // directly so there is no string round-trip between FE and the ScoreTransform enum.
+    static Status parse_score_transform(TVectorMetric::type metric, ScoreTransform* transform);
+
+    // Rewrite the reader-produced __paimon_search_score_to_dis column in place,
+    // turning paimon-rust's score back into the distance the Doris function is
+    // defined to return (see ScoreTransform for the per-metric formulas).
+    //
+    // Only rewrites [row_offset, row_offset + num_rows): block columns are
+    // append-only and a block is reused across get_block calls, so starting at 0
+    // would square-root earlier rows again. Depends on nothing but its arguments,
+    // hence static and public (also what makes it unit-testable without the
+    // paimon-rust FFI handles).
+    //
+    // A score outside l2's legal (0, 1] range is index corruption, and computing
+    // the inverse anyway would yield a non-finite distance (see the two cases
+    // below). Such rows are emitted as +inf instead: +inf sorts LAST in an ASC
+    // Top-K, so a corrupt or infinite row can never displace a valid closer row
+    // from the coordinator's final ordering. A score of exactly 0 is reachable
+    // from a genuine (huge) distance — the exact fallback in
+    // vector-index.search-mode=full computes the squared L2 in f32, so norms
+    // above ~1.8e19 overflow to +inf and score = 1/(1+d^2) rounds to 0 — which
+    // is why the boundary maps to +inf rather than to an error.
+    static constexpr float kCorruptScoreDistance = std::numeric_limits<float>::infinity();
+    // paimon-rust's lumina kernel computes the squared L2 distance as
+    // ||a||^2 + ||b||^2 - 2<a,b> (vectorized dot), which suffers catastrophic
+    // cancellation when a ~= b: the three terms nearly cancel and the residual
+    // can be a small NEGATIVE number, so score = 1/(1+d^2) lands slightly ABOVE 1
+    // (e.g. 1.00294). That is a numerical artifact for an ~exact hit, not index
+    // corruption -- clamp it to distance 0. Only a score clearly past 1 (beyond
+    // this tolerance) is treated as a genuine corrupt row and sentineled.
+    static constexpr float kScoreAboveOneTolerance = 1e-2F;
+    static void _convert_score_to_distance(Block* block, uint32_t block_idx, size_t row_offset,
+                                           size_t num_rows, ScoreTransform transform);
+
     Status init(format::TableReadOptions&& options) override;
     // Each split's Rust read builder applies the batch size stored by the base class
     // before opening the Arrow reader.
@@ -85,6 +157,9 @@ class PaimonRustTableReader final : public format::TableReader {
     void TEST_set_projected_columns(std::vector<format::ColumnDefinition> columns) {
         _projected_columns = std::move(columns);
     }
+    // The projection the reader would request for the given range; exposed so the
+    // synthetic-distance-column exclusion (vector mode only) can be unit-tested.
+    std::vector<std::string> TEST_build_read_columns(const TFileRangeDesc& range) const;
     void TEST_set_partition_values(std::map<std::string, Field> values) {
         _partition_values = std::move(values);
     }
@@ -108,9 +183,31 @@ class PaimonRustTableReader final : public format::TableReader {
     // Fail fast (without filesystem IO) when the schema-json pipeline fields are
     // missing from the split; the error text points at an FE/BE protocol mismatch.
     Status _validate_rust_split(const TFileRangeDesc& range) const;
+    // Validate the external_search_request thrift fields before any use (mirrors
+    // LanceTableReader::_validate_external_search_request): one pass over __isset
+    // and value-range checks so the vector build path can use the fields unchecked.
+    Status _validate_external_search_request() const;
+    // True when this split carries a PK-vector search payload and the reader must
+    // open the vector-search read path instead of the normal split read.
+    static bool _is_vector_mode(const TFileRangeDesc& range);
+    // Resolve the table root path + schema json and open (or reuse the cached)
+    // paimon_table handle from them. Shared by the normal-read and vector-search
+    // build paths; fills _handles->table and _opened_table_key.
+    Status _open_paimon_table(const TFileRangeDesc& range);
     // Open (or reuse the cached) paimon_table and build the per-split read
     // pipeline: read_builder -> projection -> filter -> plan -> arrow reader.
     Status _open_split_reader(const TFileRangeDesc& range);
+    // Open the table and build a primary-key vector (ANN) search reader over the
+    // FE-planned bucket split. Used when the scan range carries a
+    // TPaimonVectorPayload: the returned Arrow stream is best-first and carries the
+    // reader-produced distance column (__paimon_search_score_to_dis, converted
+    // from paimon-rust's __paimon_search_score field).
+    Status _open_vector_split_reader(const TFileRangeDesc& range);
+    // Fill the output block from one arrow record batch: arrow columns are
+    // matched by name (exact then lower-case, v1 semantics) onto fixed projected
+    // positions, then _fill_non_arrow_columns back-fills the rest.
+    Status _fill_block_from_record_batch(const std::shared_ptr<arrow::RecordBatch>& batch,
+                                         Block* block, size_t rows);
     // Drop the per-split pipeline (read_builder .. record batch reader), keeping
     // the cached table handle for the next split of the same table.
     void _close_split_reader();
@@ -120,11 +217,6 @@ class PaimonRustTableReader final : public format::TableReader {
     // apply it to the read builder. Best effort: non-convertible conjuncts are
     // dropped silently and re-applied by the scanner row-level filter.
     Status _apply_predicate();
-    // Fill the output block from one arrow record batch: arrow columns are
-    // matched by name (exact then lower-case, v1 semantics) onto fixed projected
-    // positions, then _fill_non_arrow_columns back-fills the rest.
-    Status _fill_block_from_record_batch(const std::shared_ptr<arrow::RecordBatch>& batch,
-                                         Block* block, size_t rows);
     // Materialize projected columns the arrow batch did not contain: partition
     // constants, default expressions, or a default/NULL fill as a last resort.
     Status _fill_non_arrow_columns(Block* block, size_t rows,
@@ -169,6 +261,12 @@ class PaimonRustTableReader final : public format::TableReader {
     std::unordered_map<std::string, size_t> _output_name_to_idx;
     TFileRangeDesc _current_range;
     cctz::time_zone _ctz;
+    // Which score transform the incoming __paimon_search_score field needs undone,
+    // derived from the FE-supplied metric when the vector reader is built. Empty
+    // outside vector mode (and only then): an unset or unknown metric fails the
+    // scan rather than defaulting, since guessing l2 would wrongly sqrt
+    // inner_product results.
+    std::optional<ScoreTransform> _score_transform;
     // EOF belongs to the current split; reset when a new split is prepared.
     bool _split_eof = false;