Gabriel39 commented on code in PR #66227: URL: https://github.com/apache/doris/pull/66227#discussion_r4015757770
########## be/src/format_v2/table/paimon_rust_table_reader.cpp: ########## @@ -0,0 +1,771 @@ +// 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_table_reader.h" + +#include <algorithm> +#include <string_view> +#include <utility> + +#include "arrow/c/abi.h" +#include "arrow/c/bridge.h" +#include "arrow/record_batch.h" +#include "arrow/result.h" +#include "common/logging.h" +#include "core/block/block.h" +#include "core/block/column_with_type_and_name.h" +#include "core/column/column_const.h" +#include "exprs/vexpr_context.h" +#include "exprs/vliteral.h" +#include "format/table/paimon_rust_predicate_converter.h" +#include "format_v2/column_mapper.h" +#include "runtime/descriptors.h" +#include "runtime/file_scan_profile.h" +#include "runtime/runtime_state.h" +#include "util/string_util.h" +#include "util/timezone_utils.h" +#include "util/url_coding.h" + +extern "C" { +#include "paimon_rust/paimon.h" +} + +namespace doris::format::paimon { + +namespace { +constexpr const char* VALUE_KIND_FIELD = "_VALUE_KIND"; + +// --------------------------------------------------------------------------- +// RAII wrappers over the paimon-rust C handles. Each handle is an opaque +// pointer owned by Rust and released by a matching paimon_*_free function. +// --------------------------------------------------------------------------- +#define PAIMON_OWNED(type, freefn) \ + struct type##_deleter { \ + void operator()(paimon_##type* p) const { \ + if (p) { \ + freefn(p); \ + } \ + } \ + }; \ + using type##_ptr = std::unique_ptr<paimon_##type, type##_deleter> + +PAIMON_OWNED(table, paimon_table_free); +PAIMON_OWNED(read_builder, paimon_read_builder_free); +PAIMON_OWNED(plan, paimon_plan_free); +PAIMON_OWNED(table_read, paimon_table_read_free); +PAIMON_OWNED(record_batch_reader, paimon_record_batch_reader_free); +PAIMON_OWNED(error, paimon_error_free); + +#undef PAIMON_OWNED + +// One Arrow batch (schema + array containers). Owning it requires a two-step +// teardown that the unique_ptr deleters above can't express: first invoke the +// Arrow C Data Interface `release` callback on each struct (hands buffers back +// to the producer), then free the container structs via paimon_arrow_batch_free. +class ArrowBatch { +public: + explicit ArrowBatch(paimon_arrow_batch batch) : batch_(batch) {} + ~ArrowBatch() { + auto* schema = static_cast<ArrowSchema*>(batch_.schema); + auto* array = static_cast<ArrowArray*>(batch_.array); + if (array && array->release) { + array->release(array); + } + if (schema && schema->release) { + schema->release(schema); + } + paimon_arrow_batch_free(batch_); + } + + ArrowBatch(const ArrowBatch&) = delete; + ArrowBatch& operator=(const ArrowBatch&) = delete; + + ArrowSchema* schema() const { return static_cast<ArrowSchema*>(batch_.schema); } + ArrowArray* array() const { return static_cast<ArrowArray*>(batch_.array); } + +private: + paimon_arrow_batch batch_; +}; + +// 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_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; +} + +// Render storage options for diagnostics. Values of sensitive keys (secret / +// password / token / access key) are masked so credentials never hit the log. +std::string format_options(const std::map<std::string, std::string>& options) { + std::string out; + for (const auto& kv : options) { + if (!out.empty()) { + out += ", "; + } + std::string_view key = kv.first; + const bool sensitive = key.find("secret") != std::string_view::npos || + key.find("password") != std::string_view::npos || + key.find("token") != std::string_view::npos || + key.find("access.key") != std::string_view::npos || + key.find("access-key") != std::string_view::npos; + out += kv.first; + out += '='; + out += sensitive ? "***" : kv.second; + } + return out; +} + +} // namespace + +// Paimon-rust handles. Order of members matters: destruction runs in reverse +// declaration order, and the read_builder depends on the table while the arrow +// reader depends on the whole pipeline above it. So the table MUST be declared +// first (destroyed last) and the record batch reader last. +struct PaimonRustTableReader::PaimonHandles { + table_ptr table; + read_builder_ptr read_builder; + plan_ptr plan; + table_read_ptr table_read; + record_batch_reader_ptr reader; +}; + +PaimonRustTableReader::PaimonRustTableReader() = default; + +PaimonRustTableReader::~PaimonRustTableReader() = default; + +Status PaimonRustTableReader::init(format::TableReadOptions&& options) { + RETURN_IF_ERROR(format::TableReader::init(std::move(options))); + { + // Base and derived scopes must not overlap on the same counter: RuntimeProfile timers + // add deltas, so nested use would double-count instead of extending lifecycle coverage. + SCOPED_TIMER(_profile.total_timer); + SCOPED_TIMER(_profile.init_timer); + TimezoneUtils::find_cctz_time_zone(TimezoneUtils::default_time_zone, _ctz); Review Comment: [P1] Use the session timezone when materializing TIMESTAMP_LTZ `TimezoneUtils::default_time_zone` is fixed at `+08:00`; it does not reflect the query's timezone. With the default catalog setting `enable.mapping.timestamp_tz=false`, Paimon TIMESTAMP_LTZ is mapped to DATETIMEV2. Rust emits an Arrow timestamp with UTC timezone metadata, and `DataTypeDateTimeV2SerDe::read_column_from_arrow` uses the supplied `_ctz` for that timezone-aware input. Consequently, a session using `time_zone='+00:00'` reads epoch 0 as `1970-01-01 08:00:00` through this reader, whereas the existing JNI path uses the session timezone and returns `1970-01-01 00:00:00`. This can also change filtering and joins, not just displayed formatting. Please initialize `_ctz` from RuntimeState's session timezone and add JNI/Rust comparisons for LTZ under UTC and a non-default timezone, while preserving NTZ wall-clock semantics. ########## fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/source/PaimonScanNode.java: ########## @@ -411,10 +412,52 @@ 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; + // paimon-rust additionally requires 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. + Table paimonTable = source.getPaimonTable(); + boolean canUseRust = sessionVariable.isEnablePaimonRustReader() && nativeSplit Review Comment: [P2] Align Rust selection with Scanner V2 availability `canUseRust` does not check `enable_file_scanner_v2`, but the V1 FileScanner explicitly returns NotSupported for PAIMON_RUST. With Rust enabled and V2 disabled, a logical DataSplit (for example, a merge-on-read table, or a scan forced through `force_jni_scanner=true`) is therefore encoded into a request that the selected BE scanner cannot consume. The new `test_paimon_rust_reader_v2` suite explicitly disables V2 while leaving Rust enabled at lines 83-87 and expects successful fallback, which contradicts the implementation. Please make these contracts consistent: select JNI in FE when V2 is disabled if fallback is intended, or explicitly reject the configuration combination and change the regression to expect that error. The PR description should reflect the chosen behavior as well. ########## be/src/format/table/paimon_rust_predicate_converter.cpp: ########## @@ -0,0 +1,725 @@ +// 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/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/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" +#include "util/timezone_utils.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])); + } + if (!TimezoneUtils::find_cctz_time_zone("GMT", _gmt_tz)) { + TimezoneUtils::find_cctz_time_zone(TimezoneUtils::default_time_zone, _gmt_tz); + } +} + +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()) { + root = impl; + } + } + 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(); +} + +paimon_predicate* PaimonRustPredicateConverter::_convert_expr(const VExprSPtr& expr) { + if (!expr) { + return nullptr; + } + + auto uncast = VExpr::expr_without_cast(expr); + + if (auto* direct_in = dynamic_cast<VDirectInPredicate*>(uncast.get())) { + VExprSPtr in_expr; + if (direct_in->get_slot_in_expr(in_expr)) { + return _convert_in(in_expr); + } + return nullptr; + } + + if (dynamic_cast<VInPredicate*>(uncast.get()) != nullptr) { + return _convert_in(uncast); + } + + switch (uncast->op()) { + case TExprOpcode::COMPOUND_AND: + case TExprOpcode::COMPOUND_OR: + return _convert_compound(uncast); + 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(uncast); + default: + break; + } + + if (auto* fn = dynamic_cast<VectorizedFnCall*>(uncast.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(uncast, fn_name); + } + if (fn_name == "like") { + return _convert_like_prefix(uncast); + } + } + + 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) { + 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(); + + if (expr->op() == TExprOpcode::EQ_FOR_NULL) { + return _take(paimon_predicate_is_null(_table, column)); Review Comment: [P1] Preserve column-to-column NULL-safe equality semantics This branch generates `IS NULL` before checking the RHS. For two nullable data columns, `WHERE a <=> b` with `(a, b) = (1, 1)` must retain the row, but the Rust predicate becomes `a IS NULL` and can discard it. The existing FE converter rejects this column-to-column case because the RHS is not a literal; FileScannerV2 still passes the original conjunct to this new converter, so this is a new failure path rather than the pre-existing FE handling of literal NULL-safe comparisons. Please only generate `IS NULL` for a NULL literal, use equality for a supported non-NULL literal, and leave column-to-column comparisons to the Doris residual. Re-evaluating the original predicate after the Rust scan cannot recover discarded rows. A test with `(NULL, NULL)`, `(1, 1)`, and `(1, 2)` would cover the distinction. ########## be/src/format/table/paimon_rust_predicate_converter.cpp: ########## @@ -0,0 +1,725 @@ +// 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/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/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" +#include "util/timezone_utils.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])); + } + if (!TimezoneUtils::find_cctz_time_zone("GMT", _gmt_tz)) { + TimezoneUtils::find_cctz_time_zone(TimezoneUtils::default_time_zone, _gmt_tz); + } +} + +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()) { + root = impl; + } + } + 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(); +} + +paimon_predicate* PaimonRustPredicateConverter::_convert_expr(const VExprSPtr& expr) { + if (!expr) { + return nullptr; + } + + auto uncast = VExpr::expr_without_cast(expr); + + if (auto* direct_in = dynamic_cast<VDirectInPredicate*>(uncast.get())) { + VExprSPtr in_expr; + if (direct_in->get_slot_in_expr(in_expr)) { + return _convert_in(in_expr); + } + return nullptr; + } + + if (dynamic_cast<VInPredicate*>(uncast.get()) != nullptr) { + return _convert_in(uncast); + } + + switch (uncast->op()) { + case TExprOpcode::COMPOUND_AND: + case TExprOpcode::COMPOUND_OR: + return _convert_compound(uncast); + 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(uncast); + default: + break; + } + + if (auto* fn = dynamic_cast<VectorizedFnCall*>(uncast.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(uncast, fn_name); + } + if (fn_name == "like") { + return _convert_like_prefix(uncast); + } + } + + 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) { + 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(); + + if (expr->op() == TExprOpcode::EQ_FOR_NULL) { + return _take(paimon_predicate_is_null(_table, column)); + } + + auto holder = _convert_literal(expr->get_child(1), field_meta->type); + if (!holder) { + return nullptr; + } + // `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_prefix(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; + // Only prefix matches (`abc%`) are convertible to a range scan. + if (!pattern.empty() && pattern.front() == '%') { + return nullptr; + } + if (pattern.empty() || pattern.back() != '%') { + return nullptr; + } + + const char* column = field_meta->column.c_str(); + std::string prefix = pattern.substr(0, pattern.size() - 1); + + // lower bound: column >= prefix + paimon_datum lower {}; + lower.tag = kTagString; + _bind_datum_storage(&lower, prefix); + predicate_ptr lower_pred(_take(paimon_predicate_greater_or_equal(_table, column, lower))); + if (!lower_pred) { + return nullptr; + } + + auto upper_prefix = _next_prefix(prefix); + if (!upper_prefix) { + return lower_pred.release(); + } + + // upper bound: column < next_prefix + paimon_datum upper {}; + upper.tag = kTagString; + _bind_datum_storage(&upper, *upper_prefix); + predicate_ptr upper_pred(_take(paimon_predicate_less_than(_table, column, upper))); + if (!upper_pred) { + // No usable upper bound: fall back to the (still correct) lower bound. + return lower_pred.release(); + } + return paimon_predicate_and(lower_pred.release(), upper_pred.release()); +} + +std::optional<PaimonRustPredicateConverter::FieldMeta> PaimonRustPredicateConverter::_resolve_field( + const VExprSPtr& expr) const { + if (!expr) { + return std::nullopt; + } + auto slot_expr = VExpr::expr_without_cast(expr); + auto* slot_ref = dynamic_cast<VSlotRef*>(slot_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 { + auto literal_expr = VExpr::expr_without_cast(expr); + auto* literal = dynamic_cast<VLiteral*>(literal_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, _gmt_tz); + } else { + const auto& dt = field.get<TYPE_DATEV2>(); + if (!dt.is_valid_date()) { + return std::nullopt; + } + dt.unix_timestamp(&seconds, _gmt_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; + // nanos is left at 0 to match paimon-cpp's millisecond-granularity + // Timestamp::FromEpochMillis behaviour. + 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, _gmt_tz); + datum.int_val = seconds * 1000; + } else { + const auto& dt = field.get<TYPE_DATETIMEV2>(); + if (!dt.is_valid_date()) { + return std::nullopt; + } + std::pair<int64_t, int64_t> ts; + dt.unix_timestamp(&ts, _gmt_tz); + datum.int_val = ts.first * 1000 + ts.second / 1000; Review Comment: [P1] Preserve fractional timestamps in runtime-filter literals This truncates microseconds to milliseconds while `DatumHolder` leaves `int_val2` (the Rust timestamp's nanos field) at zero. A concrete new-path trigger is an equality join on TIMESTAMP(6): if an IN runtime filter containing `2024-01-01 00:00:00.123456` arrives before the Rust split opens, `build -> get_impl -> VDirectInPredicate::get_slot_in_expr -> _convert_in -> _convert_literal` changes that value to `.123000`, allowing Rust to reject a matching probe row. This runtime value is created after FE planning, so the scenario is independent of the existing FE timestamp-literal truncation. In the pinned Rust C binding, `datum_from_c` reads nanos from `int_val2`, and timestamp comparison includes both millis and nanos. Please preserve the sub-millisecond remainder, and conservatively skip pushdown where external/Doris precision semantics cannot be matched. Add a fractional timestamp IN-runtime-filter join test with the filter available before split opening. -- 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]
