From 19215661377b83ab7d8c9dadbefc4e9ebc52f408 Mon Sep 17 00:00:00 2001 From: dengfangyuan Date: Wed, 29 Jul 2026 14:24:27 +0800 Subject: [PATCH 1/4] [chore](thirdparty) introduce paimon-rust Add PAIMON_RUST source tarball (paimon-rust-2fb5e49) and a build_paimon_rust helper that builds paimon-c via cargo, generates the C header from the Rust extern "C" surface via cbindgen (installed via cargo install when missing), and installs libpaimon_c.a + paimon.h under ${TP_INSTALL_DIR}. The build script writes a temporary cbindgen.toml at build time so we don't have to patch the upstream tree, and pins Rust/Cargo to 1.91.0 (auto-installed via rustup when available) so builds are reproducible across environments. --- thirdparty/build-thirdparty.sh | 82 ++++++++++++++++++++++++++++++++++ thirdparty/vars.sh | 7 +++ 2 files changed, 89 insertions(+) diff --git a/thirdparty/build-thirdparty.sh b/thirdparty/build-thirdparty.sh index 0c83f56ba0f1fa..f67616a2b3c32a 100755 --- a/thirdparty/build-thirdparty.sh +++ b/thirdparty/build-thirdparty.sh @@ -2123,6 +2123,86 @@ build_paimon_cpp() { echo "Paimon-cpp internal dependencies installed successfully" } +# 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.91.0 or set PAIMON_RUST_CARGO." + exit 1 + fi + + local required_rust_version="1.91.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 + if ! rustup toolchain list | grep -Eq '^1\.91\.0([[: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 + if [[ "${cargo_version}" != "${required_rust_version}" ]]; then + echo "paimon-rust requires Rust/Cargo ${required_rust_version}, but found ${cargo_version}." + echo "Install Rust ${required_rust_version} or set PAIMON_RUST_CARGO/RUSTUP_TOOLCHAIN." + exit 1 + fi + + if [[ "${KERNEL}" != 'Darwin' ]]; then + cargo_env+=("CFLAGS=${CFLAGS:-} -std=gnu17") + fi + + local cargo_args=(build --release --locked -p paimon-c --features paimon/storage-hdfs) + if [[ "$(echo "${PAIMON_RUST_CARGO_OFFLINE}" | tr '[:lower:]' '[:upper:]')" == "ON" ]]; then + cargo_args+=(--offline) + fi + env "${cargo_env[@]}" "${cargo_bin}" "${cargo_args[@]}" + + # Generate the C header from the Rust extern "C" surface via cbindgen. + # Auto-install cbindgen if it's not already on PATH. + local cbindgen_bin="${PAIMON_RUST_CBINDGEN:-cbindgen}" + if ! command -v "${cbindgen_bin}" >/dev/null 2>&1; then + echo "cbindgen not found; installing via cargo install ..." + env "${cargo_env[@]}" "${cargo_bin}" install cbindgen --locked + cbindgen_bin="cbindgen" + fi + # Write a temporary cbindgen.toml so the generated header carries our + # include-guard / cpp-compat settings without touching the upstream tree. + local cbindgen_toml="${BUILD_DIR}/cbindgen.toml" + mkdir -p "${BUILD_DIR}" + cat >"${cbindgen_toml}" <<'EOF' +language = "C" +include_guard = "PAIMON_C_H" +pragma_once = true +cpp_compat = true +EOF + env "${cargo_env[@]}" "${cbindgen_bin}" bindings/c \ + --config "${cbindgen_toml}" \ + --output "${BUILD_DIR}/release/paimon.h" + + mkdir -p "${TP_INSTALL_DIR}/include" "${TP_INSTALL_DIR}/lib64" + rm -rf "${TP_INSTALL_DIR}/include/paimon_rust" + mkdir -p "${TP_INSTALL_DIR}/include/paimon_rust" + cp -v "${BUILD_DIR}/release/paimon.h" "${TP_INSTALL_DIR}/include/paimon_rust/" + cp -v "${BUILD_DIR}/release/libpaimon_c.a" "${TP_INSTALL_DIR}/lib64/" + + if [[ "${STRIP_TP_LIB}" = "ON" && "${KERNEL}" != 'Darwin' ]]; then + strip --strip-debug --strip-unneeded "${TP_INSTALL_DIR}/lib64/libpaimon_c.a" + fi +} + if [[ "${#packages[@]}" -eq 0 ]]; then packages=( jindofs @@ -2198,6 +2278,7 @@ if [[ "${#packages[@]}" -eq 0 ]]; then icu pugixml paimon_cpp + paimon_rust ) if [[ "$(uname -s)" == 'Darwin' ]]; then read -r -a packages <<<"binutils gettext ${packages[*]}" @@ -2297,6 +2378,7 @@ cleanup_package_source() { juicefs) src_var="JUICEFS_SOURCE" ;; pugixml) src_var="PUGIXML_SOURCE" ;; paimon_cpp) src_var="PAIMON_CPP_SOURCE" ;; + paimon_rust) src_var="PAIMON_RUST_SOURCE" ;; aws_sdk) src_var="AWS_SDK_SOURCE" ;; lzma) src_var="LZMA_SOURCE" ;; xml2) src_var="XML2_SOURCE" ;; diff --git a/thirdparty/vars.sh b/thirdparty/vars.sh index 77ed4e32e35b51..80e7c4d3815934 100644 --- a/thirdparty/vars.sh +++ b/thirdparty/vars.sh @@ -574,6 +574,12 @@ PAIMON_CPP_NAME="paimon-cpp-0a4f4e2.tar.gz" PAIMON_CPP_SOURCE="doris-thirdparty-paimon-cpp-0a4f4e2" PAIMON_CPP_MD5SUM="b8599a0421dbf1ec05e2f1a481d64e87" +# paimon-rust +PAIMON_RUST_DOWNLOAD="https://github.com/pzhdfy/paimon-rust/archive/refs/tags/paimon-rust-2fb5e49.tar.gz" +PAIMON_RUST_NAME="paimon-rust-2fb5e49.tar.gz" +PAIMON_RUST_SOURCE="paimon-rust-paimon-rust-2fb5e49" +PAIMON_RUST_MD5SUM="341aef6bd75743e51cdb960260645d69" + # all thirdparties which need to be downloaded is set in array TP_ARCHIVES export TP_ARCHIVES=( 'LIBEVENT' @@ -659,6 +665,7 @@ export TP_ARCHIVES=( 'JUICEFS' 'PUGIXML' 'PAIMON_CPP' + 'PAIMON_RUST' ) if [[ "$(uname -s)" == 'Darwin' ]]; then From bdcde4adbf4a829820cc0f9194e4ffb46cc2418a Mon Sep 17 00:00:00 2001 From: dengfangyuan Date: Wed, 29 Jul 2026 14:27:18 +0800 Subject: [PATCH 2/4] [chore](thirdparty) link be to paimon-rust libpaimon_c.a (built from paimon-rust) brings in Rust compiler_builtins that would otherwise steal libm symbols. Place libm before it so the final linker resolves C math symbols from the system library first. --- be/cmake/thirdparty.cmake | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/be/cmake/thirdparty.cmake b/be/cmake/thirdparty.cmake index 8aa4ae73020489..e3d92c1d159988 100644 --- a/be/cmake/thirdparty.cmake +++ b/be/cmake/thirdparty.cmake @@ -110,6 +110,11 @@ add_thirdparty(arrow_flight_sql LIB64) add_thirdparty(arrow_dataset LIB64) add_thirdparty(arrow_acero LIB64) add_thirdparty(parquet LIB64) +# libpaimon_c.a (built from paimon-rust) brings in Rust compiler_builtins that +# would otherwise steal libm symbols. Place libm before it so the final linker +# resolves C math symbols from the system library first. +add_thirdparty(paimon_c LIB64 NOTADD) +list(APPEND COMMON_THIRDPARTY m paimon_c) add_thirdparty(brpc LIB64) add_thirdparty(rocksdb) add_thirdparty(cyrus-sasl LIBNAME "lib/libsasl2.a") From ff96b79f1d0927a6ec48b5ecd65ebc2988ca6da3 Mon Sep 17 00:00:00 2001 From: dengfangyuan Date: Wed, 29 Jul 2026 14:28:00 +0800 Subject: [PATCH 3/4] [feat][paimon] support paimon rust MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add PaimonRustReader that reads Paimon splits on BE via the paimon-rust C bindings (libpaimon_c), plumbed into the FORMAT_JNI paimon dispatch alongside PaimonJniReader and PaimonCppReader. Pipeline on BE: catalog_create(warehouse) -> catalog_get_table(db, table) -> read_builder -> projection -> filter -> plan_from_split_bytes(FE split) -> read -> arrow record batch stream FE emits PAIMON_RUST as the reader type when enable_paimon_rust_reader is set (higher priority than enable_paimon_cpp_reader). Both native paimon readers consume the same DataSplit binary wire format, so non-DataSplit inputs transparently fall back to PAIMON_JNI. FE additionally ships db/table so BE can open the paimon catalog. The FE-planned split bytes are deserialized directly into a one-split plan via paimon_plan_from_split_bytes so each scanner reads exactly the split it was assigned rather than replanning the whole table. Push-down conjuncts are converted to a paimon-rust filter predicate via PaimonRustPredicateConverter after the table opens (paimon_predicate_* functions need the live table handle to resolve fields from the schema). Conjuncts that cannot be represented are dropped silently — the engine still re-applies the full conjunct list. Adds: - BE: format/table/paimon_rust_reader.{h,cpp} format/table/paimon_rust_predicate_converter.{h,cpp} test/format/table/paimon_rust_reader_test.cpp - FE: SessionVariable.enable_paimon_rust_reader (session var, fuzzy) PaimonScanNode dispatch (RUST > CPP > JNI when native split) - Thrift: TPaimonReaderType.PAIMON_RUST = 3 TQueryOptions.enable_paimon_rust_reader = 227 --- be/src/exec/scan/file_scanner.cpp | 58 +- .../table/paimon_rust_predicate_converter.cpp | 726 ++++++++++++++++++ .../table/paimon_rust_predicate_converter.h | 131 ++++ be/src/format/table/paimon_rust_reader.cpp | 568 ++++++++++++++ be/src/format/table/paimon_rust_reader.h | 123 +++ .../format/table/paimon_rust_reader_test.cpp | 157 ++++ .../paimon/source/PaimonScanNode.java | 30 +- .../org/apache/doris/qe/SessionVariable.java | 18 + gensrc/thrift/PaloInternalService.thrift | 4 + gensrc/thrift/PlanNodes.thrift | 1 + 10 files changed, 1810 insertions(+), 6 deletions(-) create mode 100644 be/src/format/table/paimon_rust_predicate_converter.cpp create mode 100644 be/src/format/table/paimon_rust_predicate_converter.h create mode 100644 be/src/format/table/paimon_rust_reader.cpp create mode 100644 be/src/format/table/paimon_rust_reader.h create mode 100644 be/test/format/table/paimon_rust_reader_test.cpp diff --git a/be/src/exec/scan/file_scanner.cpp b/be/src/exec/scan/file_scanner.cpp index 5c7beb3e87e268..0cc693384d65b2 100644 --- a/be/src/exec/scan/file_scanner.cpp +++ b/be/src/exec/scan/file_scanner.cpp @@ -77,6 +77,7 @@ #include "format/table/paimon_jni_reader.h" #include "format/table/paimon_predicate_converter.h" #include "format/table/paimon_reader.h" +#include "format/table/paimon_rust_reader.h" #include "format/table/partition_column_filler.h" #include "format/table/remote_doris_reader.h" #include "format/table/transactional_hive_reader.h" @@ -1111,8 +1112,61 @@ Status FileScanner::_get_next_reader() { _cur_reader = std::move(mc_reader); } else if (range.__isset.table_format_params && range.table_format_params.table_format_type == "paimon") { - if (_state->query_options().__isset.enable_paimon_cpp_reader && - _state->query_options().enable_paimon_cpp_reader) { + const auto& paimon_params = range.table_format_params.paimon_params; + bool use_paimon_rust_reader = false; + bool use_paimon_cpp_reader = false; + if (paimon_params.__isset.reader_type) { + switch (paimon_params.reader_type) { + case TPaimonReaderType::PAIMON_RUST: + use_paimon_rust_reader = true; + break; + case TPaimonReaderType::PAIMON_CPP: + use_paimon_cpp_reader = true; + break; + case TPaimonReaderType::PAIMON_JNI: + break; + case TPaimonReaderType::PAIMON_NATIVE: + return Status::InternalError( + "invalid PAIMON_NATIVE reader_type for paimon FORMAT_JNI split, " + "possibly caused by FE/BE protocol mismatch"); + default: + return Status::InternalError( + "unknown paimon reader_type for paimon FORMAT_JNI split, possibly " + "caused by FE/BE protocol mismatch"); + } + } else { + // TODO: Remove this fallback after all FE versions set TPaimonReaderType. + // Rust has higher priority than cpp. + use_paimon_rust_reader = + _state->query_options().__isset.enable_paimon_rust_reader && + _state->query_options().enable_paimon_rust_reader; + if (!use_paimon_rust_reader) { + use_paimon_cpp_reader = + _state->query_options().__isset.enable_paimon_cpp_reader && + _state->query_options().enable_paimon_cpp_reader; + } + } + // Native paimon readers (rust/cpp) push conjuncts down, but they live + // under FORMAT_JNI where push_down_predicates is false, so the generic + // parquet/orc path never ran _process_late_arrival_conjuncts() and + // _push_down_conjuncts is still empty here. Populate it now (same call + // parquet/orc make) so the predicate converters see the conjuncts. + if (!_is_load && (use_paimon_rust_reader || use_paimon_cpp_reader)) { + RETURN_IF_ERROR(_process_late_arrival_conjuncts()); + } + if (use_paimon_rust_reader) { + auto rust_reader = PaimonRustReader::create_unique(_file_slot_descs, _state, + _profile, range, _params); + rust_reader->set_push_down_agg_type(_get_push_down_agg_type()); + if (!_is_load && !_push_down_conjuncts.empty()) { + // paimon-rust needs the live table handle to build predicates, + // so the converter runs inside init_reader. Just hand over the + // conjuncts here. + rust_reader->set_push_down_conjuncts(_push_down_conjuncts); + } + init_status = rust_reader->init_reader(); + _cur_reader = std::move(rust_reader); + } else if (use_paimon_cpp_reader) { auto cpp_reader = PaimonCppReader::create_unique(_file_slot_descs, _state, _profile, range, _params); cpp_reader->set_push_down_agg_type(_get_push_down_agg_type()); diff --git a/be/src/format/table/paimon_rust_predicate_converter.cpp b/be/src/format/table/paimon_rust_predicate_converter.cpp new file mode 100644 index 00000000000000..f6fc05512fae0c --- /dev/null +++ b/be/src/format/table/paimon_rust_predicate_converter.cpp @@ -0,0 +1,726 @@ +// 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 +#include +#include +#include + +#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 "runtime/descriptors.h" +#include "runtime/runtime_state.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; + +// 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; + +// 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(owned->message.data), owned->message.len); + } + return "code=" + std::to_string(owned->code) + ", msg=" + msg; +} +} // namespace + +PaimonRustPredicateConverter::PaimonRustPredicateConverter( + const std::vector& file_slot_descs, RuntimeState* state, + const paimon_table* table) + : _state(state), _table(table) { + _file_columns.reserve(file_slot_descs.size()); + for (const auto& slot : file_slot_descs) { + _file_columns.insert(_normalize_name(slot->col_name())); + } + 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(uncast.get())) { + VExprSPtr in_expr; + if (direct_in->get_slot_in_expr(in_expr)) { + return _convert_in(in_expr); + } + return nullptr; + } + + if (dynamic_cast(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(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(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 storages; + std::vector 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->slot_desc); + 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->slot_desc); + 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->slot_desc->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::_resolve_field( + const VExprSPtr& expr) const { + if (!_state || !expr) { + return std::nullopt; + } + auto slot_expr = VExpr::expr_without_cast(expr); + auto* slot_ref = dynamic_cast(slot_expr.get()); + if (!slot_ref) { + return std::nullopt; + } + auto* slot_desc = _state->desc_tbl().get_slot_descriptor(slot_ref->slot_id()); + if (!slot_desc) { + return std::nullopt; + } + if (_file_columns.find(_normalize_name(slot_desc->col_name())) == _file_columns.end()) { + return std::nullopt; + } + auto slot_type = slot_desc->type(); + if (!_is_supported_slot_type(slot_type->get_primitive_type(), slot_type->get_precision())) { + return std::nullopt; + } + return FieldMeta {slot_desc->col_name(), slot_desc}; +} + +std::optional +PaimonRustPredicateConverter::_convert_literal(const VExprSPtr& expr, + const SlotDescriptor& slot_desc) const { + auto literal_expr = VExpr::expr_without_cast(expr); + auto* literal = dynamic_cast(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 = slot_desc.type()->get_primitive_type(); + + ColumnPtr col = literal->get_column_ptr()->convert_to_full_column_if_const(); + if (const auto* nullable = check_and_get_column(*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(field.get()) ? 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(); + break; + case TYPE_SMALLINT: + value = field.get(); + break; + case TYPE_INT: + value = field.get(); + break; + case TYPE_BIGINT: + value = field.get(); + 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(field.get()) + : field.get(); + 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(); + if (!dt.is_valid_date()) { + return std::nullopt; + } + dt.unix_timestamp(&seconds, _gmt_tz); + } else { + const auto& dt = field.get(); + 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(); + 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(); + if (!dt.is_valid_date()) { + return std::nullopt; + } + std::pair ts; + dt.unix_timestamp(&ts, _gmt_tz); + datum.int_val = ts.first * 1000 + ts.second / 1000; + } + return holder; + } + case TYPE_VARCHAR: + case TYPE_STRING: { + if (!_is_string_type(literal_primitive)) { + return std::nullopt; + } + const auto& value = field.get(); + 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(literal_type->get_precision()); + int32_t scale = static_cast(literal_type->get_scale()); + if (precision <= 0 || precision > kPaimonDecimalMaxPrecision) { + return std::nullopt; + } + + __int128 value = 0; + switch (literal_primitive) { + case TYPE_DECIMALV2: + value = field.get().value(); + break; + case TYPE_DECIMAL32: + value = field.get().value; + break; + case TYPE_DECIMAL64: + value = field.get().value; + break; + case TYPE_DECIMAL128I: + value = field.get().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(static_cast(value)); + datum.int_val2 = static_cast(value >> 64); + datum.uint_val = static_cast(precision); + datum.uint_val2 = static_cast(scale); + return holder; + } + default: + break; + } + return std::nullopt; +} + +std::optional PaimonRustPredicateConverter::_extract_string_literal( + const VExprSPtr& expr) const { + auto literal_expr = VExpr::expr_without_cast(expr); + auto* literal = dynamic_cast(literal_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(*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(); + 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(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(std::tolower(c)); }); + return out; +} + +std::optional PaimonRustPredicateConverter::_next_prefix(const std::string& prefix) { + if (prefix.empty()) { + return std::nullopt; + } + std::string upper = prefix; + for (int i = static_cast(upper.size()) - 1; i >= 0; --i) { + auto c = static_cast(upper[i]); + if (c != 0xFF) { + upper[i] = static_cast(c + 1); + upper.resize(i + 1); + return upper; + } + } + return std::nullopt; +} + +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(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_DOUBLE: + case TYPE_VARCHAR: + case TYPE_STRING: + case TYPE_DATE: + case TYPE_DATEV2: + case TYPE_DATETIME: + case TYPE_DATETIMEV2: + return true; + case TYPE_DECIMALV2: + case TYPE_DECIMAL32: + case TYPE_DECIMAL64: + case TYPE_DECIMAL128I: + case TYPE_DECIMAL256: + // precision == 0 means "unset"; only a positive precision above the + // paimon ceiling is unrepresentable. + return precision <= static_cast(kPaimonDecimalMaxPrecision); + case TYPE_FLOAT: + case TYPE_CHAR: + default: + return false; + } +} + +} // namespace doris diff --git a/be/src/format/table/paimon_rust_predicate_converter.h b/be/src/format/table/paimon_rust_predicate_converter.h new file mode 100644 index 00000000000000..d4d7d0afd4cab1 --- /dev/null +++ b/be/src/format/table/paimon_rust_predicate_converter.h @@ -0,0 +1,131 @@ +// 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. + +#pragma once + +#include +#include +#include +#include +#include + +#include "cctz/time_zone.h" +#include "core/data_type/define_primitive_type.h" +#include "exprs/vexpr_fwd.h" + +// paimon-rust C bindings. This header (and the paimon_* C API) is only pulled +// into the rust-specific translation units (the converter, the rust reader and +// its test); it is intentionally kept out of file_scanner and other generic BE +// code, mirroring how paimon-cpp's headers are kept isolated. +extern "C" { +#include "paimon_rust/paimon.h" +} + +namespace doris { +class RuntimeState; +class SlotDescriptor; +} // namespace doris + +namespace doris { + +// Converts Doris push-down conjuncts into a paimon-rust filter predicate. +// +// This mirrors PaimonPredicateConverter (the paimon-cpp variant) but targets the +// paimon-rust C bindings. The key API differences drive the design: +// * paimon-cpp builds predicates standalone via PredicateBuilder using a field +// index + paimon::FieldType + paimon::Literal. +// * paimon-rust builds predicates from a live paimon_table handle and a column +// name (`paimon_predicate_equal(table, column, datum)`), resolving the field +// index and type from the table schema itself, and takes typed values as a +// flat tagged `paimon_datum`. +// +// Because the predicate functions need the opened table, the converter is +// constructed with the table handle and must run after the table is open. The +// returned `paimon_predicate*` is owned by the caller, who must either transfer +// it to `paimon_read_builder_with_filter` (which consumes it) or release it via +// `paimon_predicate_free`. +// +// Conversion is best effort: any conjunct that cannot be represented is dropped +// (the engine still re-applies the full conjunct list to the scanned block), so +// a partial filter only ever prunes a superset and never changes results. +class PaimonRustPredicateConverter { +public: + PaimonRustPredicateConverter(const std::vector& file_slot_descs, + RuntimeState* state, const paimon_table* table); + + // Builds an owned predicate AND-combining every convertible conjunct, or + // nullptr when nothing can be pushed down. + paimon_predicate* build(const VExprContextSPtrs& conjuncts); + +private: + struct FieldMeta { + // Original file column name passed verbatim to the paimon_predicate_* + // functions (rust resolves the field index/type from the schema by name). + std::string column; + const SlotDescriptor* slot_desc = nullptr; + }; + + // A datum plus its backing byte storage. String/Bytes datums hold a borrowed + // pointer into `storage`; the storage must outlive the predicate build call, + // so callers keep the holder alive and (re)bind str_data after final placement. + struct DatumHolder { + // The `{}` value-initializes every paimon_datum field to zero. Conversion + // code relies on this: each branch only sets the fields its tag uses (e.g. + // a Timestamp sets int_val but leaves int_val2/nanos at 0), so the rest + // must already be zeroed. Do not drop the `{}`. + paimon_datum datum {}; + std::string storage; + }; + + // Each returns an owned paimon_predicate* (transfer or free) or nullptr. + paimon_predicate* _convert_expr(const VExprSPtr& expr); + paimon_predicate* _convert_compound(const VExprSPtr& expr); + paimon_predicate* _convert_in(const VExprSPtr& expr); + paimon_predicate* _convert_binary(const VExprSPtr& expr); + paimon_predicate* _convert_is_null(const VExprSPtr& expr, const std::string& fn_name); + paimon_predicate* _convert_like_prefix(const VExprSPtr& expr); + + std::optional _resolve_field(const VExprSPtr& expr) const; + std::optional _convert_literal(const VExprSPtr& expr, + const SlotDescriptor& slot_desc) const; + std::optional _extract_string_literal(const VExprSPtr& expr) const; + + // Consumes a paimon_result_predicate: frees the error (logging it) and + // returns the predicate, or nullptr on error. + static paimon_predicate* _take(paimon_result_predicate result); + // Point a String/Bytes datum's str_data/str_len at the given stable storage. + static void _bind_datum_storage(paimon_datum* datum, const std::string& storage); + + static std::string _normalize_name(std::string_view name); + static std::optional _next_prefix(const std::string& prefix); + static int32_t _seconds_to_days(int64_t seconds); + static bool _is_integer_type(PrimitiveType type); + static bool _is_string_type(PrimitiveType type); + static bool _is_decimal_type(PrimitiveType type); + static bool _is_date_type(PrimitiveType type); + static bool _is_datetime_type(PrimitiveType type); + // Whether a slot of this type can be represented as a paimon_datum (mirrors + // the type coverage of paimon-cpp's _to_paimon_field_type). + static bool _is_supported_slot_type(PrimitiveType type, uint32_t precision); + + std::unordered_set _file_columns; // normalized file column names + RuntimeState* _state = nullptr; + const paimon_table* _table = nullptr; + cctz::time_zone _gmt_tz; +}; + +} // namespace doris diff --git a/be/src/format/table/paimon_rust_reader.cpp b/be/src/format/table/paimon_rust_reader.cpp new file mode 100644 index 00000000000000..ecaab5bda367d9 --- /dev/null +++ b/be/src/format/table/paimon_rust_reader.cpp @@ -0,0 +1,568 @@ +// 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_reader.h" + +#include +#include +#include + +#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 "exprs/vexpr.h" +#include "format/table/paimon_rust_predicate_converter.h" +#include "runtime/descriptors.h" +#include "runtime/runtime_state.h" +#include "util/string_util.h" +#include "util/url_coding.h" + +extern "C" { +#include "paimon_rust/paimon.h" +} + +namespace doris { + +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_OWNED(catalog, paimon_catalog_free); +PAIMON_OWNED(identifier, paimon_identifier_free); +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(batch_.schema); + auto* array = static_cast(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(batch_.schema); } + ArrowArray* array() const { return static_cast(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(owned->message.data), owned->message.len); + } + return "code=" + std::to_string(owned->code) + ", msg=" + msg; +} + +// Render catalog 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& 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 + +// Long-lived paimon-rust handles. Order of members matters: destruction runs +// in reverse declaration order, and the table depends on the catalog, the +// read_builder on the table, and so on down to the record batch reader. So the +// catalog MUST be declared first (destroyed last) and the reader last. +struct PaimonRustReader::PaimonHandles { + catalog_ptr catalog; + table_ptr table; + read_builder_ptr read_builder; + plan_ptr plan; + table_read_ptr table_read; + record_batch_reader_ptr reader; +}; + +PaimonRustReader::PaimonRustReader(const std::vector& file_slot_descs, + RuntimeState* state, RuntimeProfile* profile, + const TFileRangeDesc& range, + const TFileScanRangeParams* range_params) + : _file_slot_descs(file_slot_descs), + _state(state), + _profile(profile), + _range(range), + _range_params(range_params) { + TimezoneUtils::find_cctz_time_zone(TimezoneUtils::default_time_zone, _ctzz); + if (range.__isset.table_format_params && + range.table_format_params.__isset.table_level_row_count) { + _remaining_table_level_row_count = range.table_format_params.table_level_row_count; + } else { + _remaining_table_level_row_count = -1; + } +} + +PaimonRustReader::~PaimonRustReader() = default; + +Status PaimonRustReader::init_reader() { + if (_push_down_agg_type == TPushAggOp::type::COUNT && _remaining_table_level_row_count >= 0) { + return Status::OK(); + } + return _init_paimon_reader(); +} + +Status PaimonRustReader::get_next_block(Block* block, size_t* read_rows, bool* eof) { + if (_push_down_agg_type == TPushAggOp::type::COUNT && _remaining_table_level_row_count >= 0) { + auto rows = std::min(_remaining_table_level_row_count, + (int64_t)_state->query_options().batch_size); + _remaining_table_level_row_count -= rows; + auto mutable_columns_guard = block->mutate_columns_scoped(); + auto& mutate_columns = mutable_columns_guard.mutable_columns(); + for (auto& col : mutate_columns) { + col->resize(rows); + } + *read_rows = rows; + *eof = false; + if (_remaining_table_level_row_count == 0) { + *eof = true; + } + return Status::OK(); + } + + // num_splits == 0 yields an empty (but valid) stream: report EOF. + if (_reader_eof) { + *read_rows = 0; + *eof = true; + return Status::OK(); + } + if (!_handles || !_handles->reader) { + return Status::InternalError("paimon-rust reader is not initialized"); + } + + if (_col_name_to_block_idx.empty()) { + _col_name_to_block_idx = block->get_name_to_pos_map(); + } + + paimon_result_next_batch next = paimon_record_batch_reader_next(_handles->reader.get()); + if (next.error != nullptr) { + return Status::InternalError("paimon-rust read batch failed: {}", + consume_error(next.error)); + } + // End of stream: both pointers are null. + if (next.batch.array == nullptr && next.batch.schema == nullptr) { + _reader_eof = true; + *read_rows = 0; + *eof = true; + return Status::OK(); + } + + // RAII: the batch's Arrow release callbacks + container free run when + // `batch` leaves this scope, including on any early break/continue. + ArrowBatch batch(next.batch); + + auto* c_array = batch.array(); + auto* c_schema = batch.schema(); + arrow::Result> import_result = + arrow::ImportRecordBatch(c_array, c_schema); + if (!import_result.ok()) { + return Status::InternalError("failed to import paimon-rust arrow batch: {}", + import_result.status().message()); + } + + auto record_batch = std::move(import_result).ValueUnsafe(); + const auto num_rows = static_cast(record_batch->num_rows()); + const auto num_columns = record_batch->num_columns(); + auto columns_guard = block->mutate_columns_scoped(); + auto& columns = columns_guard.mutable_columns(); + for (int c = 0; c < num_columns; ++c) { + const auto& field = record_batch->schema()->field(c); + if (field->name() == VALUE_KIND_FIELD) { + continue; + } + // Block column names are FE-normalized to lowercase. paimon-rust's + // case_sensitive=false setting also case-folds column names in the + // schema output, so exact match works — but we tolerate mixed-case + // Rust output by folding here as well. + auto it = _col_name_to_block_idx.find(field->name()); + if (it == _col_name_to_block_idx.end()) { + it = _col_name_to_block_idx.find(to_lower(field->name())); + } + if (it == _col_name_to_block_idx.end()) { + // Skip columns that are not in the block (e.g., partition columns). + continue; + } + const auto block_pos = it->second; + try { + RETURN_IF_ERROR(columns_guard.get_datatype_by_position(block_pos) + ->get_serde() + ->read_column_from_arrow(*columns[block_pos], + record_batch->column(c).get(), 0, + num_rows, _ctzz)); + } catch (Exception& e) { + return Status::InternalError("Failed to convert from arrow to block: {}", e.what()); + } + } + + *read_rows = num_rows; + *eof = false; + return Status::OK(); +} + +Status PaimonRustReader::get_columns(std::unordered_map* name_to_type, + std::unordered_set* missing_cols) { + for (const auto& slot : _file_slot_descs) { + name_to_type->emplace(slot->col_name(), slot->type()); + } + return Status::OK(); +} + +Status PaimonRustReader::close() { + _handles.reset(); + return Status::OK(); +} + +Status PaimonRustReader::_init_paimon_reader() { + _handles = std::make_unique(); + return _open_table_and_build_reader(); +} + +Status PaimonRustReader::_open_table_and_build_reader() { + // 1. Decode the FE-planned split first so we fail fast (and without any + // filesystem IO) when it is missing or malformed. The bytes themselves are + // not consumed yet by the community API — see + // _paimon_plan_from_split_bytes_mock — but the fail-fast check still + // guards against a caller that "forgot" to attach a split. + std::string split_bytes; + RETURN_IF_ERROR(_decode_split_bytes(&split_bytes)); + + // 2. Resolve identifier and warehouse (paimon catalog convention: the + // table root is `/.db/`). + auto table_path_opt = _resolve_table_path(); + if (!table_path_opt.has_value()) { + return Status::InternalError( + "paimon-rust missing paimon_table; cannot resolve paimon table location"); + } + auto db_name_opt = _resolve_db_name(); + if (!db_name_opt.has_value()) { + return Status::InternalError( + "paimon-rust missing db_name; cannot open paimon table via catalog"); + } + auto table_name_opt = _resolve_table_name(); + if (!table_name_opt.has_value()) { + return Status::InternalError( + "paimon-rust missing table_name; cannot open paimon table via catalog"); + } + const std::string& table_path = table_path_opt.value(); + const std::string& db_name = db_name_opt.value(); + const std::string& table_name = table_name_opt.value(); + + auto warehouse_opt = _derive_warehouse(table_path, db_name, table_name); + if (!warehouse_opt.has_value()) { + return Status::InternalError( + "paimon-rust cannot derive warehouse from table_path='{}' (expected suffix " + "'/{}.db/{}')", + table_path, db_name, table_name); + } + + // 3. Assemble catalog options: warehouse + optional storage credentials + // from properties/paimon_options/hadoop_conf. Default to metastore=filesystem + // for path-based paimon layouts. + auto options = _build_options(); + options["warehouse"] = warehouse_opt.value(); + if (options.find("metastore") == options.end()) { + options["metastore"] = "filesystem"; + } + + std::vector c_options; + c_options.reserve(options.size()); + for (const auto& kv : options) { + c_options.push_back(paimon_option {kv.first.c_str(), kv.second.c_str()}); + } + + LOG(INFO) << "paimon-rust opening table via catalog: db=" << db_name + << " table=" << table_name << " warehouse=" << options["warehouse"] + << " options=[" << format_options(options) << "]"; + + // 4. Create catalog. + paimon_result_catalog_new cat_res = + paimon_catalog_create(c_options.data(), c_options.size()); + if (cat_res.error != nullptr) { + return Status::InternalError("paimon-rust catalog_create failed: {}; options=[{}]", + consume_error(cat_res.error), format_options(options)); + } + _handles->catalog.reset(cat_res.catalog); + + // 5. Build identifier and look up the table. + paimon_result_identifier_new id_res = + paimon_identifier_new(db_name.c_str(), table_name.c_str()); + if (id_res.error != nullptr) { + return Status::InternalError("paimon-rust identifier_new failed: {}", + consume_error(id_res.error)); + } + identifier_ptr identifier(id_res.identifier); + + paimon_result_get_table tbl_res = + paimon_catalog_get_table(_handles->catalog.get(), identifier.get()); + if (tbl_res.error != nullptr) { + return Status::InternalError( + "paimon-rust get_table failed: db={} table={} err={}", db_name, table_name, + consume_error(tbl_res.error)); + } + _handles->table.reset(tbl_res.table); + + // 6. Build the read pipeline: read_builder -> case-insensitive -> projection. + paimon_result_read_builder rb_res = paimon_table_new_read_builder(_handles->table.get()); + if (rb_res.error != nullptr) { + return Status::InternalError("paimon-rust new read builder failed: {}", + consume_error(rb_res.error)); + } + _handles->read_builder.reset(rb_res.read_builder); + + // Fold column casing on the Rust side so FE-normalized lowercase names + // resolve against tables with mixed-case column definitions. + if (paimon_error* case_err = + paimon_read_builder_with_case_sensitive(_handles->read_builder.get(), false)) { + return Status::InternalError("paimon-rust set case_sensitive failed: {}", + consume_error(case_err)); + } + + auto read_columns = _build_read_columns(); + std::vector projection; + projection.reserve(read_columns.size() + 1); + for (const auto& col : read_columns) { + projection.push_back(col.c_str()); + } + projection.push_back(nullptr); + if (paimon_error* proj_err = paimon_read_builder_with_projection(_handles->read_builder.get(), + projection.data())) { + return Status::InternalError("paimon-rust set projection failed: {}", + consume_error(proj_err)); + } + + // Convert the FE push-down conjuncts into a paimon-rust filter and apply it. + RETURN_IF_ERROR(_apply_predicate()); + + // 7. Deserialize the FE-planned split into a one-split plan, so this + // scanner reads exactly the split it was assigned rather than replanning + // the whole table. The wire form is identical to what paimon-cpp consumes + // (`paimon::table::DataSplit::serialize`). + paimon_result_plan plan_res = paimon_plan_from_split_bytes( + reinterpret_cast(split_bytes.data()), split_bytes.size()); + if (plan_res.error != nullptr) { + return Status::InternalError("paimon-rust build plan failed: {}", + consume_error(plan_res.error)); + } + _handles->plan.reset(plan_res.plan); + + size_t num_splits = paimon_plan_num_splits(_handles->plan.get()); + if (num_splits == 0) { + _reader_eof = true; + return Status::OK(); + } + + // 8. Open the arrow stream over the plan. + paimon_result_new_read read_res = paimon_read_builder_new_read(_handles->read_builder.get()); + if (read_res.error != nullptr) { + return Status::InternalError("paimon-rust new read failed: {}", + consume_error(read_res.error)); + } + _handles->table_read.reset(read_res.read); + + paimon_result_record_batch_reader rdr_res = paimon_table_read_to_arrow( + _handles->table_read.get(), _handles->plan.get(), /*offset=*/0, /*length=*/num_splits); + if (rdr_res.error != nullptr) { + return Status::InternalError("paimon-rust open arrow reader failed: {}", + consume_error(rdr_res.error)); + } + _handles->reader.reset(rdr_res.reader); + return Status::OK(); +} + +Status PaimonRustReader::_apply_predicate() { + if (_push_down_conjuncts.empty() || !_handles || !_handles->table || !_handles->read_builder) { + return Status::OK(); + } + LOG(INFO) << "paimon-rust predicate pushdown: " << _push_down_conjuncts.size() + << " conjunct(s) input"; + PaimonRustPredicateConverter converter(_file_slot_descs, _state, _handles->table.get()); + paimon_predicate* predicate = converter.build(_push_down_conjuncts); + if (predicate == nullptr) { + LOG(INFO) << "paimon-rust predicate pushdown: nothing convertible, no filter applied"; + return Status::OK(); + } + // paimon_read_builder_with_filter consumes the predicate (ownership moves to + // the builder) on every path, so we must not free it here. + if (paimon_error* err = + paimon_read_builder_with_filter(_handles->read_builder.get(), predicate)) { + return Status::InternalError("paimon-rust apply filter failed: {}", consume_error(err)); + } + LOG(INFO) << "paimon-rust predicate pushdown: applied"; + return Status::OK(); +} + +Status PaimonRustReader::_decode_split_bytes(std::string* out) const { + if (!_range.__isset.table_format_params || !_range.table_format_params.__isset.paimon_params || + !_range.table_format_params.paimon_params.__isset.paimon_split) { + return Status::InternalError("paimon-rust missing paimon_split in scan range"); + } + const auto& encoded_split = _range.table_format_params.paimon_params.paimon_split; + if (!base64_decode(encoded_split, out)) { + return Status::InternalError("paimon-rust base64 decode paimon_split failed"); + } + if (out->empty()) { + return Status::InternalError("paimon-rust decoded paimon_split is empty"); + } + return Status::OK(); +} + +std::optional PaimonRustReader::_resolve_table_path() const { + if (_range.__isset.table_format_params && _range.table_format_params.__isset.paimon_params && + _range.table_format_params.paimon_params.__isset.paimon_table && + !_range.table_format_params.paimon_params.paimon_table.empty()) { + return _range.table_format_params.paimon_params.paimon_table; + } + return std::nullopt; +} + +std::optional PaimonRustReader::_resolve_db_name() const { + if (_range.__isset.table_format_params && _range.table_format_params.__isset.paimon_params && + _range.table_format_params.paimon_params.__isset.db_name && + !_range.table_format_params.paimon_params.db_name.empty()) { + return _range.table_format_params.paimon_params.db_name; + } + return std::nullopt; +} + +std::optional PaimonRustReader::_resolve_table_name() const { + if (_range.__isset.table_format_params && _range.table_format_params.__isset.paimon_params && + _range.table_format_params.paimon_params.__isset.table_name && + !_range.table_format_params.paimon_params.table_name.empty()) { + return _range.table_format_params.paimon_params.table_name; + } + return std::nullopt; +} + +std::optional PaimonRustReader::_derive_warehouse(std::string_view table_path, + std::string_view db_name, + std::string_view table_name) { + // paimon convention: /.db/
[/] + // Strip a trailing slash first so both forms work. + while (!table_path.empty() && table_path.back() == '/') { + table_path.remove_suffix(1); + } + // Suffix should end with `/.db/
`. + std::string suffix; + suffix.reserve(db_name.size() + 4 + table_name.size() + 1); + suffix.append("/"); + suffix.append(db_name); + suffix.append(".db/"); + suffix.append(table_name); + if (table_path.size() <= suffix.size()) { + return std::nullopt; + } + if (table_path.substr(table_path.size() - suffix.size()) != suffix) { + return std::nullopt; + } + return std::string(table_path.substr(0, table_path.size() - suffix.size())); +} + +std::vector PaimonRustReader::_build_read_columns() const { + std::vector columns; + columns.reserve(_file_slot_descs.size()); + for (const auto& slot : _file_slot_descs) { + columns.emplace_back(slot->col_name()); + } + return columns; +} + +std::map PaimonRustReader::_build_options() const { + std::map options; + if (_range_params && _range_params->__isset.paimon_options && + !_range_params->paimon_options.empty()) { + options.insert(_range_params->paimon_options.begin(), _range_params->paimon_options.end()); + } else if (_range.__isset.table_format_params && + _range.table_format_params.__isset.paimon_params && + _range.table_format_params.paimon_params.__isset.paimon_options) { + options.insert(_range.table_format_params.paimon_params.paimon_options.begin(), + _range.table_format_params.paimon_params.paimon_options.end()); + } + + if (_range_params && _range_params->__isset.properties && !_range_params->properties.empty()) { + for (const auto& kv : _range_params->properties) { + options[kv.first] = kv.second; + } + } else if (_range.__isset.table_format_params && + _range.table_format_params.__isset.paimon_params && + _range.table_format_params.paimon_params.__isset.hadoop_conf) { + for (const auto& kv : _range.table_format_params.paimon_params.hadoop_conf) { + options[kv.first] = kv.second; + } + } + return options; +} + +} // namespace doris diff --git a/be/src/format/table/paimon_rust_reader.h b/be/src/format/table/paimon_rust_reader.h new file mode 100644 index 00000000000000..98afb422005c29 --- /dev/null +++ b/be/src/format/table/paimon_rust_reader.h @@ -0,0 +1,123 @@ +// 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. + +#pragma once + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "cctz/time_zone.h" +#include "common/status.h" +#include "exprs/vexpr_fwd.h" +#include "format/generic_reader.h" + +namespace doris { +class RuntimeProfile; +class RuntimeState; +class SlotDescriptor; +} // namespace doris + +namespace doris { + +class Block; + +// Reads Paimon data on BE by calling into the paimon-rust C bindings +// (libpaimon_c) through the catalog-based read pipeline: +// +// catalog_create(warehouse) -> catalog_get_table(db, table) +// -> read_builder -> projection -> filter +// -> plan_from_split_bytes(FE split) -> read -> arrow record batch stream +// +// The FE-planned split bytes are deserialized directly into a one-split plan +// via `paimon_plan_from_split_bytes` (wire form is identical to the one +// paimon-cpp consumes), so each scanner reads exactly the split assigned to +// it rather than re-planning the whole table. +class PaimonRustReader : public GenericReader { + ENABLE_FACTORY_CREATOR(PaimonRustReader); + +public: + PaimonRustReader(const std::vector& file_slot_descs, RuntimeState* state, + RuntimeProfile* profile, const TFileRangeDesc& range, + const TFileScanRangeParams* range_params); + ~PaimonRustReader() override; + + Status init_reader(); + Status get_next_block(Block* block, size_t* read_rows, bool* eof) override; + Status get_columns(std::unordered_map* name_to_type, + std::unordered_set* missing_cols) override; + Status close() override; + + // Stores the FE push-down conjuncts. They are converted to a paimon-rust + // filter predicate (via PaimonRustPredicateConverter) inside init_reader, + // after the table is opened, because the paimon_predicate_* functions need + // the live table handle to resolve fields from the schema. + void set_push_down_conjuncts(const VExprContextSPtrs& conjuncts) { + _push_down_conjuncts = conjuncts; + } + +private: + // Opaque holder for the paimon-rust C handles (defined in the .cpp so that + // paimon_rust/paimon.h does not leak into other translation units). + struct PaimonHandles; + + Status _init_paimon_reader(); + // Open the table via the catalog and build the scan plan + arrow reader. + Status _open_table_and_build_reader(); + std::optional _resolve_table_path() const; + std::optional _resolve_db_name() const; + std::optional _resolve_table_name() const; + // Derive `warehouse` from the table root path by stripping the trailing + // `/.db/
` segments (paimon catalog convention). Returns nullopt + // when the path shape is unexpected. + static std::optional _derive_warehouse(std::string_view table_path, + std::string_view db_name, + std::string_view table_name); + // Base64-decode the FE-planned split into the raw serialized DataSplit bytes. + // Kept even though the community API cannot deserialize them yet, so we + // fail fast on missing / malformed splits without any filesystem IO. + Status _decode_split_bytes(std::string* out) const; + std::vector _build_read_columns() const; + std::map _build_options() const; + // Builds a paimon-rust filter from _push_down_conjuncts and applies it to the + // read builder. Conjuncts that cannot be represented are dropped silently + // (the engine still re-applies the full conjunct list), but an actual failure + // of the paimon_read_builder_with_filter C call is surfaced as an error. + Status _apply_predicate(); + + const std::vector& _file_slot_descs; + RuntimeState* _state = nullptr; + [[maybe_unused]] RuntimeProfile* _profile = nullptr; + const TFileRangeDesc& _range; + const TFileScanRangeParams* _range_params = nullptr; + + VExprContextSPtrs _push_down_conjuncts; + std::unique_ptr _handles; + + std::unordered_map _col_name_to_block_idx; + int64_t _remaining_table_level_row_count = -1; + bool _reader_eof = false; + cctz::time_zone _ctzz; +}; + +} // namespace doris diff --git a/be/test/format/table/paimon_rust_reader_test.cpp b/be/test/format/table/paimon_rust_reader_test.cpp new file mode 100644 index 00000000000000..0a4038bfeb6bee --- /dev/null +++ b/be/test/format/table/paimon_rust_reader_test.cpp @@ -0,0 +1,157 @@ +// 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_reader.h" + +#include + +#include +#include + +#include "core/block/block.h" +#include "runtime/runtime_profile.h" +#include "runtime/runtime_state.h" +#include "util/url_coding.h" + +namespace doris { + +class PaimonRustReaderTest : public testing::Test { +protected: + void SetUp() override { + _query_options.__set_batch_size(3); + _runtime_state = RuntimeState::create_unique(_query_options, _query_globals); + } + + TFileRangeDesc _build_range_with_table_level_row_count(int64_t row_count) { + TFileRangeDesc range; + range.__isset.table_format_params = true; + range.table_format_params.__isset.table_level_row_count = true; + range.table_format_params.table_level_row_count = row_count; + return range; + } + + // Build a scan range with a non-empty (but arbitrary) base64-encoded split + // so _decode_split_bytes() succeeds. The bytes are ignored by the mock + // plan-from-split helper, but they must be present and decodable so the + // reader gets past the fail-fast validation. + TFileRangeDesc _build_range_with_split() { + TFileRangeDesc range; + range.__isset.table_format_params = true; + range.table_format_params.__isset.paimon_params = true; + range.table_format_params.paimon_params.__isset.paimon_split = true; + std::string dummy = "dummy-split-bytes"; + std::string encoded; + base64_encode(dummy, &encoded); + range.table_format_params.paimon_params.paimon_split = encoded; + return range; + } + + TQueryOptions _query_options; + TQueryGlobals _query_globals; + std::unique_ptr _runtime_state; + RuntimeProfile _profile {"paimon_rust_reader_test"}; + std::vector _file_slot_descs; +}; + +TEST_F(PaimonRustReaderTest, CountPushDownUsesTableLevelRowCount) { + auto range = _build_range_with_table_level_row_count(5); + PaimonRustReader reader(_file_slot_descs, _runtime_state.get(), &_profile, range, nullptr); + reader.set_push_down_agg_type(TPushAggOp::type::COUNT); + + auto init_status = reader.init_reader(); + ASSERT_TRUE(init_status.ok()) << init_status; + + Block block; + size_t read_rows = 0; + bool eof = false; + + auto first_status = reader.get_next_block(&block, &read_rows, &eof); + ASSERT_TRUE(first_status.ok()) << first_status; + EXPECT_EQ(3, read_rows); + EXPECT_FALSE(eof); + + auto second_status = reader.get_next_block(&block, &read_rows, &eof); + ASSERT_TRUE(second_status.ok()) << second_status; + EXPECT_EQ(2, read_rows); + EXPECT_TRUE(eof); + + auto third_status = reader.get_next_block(&block, &read_rows, &eof); + ASSERT_TRUE(third_status.ok()) << third_status; + EXPECT_EQ(0, read_rows); + EXPECT_TRUE(eof); +} + +TEST_F(PaimonRustReaderTest, InitReaderFailsWithoutSplit) { + // No paimon_split set on the range: _decode_split_bytes must reject it + // before any filesystem IO happens. + TFileRangeDesc range; + range.__isset.table_format_params = true; + range.table_format_params.__isset.paimon_params = true; + range.table_format_params.paimon_params.__isset.paimon_table = true; + range.table_format_params.paimon_params.paimon_table = + "file:///tmp/doris_paimon_rust_ut_missing/db.db/tbl"; + // paimon_split intentionally left unset. + + PaimonRustReader reader(_file_slot_descs, _runtime_state.get(), &_profile, range, nullptr); + auto status = reader.init_reader(); + + ASSERT_FALSE(status.ok()); + EXPECT_NE(status.to_string().find("missing paimon_split"), std::string::npos) << status; +} + +TEST_F(PaimonRustReaderTest, InitReaderFailsWithoutTablePath) { + auto range = _build_range_with_split(); + // paimon_table intentionally left unset. + + PaimonRustReader reader(_file_slot_descs, _runtime_state.get(), &_profile, range, nullptr); + auto status = reader.init_reader(); + + ASSERT_FALSE(status.ok()); + EXPECT_NE(status.to_string().find("missing paimon_table"), std::string::npos) << status; +} + +TEST_F(PaimonRustReaderTest, InitReaderFailsWithoutDbName) { + auto range = _build_range_with_split(); + range.table_format_params.paimon_params.__isset.paimon_table = true; + range.table_format_params.paimon_params.paimon_table = + "file:///tmp/doris_paimon_rust_ut_missing/db.db/tbl"; + // db_name / table_name intentionally left unset. + + PaimonRustReader reader(_file_slot_descs, _runtime_state.get(), &_profile, range, nullptr); + auto status = reader.init_reader(); + + ASSERT_FALSE(status.ok()); + EXPECT_NE(status.to_string().find("missing db_name"), std::string::npos) << status; +} + +TEST_F(PaimonRustReaderTest, InitReaderFailsWithoutTableName) { + auto range = _build_range_with_split(); + range.table_format_params.paimon_params.__isset.paimon_table = true; + range.table_format_params.paimon_params.paimon_table = + "file:///tmp/doris_paimon_rust_ut_missing/db.db/tbl"; + range.table_format_params.paimon_params.__isset.db_name = true; + range.table_format_params.paimon_params.db_name = "db"; + // table_name intentionally left unset. + + PaimonRustReader reader(_file_slot_descs, _runtime_state.get(), &_profile, range, nullptr); + auto status = reader.init_reader(); + + ASSERT_FALSE(status.ok()); + EXPECT_NE(status.to_string().find("missing table_name"), std::string::npos) << status; +} + +} // namespace doris diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/source/PaimonScanNode.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/source/PaimonScanNode.java index aab64aeb034079..9d9934489b917d 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/source/PaimonScanNode.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/source/PaimonScanNode.java @@ -302,11 +302,33 @@ 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 - // until the C++ path has a split-aware V2 adapter. - 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; + if (sessionVariable.isEnablePaimonRustReader() && nativeSplit) { + fileDesc.setReaderType(TPaimonReaderType.PAIMON_RUST); + fileDesc.setPaimonSplit(PaimonUtil.encodeDataSplitToString((DataSplit) split)); + } else if (sessionVariable.isEnablePaimonCppReader() && nativeSplit) { + fileDesc.setReaderType(TPaimonReaderType.PAIMON_CPP); + fileDesc.setPaimonSplit(PaimonUtil.encodeDataSplitToString((DataSplit) split)); + } else { + // A logical DataSplit may span multiple files, so keep it intact for the JNI reader + // until the C++ path has a split-aware V2 adapter. + fileDesc.setReaderType(TPaimonReaderType.PAIMON_JNI); + fileDesc.setPaimonSplit(PaimonUtil.encodeObjectToString(split)); + } + // Set table location for paimon-cpp / paimon-rust reader + String tableLocation = source.getTableLocation(); + if (tableLocation != null) { + fileDesc.setPaimonTable(tableLocation); + } + // paimon-rust reader needs db/table to open the table through the catalog. + if (sessionVariable.isEnablePaimonRustReader()) { + fileDesc.setDbName(source.getExternalTable().getDbName()); + fileDesc.setTableName(source.getExternalTable().getName()); + } rangeDesc.setSelfSplitWeight(paimonSplit.getSelfSplitWeight()); } else { // use native reader diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java index fedf7a126c02cf..36e4e6550d9312 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java +++ b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java @@ -792,6 +792,8 @@ public String toString() { public static final String ENABLE_PAIMON_CPP_READER = "enable_paimon_cpp_reader"; + public static final String ENABLE_PAIMON_RUST_READER = "enable_paimon_rust_reader"; + public static final String ENABLE_COUNT_PUSH_DOWN_FOR_EXTERNAL_TABLE = "enable_count_push_down_for_external_table"; public static final String FETCH_ALL_FE_FOR_SYSTEM_TABLE = "fetch_all_fe_for_system_table"; @@ -2954,6 +2956,12 @@ public static boolean isEagerAggregationOnJoin() { description = {"Paimon 非原生文件读取使用 paimon-cpp", "Use paimon-cpp for non-native Paimon reads"}) private boolean enablePaimonCppReader = false; + @VariableMgr.VarAttr(name = ENABLE_PAIMON_RUST_READER, + fuzzy = true, + description = {"Paimon 非原生文件读取使用 paimon-rust,优先级高于 paimon-cpp", + "Use paimon-rust for non-native Paimon reads, higher priority than paimon-cpp"}) + private boolean enablePaimonRustReader = false; + @VariableMgr.VarAttr(name = ENABLE_COUNT_PUSH_DOWN_FOR_EXTERNAL_TABLE, fuzzy = true, description = {"对外表启用 count(*) 下推优化", "enable count(*) pushdown optimization for external table"}) @@ -3858,6 +3866,7 @@ private void setFuzzyForCatalog(Random random) { // jni this.forceJniScanner = random.nextBoolean(); this.enablePaimonCppReader = random.nextBoolean(); + this.enablePaimonRustReader = random.nextBoolean(); // statistics this.fetchHiveRowCountSync = random.nextBoolean(); @@ -5579,6 +5588,7 @@ public TQueryOptions toThrift() { tResult.setEnableOrcFilterByMinMax(enableOrcFilterByMinMax); tResult.setEnableExprZonemapFilter(enableExprZonemapFilter); tResult.setEnablePaimonCppReader(enablePaimonCppReader); + tResult.setEnablePaimonRustReader(enablePaimonRustReader); tResult.setCheckOrcInitSargsSuccess(checkOrcInitSargsSuccess); tResult.setTruncateCharOrVarcharColumns(truncateCharOrVarcharColumns); @@ -6369,6 +6379,10 @@ public boolean isEnablePaimonCppReader() { return enablePaimonCppReader; } + public boolean isEnablePaimonRustReader() { + return enablePaimonRustReader; + } + public String getIgnoreSplitType() { return ignoreSplitType; } @@ -6396,6 +6410,10 @@ public void setEnablePaimonCppReader(boolean enable) { enablePaimonCppReader = enable; } + public void setEnablePaimonRustReader(boolean enable) { + enablePaimonRustReader = enable; + } + public boolean isEnableCountPushDownForExternalTable() { return enableCountPushDownForExternalTable; } diff --git a/gensrc/thrift/PaloInternalService.thrift b/gensrc/thrift/PaloInternalService.thrift index 6add308255d921..6764505c607525 100644 --- a/gensrc/thrift/PaloInternalService.thrift +++ b/gensrc/thrift/PaloInternalService.thrift @@ -500,6 +500,10 @@ struct TQueryOptions { 224: optional bool enable_expr_zonemap_filter = true 225: optional i64 runtime_filter_tree_publish_max_send_bytes = 268435456 + + // Use paimon-rust to read Paimon splits on BE. Higher priority than + // enable_paimon_cpp_reader. + 227: optional bool enable_paimon_rust_reader = false; // For cloud, to control if the content would be written into file cache // In write path, to control if the content would be written into file cache. // In read path, read from file cache or remote storage when execute query. diff --git a/gensrc/thrift/PlanNodes.thrift b/gensrc/thrift/PlanNodes.thrift index a76e4543b38f30..64739135fb2cd0 100644 --- a/gensrc/thrift/PlanNodes.thrift +++ b/gensrc/thrift/PlanNodes.thrift @@ -352,6 +352,7 @@ enum TPaimonReaderType { PAIMON_NATIVE = 0, PAIMON_JNI = 1, PAIMON_CPP = 2, + PAIMON_RUST = 3, } struct TPaimonFileDesc { From b07fa001750e64defa1b15ca974e2f9edba9d546 Mon Sep 17 00:00:00 2001 From: dengfangyuan Date: Wed, 29 Jul 2026 14:30:27 +0800 Subject: [PATCH 4/4] [feat][paimon] paimon-rust reader s3/oss key remap Remap common OSS/S3 hadoop-style credential keys (fs.oss.*, fs.s3a.*) to AWS_ACCESS_KEY / AWS_SECRET_KEY / AWS_ENDPOINT / ... in _build_options so OSS/S3-backed paimon tables can authenticate through the paimon-rust FileIO. --- be/src/format/table/paimon_rust_reader.cpp | 25 ++++++++++++++++++++++ 1 file changed, 25 insertions(+) diff --git a/be/src/format/table/paimon_rust_reader.cpp b/be/src/format/table/paimon_rust_reader.cpp index ecaab5bda367d9..9e9896ba3ee020 100644 --- a/be/src/format/table/paimon_rust_reader.cpp +++ b/be/src/format/table/paimon_rust_reader.cpp @@ -562,6 +562,31 @@ std::map PaimonRustReader::_build_options() const { options[kv.first] = kv.second; } } + + auto copy_if_missing = [&](const char* from_key, const char* to_key) { + if (options.find(to_key) != options.end()) { + return; + } + auto it = options.find(from_key); + if (it != options.end() && !it->second.empty()) { + options[to_key] = it->second; + } + }; + + // Map common OSS/S3 Hadoop configs to Doris/paimon-native S3 property keys + // that the paimon-rust FileIO recognizes. + copy_if_missing("fs.oss.accessKeyId", "AWS_ACCESS_KEY"); + copy_if_missing("fs.oss.accessKeySecret", "AWS_SECRET_KEY"); + copy_if_missing("fs.oss.sessionToken", "AWS_TOKEN"); + copy_if_missing("fs.oss.endpoint", "AWS_ENDPOINT"); + copy_if_missing("fs.oss.region", "AWS_REGION"); + copy_if_missing("fs.s3a.access.key", "AWS_ACCESS_KEY"); + copy_if_missing("fs.s3a.secret.key", "AWS_SECRET_KEY"); + copy_if_missing("fs.s3a.session.token", "AWS_TOKEN"); + copy_if_missing("fs.s3a.endpoint", "AWS_ENDPOINT"); + copy_if_missing("fs.s3a.region", "AWS_REGION"); + copy_if_missing("fs.s3a.path.style.access", "use_path_style"); + return options; }