github-actions[bot] commented on code in PR #68825: URL: https://github.com/apache/doris/pull/68825#discussion_r4280224754
########## be/src/format_v2/table/paimon_rust_predicate_converter.cpp: ########## @@ -0,0 +1,837 @@ +// 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 <limits> +#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) { + _converted_conjuncts = 0; + _converted_runtime_filters = 0; + if (_table == nullptr) { + return nullptr; + } + predicate_ptr result; + for (const auto& conjunct : conjuncts) { + if (!conjunct || !conjunct->root()) { + continue; + } + auto root = conjunct->root(); + const bool is_runtime_filter = root->is_rf_wrapper(); + 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; + } + ++_converted_conjuncts; + _converted_runtime_filters += is_runtime_filter; + 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) { + _converted_conjuncts = 0; + _converted_runtime_filters = 0; + 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)) { Review Comment: [P2] Reject residual-only IN filters before expanding the set. An arrived IN runtime filter reaches this branch as VDirectInPredicate, and get_slot_in_expr walks its entire HybridSet and allocates one VLiteral per key. Only afterward does _convert_in reject oversized lists or unsupported slot types. With runtime_filter_max_in_num=70000, the 65535-key case added in this PR therefore creates and discards 65535 literals for every logical split while pushing nothing; 10000 splits cause over 655 million unnecessary literal allocations. Check the original set size and slot support before expanding it (the native ORC path already does the size check), and cover the actual VDirectInPredicate path. -- 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]
