Gabriel39 commented on code in PR #66227:
URL: https://github.com/apache/doris/pull/66227#discussion_r4119331073
##########
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
+}
+
+# paimon-rust
+build_paimon_rust() {
+ check_if_source_exist "${PAIMON_RUST_SOURCE}"
+ cd "${TP_SOURCE_DIR}/${PAIMON_RUST_SOURCE}"
+
+ rm -rf "${BUILD_DIR}"
+ mkdir -p "${BUILD_DIR}"
+
+ local cargo_bin="${PAIMON_RUST_CARGO:-${CARGO:-cargo}}"
+ if ! command -v "${cargo_bin}" >/dev/null 2>&1; then
+ echo "cargo is required to build paimon-rust. Install Rust 1.94.0 or
set PAIMON_RUST_CARGO."
+ exit 1
+ fi
+
+ local required_rust_version="1.94.0"
+ local cargo_env=(
+ "CARGO_BUILD_JOBS=${PARALLEL}"
+ "CARGO_TARGET_DIR=${PWD}/${BUILD_DIR}"
+ )
+ if command -v rustup >/dev/null 2>&1 && [[ -z "${RUSTUP_TOOLCHAIN}" ]];
then
+ # The presence check must look for the toolchain the minimum actually
+ # requires, not a literal: with only an older toolchain installed the
+ # stale check would skip the install below and then force
+ # RUSTUP_TOOLCHAIN to a version rustup cannot dispatch, failing the
+ # build before any archive is produced.
+ local required_rust_regex="${required_rust_version//./\\.}"
+ if ! rustup toolchain list | grep -Eq
"^${required_rust_regex}([[:space:]-]|$)"; then
+ rustup toolchain install "${required_rust_version}" --profile
minimal
+ fi
+ cargo_env+=("RUSTUP_TOOLCHAIN=${required_rust_version}")
+ fi
+
+ local cargo_version
+ if ! cargo_version="$(env "${cargo_env[@]}" "${cargo_bin}" --version | awk
'{print $2}')"; then
+ echo "failed to get cargo version for paimon-rust. Install Rust
${required_rust_version} or set PAIMON_RUST_CARGO/RUSTUP_TOOLCHAIN."
+ exit 1
+ fi
+ # Rust 1.94.0 is the minimum supported version. Allow newer toolchains when
+ # callers explicitly select one or rustup is unavailable on the system.
+ # NOTE: paimon_c and lance_c are both Rust staticlibs linked into the same
+ # BE binary; they must be built with the SAME rustc toolchain so the linker
+ # resolves both crates' std references against a single std copy. Mixing
+ # toolchains makes the precompiled std hashes differ and the linker pulls
+ # both std copies in, colliding on the unmangled `rust_eh_personality`
+ # (duplicate symbol). Build lance_c and paimon_rust with one toolchain.
+ if ! awk -v required="${required_rust_version}" -v
actual="${cargo_version}" 'BEGIN {
+ split(required, r, ".");
+ split(actual, a, ".");
+ for (i = 1; i <= 3; i++) {
+ if ((a[i] + 0) > (r[i] + 0)) {
+ exit 0;
+ }
+ if ((a[i] + 0) < (r[i] + 0)) {
+ exit 1;
+ }
+ }
+ exit 0;
+ }'; then
+ echo "paimon-rust requires Rust/Cargo ${required_rust_version} or
newer, but found ${cargo_version}."
+ echo "Install Rust ${required_rust_version} or set
PAIMON_RUST_CARGO/RUSTUP_TOOLCHAIN."
+ exit 1
+ fi
+
+ check_rust_toolchain_identity paimon_rust "${cargo_bin}" "${cargo_env[@]}"
+
+ if [[ "${KERNEL}" != 'Darwin' ]]; then
+ cargo_env+=("CFLAGS=${CFLAGS:-} -std=gnu17")
+ fi
+
+ # paimon-vindex-core 0.4.0 uses the unstable `stdarch_neon_f16` intrinsics
+ # (vcvt_f32_f16 / vreinterpret_f16_u16) in its aarch64 NEON fast path,
which
+ # do not compile on stable Rust. On aarch64/arm64, replace the registry
+ # crate with a patched path source so the build succeeds. The patch swaps
+ # the unstable f16->f32 NEON convert for a stable scalar conversion; the
+ # rest of the NEON accumulation is left untouched. x86_64 and other arches
+ # are unaffected and keep using the pristine registry crate.
+ if [[ "$(uname -m)" == "aarch64" || "$(uname -m)" == "arm64" ]]; then
+ # Retry safety: the patch + cargo update below mutate the shared
+ # workspace tree (Cargo.toml gains the [patch.crates-io] section and
+ # Cargo.lock's paimon-vindex-core entry loses its registry checksum).
+ # If a later step fails before cleanup_package_source removes the
+ # tree, the next invocation would reuse the mutated lock and the
+ # checksum extraction below would find nothing. Preserve pristine
+ # copies up front (inside the tree, so a successful cleanup removes
+ # them) and restore them before mutating again, making the whole
+ # block idempotent across retries.
+ local vindex_pristine_lock="${PWD}/.doris-vindex-pristine.lock"
+ local vindex_pristine_toml="${PWD}/.doris-vindex-pristine.toml"
+ if [[ ! -f "${vindex_pristine_lock}" ]]; then
+ cp Cargo.lock "${vindex_pristine_lock}"
+ fi
+ if [[ ! -f "${vindex_pristine_toml}" ]]; then
+ cp Cargo.toml "${vindex_pristine_toml}"
+ fi
+ cp "${vindex_pristine_lock}" Cargo.lock
+ cp "${vindex_pristine_toml}" Cargo.toml
+ local vindex_override="${PWD}/.doris-vindex-override"
+ local vindex_crate
+ vindex_crate="$(find "${CARGO_HOME:-$HOME/.cargo}/registry/cache" \
Review Comment:
Addressed in 6009fdb795. Fixed the cache bootstrap/recovery path. The helper
validates all candidates against Cargo.lock, tolerates an absent cache
directory, fetches and rescans online when no valid candidate exists, and fails
without fetching in offline mode. Shell tests cover missing, corrupt, and
multiple-registry caches.
##########
regression-test/suites/external_table_p0/paimon/test_paimon_rust_reader_v2.groovy:
##########
@@ -0,0 +1,114 @@
+// 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.
+
+// Paimon rust reader under FileScannerV2: enable_paimon_rust_reader=true must
work with
+// the default enable_file_scanner_v2=true (before the v2 port, a rust split
was rejected by
+// FileScannerV2::is_supported and the reader was only reachable with v2
disabled).
+suite("test_paimon_rust_reader_v2", "p0,external") {
+ String enabled = context.config.otherConfigs.get("enablePaimonTest")
+ if (enabled == null || !enabled.equalsIgnoreCase("true")) {
+ logger.info("disabled paimon test")
+ return
+ }
+
+ String catalogName = "test_paimon_rust_reader_v2"
+ String hdfsPort = context.config.otherConfigs.get("hive2HdfsPort")
+ String externalEnvIp = context.config.otherConfigs.get("externalEnvIp")
+
+ // Capture the settings this suite overrides before the try block, so the
+ // finally block can restore them even when the try body failed midway:
+ // a def inside the try body is out of scope in finally, and interpolating
+ // it there would throw and mask the original failure (the sibling suites
+ // declare the captures outside the try for the same reason).
+ def originalForceJni = sql("select @@force_jni_scanner")[0][0]
+ def originalV2 = sql("select @@enable_file_scanner_v2")[0][0]
+
+ try {
+ sql """drop catalog if exists ${catalogName}"""
+ sql """create catalog if not exists ${catalogName} properties (
+ "type" = "paimon",
+ "paimon.catalog.type" = "filesystem",
+ "warehouse" =
"hdfs://${externalEnvIp}:${hdfsPort}/user/doris/paimon1"
+ );"""
+
+ sql """switch ${catalogName}"""
+ sql """use db1"""
+ // Force the logical (JNI / rust) reader path: the append parquet
tables
+ // below (all_table, all_table_with_parquet, append_table, ...) convert
+ // to raw native splits, which getSplits() would otherwise prefer —
+ // both differential legs would silently use the native reader and
+ // never reach the JNI / rust converters.
+ sql """set force_jni_scanner=true"""
+ // FileScannerV2 is the default; make the intent of this suite
explicit.
+ sql """set enable_file_scanner_v2=true"""
+
+ def testQueries = [
+ """select c1 from complex_all order by c1""",
+ // Filter on a data column exercises the v2 predicate pushdown
into the
+ // paimon-rust filter (PaimonRustPredicateConverter
global-index mode).
+ """select c1 from complex_all where c1 >= 2 order by c1""",
+ """select * from all_table order by c1""",
+ """select * from all_table_with_parquet where c13 like '13%'
order by c1""",
+ """select * from complex_tab order by c1""",
+ """select c3['a_test'], c3['b_test'], c3['bbb'], c3['ccc']
from complex_tab order by c3['a_test'], c3['b_test']""",
+ """select array_max(c2) c from complex_tab order by c""",
+ """select c20[0] c from complex_all order by c""",
+ // ORDER BY keeps the JNI/rust differential deterministic: the
two
+ // readers may return the same rows in different valid
file/batch
+ // order, and the suite compares the result lists as strings.
+ """select * from deletion_vector_orc order by id""",
+ """select * from deletion_vector_parquet order by id""",
+ // count(*) exercises the table-level metadata count path of
the v2 leaf reader.
+ """select count(*) from append_table""",
+ """select count(*) from merge_on_read_table""",
+ """select count(*) from deletion_vector_parquet""",
+ // count(*) with a filter must fall back to the real rust scan.
+ """select count(*) from all_table where c1 >= 2""",
+ // Full scans of the count-suite tables exercise merge-on-read
and
+ // deletion-vector batches through the v2 leaf reader.
+ """select * from append_table order by product_id""",
+ """select * from merge_on_read_table order by product_id""",
+ """select * from deletion_vector_parquet order by id"""
+ ]
+
+ // Default path is JNI when enable_paimon_rust_reader=false.
+ sql """set enable_paimon_rust_reader=false"""
+ def jniResults = testQueries.collect { query -> sql(query) }
+
+ sql """set enable_paimon_rust_reader=true"""
Review Comment:
Addressed in 6009fdb795. Added profile assertions to both the V2 and
schema-evolution differentials. They use ProfileAction for readiness polling
and configured HTTP credentials, require FileScannerV2, reject a Rust timer in
the disabled leg, and require PaimonRustReader in the enabled leg.
##########
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:
Review Comment:
Addressed in 6009fdb795. Bounded VARCHAR predicates, including arrived IN
runtime filters, now remain residual because the converter has no per-file
bound-equivalence metadata. Added converter coverage and a historical VARCHAR
narrowing join differential with nonempty JNI/Rust result comparison,
arrived-filter counters, and zero Rust-applied filters.
--
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]
