github-actions[bot] commented on code in PR #68988:
URL: https://github.com/apache/doris/pull/68988#discussion_r4311684744
##########
gensrc/thrift/PlanNodes.thrift:
##########
@@ -385,6 +406,10 @@ struct TPaimonFileDesc {
// paimon-rust: non-default branch name. Unset means main branch (matches
// upstream paimon commit 742da63: null-if-DEFAULT_MAIN_BRANCH).
19: optional string paimon_branch
+ // Primary-key vector (ANN) search payload for this split. When set, BE
+ // routes to the paimon-rust vector-search read path instead of the normal
+ // split read; paimon_split is set but empty in that case.
+ 20: optional TPaimonVectorPayload vector_payload;
Review Comment:
[P1] Reserve field IDs that are unused across release branches
Field 20 already means `optional string original_file_path` in
`TPaimonFileDesc` on master (`e0bf4874070ba0b6837ad93f695ea63532b77a47`). This
PR reuses it for a struct. The new
`TFileScanRangeParams.external_search_request` similarly reuses field 41,
already `map<string,string> fluss_properties` on master. These are incompatible
wire identities: mixed-version readers skip the mismatched types and lose the
payload/request instead of receiving the intended fields. Allocate fresh shared
IDs for both additions while preserving master's existing reservations; marking
them optional does not make ID reuse compatible.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/source/PaimonScanNode.java:
##########
@@ -982,72 +1008,156 @@ && splitHasOrcFile((DataSplit) split)
// rust. Gate until the rust leaf has a Variant Arrow decoder.
boolean projectedVariant = desc.getSlots().stream()
.anyMatch(slot ->
PaimonUtil.containsVariant(slot.getType()));
- // Scheme capability gate: the pinned paimon-rust storage
- // dispatcher (io/storage.rs) selects the FileIO parser from the
- // table location's URI scheme, and libpaimon_c.a compiles in
- // separate COS, OBS, GCS and Azdls parsers besides the OSS and S3
- // ones. Doris normalizes every object store's credentials into
- // the AWS_* / use_path_style aliases (see the *Properties storage
- // classes), which the BE rust bridge translates only into the
- // fs.oss.* and s3.* key families — a cosn:// / obs:// / gs:// /
- // abfs:// warehouse would reach its scheme's parser without the
- // key family it reads (fs.cosn.userinfo.*, fs.obs.*, gcs.*,
- // azure.*) and fail the open instead of using JNI. Only the
- // schemes whose property translation is implemented and
- // open-tested (s3 / s3a / oss, via RUST_VERIFIED_LOCATION_SCHEMES)
- // plus the credential-free hdfs and local-filesystem parsers stay
- // rust-eligible; every other scheme falls back to JNI. A null
- // location also routes to JNI: the rust path needs the
- // paimon_table that only a real location can provide (BE rejects
- // a split without it).
- boolean schemeCapabilityVerified =
isRustVerifiedLocationScheme(source.getTableLocation());
- // An hdfs:// location is scheme-verified only together with the
credential-free
- // backend shape: the backend storage properties that ship to BE
also carry an
- // HDFS catalog's authentication (kerberos principal / keytab,
proxy user, HA
- // nameservice config), none of which the pinned rust HDFS parser
reads — the
- // scan would open as the BE process's ambient identity instead of
the
- // catalog's configured one and fail the access JNI honors. See
- // isRustVerifiedHdfsBackend.
- boolean hdfsBackendVerified =
!isHdfsLocationScheme(source.getTableLocation())
- || isRustVerifiedHdfsBackend(backendStorageProperties);
- // With merge-on-read=true the whole table already routes to JNI
(dvMergeOnRead);
- // for the remaining partial-update/aggregation DV tables, a split
that is
+ // With merge-on-read=true the whole table already routes to JNI
+ // (rustEligibility.dvMergeOnRead); for the remaining
+ // partial-update/aggregation DV tables, a split that is
// not fully materialized (uncompacted level-0 data, or
retractions not
// known to be applied — even a split with no deletion file
attached yet)
// fails the rust is_fully_materialized_pk_dv guard, so it falls
back per
// split instead of turning into a BE-open failure. nativeSplit and
// !fallbackRead only guard the cast — those splits already route
to JNI.
- boolean splitDvNotMaterialized = puAggDeletionVectors &&
!dvMergeOnRead
+ boolean splitDvNotMaterialized =
rustEligibility.puAggDeletionVectors
+ && !rustEligibility.dvMergeOnRead
&& nativeSplit && !fallbackRead
&& !isFullyMaterializedPkDvSplit((DataSplit) split);
+ boolean vectorSearchSplit = paimonSplit.getVectorPayloadBytes() !=
null;
+ // Session, shared table / storage / backend, and scan-shape
gates. The
+ // shared PaimonRustEligibility computation is also the
+ // PushDownVectorTopNIntoPaimonScan rewrite's pre-flight
+ // (PaimonVectorSearch.vectorScanEligible): the rewrite is
one-way, so both
+ // sites must agree on the same gates, and the scan node stays
authoritative
+ // — it evaluates the relation-option-processed table, so a
t@options
+ // override the raw-table pre-flight cannot see still rejects here
(loudly,
+ // as the vector path has no JNI fallback). !isIncremental and the
+ // split-half of the fallback-read gate are the same backstops: an
+ // incremental scan has no PK-vector semantics (the VectorScan
builder has
+ // no incremental mode), and a FallbackSplit shape cannot ride the
vector
+ // payload either — both fail loudly here rather than silently
returning
+ // snapshot rows for a query that asked for changes. Each gate's
value is
+ // also what the fallback logging below reports.
boolean canUseRust = sessionVariable.isEnablePaimonRustReader()
- && sessionVariable.enableFileScannerV2 && nativeSplit &&
!fallbackRead
- && !isIncremental && providerModeTranslatable &&
!queryAuthTable
- && !restTokenTable
- && !dvMergeOnRead && !splitDvNotMaterialized
- && !orcLtzSchema && !projectedVariant && !externalFileSplit
- && !deduplicateIgnoreDelete && !rustUnsupportedMergeOption
- && schemeCapabilityVerified && hdfsBackendVerified
- && paimonFileStoreTable != null;
- if (canUseRust) {
- // Branch tables can retain persisted modes without a
selector. Validate the
- // transported options too: the pinned Rust builder only
accepts default here.
- String scanMode =
PaimonScanParams.withoutTimeTravelSelectors(paimonFileStoreTable.schema())
- .options().get(CoreOptions.SCAN_MODE.key());
- canUseRust = scanMode == null ||
"default".equalsIgnoreCase(scanMode);
- }
- if (canUseRust) {
- // Every candidate can receive this range; older BEs cannot
decode reader type 3.
- canUseRust = !backendPolicy.getBackends().isEmpty()
- &&
backendPolicy.getBackends().stream().allMatch(Backend::isPaimonRustReaderSupported);
- }
- if (canUseRust) {
- if (rustReaderCapabilities == null) {
- rustReaderCapabilities = new
PaimonRustReaderCapabilities(paimonFileStoreTable, desc);
+ && sessionVariable.enableFileScannerV2
+ && rustEligibility.eligible()
+ && !fallbackRead && !isIncremental
+ && !orcLtzSchema && !projectedVariant;
+ // The remaining gates judge the DataSplit the BATCH reader opens:
the
+ // native DataSplit shape, unmaterialized partial-update /
aggregation
+ // deletion vectors, external-path files, and the
+ // PaimonRustReaderCapabilities whitelist. A PK-vector search
split is
+ // never read as a DataSplit (it rides serialized in
vector_payload and
+ // executes paimon-rust's eager Top-K chain, which applies deletion
+ // vectors and the merge inside the search itself), so it does not
+ // consult them.
+ if (!vectorSearchSplit && canUseRust) {
Review Comment:
[P1] Keep the nested-projection gate on vector materialization
With nested pruning enabled (the default), this reduced plan is accepted:
```
TopN(dis)
Project(struct_element(payload,'b'), dis := l2_distance_approximate(vec,q))
PaimonScan(payload ROW<a INT,b INT>, indexed vec)
```
The scan slot is pruned to `ROW<b INT>`, but vector materialization projects
only the root name `payload` and returns the full Arrow `ROW<a,b>`. Skipping
`PaimonRustReaderCapabilities.canRead` also skips `hasFullNestedProjection`;
`DataTypeStructSerDe::read_column_from_arrow` then binds child 0 by ordinal,
returning `a` as `b` in release builds (or failing its field-count DCHECK in
debug). The ordinary path explicitly falls back for this shape. Check
projection compatibility before committing the rewrite, or carry a recursive
projection into the vector reader. This is a distinct instance from the prior
VARIANT thread: an ordinary STRUCT is decoded with the wrong child layout.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/source/PaimonScanNode.java:
##########
@@ -1413,6 +1531,87 @@ public Map<String, String> getIncrReadParams() throws
UserException {
return paimonScanParams;
}
+ /**
+ * Plan a primary-key vector (ANN) search. Produces one Doris split per
+ * (partition, bucket) {@link BucketVectorSearchSplit}, serialized with the
+ * native "PKVSPLIT"/v1 protocol so paimon-rust on BE can decode it.
Planning
+ * only needs the vector column and the partition-half of the predicate;
the
+ * query vector and data (residual) predicates are handled by BE.
+ */
+ @VisibleForTesting
+ public List<Split> getVectorSearchSplits() throws UserException {
+ Table paimonTable = getProcessedTable();
+ // Only the partition-half of the predicate is pushed into Paimon
planning.
+ // The data (residual) half is left for BE's vector-search filter, and
passing
+ // it here would trip PrimaryKeyVectorScan's DV/merge-on-read
pre-filter guard.
+ PartitionPredicate partitionPredicate = null;
+ if (predicates != null && !predicates.isEmpty()) {
+ Pair<Optional<PartitionPredicate>, List<Predicate>> partSplit =
+
PartitionPredicate.splitPartitionPredicatesAndDataPredicates(
+ predicates, paimonTable.rowType(),
paimonTable.partitionKeys());
+ partitionPredicate = partSplit.getLeft().orElse(null);
+ }
+
+ // Resolve the actual Paimon field name (case-insensitive); the
pushed-down
+ // column name comes from the Doris column, which may differ in case.
+ String vectorColumn = resolvePaimonFieldName(paimonTable,
annSortColumnName);
+
+ VectorScan.Plan plan;
+ try {
+ plan = paimonTable.newBatchVectorSearchBuilder()
+ .withVectorColumn(vectorColumn)
+ .withPartitionFilter(partitionPredicate)
+ .newVectorScan()
+ .scan();
+ } catch (Exception e) {
+ throw new UserException("Paimon PK-vector search planning failed:
" + e.getMessage(), e);
+ }
+
+ List<Split> splits = new ArrayList<>();
+ Set<BinaryRow> partitions = new HashSet<>();
+ for (VectorSearchSplit vectorSplit : plan.splits()) {
+ if (!(vectorSplit instanceof BucketVectorSearchSplit)) {
+ throw new UserException("Unexpected PK-vector split type: "
+ + vectorSplit.getClass().getName());
+ }
+ BucketVectorSearchSplit bucketSplit = (BucketVectorSearchSplit)
vectorSplit;
+ // Serialize with Paimon's own DataOutputSerializer: the bytes
must be
+ // exactly BucketVectorSearchSplit.serialize's DataOutputView form
+ // (magic "PKVSPLIT", version 1, big-endian), which
+ // paimon_bucket_vector_search_split_deserialize on BE decodes.
This
+ // repository's Paimon no longer ships a
BucketVectorSearchSplitSerializer.
+ byte[] payloadBytes;
+ try {
+ DataOutputSerializer output = new DataOutputSerializer(256);
+ bucketSplit.serialize(output);
+ payloadBytes = output.getCopyOfBuffer();
+ } catch (IOException e) {
+ throw new UserException("Failed to serialize PK-vector split:
" + e.getMessage(), e);
+ }
+ DataSplit dataSplit = bucketSplit.dataSplit();
+ PaimonSplit split = new PaimonSplit(dataSplit);
+ split.setVectorPayloadBytes(payloadBytes);
Review Comment:
[P1] Populate partition values on each vector split
These splits never receive `setPaimonPartitionValues`, so their connector
partition map remains null. During range construction, `setPaimonParams`
unconditionally calls `setPartitionValues`, which throws `Missing partition
values for Paimon partitioned table` whenever the table has partition keys.
Thus a nonempty ANN query on a partitioned PK-vector table fails even if it
only projects id/distance. The ordinary split path derives this map with
`PaimonUtil.getPartitionInfoMap`; populate the same metadata here and cover
partitioned vector range construction.
##########
be/test/format_v2/table/paimon_rust_table_reader_test.cpp:
##########
@@ -38,6 +40,7 @@
#include "core/field.h"
#include "exprs/vexpr_context.h"
#include "exprs/vliteral.h"
+#include "format/table/paimon_rust_predicate_converter.h"
Review Comment:
[P1] Include the predicate header from its existing v2 path
`format/table/paimon_rust_predicate_converter.h` does not exist in this
checkout; the header is
`be/src/format_v2/table/paimon_rust_predicate_converter.h` (also the path used
by the production reader). `be/test/CMakeLists.txt` recursively includes this
test in `doris_be_test`, so its new include prevents that target from
preprocessing. Change it to `format_v2/table/paimon_rust_predicate_converter.h`.
##########
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:
[P1] Preserve the ordering of zero L2 scores
On a table with `vector-index.search-mode=full` and active files not yet
indexed, a zero score can come from valid finite vectors: the pinned Rust exact
fallback computes squared L2 in f32, so `[1e20f,0,...]` against `[0,0,...]`
overflows to `+inf` and its `1/(1+d)` score is zero. With that row alone in one
bucket and `[1f,0,...]` in another, `ORDER BY l2_distance_approximate(vec,q)
LIMIT 1` receives both bucket candidates. This branch changes the far row's
distance to `-2`, making the coordinator choose it ahead of the distance-1 row;
the ordinary function returns `+inf` and 1 respectively. Preserve `+inf` for
the zero-score boundary, or fail explicitly on invalid scores, rather than
publishing negative distances that reverse the final TopN.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/PushDownVectorTopNIntoPaimonScan.java:
##########
@@ -0,0 +1,513 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.nereids.rules.rewrite;
+
+import org.apache.doris.analysis.TableScanParams;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.catalog.Type;
+import org.apache.doris.datasource.credentials.CredentialUtils;
+import org.apache.doris.datasource.credentials.VendedCredentialsFactory;
+import org.apache.doris.datasource.paimon.PaimonExternalCatalog;
+import org.apache.doris.datasource.paimon.PaimonExternalTable;
+import org.apache.doris.datasource.paimon.PaimonUtil;
+import org.apache.doris.datasource.paimon.source.PaimonVectorSearch;
+import org.apache.doris.datasource.property.storage.StorageProperties;
+import org.apache.doris.nereids.rules.Rule;
+import org.apache.doris.nereids.rules.RuleType;
+import org.apache.doris.nereids.trees.expressions.Alias;
+import org.apache.doris.nereids.trees.expressions.ComparisonPredicate;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.InPredicate;
+import org.apache.doris.nereids.trees.expressions.IsNull;
+import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.Not;
+import org.apache.doris.nereids.trees.expressions.SlotReference;
+import org.apache.doris.nereids.trees.expressions.StatementScopeIdGenerator;
+import
org.apache.doris.nereids.trees.expressions.functions.scalar.InnerProductApproximate;
+import
org.apache.doris.nereids.trees.expressions.functions.scalar.L2DistanceApproximate;
+import org.apache.doris.nereids.trees.expressions.literal.ArrayLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.Literal;
+import org.apache.doris.nereids.trees.plans.AnnTopNInfo;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalFileScan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalFilter;
+import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
+import org.apache.doris.nereids.trees.plans.logical.LogicalTopN;
+import org.apache.doris.nereids.util.ExpressionUtils;
+import org.apache.doris.qe.ConnectContext;
+import org.apache.doris.thrift.TVectorMetric;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Maps;
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.Table;
+import org.apache.paimon.types.ArrayType;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DataTypeRoot;
+import org.apache.paimon.types.MapType;
+import org.apache.paimon.types.MultisetType;
+import org.apache.paimon.types.RowType;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+
+/**
+ * Push a primary-key vector (ANN) top-N down into the Paimon scan
(index-only).
+ *
+ * <p>Recognizes {@code TopN[order by dis] -> Project[dis := dist_fn(vec, q)]
-> [Filter] -> PaimonScan}
+ * where {@code dist_fn} is {@code l2_distance_approximate} (ASC) or
+ * {@code inner_product_approximate} (DESC) and the Paimon table has a
PK-vector
+ * index on {@code vec} with a matching distance metric.
+ *
+ * <p>The rewrite mirrors the internal-table ANN path (see
+ * {@code PushDownVectorTopNIntoOlapScan}): the scan is told to produce a
+ * reader-generated {@code __paimon_search_score_to_dis} column via the
PK-vector index, and
+ * every occurrence of {@code dist_fn(vec, q)} is replaced by that column's
slot. The
+ * order key slot keeps its identity, so the surrounding TopN merges/limits
unchanged.
+ * The distance always comes from the index, never from the raw vector.
+ *
+ * <p>No arithmetic is emitted here even though paimon-rust reports a score
rather than
+ * a distance: the BE reader undoes the metric-dependent score transform in
place while
+ * filling the block, exactly as the internal table sqrt's faiss's squared
distance inside
+ * the index layer. So the column arrives already holding what {@code dist_fn}
would
+ * return, and the plan stays a bare slot reference -- which also keeps {@code
dis}
+ * non-nullable (an expression over the score would not be, since {@code
Divide} and
+ * {@code Sqrt} are always nullable) and makes predicates like
+ * {@code dist_fn(vec, q) < 10} correct without any direction flip.
+ *
+ * <p>Being index-only is just an optimization here, not a precondition: if
nothing
+ * else references the raw vector column it is left unreferenced and dropped
by column
+ * pruning (pure index-only read); if the query still needs it ({@code SELECT
vec} or a
+ * predicate on {@code vec}), it stays a normal projected column and the BE
vector
+ * reader materializes it alongside the score. Either way ANN is applied --
the rewrite
+ * never falls back merely because the vector is read.
+ */
+public class PushDownVectorTopNIntoPaimonScan implements RewriteRuleFactory {
+
+ /** Doris column types the BE paimon-rust predicate converter can push; see
+ * PaimonRustPredicateConverter._is_supported_slot_type on the BE side. */
+ private static final Set<Type> VECTOR_FILTER_SUPPORTED_COLUMN_TYPES =
ImmutableSet.of(
+ Type.BOOLEAN, Type.TINYINT, Type.SMALLINT, Type.INT, Type.BIGINT,
+ Type.STRING, Type.DATE, Type.DATEV2);
+
+ @Override
+ public List<Rule> buildRules() {
+ return ImmutableList.of(
+ logicalTopN(logicalProject(logicalFileScan())).when(t ->
t.getOrderKeys().size() == 1).then(topN -> {
+ LogicalProject<LogicalFileScan> project = topN.child();
+ LogicalFileScan scan = project.child();
+ return pushDown(topN, project, scan, Optional.empty());
+ }).toRule(RuleType.PUSH_DOWN_VECTOR_TOPN_INTO_PAIMON_SCAN),
+ logicalTopN(logicalProject(logicalFilter(logicalFileScan())))
+ .when(t -> t.getOrderKeys().size() == 1).then(topN -> {
+ LogicalProject<LogicalFilter<LogicalFileScan>>
project = topN.child();
+ LogicalFilter<LogicalFileScan> filter =
project.child();
+ LogicalFileScan scan = filter.child();
+ return pushDown(topN, project, scan,
Optional.of(filter));
+
}).toRule(RuleType.PUSH_DOWN_VECTOR_TOPN_INTO_PAIMON_SCAN)
+ );
+ }
+
+ private Plan pushDown(
+ LogicalTopN<?> topN,
+ LogicalProject<?> project,
+ LogicalFileScan scan,
+ Optional<LogicalFilter<?>> optionalFilter) {
+ // Only Paimon external tables carry PK-vector indexes.
+ if (!(scan.getTable() instanceof PaimonExternalTable)) {
+ return null;
+ }
+
+ // The index-only search path is executed exclusively by the BE
paimon-rust reader,
+ // which only FileScannerV2 can select (the V1 FileScanner explicitly
rejects
+ // PAIMON_RUST). If the rust reader or the V2 scanner is disabled, the
split would
+ // fall through to the JNI reader, which ignores the vector payload
and would
+ // return a non-ANN scan with an empty distance column. Fall back to a
normal
+ // scan then.
+ ConnectContext ctx = ConnectContext.get();
+ if (ctx == null || ctx.getSessionVariable() == null
+ || !ctx.getSessionVariable().isEnablePaimonRustReader()
+ || !ctx.getSessionVariable().enableFileScannerV2) {
+ return null;
+ }
+
+ // An incremental scan (t@incr / BETWEEN) must not be rewritten into a
+ // PK-vector search: the VectorScan builder has no incremental mode,
so a
+ // vector payload would silently return snapshot rows for a query that
asked
+ // for changes. Keeping the query un-rewritten runs it as the ordinary
+ // incremental scan the user requested, with the distance computed in
Doris.
+ // (The scan node's vector gate keeps !isIncremental as the
authoritative
+ // backstop for any path that produces a vector split regardless.)
+ //
+ // The same holds for a pinned file-creation-time read state: the
ordinary
+ // path enforces `scan.file-creation-time-millis` by dropping manifest
+ // entries older than the bound, but the vector search plans its own
splits
+ // and never consults that bound, so old files' nearest neighbors
could leak
+ // into the result. Snapshot/tag/version selectors are fine -- they are
+ // resolved into the table handle before the vector search -- so only
the
+ // file-level bound forces the decline.
+ if (scan.getScanParams().isPresent()) {
+ TableScanParams scanParams = scan.getScanParams().get();
+ if (scanParams.incrementalRead()) {
+ return null;
+ }
+ Map<String, String> rawParams = scanParams.getMapParams();
+ if
(rawParams.containsKey(CoreOptions.SCAN_FILE_CREATION_TIME_MILLIS.key())) {
+ return null;
+ }
+ }
+
+ // A bounded vector top-N is only correct when every filter conjunct is
+ // applied inside the search, before the limit is exhausted. The BE
+ // paimon-rust predicate converter skips conjuncts it cannot represent
(and
+ // the synthetic distance column is not in its field registry at all),
and
+ // the engine re-applies those residuals on the returned block --
after the
+ // top-K rows were already selected. A skipped residual can therefore
drop
+ // rows out of a bounded result: with candidate distances 0, 2 and 3, a
+ // `WHERE dis > 1 ORDER BY dis LIMIT 1` would fetch only the
distance-0 row,
+ // filter it out and return nothing instead of the distance-2 row. So
the
+ // rewrite only fires when each conjunct is provably of a form the
converter
+ // pushes: a simple comparison / IN / null-check on a table column
whose
+ // Doris type is in the converter's supported set (see
+ // PaimonRustPredicateConverter._is_supported_slot_type on BE).
Anything
+ // else -- an expression predicate, a predicate on the distance
itself, an
+ // unsupported column type -- declines the rewrite and the query runs
as an
+ // ordinary scan with the distance computed in Doris, which is always
+ // correct.
+ if (optionalFilter.isPresent() &&
!conjunctsRideIntoVectorSearch(optionalFilter.get())) {
+ return null;
+ }
+
+ // The order key must be a SlotReference produced by a Project alias.
+ Expression orderKey = topN.getOrderKeys().get(0).getExpr();
+ if (!(orderKey instanceof SlotReference)) {
+ return null;
+ }
+ SlotReference keySlot = (SlotReference) orderKey;
+ Alias orderKeyAlias = null;
+ Expression orderKeyExpr = null;
+ for (NamedExpression projection : project.getProjects()) {
+ if (projection.toSlot().equals(keySlot) && projection instanceof
Alias) {
+ orderKeyAlias = (Alias) projection;
+ orderKeyExpr = orderKeyAlias.child();
+ break;
+ }
+ }
+ if (orderKeyExpr == null) {
+ return null;
+ }
+
+ // l2 must sort ascending (smaller distance = closer); inner_product
descending.
+ boolean isAsc = topN.getOrderKeys().get(0).isAsc();
+ boolean l2Dist = orderKeyExpr instanceof L2DistanceApproximate;
+ boolean innerProduct = orderKeyExpr instanceof InnerProductApproximate;
+ if (!(l2Dist && isAsc) && !(innerProduct && !isAsc)) {
+ return null;
+ }
+
+ // The indexed column must be a direct slot reference. A Cast chain
over the
+ // column is NOT stripped: a legal cast can change vector values (for
example
+ // an ARRAY<FLOAT> -> ARRAY<INT> -> ARRAY<FLOAT> round trip loses
precision),
+ // and searching the original vectors while the query means the casted
ones
+ // changes both the distance and the row order. Doris's implicit
coercions do
+ // not produce a cast here for a matching ARRAY<FLOAT> column, so
anything
+ // wrapped in a Cast simply declines the rewrite.
+ Expression left;
+ Expression right;
+ if (l2Dist) {
+ left = ((L2DistanceApproximate) orderKeyExpr).left();
+ right = ((L2DistanceApproximate) orderKeyExpr).right();
+ } else {
+ left = ((InnerProductApproximate) orderKeyExpr).left();
+ right = ((InnerProductApproximate) orderKeyExpr).right();
+ }
+ if (!(left instanceof SlotReference && right.isConstant() && right
instanceof ArrayLiteral)) {
+ return null;
+ }
+ SlotReference vectorSlot = (SlotReference) left;
+ if (!vectorSlot.getOriginalColumn().isPresent() ||
!vectorSlot.getOriginalTable().isPresent()) {
+ return null;
+ }
+ String annColumnName = vectorSlot.getOriginalColumn().get().getName();
+
+ // Validate the PK-vector index and metric on the Paimon table.
+ PaimonExternalTable paimonTable = (PaimonExternalTable)
scan.getTable();
+ CoreOptions coreOptions;
+ try {
+ Table basePaimonTable =
paimonTable.getPaimonTable(Optional.empty());
+ coreOptions = CoreOptions.fromMap(basePaimonTable.options());
+ } catch (Throwable t) {
+ return null;
+ }
+ if (!coreOptions.primaryKeyVectorIndexEnabled()) {
+ return null;
+ }
+ boolean columnIndexed =
coreOptions.primaryKeyVectorIndexColumns().stream()
+ .anyMatch(c -> c.equalsIgnoreCase(annColumnName));
+ if (!columnIndexed) {
+ return null;
+ }
+ // v1 supports only l2 / inner_product; the function must match the
index metric.
+ String metric =
coreOptions.primaryKeyVectorDistanceMetric(annColumnName);
+ String expectedMetric = l2Dist ? "l2" : "inner_product";
+ if (!expectedMetric.equals(metric)) {
+ return null;
+ }
+
+ // Pre-flight the projection-dependent bounds the vector split would
hit at
+ // the scan node: a PK-vector split has no JNI fallback (it would
return an
+ // empty distance column), so a query whose projected columns would set
+ // projectedVariant -- or an ORC table with TIMESTAMP_LTZ in its
schema, the
+ // orc-ltz-shape bound -- must not be rewritten at all. Without the
rewrite
+ // those queries run through the ordinary paths, where they are
supported.
+ try {
+ if (projectedColumnsContainVariant(project, optionalFilter)
+ || orcSchemaContainsTimestampLtz(paimonTable)) {
+ return null;
+ }
+ } catch (Throwable t) {
+ return null;
+ }
+
+ // The thrift-level enum the AnnTopNInfo will carry. expectedMetric is
kept as a
+ // string for the comparison above (coreOptions returns a string), but
the
+ // AnnTopNInfo / wire contract uses TVectorMetric, so map once here.
DOT_PRODUCT
+ // is the thrift enum that paimon-rust's "inner_product" score
transform maps to
+ // (see the BE reader's parse_score_transform).
+ TVectorMetric expectedMetricEnum = l2Dist ? TVectorMetric.L2 :
TVectorMetric.DOT_PRODUCT;
+
+ // Pre-flight the rust eligibility the scan node will gate on. The
rewrite is
+ // one-way: a PK-vector split has no JNI fallback (the JNI reader
ignores the
+ // vector payload and would return a non-ANN scan with an empty
distance
+ // column), so the scan node can only fail loudly — a table whose gate
would
+ // fail (credential provider shape, REST-token IO, query-auth,
merge-option
+ // matrix, deletion-vector merge-on-read, scan-mode, storage scheme or
HDFS
+ // credentials, no rust-capable backend) must not be rewritten at all.
Without
+ // the rewrite the query still runs, computing the distance in Doris.
The
+ // projection-dependent bounds (ORC LTZ, projected variant) need the
final scan
+ // tuple and stay scan-node-side, and the scan node re-evaluates on the
+ // relation-option-processed table, so it stays authoritative for
t@options
+ // overrides this raw-table pre-flight cannot see.
+ try {
+ Table baseTable = paimonTable.getPaimonTable(Optional.empty());
+ PaimonExternalCatalog catalog = (PaimonExternalCatalog)
paimonTable.getCatalog();
+ Map<StorageProperties.Type, StorageProperties>
storagePropertiesMap =
+
VendedCredentialsFactory.getStoragePropertiesMapWithVendedCredentials(
+
catalog.getCatalogProperty().getMetastoreProperties(),
+
catalog.getCatalogProperty().getStoragePropertiesMap(),
+ baseTable);
+ Map<String, String> backendStorageProperties =
+
CredentialUtils.getBackendPropertiesFromStorageMap(storagePropertiesMap);
+ String tableLocation = baseTable instanceof FileStoreTable
+ ? ((FileStoreTable) baseTable).location().toString()
+ : baseTable.options().get("path");
+ if (!PaimonVectorSearch.vectorScanEligible(baseTable,
tableLocation,
+ backendStorageProperties,
+ Env.getCurrentSystemInfo().getAllClusterBackends(true))) {
+ return null;
+ }
+ } catch (Throwable t) {
+ return null;
+ }
+ // NOTE: the FE-side data-predicate interception is intentionally not
done. The
+ // data predicate is handed in full to the BE via the vector search
builder's
+ // residual filter, where the Rust candidate search stage applies it
together
+ // with deletion vectors. If some table shape cannot be handled
correctly, it is
+ // Rust's responsibility to reject it or return correct data, not the
planner's.
+
+ // paimon-rust's vector index operates on f32, so carry the query as
floats
+ // rather than List<Double>: it avoids a double→float narrowing on the
wire
+ // and lets PaimonScanNode encode straight into a FLOAT32
TSearchVector.
+ List<Literal> vectorLiterals = ((ArrayLiteral) right).getValue();
+ float[] annQueryVector = new float[vectorLiterals.size()];
+ for (int i = 0; i < vectorLiterals.size(); i++) {
+ annQueryVector[i] = (float) vectorLiterals.get(i).getDouble();
Review Comment:
[P2] Reject NULL coordinates before encoding the query vector
An `ArrayLiteral` can contain a typed `NullLiteral`, whose `getDouble()`
returns zero. Consequently a correctly dimensioned query vector such as
`[CAST(NULL AS FLOAT),0.0,...]` becomes a search against a zero coordinate, and
the rewrite removes the original distance function. On a nonempty table the
ordinary `FunctionArrayDistance::_validate_no_nulls` throws `Second argument
... cannot have null`; enabling ANN instead returns rows for a different input.
Decline the rewrite for NULL elements or raise the equivalent error before
replacing the expression. Both supported metrics have this issue.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/source/PaimonRustEligibility.java:
##########
@@ -0,0 +1,298 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.datasource.paimon.source;
+
+import org.apache.doris.datasource.paimon.PaimonScanParams;
+import org.apache.doris.system.Backend;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.rest.RESTTokenFileIO;
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.table.FallbackReadFileStoreTable;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.Table;
+
+import java.util.Collection;
+import java.util.Locale;
+import java.util.Map;
+
+/**
+ * The table-, storage- and backend-level gates of the paimon-rust reader,
+ * computed once from the table handle and shared by every call site that must
+ * agree on them: {@code PaimonScanNode}'s per-split gate (both the batch
+ * reader and the PK-vector search) and the {@code
PushDownVectorTopNIntoPaimonScan}
+ * rewrite's pre-flight via {@code PaimonVectorSearch.vectorScanEligible}. The
+ * rewrite is one-way (a rewritten PK-vector split has no JNI fallback), so
+ * both sites must evaluate the same conditions; this class is that single
+ * definition.
+ *
+ * <p>Covers only what is a property of the table handle, the storage shape or
+ * the cluster's backends. Everything that depends on the individual split
+ * (native DataSplit shape, unmaterialized deletion vectors, external-path
+ * files, the PaimonRustReaderCapabilities whitelist) or on the final scan
+ * tuple (ORC TIMESTAMP_LTZ, projected VARIANT) stays with the scan node.
+ */
+final class PaimonRustEligibility {
+
+ // BE opens the table via paimon_table_from_schema_json, which needs the
resolved
+ // TableSchema that only FileStoreTable exposes via schema(). A table that
is not a
+ // FileStoreTable (e.g. a sys table backed by DataSplit) cannot ship a
schema JSON,
+ // so it falls back to JNI rather than sending an incomplete PAIMON_RUST
request
+ // that BE would reject.
+ final boolean isFileStoreTable;
+
+ // Both sides of a FallbackReadFileStoreTable wrap their splits, and the
pinned
+ // rust decoder rejects a FallbackDataSplit outright ("trailing bytes after
+ // DataSplit" — it requires full-buffer consumption); the wrapper is gated
as a
+ // whole (any split from it routes to JNI) until the rust ABI represents
both
+ // sides. The per-split FallbackSplit shape is judged separately in
PaimonScanNode.
+ final boolean fallbackReadTable;
+
+ // A renewable REST token reached the shipped properties as a plain value;
only
+ // the table's FileIO type reveals it expires: doInitialize snapshots
+ // RESTTokenFileIO.validToken().token() into the backend storage
properties,
+ // discarding expireAtMillis and the REST refresh context, so the shipped
+ // credentials look static — but the pinned rust table reuses one option
map with
+ // no refresh callback, while the JNI RESTTokenFileIO checks expiry before
each
+ // file operation and obtains a replacement token. A queued or long scan
that
+ // crosses the token TTL would start on rust and later fail authentication.
+ // Null-safe: a table handle whose FileIO is not resolved stays
rust-eligible,
+ // mirroring the CoreOptions null-safety below.
+ final boolean restTokenTable;
+
+ // query-auth.enabled tables stay on JNI: when catalog authorization
succeeds with
+ // no row filter or column mask, Paimon still leaves an ordinary DataSplit
+ // (restricted results use QueryAuthSplit and are already handled by the
+ // nativeSplit gate), so this table shape passes the compound gate — but
the
+ // shipped schema keeps query-auth.enabled=true and the pinned rust
ReadBuilder
+ // rejects every such table (its CoreOptions::ensure_read_authorized fails
closed
+ // because the client cannot enforce the row filter / column masking),
turning a
+ // valid authorized scan into a BE-open failure. Until the authorization
result
+ // can be transported and enforced by the rust ABI, these tables route to
JNI.
+ final boolean queryAuthTable;
+
+ // Partial-update / aggregation tables with deletion vectors: the pinned
rust
+ // read_pk rejects merge-engine=partial-update/aggregation with
+ // deletion-vectors.merge-on-read=true outright, so that table option
routes the
+ // whole table to JNI. merge-on-read has no Java accessor in the pinned
paimon, so
+ // it is read raw from the schema options map — the one the BE rust reader
+ // deserializes from the shipped schema JSON, so it cannot diverge from
what BE
+ // sees — with the rust parsing semantics (any case-insensitive "true" is
on,
+ // default false). For the remaining PU/AGG DV tables, per-split
materialization
+ // is judged in PaimonScanNode (splitDvNotMaterialized); Deduplicate stays
+ // rust-eligible (its read_pk routes uncompacted splits to the KV reader,
which
+ // applies the attached per-file DVs).
+ final boolean puAggDeletionVectors;
+ final boolean dvMergeOnRead;
+
+ // deduplicate.ignore-delete=true tables stay on JNI: Java's
+ // DeduplicateMergeFunction skips retract records when the option is set —
+ // including old, uncompacted files that still contain them — but the
pinned rust
+ // read_pk does not pass table options into its deduplicate merge: it
picks the
+ // latest row and omits the key when that row is DELETE/UPDATE_BEFORE. An
+ // uncompacted insert followed by a delete therefore returns the insert
through
+ // JNI but silently disappears through rust. Gate the option until the rust
+ // merge implements it.
+ final boolean deduplicateIgnoreDelete;
+
+ // Non-DV merge options the pinned rust read rejects: Java supports
+ // partial-update.remove-record-on-delete /
aggregation.remove-record-on-delete
+ // and the wider per-field retract matrix, but the rust
PartialUpdateConfig /
+ // AggregationConfig validations return Unsupported for them — and the
DV-derived
+ // gates above only cover deletion-vector tables, so an ordinary non-DV
DataSplit
+ // with one of these options would pass the compound gate and fail during
the
+ // rust merge construction. Mirrors the exact rust key matrix (presence,
not
+ // values) against the same schema options map BE deserializes.
+ final boolean rustUnsupportedMergeOption;
+
+ // Branch tables can retain persisted modes without a selector: the shipped
+ // schema options carry scan.mode, and the pinned Rust builders only
accept the
+ // default.
+ final boolean scanModeNotDefault;
+
+ // The credential provider shape is evaluated by providerModeTranslatable
below.
+ final boolean providerModeTranslatable;
+
+ // Scheme capability gate: the pinned paimon-rust storage dispatcher
(io/storage.rs)
+ // selects the FileIO parser from the table location's URI scheme, and
+ // libpaimon_c.a compiles in separate COS, OBS, GCS and Azdls parsers
besides the
+ // OSS and S3 ones. Doris normalizes every object store's credentials into
the
+ // AWS_* / use_path_style aliases, which the BE rust bridge translates
only into
+ // the fs.oss.* and s3.* key families — a cosn:// / obs:// / gs:// /
abfs://
+ // warehouse would reach its scheme's parser without the key family it
reads and
+ // fail the open instead of using JNI. Only the schemes whose property
translation
+ // is implemented and open-tested (s3 / s3a / oss) plus the
credential-free hdfs
+ // and local-filesystem parsers stay rust-eligible. A null location also
routes to
+ // JNI: the rust path needs the paimon_table that only a real location can
provide.
+ final boolean schemeCapabilityVerified;
+
+ // An hdfs:// location is scheme-verified only together with the
credential-free
+ // backend shape: the backend storage properties that ship to BE also
carry an
+ // HDFS catalog's authentication (kerberos principal / keytab, proxy user,
HA
+ // nameservice config), none of which the pinned rust HDFS parser reads —
the
+ // scan would open as the BE process's ambient identity instead of the
catalog's
+ // configured one and fail the access JNI honors. See
isRustVerifiedHdfsBackend.
+ final boolean hdfsBackendVerified;
+
+ // The BE capability negotiation requires every candidate backend to report
+ // supports_paimon_rust_reader.
+ final boolean allBackendsRustCapable;
+
+ private PaimonRustEligibility(boolean isFileStoreTable, boolean
fallbackReadTable,
+ boolean restTokenTable, boolean queryAuthTable, boolean
puAggDeletionVectors,
+ boolean dvMergeOnRead, boolean deduplicateIgnoreDelete, boolean
rustUnsupportedMergeOption,
+ boolean scanModeNotDefault, boolean providerModeTranslatable,
boolean schemeCapabilityVerified,
+ boolean hdfsBackendVerified, boolean allBackendsRustCapable) {
+ this.isFileStoreTable = isFileStoreTable;
+ this.fallbackReadTable = fallbackReadTable;
+ this.restTokenTable = restTokenTable;
+ this.queryAuthTable = queryAuthTable;
+ this.puAggDeletionVectors = puAggDeletionVectors;
+ this.dvMergeOnRead = dvMergeOnRead;
+ this.deduplicateIgnoreDelete = deduplicateIgnoreDelete;
+ this.rustUnsupportedMergeOption = rustUnsupportedMergeOption;
+ this.scanModeNotDefault = scanModeNotDefault;
+ this.providerModeTranslatable = providerModeTranslatable;
+ this.schemeCapabilityVerified = schemeCapabilityVerified;
+ this.hdfsBackendVerified = hdfsBackendVerified;
+ this.allBackendsRustCapable = allBackendsRustCapable;
+ }
+
+ /**
+ * Whether the credential provider shape in the backend storage properties
can be
+ * translated to the pinned paimon-rust reader without changing the
identity the
+ * catalog configured. The rust S3 bridge maps static credentials,
anonymous access
+ * (AWS_CREDENTIALS_PROVIDER_TYPE=ANONYMOUS -> s3.anonymous) and
assume-role, but
+ * the remaining provider modes are ambient JVM provider chains (ENV,
+ * SYSTEM_PROPERTIES, WEB_IDENTITY, CONTAINER, INSTANCE_PROFILE) with no
paimon-rust
+ * equivalent — rust would silently sign with whatever its own ambient
chain
+ * resolves to. Lives here as one of the shared eligibility gates; also
called
+ * directly for the JNI fallback-reason logging.
+ */
+ static boolean providerModeTranslatable(String location,
+ Map<String, String> backendStorageProperties) {
+ String providerType = backendStorageProperties == null
+ ? null :
backendStorageProperties.get("AWS_CREDENTIALS_PROVIDER_TYPE");
+ String mode = providerType == null ? "DEFAULT" :
providerType.trim().toUpperCase(Locale.ROOT);
+ if (mode.isEmpty()) {
+ mode = "DEFAULT";
+ }
+ if (location != null && (location.startsWith("s3://") ||
location.startsWith("s3a://"))) {
+ // Java DEFAULT may resolve JVM properties or anonymous
credentials;
+ // Rust's ambient chain is different. Only explicit keys or
explicit
+ // anonymous access can cross this boundary without changing
identity.
+ String accessKey = backendStorageProperties == null
+ ? null : backendStorageProperties.get("AWS_ACCESS_KEY");
+ String secretKey = backendStorageProperties == null
+ ? null : backendStorageProperties.get("AWS_SECRET_KEY");
+ boolean staticKeys = accessKey != null &&
!accessKey.trim().isEmpty()
+ && secretKey != null && !secretKey.trim().isEmpty();
+ // Hadoop's Simple provider gives static keys precedence over
anonymous,
+ // role and token settings. Rust interprets those settings
differently.
+ boolean conflictingStaticSettings = staticKeys &&
(mode.equals("ANONYMOUS")
+ ||
StringUtils.isNotEmpty(backendStorageProperties.get("AWS_ROLE_ARN"))
+ ||
StringUtils.isNotEmpty(backendStorageProperties.get("AWS_TOKEN")));
+ return !conflictingStaticSettings
+ && (mode.equals("ANONYMOUS") || (mode.equals("DEFAULT") &&
staticKeys));
+ }
+ if (providerType != null) {
+ // OSS has no anonymous FileIO mode in the pinned Rust dependency.
+ return mode.equals("DEFAULT")
+ || (mode.equals("ANONYMOUS") && (location == null
+ || !location.startsWith("oss://")));
+ }
+ return true;
+ }
+
+ static PaimonRustEligibility of(Table table, String tableLocation,
+ Map<String, String> backendStorageProperties, Collection<Backend>
backends) {
+ boolean isFileStoreTable = table instanceof FileStoreTable;
+ boolean fallbackReadTable = table instanceof
FallbackReadFileStoreTable;
+ boolean restTokenTable = false;
+ boolean queryAuthTable = false;
+ boolean puAggDeletionVectors = false;
+ boolean dvMergeOnRead = false;
+ boolean deduplicateIgnoreDelete = false;
+ boolean rustUnsupportedMergeOption = false;
+ if (isFileStoreTable) {
+ FileStoreTable fileStoreTable = (FileStoreTable) table;
+ restTokenTable = fileStoreTable.fileIO() instanceof
RESTTokenFileIO;
+ CoreOptions resolvedCoreOptions = fileStoreTable.coreOptions();
+ // Null-safe: a table handle whose CoreOptions is not resolved
(e.g. some
+ // wrapper shapes) stays rust-eligible rather than failing here —
the rust
+ // open itself rejects such a table if the option is really set.
+ if (resolvedCoreOptions != null) {
+ queryAuthTable = resolvedCoreOptions.queryAuthEnabled();
+ CoreOptions.MergeEngine mergeEngine =
resolvedCoreOptions.mergeEngine();
+ if (resolvedCoreOptions.deletionVectorsEnabled()
+ && (mergeEngine ==
CoreOptions.MergeEngine.PARTIAL_UPDATE
+ || mergeEngine ==
CoreOptions.MergeEngine.AGGREGATE)) {
+ puAggDeletionVectors = true;
+ TableSchema dvSchema = fileStoreTable.schema();
+ Map<String, String> dvOptions = dvSchema == null ? null :
dvSchema.options();
+ String mergeOnRead = dvOptions == null
+ ? null :
dvOptions.get(PaimonScanNode.DELETION_VECTORS_MERGE_ON_READ);
+ dvMergeOnRead = "true".equalsIgnoreCase(mergeOnRead);
+ }
+ if (mergeEngine == CoreOptions.MergeEngine.DEDUPLICATE
+ && resolvedCoreOptions.ignoreDelete()) {
+ deduplicateIgnoreDelete = true;
+ }
+ if (mergeEngine == CoreOptions.MergeEngine.PARTIAL_UPDATE
+ || mergeEngine == CoreOptions.MergeEngine.AGGREGATE) {
+ TableSchema mergeSchema = fileStoreTable.schema();
+ Map<String, String> mergeOptions =
+ mergeSchema == null ? null : mergeSchema.options();
+ rustUnsupportedMergeOption = mergeOptions != null
+ &&
PaimonScanNode.hasRustUnsupportedMergeOption(mergeOptions, mergeEngine);
+ }
+ }
+ }
+ // Null-safe on a null schema, mirroring the CoreOptions null-safety
above:
+ // a fallback-shaped table can carry a null schema.
+ boolean scanModeNotDefault = false;
+ if (isFileStoreTable && ((FileStoreTable) table).schema() != null) {
+ TableSchema schema = ((FileStoreTable) table).schema();
+ String scanMode =
PaimonScanParams.withoutTimeTravelSelectors(schema)
+ .options().get(CoreOptions.SCAN_MODE.key());
+ scanModeNotDefault = scanMode != null &&
!"default".equalsIgnoreCase(scanMode);
+ }
+ boolean schemeCapabilityVerified =
PaimonScanNode.isRustVerifiedLocationScheme(tableLocation);
+ boolean hdfsBackendVerified =
!PaimonScanNode.isHdfsLocationScheme(tableLocation)
+ ||
PaimonScanNode.isRustVerifiedHdfsBackend(backendStorageProperties);
+ boolean providerModeTranslatable = providerModeTranslatable(
+ tableLocation, backendStorageProperties);
+ boolean allBackendsRustCapable = backends != null &&
!backends.isEmpty()
+ &&
backends.stream().allMatch(Backend::isPaimonRustReaderSupported);
Review Comment:
[P2] Negotiate vector support separately from the batch Rust reader
This heartbeat bit is already true on BEs from the PR base, but those BEs
only implement ordinary Rust splits. An upgraded FE therefore passes this
preflight with an older Rust-capable BE, rewrites the query, and sends
`paimon_split=""` plus the new vector payload. The older `_validate_rust_split`
rejects that range as `missing paimon_split`, so enabling the optimization
breaks a query that could still run normally during a mixed-version rollout.
Gate the one-way rewrite on a vector-specific capability or a negotiated
protocol version, while retaining the existing batch capability check.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/PushDownVectorTopNIntoPaimonScan.java:
##########
@@ -0,0 +1,513 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.nereids.rules.rewrite;
+
+import org.apache.doris.analysis.TableScanParams;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.catalog.Type;
+import org.apache.doris.datasource.credentials.CredentialUtils;
+import org.apache.doris.datasource.credentials.VendedCredentialsFactory;
+import org.apache.doris.datasource.paimon.PaimonExternalCatalog;
+import org.apache.doris.datasource.paimon.PaimonExternalTable;
+import org.apache.doris.datasource.paimon.PaimonUtil;
+import org.apache.doris.datasource.paimon.source.PaimonVectorSearch;
+import org.apache.doris.datasource.property.storage.StorageProperties;
+import org.apache.doris.nereids.rules.Rule;
+import org.apache.doris.nereids.rules.RuleType;
+import org.apache.doris.nereids.trees.expressions.Alias;
+import org.apache.doris.nereids.trees.expressions.ComparisonPredicate;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.InPredicate;
+import org.apache.doris.nereids.trees.expressions.IsNull;
+import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.Not;
+import org.apache.doris.nereids.trees.expressions.SlotReference;
+import org.apache.doris.nereids.trees.expressions.StatementScopeIdGenerator;
+import
org.apache.doris.nereids.trees.expressions.functions.scalar.InnerProductApproximate;
+import
org.apache.doris.nereids.trees.expressions.functions.scalar.L2DistanceApproximate;
+import org.apache.doris.nereids.trees.expressions.literal.ArrayLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.Literal;
+import org.apache.doris.nereids.trees.plans.AnnTopNInfo;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalFileScan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalFilter;
+import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
+import org.apache.doris.nereids.trees.plans.logical.LogicalTopN;
+import org.apache.doris.nereids.util.ExpressionUtils;
+import org.apache.doris.qe.ConnectContext;
+import org.apache.doris.thrift.TVectorMetric;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Maps;
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.Table;
+import org.apache.paimon.types.ArrayType;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DataTypeRoot;
+import org.apache.paimon.types.MapType;
+import org.apache.paimon.types.MultisetType;
+import org.apache.paimon.types.RowType;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+
+/**
+ * Push a primary-key vector (ANN) top-N down into the Paimon scan
(index-only).
+ *
+ * <p>Recognizes {@code TopN[order by dis] -> Project[dis := dist_fn(vec, q)]
-> [Filter] -> PaimonScan}
+ * where {@code dist_fn} is {@code l2_distance_approximate} (ASC) or
+ * {@code inner_product_approximate} (DESC) and the Paimon table has a
PK-vector
+ * index on {@code vec} with a matching distance metric.
+ *
+ * <p>The rewrite mirrors the internal-table ANN path (see
+ * {@code PushDownVectorTopNIntoOlapScan}): the scan is told to produce a
+ * reader-generated {@code __paimon_search_score_to_dis} column via the
PK-vector index, and
+ * every occurrence of {@code dist_fn(vec, q)} is replaced by that column's
slot. The
+ * order key slot keeps its identity, so the surrounding TopN merges/limits
unchanged.
+ * The distance always comes from the index, never from the raw vector.
+ *
+ * <p>No arithmetic is emitted here even though paimon-rust reports a score
rather than
+ * a distance: the BE reader undoes the metric-dependent score transform in
place while
+ * filling the block, exactly as the internal table sqrt's faiss's squared
distance inside
+ * the index layer. So the column arrives already holding what {@code dist_fn}
would
+ * return, and the plan stays a bare slot reference -- which also keeps {@code
dis}
+ * non-nullable (an expression over the score would not be, since {@code
Divide} and
+ * {@code Sqrt} are always nullable) and makes predicates like
+ * {@code dist_fn(vec, q) < 10} correct without any direction flip.
+ *
+ * <p>Being index-only is just an optimization here, not a precondition: if
nothing
+ * else references the raw vector column it is left unreferenced and dropped
by column
+ * pruning (pure index-only read); if the query still needs it ({@code SELECT
vec} or a
+ * predicate on {@code vec}), it stays a normal projected column and the BE
vector
+ * reader materializes it alongside the score. Either way ANN is applied --
the rewrite
+ * never falls back merely because the vector is read.
+ */
+public class PushDownVectorTopNIntoPaimonScan implements RewriteRuleFactory {
+
+ /** Doris column types the BE paimon-rust predicate converter can push; see
+ * PaimonRustPredicateConverter._is_supported_slot_type on the BE side. */
+ private static final Set<Type> VECTOR_FILTER_SUPPORTED_COLUMN_TYPES =
ImmutableSet.of(
+ Type.BOOLEAN, Type.TINYINT, Type.SMALLINT, Type.INT, Type.BIGINT,
+ Type.STRING, Type.DATE, Type.DATEV2);
+
+ @Override
+ public List<Rule> buildRules() {
+ return ImmutableList.of(
+ logicalTopN(logicalProject(logicalFileScan())).when(t ->
t.getOrderKeys().size() == 1).then(topN -> {
+ LogicalProject<LogicalFileScan> project = topN.child();
+ LogicalFileScan scan = project.child();
+ return pushDown(topN, project, scan, Optional.empty());
+ }).toRule(RuleType.PUSH_DOWN_VECTOR_TOPN_INTO_PAIMON_SCAN),
+ logicalTopN(logicalProject(logicalFilter(logicalFileScan())))
+ .when(t -> t.getOrderKeys().size() == 1).then(topN -> {
+ LogicalProject<LogicalFilter<LogicalFileScan>>
project = topN.child();
+ LogicalFilter<LogicalFileScan> filter =
project.child();
+ LogicalFileScan scan = filter.child();
+ return pushDown(topN, project, scan,
Optional.of(filter));
+
}).toRule(RuleType.PUSH_DOWN_VECTOR_TOPN_INTO_PAIMON_SCAN)
+ );
+ }
+
+ private Plan pushDown(
+ LogicalTopN<?> topN,
+ LogicalProject<?> project,
+ LogicalFileScan scan,
+ Optional<LogicalFilter<?>> optionalFilter) {
+ // Only Paimon external tables carry PK-vector indexes.
+ if (!(scan.getTable() instanceof PaimonExternalTable)) {
+ return null;
+ }
+
+ // The index-only search path is executed exclusively by the BE
paimon-rust reader,
+ // which only FileScannerV2 can select (the V1 FileScanner explicitly
rejects
+ // PAIMON_RUST). If the rust reader or the V2 scanner is disabled, the
split would
+ // fall through to the JNI reader, which ignores the vector payload
and would
+ // return a non-ANN scan with an empty distance column. Fall back to a
normal
+ // scan then.
+ ConnectContext ctx = ConnectContext.get();
+ if (ctx == null || ctx.getSessionVariable() == null
+ || !ctx.getSessionVariable().isEnablePaimonRustReader()
+ || !ctx.getSessionVariable().enableFileScannerV2) {
+ return null;
+ }
+
+ // An incremental scan (t@incr / BETWEEN) must not be rewritten into a
+ // PK-vector search: the VectorScan builder has no incremental mode,
so a
+ // vector payload would silently return snapshot rows for a query that
asked
+ // for changes. Keeping the query un-rewritten runs it as the ordinary
+ // incremental scan the user requested, with the distance computed in
Doris.
+ // (The scan node's vector gate keeps !isIncremental as the
authoritative
+ // backstop for any path that produces a vector split regardless.)
+ //
+ // The same holds for a pinned file-creation-time read state: the
ordinary
+ // path enforces `scan.file-creation-time-millis` by dropping manifest
+ // entries older than the bound, but the vector search plans its own
splits
+ // and never consults that bound, so old files' nearest neighbors
could leak
+ // into the result. Snapshot/tag/version selectors are fine -- they are
+ // resolved into the table handle before the vector search -- so only
the
+ // file-level bound forces the decline.
+ if (scan.getScanParams().isPresent()) {
+ TableScanParams scanParams = scan.getScanParams().get();
+ if (scanParams.incrementalRead()) {
+ return null;
+ }
+ Map<String, String> rawParams = scanParams.getMapParams();
+ if
(rawParams.containsKey(CoreOptions.SCAN_FILE_CREATION_TIME_MILLIS.key())) {
+ return null;
+ }
+ }
+
+ // A bounded vector top-N is only correct when every filter conjunct is
+ // applied inside the search, before the limit is exhausted. The BE
+ // paimon-rust predicate converter skips conjuncts it cannot represent
(and
+ // the synthetic distance column is not in its field registry at all),
and
+ // the engine re-applies those residuals on the returned block --
after the
+ // top-K rows were already selected. A skipped residual can therefore
drop
+ // rows out of a bounded result: with candidate distances 0, 2 and 3, a
+ // `WHERE dis > 1 ORDER BY dis LIMIT 1` would fetch only the
distance-0 row,
+ // filter it out and return nothing instead of the distance-2 row. So
the
+ // rewrite only fires when each conjunct is provably of a form the
converter
+ // pushes: a simple comparison / IN / null-check on a table column
whose
+ // Doris type is in the converter's supported set (see
+ // PaimonRustPredicateConverter._is_supported_slot_type on BE).
Anything
+ // else -- an expression predicate, a predicate on the distance
itself, an
+ // unsupported column type -- declines the rewrite and the query runs
as an
+ // ordinary scan with the distance computed in Doris, which is always
+ // correct.
+ if (optionalFilter.isPresent() &&
!conjunctsRideIntoVectorSearch(optionalFilter.get())) {
+ return null;
+ }
+
+ // The order key must be a SlotReference produced by a Project alias.
+ Expression orderKey = topN.getOrderKeys().get(0).getExpr();
+ if (!(orderKey instanceof SlotReference)) {
+ return null;
+ }
+ SlotReference keySlot = (SlotReference) orderKey;
+ Alias orderKeyAlias = null;
+ Expression orderKeyExpr = null;
+ for (NamedExpression projection : project.getProjects()) {
+ if (projection.toSlot().equals(keySlot) && projection instanceof
Alias) {
+ orderKeyAlias = (Alias) projection;
+ orderKeyExpr = orderKeyAlias.child();
+ break;
+ }
+ }
+ if (orderKeyExpr == null) {
+ return null;
+ }
+
+ // l2 must sort ascending (smaller distance = closer); inner_product
descending.
+ boolean isAsc = topN.getOrderKeys().get(0).isAsc();
+ boolean l2Dist = orderKeyExpr instanceof L2DistanceApproximate;
+ boolean innerProduct = orderKeyExpr instanceof InnerProductApproximate;
+ if (!(l2Dist && isAsc) && !(innerProduct && !isAsc)) {
+ return null;
+ }
+
+ // The indexed column must be a direct slot reference. A Cast chain
over the
+ // column is NOT stripped: a legal cast can change vector values (for
example
+ // an ARRAY<FLOAT> -> ARRAY<INT> -> ARRAY<FLOAT> round trip loses
precision),
+ // and searching the original vectors while the query means the casted
ones
+ // changes both the distance and the row order. Doris's implicit
coercions do
+ // not produce a cast here for a matching ARRAY<FLOAT> column, so
anything
+ // wrapped in a Cast simply declines the rewrite.
+ Expression left;
+ Expression right;
+ if (l2Dist) {
+ left = ((L2DistanceApproximate) orderKeyExpr).left();
+ right = ((L2DistanceApproximate) orderKeyExpr).right();
+ } else {
+ left = ((InnerProductApproximate) orderKeyExpr).left();
+ right = ((InnerProductApproximate) orderKeyExpr).right();
+ }
+ if (!(left instanceof SlotReference && right.isConstant() && right
instanceof ArrayLiteral)) {
+ return null;
+ }
+ SlotReference vectorSlot = (SlotReference) left;
+ if (!vectorSlot.getOriginalColumn().isPresent() ||
!vectorSlot.getOriginalTable().isPresent()) {
+ return null;
+ }
+ String annColumnName = vectorSlot.getOriginalColumn().get().getName();
+
+ // Validate the PK-vector index and metric on the Paimon table.
+ PaimonExternalTable paimonTable = (PaimonExternalTable)
scan.getTable();
+ CoreOptions coreOptions;
+ try {
+ Table basePaimonTable =
paimonTable.getPaimonTable(Optional.empty());
+ coreOptions = CoreOptions.fromMap(basePaimonTable.options());
+ } catch (Throwable t) {
+ return null;
+ }
+ if (!coreOptions.primaryKeyVectorIndexEnabled()) {
+ return null;
+ }
+ boolean columnIndexed =
coreOptions.primaryKeyVectorIndexColumns().stream()
+ .anyMatch(c -> c.equalsIgnoreCase(annColumnName));
+ if (!columnIndexed) {
+ return null;
+ }
+ // v1 supports only l2 / inner_product; the function must match the
index metric.
+ String metric =
coreOptions.primaryKeyVectorDistanceMetric(annColumnName);
+ String expectedMetric = l2Dist ? "l2" : "inner_product";
+ if (!expectedMetric.equals(metric)) {
+ return null;
+ }
+
+ // Pre-flight the projection-dependent bounds the vector split would
hit at
+ // the scan node: a PK-vector split has no JNI fallback (it would
return an
+ // empty distance column), so a query whose projected columns would set
+ // projectedVariant -- or an ORC table with TIMESTAMP_LTZ in its
schema, the
+ // orc-ltz-shape bound -- must not be rewritten at all. Without the
rewrite
+ // those queries run through the ordinary paths, where they are
supported.
+ try {
+ if (projectedColumnsContainVariant(project, optionalFilter)
+ || orcSchemaContainsTimestampLtz(paimonTable)) {
+ return null;
+ }
+ } catch (Throwable t) {
+ return null;
+ }
+
+ // The thrift-level enum the AnnTopNInfo will carry. expectedMetric is
kept as a
+ // string for the comparison above (coreOptions returns a string), but
the
+ // AnnTopNInfo / wire contract uses TVectorMetric, so map once here.
DOT_PRODUCT
+ // is the thrift enum that paimon-rust's "inner_product" score
transform maps to
+ // (see the BE reader's parse_score_transform).
+ TVectorMetric expectedMetricEnum = l2Dist ? TVectorMetric.L2 :
TVectorMetric.DOT_PRODUCT;
+
+ // Pre-flight the rust eligibility the scan node will gate on. The
rewrite is
+ // one-way: a PK-vector split has no JNI fallback (the JNI reader
ignores the
+ // vector payload and would return a non-ANN scan with an empty
distance
+ // column), so the scan node can only fail loudly — a table whose gate
would
+ // fail (credential provider shape, REST-token IO, query-auth,
merge-option
+ // matrix, deletion-vector merge-on-read, scan-mode, storage scheme or
HDFS
+ // credentials, no rust-capable backend) must not be rewritten at all.
Without
+ // the rewrite the query still runs, computing the distance in Doris.
The
+ // projection-dependent bounds (ORC LTZ, projected variant) need the
final scan
+ // tuple and stay scan-node-side, and the scan node re-evaluates on the
+ // relation-option-processed table, so it stays authoritative for
t@options
+ // overrides this raw-table pre-flight cannot see.
+ try {
+ Table baseTable = paimonTable.getPaimonTable(Optional.empty());
+ PaimonExternalCatalog catalog = (PaimonExternalCatalog)
paimonTable.getCatalog();
+ Map<StorageProperties.Type, StorageProperties>
storagePropertiesMap =
+
VendedCredentialsFactory.getStoragePropertiesMapWithVendedCredentials(
+
catalog.getCatalogProperty().getMetastoreProperties(),
+
catalog.getCatalogProperty().getStoragePropertiesMap(),
+ baseTable);
+ Map<String, String> backendStorageProperties =
+
CredentialUtils.getBackendPropertiesFromStorageMap(storagePropertiesMap);
+ String tableLocation = baseTable instanceof FileStoreTable
+ ? ((FileStoreTable) baseTable).location().toString()
+ : baseTable.options().get("path");
+ if (!PaimonVectorSearch.vectorScanEligible(baseTable,
tableLocation,
+ backendStorageProperties,
+ Env.getCurrentSystemInfo().getAllClusterBackends(true))) {
+ return null;
+ }
+ } catch (Throwable t) {
+ return null;
+ }
+ // NOTE: the FE-side data-predicate interception is intentionally not
done. The
Review Comment:
[P2] Preflight the SDK's table-mode restriction for data filters
A supported PK-vector table with `merge-engine=first-row` and
`deletion-vectors.enabled=false` passes this rewrite for `WHERE tag < 3 ORDER
BY l2_distance_approximate(vec,q) LIMIT k`. Paimon 2.1 explicitly permits
FIRST_ROW vector indexes without deletion vectors, but pinned paimon-rust
`PkVectorSearchParams::resolve` rejects every data predicate unless deletion
vectors are enabled and merge-on-read is disabled. FE planning supplies only
partition filters, so the failure occurs after the expression is replaced, when
BE sends the successfully converted tag predicate. Decline this table/filter
combination before rewriting so the supported ordinary query can execute.
Unlike the prior residual-filter thread, this predicate converts completely;
the SDK rejects its table mode.
##########
be/src/format_v2/table/paimon_rust_table_reader.cpp:
##########
@@ -804,6 +1308,13 @@ std::vector<std::string>
PaimonRustTableReader::_build_read_columns() const {
if (column.is_partition_key) {
continue;
}
+ // The PK-vector search distance is reader-produced and auto-appended
by
+ // paimon-rust under its own field name; it is never a projectable
table
+ // column, so exclude it from the projection in every mode (in
non-vector
+ // mode it is simply absent).
+ if (column.name == kPaimonSearchDistanceColumn) {
Review Comment:
[P2] Skip the synthetic distance name only for a synthetic slot
This helper also builds ordinary Rust projections, and
`__paimon_search_score_to_dis` is a legal Paimon user-column name. On a
compatible logical Rust scan of `SELECT id, __paimon_search_score_to_dis FROM
t`, this unconditional exclusion requests only `id`; `_fill_non_arrow_columns`
then replaces the real nullable column's values with NULL/defaults. No ANN
rewrite or vector index is needed for this regression. Limit the synthetic
exclusion to the actual vector-produced slot and preserve physical columns with
the same name on ordinary scans; handle collisions before adding a synthetic
slot.
##########
regression-test/framework/src/main/groovy/org/apache/doris/regression/util/ResultUtils.groovy:
##########
@@ -59,8 +64,8 @@ class ResultUtils {
}
static void assertSparkDorisResultEquals(List<List<Object>> sparkRows,
List<List<Object>> dorisRows) {
- List<List<Object>> normalizedSparkRows = normalizeRows(sparkRows)
- List<List<Object>> normalizedDorisRows = normalizeRows(dorisRows)
+ List<List<Object>> normalizedSparkRows =
decodeNestedBinaryTokens(normalizeRows(sparkRows))
Review Comment:
[P2] Preserve binary identity during result normalization
`normalizeValue(byte[])` first converts scalar binary values to `0x...`
strings, so this new pass decodes scalar columns as well as nested tokens.
Distinct values such as the single bytes `0x80` and `0x81` both become the same
UTF-8 replacement character in `decodeHexString`, making
`assertSparkDorisResultEquals` pass despite different returned bytes. The same
information loss affects nested arbitrary binary data. Keep a lossless byte/hex
representation and use type-aware conversion for nested renderings instead of
decoding every matching string through replacement UTF-8.
--
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]
