Gabriel39 commented on code in PR #66227:
URL: https://github.com/apache/doris/pull/66227#discussion_r4119332169
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/source/PaimonScanNode.java:
##########
@@ -411,10 +726,315 @@ private void setPaimonParams(TFileRangeDesc rangeDesc,
PaimonSplit paimonSplit)
String fileFormat = getFileFormat(paimonSplit.getPathString());
if (split != null) {
+ // use jni reader / paimon-cpp reader / paimon-rust reader
rangeDesc.setFormatType(TFileFormatType.FORMAT_JNI);
- // A logical DataSplit may span multiple files, so keep it intact
for the JNI reader.
- fileDesc.setReaderType(TPaimonReaderType.PAIMON_JNI);
- fileDesc.setPaimonSplit(PaimonUtil.encodeObjectToString(split));
+ // paimon-cpp and paimon-rust both consume Paimon native binary
serialization,
+ // which only supports DataSplit. Any other split type falls back
to JNI.
+ boolean nativeSplit = split instanceof DataSplit;
+ // Fallback-read splits stay on JNI: FallbackDataSplit extends
+ // DataSplit, so the instanceof above passes, but its serializer
+ // appends an isFallback byte after the ordinary split that the
+ // pinned rust decoder rejects outright ("trailing bytes after
+ // DataSplit" — it requires full-buffer consumption), and even a
+ // permissive decode would still lack the second table identity
+ // needed to honor the fallback-side discriminator. Both sides of a
+ // FallbackReadFileStoreTable wrap their splits, so the table
+ // wrapper is gated as a whole (any split from it routes to JNI)
+ // until the rust ABI represents both sides; the FallbackSplit
+ // interface also catches a wrapper split regardless of how the
+ // table was resolved here.
+ boolean fallbackRead = split instanceof
FallbackReadFileStoreTable.FallbackSplit
+ || processedTable instanceof FallbackReadFileStoreTable;
+ // Serialize the same effective table that planning and the JNI
reader use.
+ // Relation options such as t@options('read.batch-size'='1') are
applied by
+ // getProcessedTable() (doInitialize caches it in processedTable),
and the
+ // rust reader derives its read batch size from the schema options
— the raw
+ // cached table would silently drop the override. Copies,
delegates and
+ // fallback wrappers of getProcessedTable() are still
FileStoreTable, so the
+ // instanceof gate keeps its semantics.
+ Table paimonTable = processedTable;
+ FileStoreTable paimonFileStoreTable =
+ paimonTable instanceof FileStoreTable ? (FileStoreTable)
paimonTable : null;
+ // 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 above), 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.
+ boolean queryAuthTable = false;
+ // REST-token tables stay on JNI: 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 paimon 1.4.2's 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. Gate until the
+ // rust ABI can refresh and atomically update credentials.
+ boolean restTokenTable = false;
+ // Partial-update / aggregation tables with deletion vectors only
pass
+ // the rust reader in the fully materialized shape: the pinned rust
+ // read_pk rejects merge-engine=partial-update/aggregation with
+ // deletion-vectors.merge-on-read=true outright, and otherwise
requires
+ // every split to be compacted and known free of retract rows
+ // (DataSplit::is_fully_materialized_pk_dv). Their ordinary
DataSplits
+ // sail through the compound gate above, so without this check a
valid
+ // Java/JNI scan reaches BE and the rust open fails. Deduplicate
stays
+ // rust-eligible: its read_pk routes uncompacted splits to the KV
+ // reader, which applies the attached per-file DVs.
merge-on-read=true
+ // is a table option, so the whole table routes to JNI;
+ // non-materialized splits are gated per split below.
+ boolean puAggDeletionVectors = false;
+ boolean dvMergeOnRead = false;
+ boolean deduplicateIgnoreDelete = false;
+ boolean rustUnsupportedMergeOption = false;
+ if (paimonFileStoreTable != null) {
+ // A renewable REST token reached the shipped properties as a
+ // plain value; only the table's FileIO type reveals it
expires.
+ // Null-safe: a table handle whose FileIO is not resolved stays
+ // rust-eligible, mirroring the CoreOptions null-safety below.
+ restTokenTable = paimonFileStoreTable.fileIO() instanceof
RESTTokenFileIO;
+ CoreOptions resolvedCoreOptions =
paimonFileStoreTable.coreOptions();
+ // Null-safe: a table handle whose CoreOptions is not resolved
+ // (e.g. some wrapper shapes) stays rust-eligible rather than
+ // failing the scan 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;
+ // The merge-engine and deletion-vectors.enabled checks
+ // above resolve through the Java CoreOptions
accessors,
+ // which the table builds from this same schema options
+ // map — the one the BE rust reader deserializes from
+ // the shipped schema JSON — so they cannot diverge
from
+ // what BE sees. merge-on-read has no Java accessor in
+ // paimon 1.4, so it is read raw from the map, with the
+ // rust parsing semantics (any case-insensitive "true"
+ // is on, default false).
+ TableSchema dvSchema = paimonFileStoreTable.schema();
+ Map<String, String> dvOptions = dvSchema == null ?
null : dvSchema.options();
+ String mergeOnRead = dvOptions == null
+ ? null :
dvOptions.get(DELETION_VECTORS_MERGE_ON_READ);
+ dvMergeOnRead = "true".equalsIgnoreCase(mergeOnRead);
+ }
+ // 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.
+ if (mergeEngine == CoreOptions.MergeEngine.DEDUPLICATE
+ && resolvedCoreOptions.ignoreDelete()) {
+ deduplicateIgnoreDelete = true;
+ }
+ // 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. Mirror the exact rust key matrix
(presence,
+ // not values) against the same schema options map BE
+ // deserializes.
+ if (mergeEngine == CoreOptions.MergeEngine.PARTIAL_UPDATE
+ || mergeEngine ==
CoreOptions.MergeEngine.AGGREGATE) {
+ TableSchema mergeSchema =
paimonFileStoreTable.schema();
+ Map<String, String> mergeOptions =
+ mergeSchema == null ? null :
mergeSchema.options();
+ rustUnsupportedMergeOption = mergeOptions != null
+ && hasRustUnsupportedMergeOption(mergeOptions,
mergeEngine);
+ }
+ }
+ }
+ // paimon-rust additionally requires (a) FileScannerV2: the V1
FileScanner
+ // explicitly rejects PAIMON_RUST, so with enable_file_scanner_v2
disabled
+ // the split falls back to JNI instead of encoding a rust request
that the
+ // selected scanner cannot consume, and (b) a FileStoreTable: BE
opens the
+ // table via paimon_table_from_schema_json, which needs the
resolved
+ // TableSchema that only FileStoreTable exposes via schema(). If
the table
+ // is not a FileStoreTable (e.g. a sys table backed by DataSplit),
we cannot
+ // ship a schema JSON, so fall back to CPP / JNI rather than
sending an
+ // incomplete PAIMON_RUST request that BE would reject.
+ //
+ // The paimon-rust S3 bridge maps static credentials, anonymous
+ // access (AWS_CREDENTIALS_PROVIDER_TYPE=ANONYMOUS -> s3.anonymous)
+ // and assume-role (AWS_ROLE_ARN / AWS_EXTERNAL_ID ->
+ // s3.assumed.role.*), but the remaining credential-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 the ambient
+ // chain resolves to. Gate those modes away from the rust reader
+ // here so the configured provider is honored via the JNI path.
+ boolean providerModeTranslatable = true;
+ String providerType = backendStorageProperties == null
+ ? null :
backendStorageProperties.get("AWS_CREDENTIALS_PROVIDER_TYPE");
+ if (providerType != null) {
+ String mode = providerType.trim().toUpperCase(Locale.ROOT);
+ providerModeTranslatable = mode.equals("DEFAULT")
Review Comment:
Addressed in 6009fdb795. S3 DEFAULT now requires both nonblank static keys
before selecting Rust; no-key, partial-key, and blank-key configurations remain
on JNI. Explicit S3 ANONYMOUS remains supported. The FE selection tests cover
these cases and pass (66 tests in PaimonScanNodeTest).
##########
be/src/format_v2/table/paimon_rust_predicate_converter.cpp:
##########
@@ -0,0 +1,832 @@
+// 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.
+
+#include "format_v2/table/paimon_rust_predicate_converter.h"
+
+#include <algorithm>
+#include <cctype>
+#include <memory>
+#include <utility>
+
+#include "common/logging.h"
+#include "core/column/column_const.h"
+#include "core/column/column_nullable.h"
+#include "core/data_type/data_type.h"
+#include "core/data_type/data_type_nullable.h"
+#include "core/field.h"
+#include "core/types.h"
+#include "core/value/decimalv2_value.h"
+#include "core/value/timestamptz_value.h"
+#include "core/value/vdatetime_value.h"
+#include "exprs/runtime_filter_expr.h"
+#include "exprs/vcompound_pred.h"
+#include "exprs/vdirect_in_predicate.h"
+#include "exprs/vectorized_fn_call.h"
+#include "exprs/vexpr.h"
+#include "exprs/vin_predicate.h"
+#include "exprs/vliteral.h"
+#include "exprs/vslot_ref.h"
+
+namespace doris {
+
+namespace {
+// paimon_datum tags (see paimon.h / bindings/c/src/table.rs::datum_from_c).
+constexpr int32_t kTagBool = 0;
+constexpr int32_t kTagTinyInt = 1;
+constexpr int32_t kTagSmallInt = 2;
+constexpr int32_t kTagInt = 3;
+constexpr int32_t kTagLong = 4;
+constexpr int32_t kTagDouble = 6;
+constexpr int32_t kTagString = 7;
+constexpr int32_t kTagDate = 8;
+constexpr int32_t kTagTimestamp = 10;
+constexpr int32_t kTagDecimal = 12;
+constexpr int32_t kTagBytes = 13;
+
+// paimon decimal precision ceiling (paimon::Decimal::MAX_PRECISION).
+constexpr int32_t kPaimonDecimalMaxPrecision = 38;
+
+// RAII for an owned paimon_predicate*. and/or/not consume their inputs, so we
+// release() before handing pointers to them.
+struct predicate_deleter {
+ void operator()(paimon_predicate* p) const {
+ if (p) {
+ paimon_predicate_free(p);
+ }
+ }
+};
+using predicate_ptr = std::unique_ptr<paimon_predicate, predicate_deleter>;
+
+// RAII for an owned paimon_error*.
+struct error_deleter {
+ void operator()(paimon_error* p) const {
+ if (p) {
+ paimon_error_free(p);
+ }
+ }
+};
+using error_ptr = std::unique_ptr<paimon_error, error_deleter>;
+
+// Render a paimon_error into a string. Takes ownership of `err` via RAII so it
+// is freed on every return path. Safe to call with nullptr.
+std::string consume_predicate_error(paimon_error* err) {
+ error_ptr owned(err);
+ if (!owned) {
+ return "unknown error";
+ }
+ std::string msg;
+ if (owned->message.data != nullptr && owned->message.len > 0) {
+ msg.assign(reinterpret_cast<const char*>(owned->message.data),
owned->message.len);
+ }
+ return "code=" + std::to_string(owned->code) + ", msg=" + msg;
+}
+} // namespace
+
+PaimonRustPredicateConverter::PaimonRustPredicateConverter(
+ const std::vector<std::string>& column_names, const
std::vector<DataTypePtr>& column_types,
+ const paimon_table* table)
+ : _table(table) {
+ DORIS_CHECK(column_names.size() == column_types.size());
+ _columns_by_name.reserve(column_names.size());
+ for (size_t i = 0; i < column_names.size(); ++i) {
+ _columns_by_name.emplace(_normalize_name(column_names[i]),
+ std::make_pair(column_names[i],
column_types[i]));
+ }
+ // Paimon TIMESTAMP (wall clock) is stored as epoch-millis-of-the-wall-time
+ // and the DateTimeV2 serde decodes timezone-naive arrow values in UTC, so
+ // timestamp literals convert wall->epoch in UTC. utc_time_zone() needs no
+ // tzdata lookup, so the conversion cannot silently fall back to a
+ // machine-local zone.
+ _utc_tz = cctz::utc_time_zone();
+}
+
+paimon_predicate* PaimonRustPredicateConverter::build(const VExprContextSPtrs&
conjuncts) {
+ if (_table == nullptr) {
+ return nullptr;
+ }
+ predicate_ptr result;
+ for (const auto& conjunct : conjuncts) {
+ if (!conjunct || !conjunct->root()) {
+ continue;
+ }
+ auto root = conjunct->root();
+ if (root->is_rf_wrapper()) {
+ if (auto impl = root->get_impl()) {
+ // A null-aware runtime filter (an EQ_FOR_NULL join) must stay
+ // residual: its wrapper execution restores NULL probe rows to
+ // true (RuntimeFilterExpr::change_null_to_true), while the
+ // unwrapped impl — rebuilt as an ordinary IN predicate through
+ // VDirectInPredicate::get_slot_in_expr — treats NULL as
+ // not-in-set and would prune the NULL probes permanently
+ // before the join sees them. Keep the wrapper itself: it fails
+ // every dispatch below, so the conjunct stays in the residual,
+ // and it is safe to execute on selected rows, so later
+ // conjuncts keep pushing. The lance pushdown declines
+ // is_null_aware() filters for the same reason.
+ // is_null_aware() is concrete on RuntimeFilterExpr (the only
+ // class whose is_rf_wrapper() is true), so the dynamic_cast
+ // never fails in practice; the null guard keeps the unwrap for
+ // any future wrapper shape.
+ auto* rf_wrapper =
dynamic_cast<RuntimeFilterExpr*>(root.get());
+ if (rf_wrapper == nullptr || !rf_wrapper->is_null_aware()) {
+ root = impl;
+ }
+ }
+ }
+ // Preserve a safe prefix of the conjunct order: a later pushed
+ // predicate (e.g. an arrived IN runtime filter) could otherwise prune
+ // rows on which an earlier error-preserving conjunct —
+ // assert_true(...), a failing cast, ... — must still raise. The v1
+ // partition-pruning path (FileScanner::_init_runtime_filter_partition_
+ // prune_ctxs) stops at is_safe_to_execute_on_selected_rows() for the
+ // same reason, so a convertible predicate after an unsafe conjunct
+ // must not be pushed. Safe conjuncts that cannot be converted keep
+ // the old skip: they cannot raise, so pruning rows before they are
+ // evaluated as the residual never loses an error.
+ if (!_is_safe_to_push(root)) {
+ break;
+ }
+ predicate_ptr pred(_convert_expr(root));
+ if (!pred) {
+ continue;
+ }
+ if (!result) {
+ result = std::move(pred);
+ } else {
+ // and consumes both inputs regardless of success.
+ result.reset(paimon_predicate_and(result.release(),
pred.release()));
+ if (!result) {
+ return nullptr;
+ }
+ }
+ }
+ return result.release();
+}
+
+bool PaimonRustPredicateConverter::_is_safe_to_push(const VExprSPtr& expr) {
+ if (expr->is_safe_to_execute_on_selected_rows()) {
+ return true;
+ }
+ // VectorizedFnCall::is_safe_to_execute_on_selected_rows() admits a fixed
+ // whitelist that does not include `like`, so a like conjunct would always
+ // stop the safe prefix here and never reach _convert_like. A like call
+ // whose children are themselves safe is just as total as the whitelisted
+ // comparisons: it only compares strings, and its regex is built by
escaping
+ // the pattern operand (FunctionLike::convert_like_pattern), so no scanned
+ // value can make it raise — a malformed pattern or escape operand fails
+ // at open() on every execution, regardless of which rows survive. Admit
+ // it here rather than widening the generic whitelist, which gates shared
+ // BE pushdown paths outside this reader's scope.
+ auto* fn = dynamic_cast<VectorizedFnCall*>(expr.get());
+ if (fn == nullptr || _normalize_name(fn->function_name()) != "like") {
+ return false;
+ }
+ for (uint16_t i = 0; i < fn->get_num_children(); ++i) {
+ if (!fn->get_child(i)->is_safe_to_execute_on_selected_rows()) {
+ return false;
+ }
+ }
+ return true;
+}
+
+paimon_predicate* PaimonRustPredicateConverter::_convert_expr(const VExprSPtr&
expr) {
+ if (!expr) {
+ return nullptr;
+ }
+
+ // Casts are not unwrapped anywhere (predicate root included): a cast node
+ // fails every dispatch below and the conjunct stays in the Doris residual,
+ // mirroring the FE converter, which keeps casted expressions unconverted.
+ if (auto* direct_in = dynamic_cast<VDirectInPredicate*>(expr.get())) {
+ VExprSPtr in_expr;
+ if (direct_in->get_slot_in_expr(in_expr)) {
+ return _convert_in(in_expr);
+ }
+ return nullptr;
+ }
+
+ if (dynamic_cast<VInPredicate*>(expr.get()) != nullptr) {
+ return _convert_in(expr);
+ }
+
+ switch (expr->op()) {
+ case TExprOpcode::COMPOUND_AND:
+ case TExprOpcode::COMPOUND_OR:
+ return _convert_compound(expr);
+ case TExprOpcode::COMPOUND_NOT:
+ return nullptr;
+ case TExprOpcode::EQ:
+ case TExprOpcode::EQ_FOR_NULL:
+ case TExprOpcode::NE:
+ case TExprOpcode::GE:
+ case TExprOpcode::GT:
+ case TExprOpcode::LE:
+ case TExprOpcode::LT:
+ return _convert_binary(expr);
+ default:
+ break;
+ }
+
+ if (auto* fn = dynamic_cast<VectorizedFnCall*>(expr.get())) {
+ auto fn_name = _normalize_name(fn->function_name());
+ if (fn_name == "is_null_pred" || fn_name == "is_not_null_pred") {
+ return _convert_is_null(expr, fn_name);
+ }
+ if (fn_name == "like") {
+ return _convert_like(expr);
+ }
+ }
+
+ return nullptr;
+}
+
+paimon_predicate* PaimonRustPredicateConverter::_convert_compound(const
VExprSPtr& expr) {
+ if (!expr || expr->get_num_children() != 2) {
+ return nullptr;
+ }
+ predicate_ptr left(_convert_expr(expr->get_child(0)));
+ if (!left) {
+ return nullptr;
+ }
+ predicate_ptr right(_convert_expr(expr->get_child(1)));
+ if (!right) {
+ return nullptr;
+ }
+
+ if (expr->op() == TExprOpcode::COMPOUND_AND) {
+ return paimon_predicate_and(left.release(), right.release());
+ }
+ if (expr->op() == TExprOpcode::COMPOUND_OR) {
+ return paimon_predicate_or(left.release(), right.release());
+ }
+ return nullptr;
+}
+
+paimon_predicate* PaimonRustPredicateConverter::_convert_in(const VExprSPtr&
expr) {
+ auto* in_pred = dynamic_cast<VInPredicate*>(expr.get());
+ if (!in_pred || expr->get_num_children() < 2) {
+ return nullptr;
+ }
+ auto field_meta = _resolve_field(expr->get_child(0));
+ if (!field_meta) {
+ return nullptr;
+ }
+
+ const auto num_values = expr->get_num_children() - 1;
+ // Reserve up front so the backing strings never reallocate: each datum's
+ // str_data points into storages[i], which must stay stable.
+ std::vector<std::string> storages;
+ std::vector<paimon_datum> datums;
+ storages.reserve(num_values);
+ datums.reserve(num_values);
+ for (uint16_t i = 1; i < expr->get_num_children(); ++i) {
+ // Casted list values are rejected by _convert_literal (the same rule
+ // as the binary RHS): with debug_skip_fold_constant the cast reaches
+ // the BE un-folded, and unwrapping it would filter on the pre-cast
+ // value — in `amount IN (CAST(1.24 AS DECIMAL(10,1)))` Doris keeps
+ // the 1.2 rows while the unwrapped 1.24 push removes them. Rejecting
+ // the value rejects the whole predicate; the residual applies the
+ // cast correctly.
+ auto holder = _convert_literal(expr->get_child(i), field_meta->type);
+ if (!holder) {
+ return nullptr;
+ }
+ storages.emplace_back(std::move(holder->storage));
+ paimon_datum datum = holder->datum;
+ _bind_datum_storage(&datum, storages.back());
+ datums.emplace_back(datum);
+ }
+
+ if (datums.empty()) {
+ return nullptr;
+ }
+ if (in_pred->is_not_in()) {
+ return _take(paimon_predicate_is_not_in(_table,
field_meta->column.c_str(), datums.data(),
+ datums.size()));
+ }
+ return _take(paimon_predicate_is_in(_table, field_meta->column.c_str(),
datums.data(),
+ datums.size()));
+}
+
+paimon_predicate* PaimonRustPredicateConverter::_convert_binary(const
VExprSPtr& expr) {
+ if (!expr || expr->get_num_children() != 2) {
+ return nullptr;
+ }
+ auto field_meta = _resolve_field(expr->get_child(0));
+ if (!field_meta) {
+ return nullptr;
+ }
+ const char* column = field_meta->column.c_str();
+
+ // Convert the RHS first so EQ_FOR_NULL (<=>) only converts when the RHS is
+ // a convertible literal, mirroring the FE converter, which rejects a
+ // non-literal RHS. A column-to-column `a <=> b` must therefore stay in the
+ // Doris residual: it has no single-column rust predicate, and pushing
+ // `a IS NULL` would wrongly discard rows like (1, 1) — rows dropped by the
+ // pushed filter cannot be recovered by the residual conjunct.
+ auto holder = _convert_literal(expr->get_child(1), field_meta->type);
+ if (!holder) {
+ return nullptr;
+ }
+
+ if (expr->op() == TExprOpcode::EQ_FOR_NULL) {
+ return _take(paimon_predicate_is_null(_table, column));
+ }
+
+ // `holder` is a local, so its storage stays put for the duration of the
call.
+ _bind_datum_storage(&holder->datum, holder->storage);
+ const paimon_datum& datum = holder->datum;
+
+ switch (expr->op()) {
+ case TExprOpcode::EQ:
+ return _take(paimon_predicate_equal(_table, column, datum));
+ case TExprOpcode::NE:
+ return _take(paimon_predicate_not_equal(_table, column, datum));
+ case TExprOpcode::GE:
+ return _take(paimon_predicate_greater_or_equal(_table, column, datum));
+ case TExprOpcode::GT:
+ return _take(paimon_predicate_greater_than(_table, column, datum));
+ case TExprOpcode::LE:
+ return _take(paimon_predicate_less_or_equal(_table, column, datum));
+ case TExprOpcode::LT:
+ return _take(paimon_predicate_less_than(_table, column, datum));
+ default:
+ break;
+ }
+ return nullptr;
+}
+
+paimon_predicate* PaimonRustPredicateConverter::_convert_is_null(const
VExprSPtr& expr,
+ const
std::string& fn_name) {
+ if (!expr || expr->get_num_children() != 1) {
+ return nullptr;
+ }
+ auto field_meta = _resolve_field(expr->get_child(0));
+ if (!field_meta) {
+ return nullptr;
+ }
+ if (fn_name == "is_not_null_pred") {
+ return _take(paimon_predicate_is_not_null(_table,
field_meta->column.c_str()));
+ }
+ return _take(paimon_predicate_is_null(_table, field_meta->column.c_str()));
+}
+
+paimon_predicate* PaimonRustPredicateConverter::_convert_like(const VExprSPtr&
expr) {
+ if (!expr || expr->get_num_children() < 2) {
+ return nullptr;
+ }
+ auto field_meta = _resolve_field(expr->get_child(0));
+ if (!field_meta ||
!_is_string_type(field_meta->type->get_primitive_type())) {
+ return nullptr;
+ }
+
+ auto pattern_opt = _extract_string_literal(expr->get_child(1));
+ if (!pattern_opt) {
+ return nullptr;
+ }
+ const std::string& pattern = *pattern_opt;
+
+ // Doris's 3-arg like(col, pattern, escape) form: the pinned rust
+ // predicate only supports the backslash escape (its builder rejects any
+ // other escape character), so only the default-escape shape converts.
+ // The 2-arg form always carries the Doris/SQL default '\'.
+ if (expr->get_num_children() >= 3) {
+ auto escape_opt = _extract_string_literal(expr->get_child(2));
+ if (!escape_opt || escape_opt->size() != 1 || (*escape_opt)[0] !=
'\\') {
+ return nullptr;
+ }
+ }
+
+ // The pinned rust like implements SQL LIKE (% = any run, _ = one char,
+ // \X = literal X for any X — arrow's like kernel), and its builder
+ // optimizes `prefix%` / `%suffix` / `%mid%` shapes internally, so every
+ // pattern without a backslash matches Doris semantics exactly. The one
+ // divergence is the escape handling for a backslash before an ordinary
+ // character: Doris keeps both characters (only \%, \_ and \\ are
+ // escapes; `a\qb` matches a backslash followed by q b) while rust
+ // consumes the backslash (`a\qb` matches aqb) — pushing such a pattern
+ // would wrongly prune the Doris-matching rows before the residual can
+ // see them. Reject exactly the divergent shapes: every backslash must
+ // precede %, _ or \ (identical literal semantics on both sides) or end
+ // the pattern (literal backslash on both sides); anything else stays in
+ // the residual.
+ for (size_t i = 0; i < pattern.size(); ++i) {
+ if (pattern[i] == '\\') {
+ if (i + 1 >= pattern.size()) {
+ continue; // trailing backslash: literal on both sides
+ }
+ char next = pattern[i + 1];
+ if (next != '%' && next != '_' && next != '\\') {
+ return nullptr;
+ }
+ ++i;
+ }
+ }
+
+ paimon_datum datum {};
+ datum.tag = kTagString;
+ // `pattern` outlives the build call (it is a const ref into the caller's
+ // optional), so the datum may point straight at it.
+ _bind_datum_storage(&datum, pattern);
+ return _take(paimon_predicate_like(_table, field_meta->column.c_str(),
datum, '\\'));
+}
+
+std::optional<PaimonRustPredicateConverter::FieldMeta>
PaimonRustPredicateConverter::_resolve_field(
+ const VExprSPtr& expr) const {
+ if (!expr) {
+ return std::nullopt;
+ }
+ // Mirror the FE converter's convertDorisExprToSlotRef: a casted column is
+ // rejected, never unwrapped. Stripping a lossy cast changes which rows
match
+ // — for a DECIMAL(10,2) column, CAST(amount AS DECIMAL(10,1)) = 1.2 keeps
+ // the row 1.24 while the unwrapped `amount = 1.2` prunes it — and rows
+ // pruned by the pushed filter cannot be recovered by the Doris residual.
+ // The conjunct stays in the residual instead.
+ auto* slot_ref = dynamic_cast<VSlotRef*>(expr.get());
+ if (!slot_ref) {
+ return std::nullopt;
+ }
+ // FileScannerV2 rewrites conjunct VSlotRefs to table global indices, so
slot_id
+ // is a position, not a slot id; resolve by the carried column name
against the
+ // projected-column registry instead of the desc table.
+ auto it = _columns_by_name.find(_normalize_name(slot_ref->column_name()));
+ if (it == _columns_by_name.end()) {
+ return std::nullopt;
+ }
+ const auto& [column, type] = it->second;
+ if (!_is_supported_slot_type(type->get_primitive_type(),
type->get_precision())) {
+ return std::nullopt;
+ }
+ return FieldMeta {column, type};
+}
+
+std::optional<PaimonRustPredicateConverter::DatumHolder>
+PaimonRustPredicateConverter::_convert_literal(const VExprSPtr& expr,
+ const DataTypePtr& column_type)
const {
+ // A casted literal is rejected, never unwrapped: the cast is not executed
+ // here (with constant folding disabled — debug_skip_fold_constant — it
+ // reaches the BE un-folded), so unwrapping would push the pre-cast value
+ // while the Doris residual compares against the cast result. A
+ // scale-reducing cast makes the two disagree — `amount =
+ // CAST(1.24 AS DECIMAL(10,1))` keeps a stored 1.20 row in Doris (1.24
+ // rounds down to 1.2) but the unwrapped `amount = 1.24` push prunes it,
and
+ // rows pruned by the pushed filter cannot be recovered by the residual.
+ // The conjunct stays in the residual, where the cast is evaluated exactly.
+ if (expr->node_type() == TExprNodeType::CAST_EXPR) {
+ return std::nullopt;
+ }
+ auto* literal = dynamic_cast<VLiteral*>(expr.get());
+ if (!literal) {
+ return std::nullopt;
+ }
+
+ auto literal_type = remove_nullable(literal->get_data_type());
+ PrimitiveType literal_primitive = literal_type->get_primitive_type();
+ PrimitiveType slot_primitive = column_type->get_primitive_type();
+
+ ColumnPtr col =
literal->get_column_ptr()->convert_to_full_column_if_const();
+ if (const auto* nullable = check_and_get_column<ColumnNullable>(*col)) {
+ if (nullable->is_null_at(0)) {
+ return std::nullopt;
+ }
+ col = nullable->get_nested_column_ptr();
+ }
+
+ Field field;
+ col->get(0, field);
+
+ DatumHolder holder;
+ paimon_datum& datum = holder.datum;
+
+ switch (slot_primitive) {
+ case TYPE_BOOLEAN: {
+ if (literal_primitive != TYPE_BOOLEAN) {
+ return std::nullopt;
+ }
+ datum.tag = kTagBool;
+ datum.int_val = static_cast<bool>(field.get<TYPE_BOOLEAN>()) ? 1 : 0;
+ return holder;
+ }
+ case TYPE_TINYINT:
+ case TYPE_SMALLINT:
+ case TYPE_INT:
+ case TYPE_BIGINT: {
+ if (!_is_integer_type(literal_primitive)) {
+ return std::nullopt;
+ }
+ int64_t value = 0;
+ switch (literal_primitive) {
+ case TYPE_TINYINT:
+ value = field.get<TYPE_TINYINT>();
+ break;
+ case TYPE_SMALLINT:
+ value = field.get<TYPE_SMALLINT>();
+ break;
+ case TYPE_INT:
+ value = field.get<TYPE_INT>();
+ break;
+ case TYPE_BIGINT:
+ value = field.get<TYPE_BIGINT>();
+ break;
+ default:
+ return std::nullopt;
+ }
+ datum.int_val = value;
+ switch (slot_primitive) {
+ case TYPE_TINYINT:
+ datum.tag = kTagTinyInt;
+ break;
+ case TYPE_SMALLINT:
+ datum.tag = kTagSmallInt;
+ break;
+ case TYPE_INT:
+ datum.tag = kTagInt;
+ break;
+ default:
+ datum.tag = kTagLong;
+ break;
+ }
+ return holder;
+ }
+ case TYPE_DOUBLE: {
+ if (literal_primitive != TYPE_DOUBLE && literal_primitive !=
TYPE_FLOAT) {
+ return std::nullopt;
+ }
+ datum.tag = kTagDouble;
+ datum.double_val = literal_primitive == TYPE_FLOAT
+ ?
static_cast<double>(field.get<TYPE_FLOAT>())
+ : field.get<TYPE_DOUBLE>();
+ return holder;
+ }
+ case TYPE_DATE:
+ case TYPE_DATEV2: {
+ if (!_is_date_type(literal_primitive)) {
+ return std::nullopt;
+ }
+ int64_t seconds = 0;
+ if (literal_primitive == TYPE_DATE) {
+ const auto& dt = field.get<TYPE_DATE>();
+ if (!dt.is_valid_date()) {
+ return std::nullopt;
+ }
+ dt.unix_timestamp(&seconds, _utc_tz);
+ } else {
+ const auto& dt = field.get<TYPE_DATEV2>();
+ if (!dt.is_valid_date()) {
+ return std::nullopt;
+ }
+ dt.unix_timestamp(&seconds, _utc_tz);
+ }
+ datum.tag = kTagDate;
+ datum.int_val = _seconds_to_days(seconds);
+ return holder;
+ }
+ case TYPE_DATETIME:
+ case TYPE_DATETIMEV2: {
+ if (!_is_datetime_type(literal_primitive)) {
+ return std::nullopt;
+ }
+ datum.tag = kTagTimestamp;
+ if (literal_primitive == TYPE_DATETIME) {
+ const auto& dt = field.get<TYPE_DATETIME>();
+ if (!dt.is_valid_date()) {
+ return std::nullopt;
+ }
+ int64_t seconds = 0;
+ dt.unix_timestamp(&seconds, _utc_tz);
+ // No sub-second part in datetime (v1): millis only, nanos stays 0.
+ datum.int_val = seconds * 1000;
+ } else {
+ const auto& dt = field.get<TYPE_DATETIMEV2>();
+ if (!dt.is_valid_date()) {
+ return std::nullopt;
+ }
+ // ts is (seconds since epoch, the microsecond-of-second part);
+ // split it into the rust timestamp's (millis, nanos): truncating
+ // the sub-millisecond remainder would make an equality/IN
predicate
+ // more selective than the original conjunct (e.g. .123456 pushed
as
+ // .123000), wrongly discarding matching rows. DATETIMEV2 carries
at
+ // most microseconds, and micros % 1000 * 1000 <= 999000 fits the
+ // nanos field, so the value is always representable exactly.
+ std::pair<int64_t, int64_t> ts;
+ dt.unix_timestamp(&ts, _utc_tz);
+ datum.int_val = ts.first * 1000 + ts.second / 1000;
+ datum.int_val2 = (ts.second % 1000) * 1000;
+ }
+ return holder;
+ }
+ case TYPE_VARCHAR:
+ case TYPE_STRING: {
+ if (!_is_string_type(literal_primitive)) {
+ return std::nullopt;
+ }
+ const auto& value = field.get<TYPE_STRING>();
+ datum.tag = kTagString;
+ holder.storage.assign(value.data(), value.size());
+ return holder;
+ }
+ case TYPE_DECIMALV2:
+ case TYPE_DECIMAL32:
+ case TYPE_DECIMAL64:
+ case TYPE_DECIMAL128I:
+ case TYPE_DECIMAL256: {
+ if (!_is_decimal_type(literal_primitive)) {
+ return std::nullopt;
+ }
+ int32_t precision =
static_cast<int32_t>(literal_type->get_precision());
+ int32_t scale = static_cast<int32_t>(literal_type->get_scale());
+ if (precision <= 0 || precision > kPaimonDecimalMaxPrecision) {
+ return std::nullopt;
+ }
+
+ __int128 value = 0;
+ switch (literal_primitive) {
+ case TYPE_DECIMALV2:
+ value = field.get<TYPE_DECIMALV2>().value();
+ break;
+ case TYPE_DECIMAL32:
+ value = field.get<TYPE_DECIMAL32>().value;
+ break;
+ case TYPE_DECIMAL64:
+ value = field.get<TYPE_DECIMAL64>().value;
+ break;
+ case TYPE_DECIMAL128I:
+ value = field.get<TYPE_DECIMAL128I>().value;
+ break;
+ default:
+ return std::nullopt;
+ }
+ datum.tag = kTagDecimal;
+ // rust reassembles as ((int_val2 as i128) << 64) | (int_val as u64 as
i128):
+ // int_val holds the low 64 bits, int_val2 the (sign-extended) high 64
bits.
+ datum.int_val = static_cast<int64_t>(static_cast<uint64_t>(value));
+ datum.int_val2 = static_cast<int64_t>(value >> 64);
+ datum.uint_val = static_cast<uint32_t>(precision);
+ datum.uint_val2 = static_cast<uint32_t>(scale);
+ return holder;
+ }
+ default:
+ break;
+ }
+ return std::nullopt;
+}
+
+std::optional<std::string>
PaimonRustPredicateConverter::_extract_string_literal(
+ const VExprSPtr& expr) const {
+ // Same rule as _convert_literal: a bare string literal converts, a casted
+ // one is rejected — the cast is not executed here, so the pushed pattern
+ // must come from the literal's own value. A truncating cast (e.g. to
+ // CHAR(k)) would otherwise push a different prefix than the residual
+ // matches. Cast trees stay in the residual.
+ if (expr->node_type() == TExprNodeType::CAST_EXPR) {
+ return std::nullopt;
+ }
+ auto* literal = dynamic_cast<VLiteral*>(expr.get());
+ if (!literal) {
+ return std::nullopt;
+ }
+ auto literal_type = remove_nullable(literal->get_data_type());
+ if (!_is_string_type(literal_type->get_primitive_type())) {
+ return std::nullopt;
+ }
+
+ ColumnPtr col =
literal->get_column_ptr()->convert_to_full_column_if_const();
+ if (const auto* nullable = check_and_get_column<ColumnNullable>(*col)) {
+ if (nullable->is_null_at(0)) {
+ return std::nullopt;
+ }
+ col = nullable->get_nested_column_ptr();
+ }
+ Field field;
+ col->get(0, field);
+ const auto& value = field.get<TYPE_STRING>();
+ return std::string(value.data(), value.size());
+}
+
+paimon_predicate* PaimonRustPredicateConverter::_take(paimon_result_predicate
result) {
+ if (result.error != nullptr) {
+ // A single leaf failing to build is not fatal: that conjunct is simply
+ // dropped from the pushed-down filter and the engine still re-applies
it.
+ // Log at WARNING so it is visible without verbose logging enabled.
+ LOG(WARNING) << "paimon-rust build predicate failed: "
+ << consume_predicate_error(result.error);
+ return nullptr;
+ }
+ return result.predicate;
+}
+
+void PaimonRustPredicateConverter::_bind_datum_storage(paimon_datum* datum,
+ const std::string&
storage) {
+ if (datum->tag == kTagString || datum->tag == kTagBytes) {
+ datum->str_data = reinterpret_cast<const uint8_t*>(storage.data());
+ datum->str_len = storage.size();
+ }
+}
+
+std::string PaimonRustPredicateConverter::_normalize_name(std::string_view
name) {
+ std::string out(name);
+ std::transform(out.begin(), out.end(), out.begin(),
+ [](unsigned char c) { return
static_cast<char>(std::tolower(c)); });
+ return out;
+}
+
+int32_t PaimonRustPredicateConverter::_seconds_to_days(int64_t seconds) {
+ static constexpr int64_t kSecondsPerDay = 24 * 60 * 60;
+ int64_t days = seconds / kSecondsPerDay;
+ if (seconds < 0 && seconds % kSecondsPerDay != 0) {
+ --days;
+ }
+ return static_cast<int32_t>(days);
+}
+
+bool PaimonRustPredicateConverter::_is_integer_type(PrimitiveType type) {
+ switch (type) {
+ case TYPE_TINYINT:
+ case TYPE_SMALLINT:
+ case TYPE_INT:
+ case TYPE_BIGINT:
+ return true;
+ default:
+ return false;
+ }
+}
+
+bool PaimonRustPredicateConverter::_is_string_type(PrimitiveType type) {
+ return type == TYPE_CHAR || type == TYPE_VARCHAR || type == TYPE_STRING;
+}
+
+bool PaimonRustPredicateConverter::_is_decimal_type(PrimitiveType type) {
+ switch (type) {
+ case TYPE_DECIMALV2:
+ case TYPE_DECIMAL32:
+ case TYPE_DECIMAL64:
+ case TYPE_DECIMAL128I:
+ case TYPE_DECIMAL256:
+ return true;
+ default:
+ return false;
+ }
+}
+
+bool PaimonRustPredicateConverter::_is_date_type(PrimitiveType type) {
+ return type == TYPE_DATE || type == TYPE_DATEV2;
+}
+
+bool PaimonRustPredicateConverter::_is_datetime_type(PrimitiveType type) {
+ return type == TYPE_DATETIME || type == TYPE_DATETIMEV2;
+}
+
+bool PaimonRustPredicateConverter::_is_supported_slot_type(PrimitiveType type,
uint32_t precision) {
+ switch (type) {
+ case TYPE_BOOLEAN:
+ case TYPE_TINYINT:
+ case TYPE_SMALLINT:
+ case TYPE_INT:
+ case TYPE_BIGINT:
+ case TYPE_VARCHAR:
+ case TYPE_STRING:
+ case TYPE_DATE:
+ case TYPE_DATEV2:
+ case TYPE_DATETIME:
+ case TYPE_DATETIMEV2:
Review Comment:
Addressed in 6009fdb795. Timestamp predicates and arrived runtime filters
now remain residual until source precision can be proven equivalent to the
Doris slot. Added converter coverage and a TIMESTAMP(9) join differential
against the existing Flink nanosecond fixture, asserting the filter arrived and
was not applied in Rust.
##########
thirdparty/build-thirdparty.sh:
##########
@@ -2167,6 +2258,257 @@ build_lance_c() {
if [[ "${STRIP_TP_LIB}" = "ON" && "${KERNEL}" != 'Darwin' ]]; then
strip --strip-debug --strip-unneeded
"${TP_INSTALL_DIR}/lib64/liblance_c.a"
fi
+
+ # Only now (archive installed) is it safe to record the rustc identity
+ # this library was built with; see check_rust_toolchain_identity.
+ commit_rust_toolchain_identity lance_c
Review Comment:
Addressed in 6009fdb795. Both archives now use staged copy/strip plus
checksum-bound per-archive compiler identities. Each file is published with an
atomic rename; an interruption between the two renames leaves a
missing/mismatched identity that the next build rejects. Paired rebuilds clear
incomplete records. Failure-injection tests cover strip and publication
failures for both first installation and replacement.
--
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]
