github-actions[bot] commented on code in PR #66227: URL: https://github.com/apache/doris/pull/66227#discussion_r4038207476
########## be/src/format_v2/table/paimon_rust_predicate_converter.cpp: ########## @@ -0,0 +1,760 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +#include "format_v2/table/paimon_rust_predicate_converter.h" + +#include <algorithm> +#include <cctype> +#include <memory> +#include <utility> + +#include "common/logging.h" +#include "core/column/column_const.h" +#include "core/column/column_nullable.h" +#include "core/data_type/data_type.h" +#include "core/data_type/data_type_nullable.h" +#include "core/field.h" +#include "core/types.h" +#include "core/value/decimalv2_value.h" +#include "core/value/timestamptz_value.h" +#include "core/value/vdatetime_value.h" +#include "exprs/vcompound_pred.h" +#include "exprs/vdirect_in_predicate.h" +#include "exprs/vectorized_fn_call.h" +#include "exprs/vexpr.h" +#include "exprs/vin_predicate.h" +#include "exprs/vliteral.h" +#include "exprs/vslot_ref.h" + +namespace doris { + +namespace { +// paimon_datum tags (see paimon.h / bindings/c/src/table.rs::datum_from_c). +constexpr int32_t kTagBool = 0; +constexpr int32_t kTagTinyInt = 1; +constexpr int32_t kTagSmallInt = 2; +constexpr int32_t kTagInt = 3; +constexpr int32_t kTagLong = 4; +constexpr int32_t kTagDouble = 6; +constexpr int32_t kTagString = 7; +constexpr int32_t kTagDate = 8; +constexpr int32_t kTagTimestamp = 10; +constexpr int32_t kTagDecimal = 12; +constexpr int32_t kTagBytes = 13; + +// paimon decimal precision ceiling (paimon::Decimal::MAX_PRECISION). +constexpr int32_t kPaimonDecimalMaxPrecision = 38; + +// RAII for an owned paimon_predicate*. and/or/not consume their inputs, so we +// release() before handing pointers to them. +struct predicate_deleter { + void operator()(paimon_predicate* p) const { + if (p) { + paimon_predicate_free(p); + } + } +}; +using predicate_ptr = std::unique_ptr<paimon_predicate, predicate_deleter>; + +// RAII for an owned paimon_error*. +struct error_deleter { + void operator()(paimon_error* p) const { + if (p) { + paimon_error_free(p); + } + } +}; +using error_ptr = std::unique_ptr<paimon_error, error_deleter>; + +// Render a paimon_error into a string. Takes ownership of `err` via RAII so it +// is freed on every return path. Safe to call with nullptr. +std::string consume_predicate_error(paimon_error* err) { + error_ptr owned(err); + if (!owned) { + return "unknown error"; + } + std::string msg; + if (owned->message.data != nullptr && owned->message.len > 0) { + msg.assign(reinterpret_cast<const char*>(owned->message.data), owned->message.len); + } + return "code=" + std::to_string(owned->code) + ", msg=" + msg; +} +} // namespace + +PaimonRustPredicateConverter::PaimonRustPredicateConverter( + const std::vector<std::string>& column_names, const std::vector<DataTypePtr>& column_types, + const paimon_table* table) + : _table(table) { + DORIS_CHECK(column_names.size() == column_types.size()); + _columns_by_name.reserve(column_names.size()); + for (size_t i = 0; i < column_names.size(); ++i) { + _columns_by_name.emplace(_normalize_name(column_names[i]), + std::make_pair(column_names[i], column_types[i])); + } + // Paimon TIMESTAMP (wall clock) is stored as epoch-millis-of-the-wall-time + // and the DateTimeV2 serde decodes timezone-naive arrow values in UTC, so + // timestamp literals convert wall->epoch in UTC. utc_time_zone() needs no + // tzdata lookup, so the conversion cannot silently fall back to a + // machine-local zone. + _utc_tz = cctz::utc_time_zone(); +} + +paimon_predicate* PaimonRustPredicateConverter::build(const VExprContextSPtrs& conjuncts) { + if (_table == nullptr) { + return nullptr; + } + predicate_ptr result; + for (const auto& conjunct : conjuncts) { + if (!conjunct || !conjunct->root()) { + continue; + } + auto root = conjunct->root(); + if (root->is_rf_wrapper()) { + if (auto impl = root->get_impl()) { + 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; + } + + // Casts are not unwrapped anywhere (predicate root included): a cast node + // fails every dispatch below and the conjunct stays in the Doris residual, + // mirroring the FE converter, which keeps casted expressions unconverted. + if (auto* direct_in = dynamic_cast<VDirectInPredicate*>(expr.get())) { + VExprSPtr in_expr; + if (direct_in->get_slot_in_expr(in_expr)) { + return _convert_in(in_expr); + } + return nullptr; + } + + if (dynamic_cast<VInPredicate*>(expr.get()) != nullptr) { + return _convert_in(expr); + } + + switch (expr->op()) { + case TExprOpcode::COMPOUND_AND: + case TExprOpcode::COMPOUND_OR: + return _convert_compound(expr); + case TExprOpcode::COMPOUND_NOT: + return nullptr; + case TExprOpcode::EQ: + case TExprOpcode::EQ_FOR_NULL: + case TExprOpcode::NE: + case TExprOpcode::GE: + case TExprOpcode::GT: + case TExprOpcode::LE: + case TExprOpcode::LT: + return _convert_binary(expr); + default: + break; + } + + if (auto* fn = dynamic_cast<VectorizedFnCall*>(expr.get())) { + auto fn_name = _normalize_name(fn->function_name()); + if (fn_name == "is_null_pred" || fn_name == "is_not_null_pred") { + return _convert_is_null(expr, fn_name); + } + if (fn_name == "like") { + return _convert_like_prefix(expr); + } + } + + return nullptr; +} + +paimon_predicate* PaimonRustPredicateConverter::_convert_compound(const VExprSPtr& expr) { + if (!expr || expr->get_num_children() != 2) { + return nullptr; + } + predicate_ptr left(_convert_expr(expr->get_child(0))); + if (!left) { + return nullptr; + } + predicate_ptr right(_convert_expr(expr->get_child(1))); + if (!right) { + return nullptr; + } + + if (expr->op() == TExprOpcode::COMPOUND_AND) { + return paimon_predicate_and(left.release(), right.release()); + } + if (expr->op() == TExprOpcode::COMPOUND_OR) { + return paimon_predicate_or(left.release(), right.release()); + } + return nullptr; +} + +paimon_predicate* PaimonRustPredicateConverter::_convert_in(const VExprSPtr& expr) { + auto* in_pred = dynamic_cast<VInPredicate*>(expr.get()); + if (!in_pred || expr->get_num_children() < 2) { + return nullptr; + } + auto field_meta = _resolve_field(expr->get_child(0)); + if (!field_meta) { + return nullptr; + } + + const auto num_values = expr->get_num_children() - 1; + // Reserve up front so the backing strings never reallocate: each datum's + // str_data points into storages[i], which must stay stable. + std::vector<std::string> storages; + std::vector<paimon_datum> datums; + storages.reserve(num_values); + datums.reserve(num_values); + for (uint16_t i = 1; i < expr->get_num_children(); ++i) { + auto holder = _convert_literal(expr->get_child(i), field_meta->type); Review Comment: [P1] Evaluate or reject casts inside IN lists `_convert_in()` accepts a CAST child and `_convert_literal()` reads its raw child without executing the cast. With `debug_skip_fold_constant=true`, a DECIMAL(10,1) predicate `amount IN (CAST(1.24 AS DECIMAL(10,1)))` reaches this path: Doris compares against 1.2, while Rust filters for 1.24 and can permanently remove the 1.2 row. FE deliberately rejects non-literal IN children. Please match that fallback or convert the evaluated cast result, with IN/NOT IN differential tests. ########## fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/source/PaimonScanNode.java: ########## @@ -411,10 +412,64 @@ 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) FileScannerV2: the V1 FileScanner + // explicitly rejects PAIMON_RUST, so with enable_file_scanner_v2 disabled + // the split falls back to JNI instead of encoding a rust request that the + // selected scanner cannot consume, and (b) a FileStoreTable: BE opens the + // table via paimon_table_from_schema_json, which needs the resolved + // TableSchema that only FileStoreTable exposes via schema(). If the table + // is not a FileStoreTable (e.g. a sys table backed by DataSplit), we cannot + // ship a schema JSON, so fall back to CPP / JNI rather than sending an + // incomplete PAIMON_RUST request that BE would reject. + // + // Serialize the same effective table that planning and the JNI reader use. + // Relation options such as t@options('read.batch-size'='1') are applied by + // getProcessedTable() (doInitialize caches it in processedTable), and the + // rust reader derives its read batch size from the schema options — the raw + // cached table would silently drop the override. Copies, delegates and + // fallback wrappers of getProcessedTable() are still FileStoreTable, so the + // instanceof gate keeps its semantics. + Table paimonTable = processedTable; + boolean canUseRust = sessionVariable.isEnablePaimonRustReader() Review Comment: [P1] Keep incremental scans off the ordinary Rust path `canUseRust` does not exclude `scanParams.incrementalRead()`, but this wire path carries only an ordinary DataSplit and invokes `TableRead::to_arrow`. Paimon 1.4 marks delta/changelog splits as streaming, which the pinned Rust deserializer rejects, while diff requires a separate IncrementalPlan; ordinary PK reads can also merge versions instead of returning changes. Please fall back to JNI for incremental scans until the C ABI transports the mode/plan, and cover delta, changelog, and diff. ########## be/src/format_v2/table/paimon_rust_table_reader.cpp: ########## @@ -0,0 +1,785 @@ +// 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_v2/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); + // Materialize TIMESTAMP_LTZ in the session timezone — the same + // convention as the JNI reader (PaimonJniScanner reads time_zone from + // its scan params) and lance_reader. Timezone-naive (paimon TIMESTAMP) + // arrow values are decoded in UTC by the DateTimeV2 serde regardless + // of _ctz, so NTZ wall-clock semantics are preserved. + DORIS_CHECK(_runtime_state != nullptr); + _ctz = _runtime_state->timezone_obj(); + if (_scanner_profile != nullptr) { + file_scan_profile::ensure_hierarchy(_scanner_profile); + _rust_total_time = ADD_CHILD_TIMER(_scanner_profile, "PaimonRustReader", + file_scan_profile::TABLE_READER); + _rust_open_split_time = ADD_CHILD_TIMER(_scanner_profile, "OpenSplitTime", + "PaimonRustReader"); + _rust_read_batch_time = + ADD_CHILD_TIMER(_scanner_profile, "ReadBatchTime", "PaimonRustReader"); + _rust_arrow_to_block_time = + ADD_CHILD_TIMER(_scanner_profile, "ArrowToBlockTime", "PaimonRustReader"); + } + // Projected column name -> fixed output position, registered with both the exact and + // the lower-case spelling so mixed-case Rust schema output still resolves (v1 + // semantics: exact match first, lower-case fallback on lookup). + _output_name_to_idx.reserve(_projected_columns.size() * 2); + for (size_t idx = 0; idx < _projected_columns.size(); ++idx) { + _output_name_to_idx.emplace(_projected_columns[idx].name, idx); + _output_name_to_idx.emplace(to_lower(_projected_columns[idx].name), idx); + } + } + return Status::OK(); +} + +Status PaimonRustTableReader::prepare_split(const format::SplitReadOptions& options) { + // EOF belongs to the previous split. Keep it set after closing that split so repeated reads + // are idempotent, and clear it only when a new split is explicitly prepared. + _close_split_reader(); + _split_eof = false; + _current_range = options.current_range; + RETURN_IF_ERROR(format::TableReader::prepare_split(options)); + if (current_split_pruned()) { + return Status::OK(); + } + if (_is_table_level_count_active()) { + // No rust pipeline is opened; get_block emits the synthetic count rows. + return Status::OK(); + } + RETURN_IF_ERROR(_validate_rust_split(options.current_range)); + { + SCOPED_TIMER(_profile.total_timer); + SCOPED_TIMER(_profile.prepare_split_timer); + SCOPED_TIMER(_rust_open_split_time); + RETURN_IF_ERROR(_open_split_reader(options.current_range)); + } + return Status::OK(); +} + +Status PaimonRustTableReader::get_block(Block* block, bool* eos) { + SCOPED_TIMER(_profile.total_timer); + SCOPED_TIMER(_profile.exec_timer); + SCOPED_TIMER(_rust_total_time); + DORIS_CHECK(block != nullptr); + DORIS_CHECK(eos != nullptr); + DORIS_CHECK(block->columns() == _projected_columns.size()); + block->clear_column_data(_projected_columns.size()); + *eos = false; + + if (_is_table_level_count_active()) { + return _read_table_level_count(block, eos); + } + + // num_splits == 0 yields an empty (but valid) stream: report EOF. + if (_split_eof) { + *eos = true; + return Status::OK(); + } + if (!_handles || !_handles->reader) { + return Status::InternalError("paimon-rust reader is not initialized"); + } + + while (true) { + // Mirror the base TableReader cancellation contract so a cancelled query does not + // drain the whole split. + if (_io_ctx != nullptr && _io_ctx->should_stop) { + _split_eof = true; + _close_split_reader(); + *eos = true; + return Status::OK(); + } + + paimon_result_next_batch next; + { + SCOPED_TIMER(_rust_read_batch_time); + 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) { + _split_eof = true; + _close_split_reader(); + *eos = true; + return Status::OK(); + } + + // RAII: the batch's Arrow release callbacks + container free run when + // `batch` leaves this scope, including on any early return. + ArrowBatch batch(next.batch); + + auto* c_array = batch.array(); + auto* c_schema = batch.schema(); + arrow::Result<std::shared_ptr<arrow::RecordBatch>> 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 rows = static_cast<size_t>(record_batch->num_rows()); + if (rows == 0) { + // Skip empty batches and keep draining the stream. + continue; + } + RETURN_IF_ERROR(_fill_block_from_record_batch(record_batch, block, rows)); + _record_scan_rows(rows); + *eos = false; + return Status::OK(); + } +} + +Status PaimonRustTableReader::abort_split() { + { + SCOPED_TIMER(_profile.total_timer); + SCOPED_TIMER(_profile.close_timer); + _close_split_reader(); + _split_eof = false; + } + return format::TableReader::abort_split(); +} + +Status PaimonRustTableReader::close() { + { + SCOPED_TIMER(_profile.total_timer); + SCOPED_TIMER(_profile.close_timer); + _close_split_reader(); + _close_table(); + } + return format::TableReader::close(); +} + +Status PaimonRustTableReader::_validate_rust_split(const TFileRangeDesc& range) const { + if (!range.__isset.table_format_params || !range.table_format_params.__isset.paimon_params) { + return Status::InternalError( + "missing paimon_params for paimon rust reader, possibly caused by FE/BE protocol " + "mismatch"); + } + const auto& params = range.table_format_params.paimon_params; + if (!params.__isset.paimon_split || params.paimon_split.empty()) { + return Status::InternalError( + "missing paimon_split for paimon rust reader, possibly caused by FE/BE protocol " + "mismatch"); + } + if (params.__isset.reader_type && params.reader_type != TPaimonReaderType::PAIMON_RUST) { + return Status::InternalError( + "invalid reader_type for paimon rust reader, possibly caused by FE/BE protocol " + "mismatch"); + } + if (!_resolve_table_path(range).has_value()) { + return Status::InternalError( + "paimon-rust missing paimon_table; cannot resolve paimon table location"); + } + if (!_resolve_db_name(range).has_value()) { + return Status::InternalError( + "paimon-rust missing db_name; cannot open paimon table via schema json"); + } + if (!_resolve_table_name(range).has_value()) { + return Status::InternalError( + "paimon-rust missing table_name; cannot open paimon table via schema json"); + } + if (!_resolve_table_schema_json(range).has_value()) { + return Status::InternalError( + "paimon-rust missing paimon_table_schema_json; cannot open paimon table via " + "schema json"); + } + return Status::OK(); +} + +Status PaimonRustTableReader::_open_split_reader(const TFileRangeDesc& range) { + // 1. Decode the FE-planned split first so we fail fast (and without any + // filesystem IO) when it is missing or malformed. + std::string split_bytes; + RETURN_IF_ERROR(_decode_split_bytes(&split_bytes)); + + // 2. Resolve identifier + table_path + FE-supplied TableSchema JSON. + auto table_path = _resolve_table_path(range).value(); + auto db_name = _resolve_db_name(range).value(); + auto table_name = _resolve_table_name(range).value(); + auto schema_json = _resolve_table_schema_json(range).value(); + auto branch_opt = _resolve_branch(range); + + // 3. Assemble storage options: FE-supplied paimon options + hadoop_conf + + // OSS/S3 → AWS_* translations. These feed FileIO only (per + // paimon_table_from_schema_json contract); they are NOT merged into the + // supplied table schema. + auto options = _build_options(); + + auto opened_table_key = std::make_tuple(table_path, schema_json, db_name, table_name, + branch_opt, options); + if (!_handles || !_handles->table || _opened_table_key != opened_table_key) { + // A paimon scan reads one table, so the handle is opened at most once per + // distinct identity (e.g. re-created after a close); splits of the same + // table reuse it and only rebuild the read pipeline below. + _close_table(); + _handles = std::make_unique<PaimonHandles>(); + + std::vector<paimon_option> 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 schema json: db=" << db_name + << " table=" << table_name << " path=" << table_path + << " branch=" << (branch_opt.has_value() ? branch_opt.value() : "main") + << " storage_options=[" << format_options(options) << "]"; + + // Build the table directly from the FE-supplied schema JSON. The Rust + // side rejects null / empty branch, so we default to paimon's canonical + // "main" sentinel when FE did not set paimon_branch (i.e. the table is + // on the main branch — matches upstream Identifier.DEFAULT_MAIN_BRANCH). + const std::string& branch_str = branch_opt.has_value() ? branch_opt.value() : "main"; + paimon_result_get_table tbl_res = paimon_table_from_schema_json( + table_path.c_str(), schema_json.c_str(), db_name.c_str(), table_name.c_str(), + branch_str.c_str(), c_options.empty() ? nullptr : c_options.data(), + c_options.size()); + if (tbl_res.error != nullptr) { + return Status::InternalError( + "paimon-rust table_from_schema_json failed: db={} table={} err={}", db_name, + table_name, consume_error(tbl_res.error)); + } + _handles->table.reset(tbl_res.table); + _opened_table_key = std::move(opened_table_key); + } + + // 4. 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)); + } + + // Partition keys are excluded: they are materialized from split metadata + // (see _fill_non_arrow_columns), and paimon-rust does not emit them. + auto read_columns = _build_read_columns(); + std::vector<const char*> 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 scanner conjuncts into a paimon-rust filter and apply it. + RETURN_IF_ERROR(_apply_predicate()); + + // 5. 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<const uint8_t*>(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) { + _split_eof = true; + return Status::OK(); + } + + // 6. 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(); +} + +void PaimonRustTableReader::_close_split_reader() { + if (!_handles) { + return; + } + // Reverse of the declaration order in PaimonHandles. + _handles->reader.reset(); + _handles->table_read.reset(); + _handles->plan.reset(); + _handles->read_builder.reset(); +} + +void PaimonRustTableReader::_close_table() { + if (!_handles) { + return; + } + _close_split_reader(); + _handles->table.reset(); + _opened_table_key.reset(); +} + +Status PaimonRustTableReader::_apply_predicate() { + if (_conjuncts.empty() || !_handles || !_handles->table || !_handles->read_builder) { + return Status::OK(); + } + LOG(INFO) << "paimon-rust predicate pushdown: " << _conjuncts.size() << " conjunct(s) input"; + // The conjunct VSlotRefs carry table global indices (positions), so the v2 + // converter mode resolves fields by the projected column names; partition + // keys are excluded because the rust reader does not read them. + std::vector<std::string> names; + std::vector<DataTypePtr> types; + names.reserve(_projected_columns.size()); + types.reserve(_projected_columns.size()); + for (const auto& col : _projected_columns) { + if (col.is_partition_key) { + continue; + } + names.push_back(col.name); + types.push_back(col.type); + } + PaimonRustPredicateConverter converter(names, types, _handles->table.get()); + paimon_predicate* predicate = converter.build(_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 PaimonRustTableReader::_fill_block_from_record_batch( + const std::shared_ptr<arrow::RecordBatch>& batch, Block* block, size_t rows) { + SCOPED_TIMER(_rust_arrow_to_block_time); + DORIS_CHECK(batch != nullptr); + DORIS_CHECK(block != nullptr); + std::unordered_set<size_t> materialized_indices; + materialized_indices.reserve(_projected_columns.size()); + { + auto columns_guard = block->mutate_columns_scoped(); + auto& columns = columns_guard.mutable_columns(); + for (int c = 0; c < batch->num_columns(); ++c) { + const auto& field = batch->schema()->field(c); + if (field->name() == VALUE_KIND_FIELD) { + continue; + } + // Projected 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 tolerate + // mixed-case Rust output by folding here as well. + auto it = _output_name_to_idx.find(field->name()); + if (it == _output_name_to_idx.end()) { + it = _output_name_to_idx.find(to_lower(field->name())); + } + if (it == _output_name_to_idx.end()) { + // Skip columns that are not in the block (e.g. columns dropped by + // slot pruning). + continue; + } + const auto output_idx = it->second; + if (!materialized_indices.emplace(output_idx).second) { + return Status::InternalError("paimon-rust returned duplicate column '{}'", + field->name()); + } + try { + RETURN_IF_ERROR(columns_guard.get_datatype_by_position(output_idx) + ->get_serde() + ->read_column_from_arrow(*columns[output_idx], + batch->column(c).get(), 0, rows, + _ctz)); + } catch (Exception& e) { + return Status::InternalError("Failed to convert from arrow to block: {}", + e.what()); + } + } + } + // Partition columns and other projected columns absent from the arrow batch + // are back-filled from split metadata / defaults. + return _fill_non_arrow_columns(block, rows, materialized_indices); +} + +Status PaimonRustTableReader::_fill_non_arrow_columns( + Block* block, size_t rows, const std::unordered_set<size_t>& materialized_indices) { + for (size_t idx = 0; idx < _projected_columns.size(); ++idx) { + if (materialized_indices.count(idx) != 0) { + continue; + } + const auto& column = _projected_columns[idx]; + VExprContextSPtr constant_expr; + if (const Field* value = find_partition_value(column, _partition_values); + column.is_partition_key && value != nullptr) { + // Partition values are split constants (same materialization the + // TableColumnMapper builds for native readers). + constant_expr = VExprContext::create_shared( + VLiteral::create_shared(column.type, *value)); + } else if (column.default_expr != nullptr) { + constant_expr = column.default_expr; + } else { + // The column is genuinely absent from the arrow batch. Schema + // evolution is handled by paimon-rust itself, so reaching here means + // an unexpected schema drift: fill defaults so the scan remains + // well-defined instead of failing the query. + LOG(WARNING) << "paimon-rust did not return projected column '" << column.name + << "'; filling with defaults"; + auto data = column.type->create_column(); + data->insert_many_defaults(rows); + block->replace_by_position(idx, std::move(data)); + continue; + } + ColumnPtr constant_column; + RETURN_IF_ERROR(_materialize_constant_column(constant_expr, column.type, column.name, rows, + &constant_column)); + block->replace_by_position(idx, std::move(constant_column)); + } + return Status::OK(); +} + +Status PaimonRustTableReader::_materialize_constant_column(const VExprContextSPtr& expr, + const DataTypePtr& type, + const std::string& name, size_t rows, + ColumnPtr* column) { + DORIS_CHECK(expr != nullptr); + DORIS_CHECK(column != nullptr); + RowDescriptor row_desc; + RETURN_IF_ERROR(expr->prepare(_runtime_state, row_desc)); + RETURN_IF_ERROR(expr->open(_runtime_state)); + // Constants evaluate per input row, so a rows-sized synthetic block yields a + // rows-sized result for both plain literals and default expressions. + Block eval_block; + eval_block.insert({type->create_column_const_with_default_value(rows), type, name}); + int result_column_id = -1; + RETURN_IF_ERROR(expr->execute(&eval_block, &result_column_id)); + DORIS_CHECK(result_column_id >= 0); + ColumnPtr result_column = eval_block.get_by_position(result_column_id).column; + if (result_column->size() == 1 && rows > 1) { + result_column = ColumnConst::create(std::move(result_column), rows); + } + *column = std::move(result_column); + return Status::OK(); +} + +Status PaimonRustTableReader::_decode_split_bytes(std::string* out) const { + if (!_current_range.__isset.table_format_params || + !_current_range.table_format_params.__isset.paimon_params || + !_current_range.table_format_params.paimon_params.__isset.paimon_split) { + return Status::InternalError("paimon-rust missing paimon_split in scan range"); + } + const auto& encoded_split = _current_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<std::string> PaimonRustTableReader::_resolve_table_path( + const TFileRangeDesc& range) 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<std::string> PaimonRustTableReader::_resolve_db_name( + const TFileRangeDesc& range) 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<std::string> PaimonRustTableReader::_resolve_table_name( + const TFileRangeDesc& range) 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<std::string> PaimonRustTableReader::_resolve_table_schema_json( + const TFileRangeDesc& range) const { + if (range.__isset.table_format_params && range.table_format_params.__isset.paimon_params && + range.table_format_params.paimon_params.__isset.paimon_table_schema_json && + !range.table_format_params.paimon_params.paimon_table_schema_json.empty()) { + return range.table_format_params.paimon_params.paimon_table_schema_json; + } + return std::nullopt; +} + +std::optional<std::string> PaimonRustTableReader::_resolve_branch( + const TFileRangeDesc& range) const { + // FE only sets paimon_branch when the branch is not `main` (matches + // upstream paimon commit 742da63: null-if-DEFAULT_MAIN_BRANCH). Unset here + // means main-branch semantics. + if (range.__isset.table_format_params && range.table_format_params.__isset.paimon_params && + range.table_format_params.paimon_params.__isset.paimon_branch && + !range.table_format_params.paimon_params.paimon_branch.empty()) { + return range.table_format_params.paimon_params.paimon_branch; + } + return std::nullopt; +} + +std::vector<std::string> PaimonRustTableReader::_build_read_columns() const { + std::vector<std::string> columns; + columns.reserve(_projected_columns.size()); + for (const auto& column : _projected_columns) { + if (column.is_partition_key) { + continue; + } + columns.emplace_back(column.name); + } + return columns; +} + +std::map<std::string, std::string> PaimonRustTableReader::_build_options() const { + std::map<std::string, std::string> options; + if (_scan_params && _scan_params->__isset.paimon_options && + !_scan_params->paimon_options.empty()) { + options.insert(_scan_params->paimon_options.begin(), _scan_params->paimon_options.end()); + } else if (_current_range.__isset.table_format_params && + _current_range.table_format_params.__isset.paimon_params && + _current_range.table_format_params.paimon_params.__isset.paimon_options) { + options.insert( + _current_range.table_format_params.paimon_params.paimon_options.begin(), + _current_range.table_format_params.paimon_params.paimon_options.end()); + } + + if (_scan_params && _scan_params->__isset.properties && !_scan_params->properties.empty()) { + for (const auto& kv : _scan_params->properties) { + options[kv.first] = kv.second; + } + } else if (_current_range.__isset.table_format_params && + _current_range.table_format_params.__isset.paimon_params && + _current_range.table_format_params.paimon_params.__isset.hadoop_conf) { + for (const auto& kv : + _current_range.table_format_params.paimon_params.hadoop_conf) { + 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; + } + }; + + // The pinned paimon-rust FileIO reads paimon-java's `s3.*` option family + // (io/storage_s3.rs normalizes the `fs.s3a.`/`s3a.`/`s3.` prefixes and the + // `s3.access.key`/`s3.path.style.access` aliases): s3.access-key, + // s3.secret-key, s3.session.token, s3.endpoint, s3.region and + // s3.path-style-access. `fs.s3a.*` keys therefore pass through natively, + // but the FE's storage-properties channel delivers the vended S3 config + // under the AWS_* / use_path_style aliases, which the crate does not read, + // and the OSS configs use their own fs.oss.* names — so remap both to the + // s3.* family. Without this the rust S3 FileIO builds with an empty region + // ("ConfigInvalid ... region is missing") and never connects. + copy_if_missing("AWS_ACCESS_KEY", "s3.access-key"); + copy_if_missing("AWS_SECRET_KEY", "s3.secret-key"); + copy_if_missing("AWS_TOKEN", "s3.session.token"); + copy_if_missing("AWS_ENDPOINT", "s3.endpoint"); + copy_if_missing("AWS_REGION", "s3.region"); + copy_if_missing("use_path_style", "s3.path-style-access"); + copy_if_missing("fs.oss.accessKeyId", "s3.access-key"); Review Comment: [P1] Supply OSS keys to the OSS FileIO For a normal `oss://` catalog, FE's `OSSProperties` backend map contains the AWS aliases, and these copies create only `s3.*` keys. The pinned Rust storage dispatcher selects the OSS parser from the URI scheme, however, and that parser requires `fs.oss.endpoint`, `fs.oss.accessKeyId`, and `fs.oss.accessKeySecret` (plus `fs.oss.securityToken` for STS). Thus valid OSS catalogs fail to open when Rust is enabled. Map by scheme or gate unverified schemes, and add a production-map OSS open test. ########## be/src/format_v2/table/paimon_rust_predicate_converter.cpp: ########## @@ -0,0 +1,760 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +#include "format_v2/table/paimon_rust_predicate_converter.h" + +#include <algorithm> +#include <cctype> +#include <memory> +#include <utility> + +#include "common/logging.h" +#include "core/column/column_const.h" +#include "core/column/column_nullable.h" +#include "core/data_type/data_type.h" +#include "core/data_type/data_type_nullable.h" +#include "core/field.h" +#include "core/types.h" +#include "core/value/decimalv2_value.h" +#include "core/value/timestamptz_value.h" +#include "core/value/vdatetime_value.h" +#include "exprs/vcompound_pred.h" +#include "exprs/vdirect_in_predicate.h" +#include "exprs/vectorized_fn_call.h" +#include "exprs/vexpr.h" +#include "exprs/vin_predicate.h" +#include "exprs/vliteral.h" +#include "exprs/vslot_ref.h" + +namespace doris { + +namespace { +// paimon_datum tags (see paimon.h / bindings/c/src/table.rs::datum_from_c). +constexpr int32_t kTagBool = 0; +constexpr int32_t kTagTinyInt = 1; +constexpr int32_t kTagSmallInt = 2; +constexpr int32_t kTagInt = 3; +constexpr int32_t kTagLong = 4; +constexpr int32_t kTagDouble = 6; +constexpr int32_t kTagString = 7; +constexpr int32_t kTagDate = 8; +constexpr int32_t kTagTimestamp = 10; +constexpr int32_t kTagDecimal = 12; +constexpr int32_t kTagBytes = 13; + +// paimon decimal precision ceiling (paimon::Decimal::MAX_PRECISION). +constexpr int32_t kPaimonDecimalMaxPrecision = 38; + +// RAII for an owned paimon_predicate*. and/or/not consume their inputs, so we +// release() before handing pointers to them. +struct predicate_deleter { + void operator()(paimon_predicate* p) const { + if (p) { + paimon_predicate_free(p); + } + } +}; +using predicate_ptr = std::unique_ptr<paimon_predicate, predicate_deleter>; + +// RAII for an owned paimon_error*. +struct error_deleter { + void operator()(paimon_error* p) const { + if (p) { + paimon_error_free(p); + } + } +}; +using error_ptr = std::unique_ptr<paimon_error, error_deleter>; + +// Render a paimon_error into a string. Takes ownership of `err` via RAII so it +// is freed on every return path. Safe to call with nullptr. +std::string consume_predicate_error(paimon_error* err) { + error_ptr owned(err); + if (!owned) { + return "unknown error"; + } + std::string msg; + if (owned->message.data != nullptr && owned->message.len > 0) { + msg.assign(reinterpret_cast<const char*>(owned->message.data), owned->message.len); + } + return "code=" + std::to_string(owned->code) + ", msg=" + msg; +} +} // namespace + +PaimonRustPredicateConverter::PaimonRustPredicateConverter( + const std::vector<std::string>& column_names, const std::vector<DataTypePtr>& column_types, + const paimon_table* table) + : _table(table) { + DORIS_CHECK(column_names.size() == column_types.size()); + _columns_by_name.reserve(column_names.size()); + for (size_t i = 0; i < column_names.size(); ++i) { + _columns_by_name.emplace(_normalize_name(column_names[i]), + std::make_pair(column_names[i], column_types[i])); + } + // Paimon TIMESTAMP (wall clock) is stored as epoch-millis-of-the-wall-time + // and the DateTimeV2 serde decodes timezone-naive arrow values in UTC, so + // timestamp literals convert wall->epoch in UTC. utc_time_zone() needs no + // tzdata lookup, so the conversion cannot silently fall back to a + // machine-local zone. + _utc_tz = cctz::utc_time_zone(); +} + +paimon_predicate* PaimonRustPredicateConverter::build(const VExprContextSPtrs& conjuncts) { + if (_table == nullptr) { + return nullptr; + } + predicate_ptr result; + for (const auto& conjunct : conjuncts) { Review Comment: [P1] Stop pushdown at unsafe conjuncts `build()` skips an unconvertible conjunct and continues pushing later ones. If an error-preserving predicate such as `assert_true(...)` precedes an arrived IN runtime filter, Rust applies the later filter first and can discard the row that should make the earlier residual raise. The common partition-pruning path and Parquet schedule already stop at `is_safe_to_execute_on_selected_rows()` for this reason. Please preserve only the safe prefix here as well, and cover an unsafe original conjunct followed by an arrived runtime filter. ########## thirdparty/build-thirdparty.sh: ########## @@ -2169,6 +2169,133 @@ build_lance_c() { fi } +# 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 + + # paimon-vindex-core 0.4.0 uses the unstable `stdarch_neon_f16` intrinsics + # (vcvt_f32_f16 / vreinterpret_f16_u16) in its aarch64 NEON fast path, which + # do not compile on stable Rust. On aarch64/arm64, replace the registry + # crate with a patched path source so the build succeeds. The patch swaps + # the unstable f16->f32 NEON convert for a stable scalar conversion; the + # rest of the NEON accumulation is left untouched. x86_64 and other arches + # are unaffected and keep using the pristine registry crate. + if [[ "$(uname -m)" == "aarch64" || "$(uname -m)" == "arm64" ]]; then + local vindex_override="${PWD}/.doris-vindex-override" + local vindex_crate + vindex_crate="$(find "${CARGO_HOME:-$HOME/.cargo}/registry/cache" \ Review Comment: [P1] Verify the aarch64 crate before making it a path dependency This selects the first ambient cache file named `paimon-vindex-core-0.4.0.crate`, extracts it without checking the Cargo.lock checksum or registry identity, and then changes it to a path dependency. `cargo build --locked` no longer authenticates those bytes, so a stale/poisoned same-named cache can enter `libpaimon_c.a` and identical Doris sources need not produce identical artifacts. Please source this crate from a Doris-checksummed archive/vendor tree, or verify the exact locked checksum before extraction. ########## be/src/format_v2/table/paimon_rust_table_reader.h: ########## @@ -0,0 +1,164 @@ +// 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 <cstddef> +#include <map> +#include <memory> +#include <optional> +#include <string> +#include <tuple> +#include <unordered_map> +#include <unordered_set> +#include <vector> + +#include "cctz/time_zone.h" +#include "common/status.h" +#include "format_v2/table_reader.h" +#include "runtime/runtime_profile.h" + +namespace arrow { +class RecordBatch; +} + +namespace doris::format::paimon { + +// Reads one FE-planned Paimon DataSplit per split through the paimon-rust C +// bindings (libpaimon_c), using the schema-json pipeline: +// +// paimon_table_from_schema_json(table_path, schema_json, db, table, branch, options) +// -> read_builder (case-insensitive, projection, filter) +// -> paimon_plan_from_split_bytes(base64-decoded FE split) +// -> paimon_read_builder_new_read -> paimon_table_read_to_arrow +// -> per-batch paimon_record_batch_reader_next (Arrow C Data Interface). +// +// Leaf reader inside PaimonHybridReader: a DataSplit is a logical multi-file +// split, not a file range, so this reader does not go through +// FileReader/TableColumnMapper and fills the table-schema output block directly +// (same shape as LanceTableReader). Partition columns that paimon-rust does not +// emit are materialized from the partition values captured by the TableReader +// base class; columns with default expressions are materialized from those +// expressions. +class PaimonRustTableReader final : public format::TableReader { +public: + // Out-of-line (in the .cpp) so TUs that construct or destroy the reader do not + // need the complete PaimonHandles pimpl type. + PaimonRustTableReader(); + ~PaimonRustTableReader() override; + + Status init(format::TableReadOptions&& options) override; + Status prepare_split(const format::SplitReadOptions& options) override; + Status get_block(Block* block, bool* eos) override; + Status abort_split() override; + Status close() override; + // paimon_table_read_to_arrow has no batch-size control and the arrow reader Review Comment: [P1] Make the Rust stream honor adaptive batch limits Scanner V2 enables adaptive batching for this FORMAT_JNI range and forwards its initial probe and later row caps, but this leaf only stores the value. The pinned Rust readers always allocate using the table schema's `read.batch-size`, so a wide/nested table can materialize a full Rust/Arrow batch despite a much smaller predicted cap and exceed the memory target. Please add a physical batch-size control (or an equivalent bounded path) and test actual returned rows/bytes rather than only the stored base value. ########## be/src/format_v2/table/paimon_rust_table_reader.cpp: ########## @@ -0,0 +1,785 @@ +// 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_v2/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 || Review Comment: [P1] Keep credentials out of the INFO log This redactor is case-sensitive, but the supported FE map supplies keys such as `AWS_SECRET_KEY`, `AWS_TOKEN`, and `fs.oss.accessKeySecret`. `_build_options()` retains those originals, and the table-open INFO message renders the whole map, so their values are logged verbatim. Please avoid logging option values; at minimum normalize keys and cover every credential alias with tests. ########## be/src/format_v2/table/paimon_rust_predicate_converter.cpp: ########## @@ -0,0 +1,760 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +#include "format_v2/table/paimon_rust_predicate_converter.h" + +#include <algorithm> +#include <cctype> +#include <memory> +#include <utility> + +#include "common/logging.h" +#include "core/column/column_const.h" +#include "core/column/column_nullable.h" +#include "core/data_type/data_type.h" +#include "core/data_type/data_type_nullable.h" +#include "core/field.h" +#include "core/types.h" +#include "core/value/decimalv2_value.h" +#include "core/value/timestamptz_value.h" +#include "core/value/vdatetime_value.h" +#include "exprs/vcompound_pred.h" +#include "exprs/vdirect_in_predicate.h" +#include "exprs/vectorized_fn_call.h" +#include "exprs/vexpr.h" +#include "exprs/vin_predicate.h" +#include "exprs/vliteral.h" +#include "exprs/vslot_ref.h" + +namespace doris { + +namespace { +// paimon_datum tags (see paimon.h / bindings/c/src/table.rs::datum_from_c). +constexpr int32_t kTagBool = 0; +constexpr int32_t kTagTinyInt = 1; +constexpr int32_t kTagSmallInt = 2; +constexpr int32_t kTagInt = 3; +constexpr int32_t kTagLong = 4; +constexpr int32_t kTagDouble = 6; +constexpr int32_t kTagString = 7; +constexpr int32_t kTagDate = 8; +constexpr int32_t kTagTimestamp = 10; +constexpr int32_t kTagDecimal = 12; +constexpr int32_t kTagBytes = 13; + +// paimon decimal precision ceiling (paimon::Decimal::MAX_PRECISION). +constexpr int32_t kPaimonDecimalMaxPrecision = 38; + +// RAII for an owned paimon_predicate*. and/or/not consume their inputs, so we +// release() before handing pointers to them. +struct predicate_deleter { + void operator()(paimon_predicate* p) const { + if (p) { + paimon_predicate_free(p); + } + } +}; +using predicate_ptr = std::unique_ptr<paimon_predicate, predicate_deleter>; + +// RAII for an owned paimon_error*. +struct error_deleter { + void operator()(paimon_error* p) const { + if (p) { + paimon_error_free(p); + } + } +}; +using error_ptr = std::unique_ptr<paimon_error, error_deleter>; + +// Render a paimon_error into a string. Takes ownership of `err` via RAII so it +// is freed on every return path. Safe to call with nullptr. +std::string consume_predicate_error(paimon_error* err) { + error_ptr owned(err); + if (!owned) { + return "unknown error"; + } + std::string msg; + if (owned->message.data != nullptr && owned->message.len > 0) { + msg.assign(reinterpret_cast<const char*>(owned->message.data), owned->message.len); + } + return "code=" + std::to_string(owned->code) + ", msg=" + msg; +} +} // namespace + +PaimonRustPredicateConverter::PaimonRustPredicateConverter( + const std::vector<std::string>& column_names, const std::vector<DataTypePtr>& column_types, + const paimon_table* table) + : _table(table) { + DORIS_CHECK(column_names.size() == column_types.size()); + _columns_by_name.reserve(column_names.size()); + for (size_t i = 0; i < column_names.size(); ++i) { + _columns_by_name.emplace(_normalize_name(column_names[i]), + std::make_pair(column_names[i], column_types[i])); + } + // Paimon TIMESTAMP (wall clock) is stored as epoch-millis-of-the-wall-time + // and the DateTimeV2 serde decodes timezone-naive arrow values in UTC, so + // timestamp literals convert wall->epoch in UTC. utc_time_zone() needs no + // tzdata lookup, so the conversion cannot silently fall back to a + // machine-local zone. + _utc_tz = cctz::utc_time_zone(); +} + +paimon_predicate* PaimonRustPredicateConverter::build(const VExprContextSPtrs& conjuncts) { + if (_table == nullptr) { + return nullptr; + } + predicate_ptr result; + for (const auto& conjunct : conjuncts) { + if (!conjunct || !conjunct->root()) { + continue; + } + auto root = conjunct->root(); + if (root->is_rf_wrapper()) { + if (auto impl = root->get_impl()) { + 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; + } + + // Casts are not unwrapped anywhere (predicate root included): a cast node + // fails every dispatch below and the conjunct stays in the Doris residual, + // mirroring the FE converter, which keeps casted expressions unconverted. + if (auto* direct_in = dynamic_cast<VDirectInPredicate*>(expr.get())) { + VExprSPtr in_expr; + if (direct_in->get_slot_in_expr(in_expr)) { + return _convert_in(in_expr); + } + return nullptr; + } + + if (dynamic_cast<VInPredicate*>(expr.get()) != nullptr) { + return _convert_in(expr); + } + + switch (expr->op()) { + case TExprOpcode::COMPOUND_AND: + case TExprOpcode::COMPOUND_OR: + return _convert_compound(expr); + case TExprOpcode::COMPOUND_NOT: + return nullptr; + case TExprOpcode::EQ: + case TExprOpcode::EQ_FOR_NULL: + case TExprOpcode::NE: + case TExprOpcode::GE: + case TExprOpcode::GT: + case TExprOpcode::LE: + case TExprOpcode::LT: + return _convert_binary(expr); + default: + break; + } + + if (auto* fn = dynamic_cast<VectorizedFnCall*>(expr.get())) { + auto fn_name = _normalize_name(fn->function_name()); + if (fn_name == "is_null_pred" || fn_name == "is_not_null_pred") { + return _convert_is_null(expr, fn_name); + } + if (fn_name == "like") { + return _convert_like_prefix(expr); + } + } + + return nullptr; +} + +paimon_predicate* PaimonRustPredicateConverter::_convert_compound(const VExprSPtr& expr) { + if (!expr || expr->get_num_children() != 2) { + return nullptr; + } + predicate_ptr left(_convert_expr(expr->get_child(0))); + if (!left) { + return nullptr; + } + predicate_ptr right(_convert_expr(expr->get_child(1))); + if (!right) { + return nullptr; + } + + if (expr->op() == TExprOpcode::COMPOUND_AND) { + return paimon_predicate_and(left.release(), right.release()); + } + if (expr->op() == TExprOpcode::COMPOUND_OR) { + return paimon_predicate_or(left.release(), right.release()); + } + return nullptr; +} + +paimon_predicate* PaimonRustPredicateConverter::_convert_in(const VExprSPtr& expr) { + auto* in_pred = dynamic_cast<VInPredicate*>(expr.get()); + if (!in_pred || expr->get_num_children() < 2) { + return nullptr; + } + auto field_meta = _resolve_field(expr->get_child(0)); + if (!field_meta) { + return nullptr; + } + + const auto num_values = expr->get_num_children() - 1; + // Reserve up front so the backing strings never reallocate: each datum's + // str_data points into storages[i], which must stay stable. + std::vector<std::string> storages; + std::vector<paimon_datum> datums; + storages.reserve(num_values); + datums.reserve(num_values); + for (uint16_t i = 1; i < expr->get_num_children(); ++i) { + auto holder = _convert_literal(expr->get_child(i), field_meta->type); + if (!holder) { + return nullptr; + } + storages.emplace_back(std::move(holder->storage)); + paimon_datum datum = holder->datum; + _bind_datum_storage(&datum, storages.back()); + datums.emplace_back(datum); + } + + if (datums.empty()) { + return nullptr; + } + if (in_pred->is_not_in()) { + return _take(paimon_predicate_is_not_in(_table, field_meta->column.c_str(), datums.data(), + datums.size())); + } + return _take(paimon_predicate_is_in(_table, field_meta->column.c_str(), datums.data(), + datums.size())); +} + +paimon_predicate* PaimonRustPredicateConverter::_convert_binary(const VExprSPtr& expr) { + if (!expr || expr->get_num_children() != 2) { + return nullptr; + } + auto field_meta = _resolve_field(expr->get_child(0)); + if (!field_meta) { + return nullptr; + } + const char* column = field_meta->column.c_str(); + + // Convert the RHS first so EQ_FOR_NULL (<=>) only converts when the RHS is + // a convertible literal, mirroring the FE converter, which rejects a + // non-literal RHS. A column-to-column `a <=> b` must therefore stay in the + // Doris residual: it has no single-column rust predicate, and pushing + // `a IS NULL` would wrongly discard rows like (1, 1) — rows dropped by the + // pushed filter cannot be recovered by the residual conjunct. + auto holder = _convert_literal(expr->get_child(1), field_meta->type); + if (!holder) { + return nullptr; + } + + if (expr->op() == TExprOpcode::EQ_FOR_NULL) { + return _take(paimon_predicate_is_null(_table, column)); + } + + // `holder` is a local, so its storage stays put for the duration of the call. + _bind_datum_storage(&holder->datum, holder->storage); + const paimon_datum& datum = holder->datum; + + switch (expr->op()) { + case TExprOpcode::EQ: + return _take(paimon_predicate_equal(_table, column, datum)); + case TExprOpcode::NE: + return _take(paimon_predicate_not_equal(_table, column, datum)); + case TExprOpcode::GE: + return _take(paimon_predicate_greater_or_equal(_table, column, datum)); + case TExprOpcode::GT: + return _take(paimon_predicate_greater_than(_table, column, datum)); + case TExprOpcode::LE: + return _take(paimon_predicate_less_or_equal(_table, column, datum)); + case TExprOpcode::LT: + return _take(paimon_predicate_less_than(_table, column, datum)); + default: + break; + } + return nullptr; +} + +paimon_predicate* PaimonRustPredicateConverter::_convert_is_null(const VExprSPtr& expr, + const std::string& fn_name) { + if (!expr || expr->get_num_children() != 1) { + return nullptr; + } + auto field_meta = _resolve_field(expr->get_child(0)); + if (!field_meta) { + return nullptr; + } + if (fn_name == "is_not_null_pred") { + return _take(paimon_predicate_is_not_null(_table, field_meta->column.c_str())); + } + return _take(paimon_predicate_is_null(_table, field_meta->column.c_str())); +} + +paimon_predicate* PaimonRustPredicateConverter::_convert_like_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; + } + // Mirror the FE converter's convertDorisExprToSlotRef: a casted column is + // rejected, never unwrapped. Stripping a lossy cast changes which rows match + // — for a DECIMAL(10,2) column, CAST(amount AS DECIMAL(10,1)) = 1.2 keeps + // the row 1.24 while the unwrapped `amount = 1.2` prunes it — and rows + // pruned by the pushed filter cannot be recovered by the Doris residual. + // The conjunct stays in the residual instead. + auto* slot_ref = dynamic_cast<VSlotRef*>(expr.get()); + if (!slot_ref) { + return std::nullopt; + } + // FileScannerV2 rewrites conjunct VSlotRefs to table global indices, so slot_id + // is a position, not a slot id; resolve by the carried column name against the + // projected-column registry instead of the desc table. + auto it = _columns_by_name.find(_normalize_name(slot_ref->column_name())); + if (it == _columns_by_name.end()) { + return std::nullopt; + } + const auto& [column, type] = it->second; + if (!_is_supported_slot_type(type->get_primitive_type(), type->get_precision())) { + return std::nullopt; + } + return FieldMeta {column, type}; +} + +std::optional<PaimonRustPredicateConverter::DatumHolder> +PaimonRustPredicateConverter::_convert_literal(const VExprSPtr& expr, + const DataTypePtr& column_type) const { + // Mirror the FE converter's convertDorisExprToLiteralExpr: a bare literal + // or a single cast wrapping a direct literal converts; anything deeper is + // rejected. Unwrapping recursively would silently apply the inner casts' + // lossy semantics (e.g. a DECIMAL scale reduction), which FE also rejects + // (its instanceof check only unwraps one CastExpr around a LiteralExpr). + VExprSPtr literal_expr = expr; + if (expr->node_type() == TExprNodeType::CAST_EXPR) { + literal_expr = expr->get_child(0); + } + 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: { Review Comment: [P1] Do not push DOUBLE predicates with incompatible NaN semantics Doris defines NaN as equal to NaN and greater than every finite value, but the pinned Rust evaluator uses `f64::partial_cmp`, so NaN is unordered and not equal to itself. A stored NaN therefore gets dropped by `column > 1.0`, and a late `IN (NaN)` runtime filter also rejects it, even though Doris retains both rows before residual evaluation. Please skip DOUBLE pushdown until the Rust evaluator matches Doris total ordering/equality, and add NaN differential coverage. ########## thirdparty/build-thirdparty.sh: ########## @@ -2169,6 +2169,133 @@ build_lance_c() { fi } +# 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 + + # paimon-vindex-core 0.4.0 uses the unstable `stdarch_neon_f16` intrinsics + # (vcvt_f32_f16 / vreinterpret_f16_u16) in its aarch64 NEON fast path, which + # do not compile on stable Rust. On aarch64/arm64, replace the registry + # crate with a patched path source so the build succeeds. The patch swaps + # the unstable f16->f32 NEON convert for a stable scalar conversion; the + # rest of the NEON accumulation is left untouched. x86_64 and other arches + # are unaffected and keep using the pristine registry crate. + if [[ "$(uname -m)" == "aarch64" || "$(uname -m)" == "arm64" ]]; then + local vindex_override="${PWD}/.doris-vindex-override" + local vindex_crate + vindex_crate="$(find "${CARGO_HOME:-$HOME/.cargo}/registry/cache" \ + -type f -name 'paimon-vindex-core-0.4.0.crate' 2>/dev/null | head -n1)" + if [[ -z "${vindex_crate}" ]]; then + # Ensure the .crate is downloaded into the registry cache first. + local fetch_args=(fetch) + if [[ "$(echo "${PAIMON_RUST_CARGO_OFFLINE}" | tr '[:lower:]' '[:upper:]')" == "ON" ]]; then + fetch_args+=(--offline) + fi + env "${cargo_env[@]}" "${cargo_bin}" "${fetch_args[@]}" + vindex_crate="$(find "${CARGO_HOME:-$HOME/.cargo}/registry/cache" \ + -type f -name 'paimon-vindex-core-0.4.0.crate' 2>/dev/null | head -n1)" + fi + if [[ -z "${vindex_crate}" ]]; then + echo "failed to locate paimon-vindex-core-0.4.0.crate in the cargo registry cache" + exit 1 + fi + rm -rf "${vindex_override}" + mkdir -p "${vindex_override}" + tar xzf "${vindex_crate}" -C "${vindex_override}" --strip-components=1 + (cd "${vindex_override}" && \ + patch -p1 -s <"${TP_PATCH_DIR}/paimon-vindex-core-0.4.0-aarch64-stable.patch") + # Inject the [patch.crates-io] override into the workspace manifest. + # Idempotent: skip if a previous run already injected it. + if ! grep -q "DORIS_PATCHED_VINDEX" Cargo.toml; then + cat >>Cargo.toml <<'EOF' + +# DORIS_PATCHED_VINDEX: override the registry crate with a build that compiles +# on stable Rust for aarch64 (avoids the unstable stdarch_neon_f16 intrinsics). +[patch.crates-io] +paimon-vindex-core = { path = ".doris-vindex-override" } +EOF + fi + # Record the path source in Cargo.lock so --locked stays satisfied. + env "${cargo_env[@]}" "${cargo_bin}" update \ + -p paimon-vindex-core --offline + 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 Review Comment: [P2] Pin the C header generator and honor offline mode When cbindgen is absent, this runs an unversioned online `cargo install` even when `PAIMON_RUST_CARGO_OFFLINE=ON`; the rustup auto-install above has the same network escape. Clean offline builds therefore fail/attempt network access, and online builds generate `paimon.h` with whichever cbindgen release is current (a custom CARGO_HOME may not even put the installed binary on PATH). Please make the toolchain a pinned Doris-controlled input and invoke its resolved path directly. ########## fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/source/PaimonScanNode.java: ########## @@ -411,10 +412,64 @@ 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) FileScannerV2: the V1 FileScanner + // explicitly rejects PAIMON_RUST, so with enable_file_scanner_v2 disabled + // the split falls back to JNI instead of encoding a rust request that the + // selected scanner cannot consume, and (b) a FileStoreTable: BE opens the + // table via paimon_table_from_schema_json, which needs the resolved + // TableSchema that only FileStoreTable exposes via schema(). If the table + // is not a FileStoreTable (e.g. a sys table backed by DataSplit), we cannot + // ship a schema JSON, so fall back to CPP / JNI rather than sending an + // incomplete PAIMON_RUST request that BE would reject. + // + // Serialize the same effective table that planning and the JNI reader use. + // Relation options such as t@options('read.batch-size'='1') are applied by + // getProcessedTable() (doInitialize caches it in processedTable), and the + // rust reader derives its read batch size from the schema options — the raw + // cached table would silently drop the override. Copies, delegates and + // fallback wrappers of getProcessedTable() are still FileStoreTable, so the + // instanceof gate keeps its semantics. + Table paimonTable = processedTable; + boolean canUseRust = sessionVariable.isEnablePaimonRustReader() + && sessionVariable.enableFileScannerV2 && nativeSplit Review Comment: [P2] Gate Rust ranges during BE rolling upgrades This decision is independent of the destination backend. During a rolling upgrade, an old BE does not recognize `PAIMON_RUST`: old Scanner V2 rejects enum value 3, while old V1 sends the native DataSplit bytes to the Java-object deserializer. Enabling the session flag can therefore fail otherwise valid queries until every BE is upgraded. Please use the backend smooth-upgrade/capability signal to fall back to JNI (or reject the feature up front), and add a mixed-version routing test. ########## be/src/format_v2/table/paimon_rust_table_reader.cpp: ########## @@ -0,0 +1,785 @@ +// 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_v2/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); + // Materialize TIMESTAMP_LTZ in the session timezone — the same + // convention as the JNI reader (PaimonJniScanner reads time_zone from + // its scan params) and lance_reader. Timezone-naive (paimon TIMESTAMP) + // arrow values are decoded in UTC by the DateTimeV2 serde regardless + // of _ctz, so NTZ wall-clock semantics are preserved. + DORIS_CHECK(_runtime_state != nullptr); + _ctz = _runtime_state->timezone_obj(); + if (_scanner_profile != nullptr) { + file_scan_profile::ensure_hierarchy(_scanner_profile); + _rust_total_time = ADD_CHILD_TIMER(_scanner_profile, "PaimonRustReader", + file_scan_profile::TABLE_READER); + _rust_open_split_time = ADD_CHILD_TIMER(_scanner_profile, "OpenSplitTime", + "PaimonRustReader"); + _rust_read_batch_time = + ADD_CHILD_TIMER(_scanner_profile, "ReadBatchTime", "PaimonRustReader"); + _rust_arrow_to_block_time = + ADD_CHILD_TIMER(_scanner_profile, "ArrowToBlockTime", "PaimonRustReader"); + } + // Projected column name -> fixed output position, registered with both the exact and + // the lower-case spelling so mixed-case Rust schema output still resolves (v1 + // semantics: exact match first, lower-case fallback on lookup). + _output_name_to_idx.reserve(_projected_columns.size() * 2); + for (size_t idx = 0; idx < _projected_columns.size(); ++idx) { + _output_name_to_idx.emplace(_projected_columns[idx].name, idx); + _output_name_to_idx.emplace(to_lower(_projected_columns[idx].name), idx); + } + } + return Status::OK(); +} + +Status PaimonRustTableReader::prepare_split(const format::SplitReadOptions& options) { + // EOF belongs to the previous split. Keep it set after closing that split so repeated reads + // are idempotent, and clear it only when a new split is explicitly prepared. + _close_split_reader(); + _split_eof = false; + _current_range = options.current_range; + RETURN_IF_ERROR(format::TableReader::prepare_split(options)); + if (current_split_pruned()) { + return Status::OK(); + } + if (_is_table_level_count_active()) { + // No rust pipeline is opened; get_block emits the synthetic count rows. + return Status::OK(); + } + RETURN_IF_ERROR(_validate_rust_split(options.current_range)); + { + SCOPED_TIMER(_profile.total_timer); + SCOPED_TIMER(_profile.prepare_split_timer); + SCOPED_TIMER(_rust_open_split_time); + RETURN_IF_ERROR(_open_split_reader(options.current_range)); + } + return Status::OK(); +} + +Status PaimonRustTableReader::get_block(Block* block, bool* eos) { + SCOPED_TIMER(_profile.total_timer); + SCOPED_TIMER(_profile.exec_timer); + SCOPED_TIMER(_rust_total_time); + DORIS_CHECK(block != nullptr); + DORIS_CHECK(eos != nullptr); + DORIS_CHECK(block->columns() == _projected_columns.size()); + block->clear_column_data(_projected_columns.size()); + *eos = false; + + if (_is_table_level_count_active()) { + return _read_table_level_count(block, eos); + } + + // num_splits == 0 yields an empty (but valid) stream: report EOF. + if (_split_eof) { + *eos = true; + return Status::OK(); + } + if (!_handles || !_handles->reader) { + return Status::InternalError("paimon-rust reader is not initialized"); + } + + while (true) { + // Mirror the base TableReader cancellation contract so a cancelled query does not + // drain the whole split. + if (_io_ctx != nullptr && _io_ctx->should_stop) { + _split_eof = true; + _close_split_reader(); + *eos = true; + return Status::OK(); + } + + paimon_result_next_batch next; + { + SCOPED_TIMER(_rust_read_batch_time); + 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) { + _split_eof = true; + _close_split_reader(); + *eos = true; + return Status::OK(); + } + + // RAII: the batch's Arrow release callbacks + container free run when + // `batch` leaves this scope, including on any early return. + ArrowBatch batch(next.batch); + + auto* c_array = batch.array(); + auto* c_schema = batch.schema(); + arrow::Result<std::shared_ptr<arrow::RecordBatch>> 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 rows = static_cast<size_t>(record_batch->num_rows()); + if (rows == 0) { + // Skip empty batches and keep draining the stream. + continue; + } + RETURN_IF_ERROR(_fill_block_from_record_batch(record_batch, block, rows)); + _record_scan_rows(rows); + *eos = false; + return Status::OK(); + } +} + +Status PaimonRustTableReader::abort_split() { + { + SCOPED_TIMER(_profile.total_timer); + SCOPED_TIMER(_profile.close_timer); + _close_split_reader(); + _split_eof = false; + } + return format::TableReader::abort_split(); +} + +Status PaimonRustTableReader::close() { + { + SCOPED_TIMER(_profile.total_timer); + SCOPED_TIMER(_profile.close_timer); + _close_split_reader(); + _close_table(); + } + return format::TableReader::close(); +} + +Status PaimonRustTableReader::_validate_rust_split(const TFileRangeDesc& range) const { + if (!range.__isset.table_format_params || !range.table_format_params.__isset.paimon_params) { + return Status::InternalError( + "missing paimon_params for paimon rust reader, possibly caused by FE/BE protocol " + "mismatch"); + } + const auto& params = range.table_format_params.paimon_params; + if (!params.__isset.paimon_split || params.paimon_split.empty()) { + return Status::InternalError( + "missing paimon_split for paimon rust reader, possibly caused by FE/BE protocol " + "mismatch"); + } + if (params.__isset.reader_type && params.reader_type != TPaimonReaderType::PAIMON_RUST) { + return Status::InternalError( + "invalid reader_type for paimon rust reader, possibly caused by FE/BE protocol " + "mismatch"); + } + if (!_resolve_table_path(range).has_value()) { + return Status::InternalError( + "paimon-rust missing paimon_table; cannot resolve paimon table location"); + } + if (!_resolve_db_name(range).has_value()) { + return Status::InternalError( + "paimon-rust missing db_name; cannot open paimon table via schema json"); + } + if (!_resolve_table_name(range).has_value()) { + return Status::InternalError( + "paimon-rust missing table_name; cannot open paimon table via schema json"); + } + if (!_resolve_table_schema_json(range).has_value()) { + return Status::InternalError( + "paimon-rust missing paimon_table_schema_json; cannot open paimon table via " + "schema json"); + } + return Status::OK(); +} + +Status PaimonRustTableReader::_open_split_reader(const TFileRangeDesc& range) { + // 1. Decode the FE-planned split first so we fail fast (and without any + // filesystem IO) when it is missing or malformed. + std::string split_bytes; + RETURN_IF_ERROR(_decode_split_bytes(&split_bytes)); + + // 2. Resolve identifier + table_path + FE-supplied TableSchema JSON. + auto table_path = _resolve_table_path(range).value(); + auto db_name = _resolve_db_name(range).value(); + auto table_name = _resolve_table_name(range).value(); + auto schema_json = _resolve_table_schema_json(range).value(); + auto branch_opt = _resolve_branch(range); + + // 3. Assemble storage options: FE-supplied paimon options + hadoop_conf + + // OSS/S3 → AWS_* translations. These feed FileIO only (per + // paimon_table_from_schema_json contract); they are NOT merged into the + // supplied table schema. + auto options = _build_options(); + + auto opened_table_key = std::make_tuple(table_path, schema_json, db_name, table_name, + branch_opt, options); + if (!_handles || !_handles->table || _opened_table_key != opened_table_key) { + // A paimon scan reads one table, so the handle is opened at most once per + // distinct identity (e.g. re-created after a close); splits of the same + // table reuse it and only rebuild the read pipeline below. + _close_table(); + _handles = std::make_unique<PaimonHandles>(); + + std::vector<paimon_option> 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 schema json: db=" << db_name + << " table=" << table_name << " path=" << table_path + << " branch=" << (branch_opt.has_value() ? branch_opt.value() : "main") + << " storage_options=[" << format_options(options) << "]"; + + // Build the table directly from the FE-supplied schema JSON. The Rust + // side rejects null / empty branch, so we default to paimon's canonical + // "main" sentinel when FE did not set paimon_branch (i.e. the table is + // on the main branch — matches upstream Identifier.DEFAULT_MAIN_BRANCH). + const std::string& branch_str = branch_opt.has_value() ? branch_opt.value() : "main"; + paimon_result_get_table tbl_res = paimon_table_from_schema_json( + table_path.c_str(), schema_json.c_str(), db_name.c_str(), table_name.c_str(), + branch_str.c_str(), c_options.empty() ? nullptr : c_options.data(), + c_options.size()); + if (tbl_res.error != nullptr) { + return Status::InternalError( + "paimon-rust table_from_schema_json failed: db={} table={} err={}", db_name, + table_name, consume_error(tbl_res.error)); + } + _handles->table.reset(tbl_res.table); + _opened_table_key = std::move(opened_table_key); + } + + // 4. 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)); + } + + // Partition keys are excluded: they are materialized from split metadata + // (see _fill_non_arrow_columns), and paimon-rust does not emit them. + auto read_columns = _build_read_columns(); + std::vector<const char*> 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 scanner conjuncts into a paimon-rust filter and apply it. + RETURN_IF_ERROR(_apply_predicate()); + + // 5. 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<const uint8_t*>(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) { + _split_eof = true; + return Status::OK(); + } + + // 6. 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(); +} + +void PaimonRustTableReader::_close_split_reader() { + if (!_handles) { + return; + } + // Reverse of the declaration order in PaimonHandles. + _handles->reader.reset(); + _handles->table_read.reset(); + _handles->plan.reset(); + _handles->read_builder.reset(); +} + +void PaimonRustTableReader::_close_table() { + if (!_handles) { + return; + } + _close_split_reader(); + _handles->table.reset(); + _opened_table_key.reset(); +} + +Status PaimonRustTableReader::_apply_predicate() { + if (_conjuncts.empty() || !_handles || !_handles->table || !_handles->read_builder) { + return Status::OK(); + } + LOG(INFO) << "paimon-rust predicate pushdown: " << _conjuncts.size() << " conjunct(s) input"; + // The conjunct VSlotRefs carry table global indices (positions), so the v2 + // converter mode resolves fields by the projected column names; partition + // keys are excluded because the rust reader does not read them. + std::vector<std::string> names; + std::vector<DataTypePtr> types; + names.reserve(_projected_columns.size()); + types.reserve(_projected_columns.size()); + for (const auto& col : _projected_columns) { + if (col.is_partition_key) { + continue; + } + names.push_back(col.name); + types.push_back(col.type); + } + PaimonRustPredicateConverter converter(names, types, _handles->table.get()); + paimon_predicate* predicate = converter.build(_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 PaimonRustTableReader::_fill_block_from_record_batch( + const std::shared_ptr<arrow::RecordBatch>& batch, Block* block, size_t rows) { + SCOPED_TIMER(_rust_arrow_to_block_time); + DORIS_CHECK(batch != nullptr); + DORIS_CHECK(block != nullptr); + std::unordered_set<size_t> materialized_indices; + materialized_indices.reserve(_projected_columns.size()); + { + auto columns_guard = block->mutate_columns_scoped(); + auto& columns = columns_guard.mutable_columns(); + for (int c = 0; c < batch->num_columns(); ++c) { + const auto& field = batch->schema()->field(c); + if (field->name() == VALUE_KIND_FIELD) { + continue; + } + // Projected 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 tolerate + // mixed-case Rust output by folding here as well. + auto it = _output_name_to_idx.find(field->name()); + if (it == _output_name_to_idx.end()) { + it = _output_name_to_idx.find(to_lower(field->name())); + } + if (it == _output_name_to_idx.end()) { + // Skip columns that are not in the block (e.g. columns dropped by + // slot pruning). + continue; + } + const auto output_idx = it->second; + if (!materialized_indices.emplace(output_idx).second) { + return Status::InternalError("paimon-rust returned duplicate column '{}'", + field->name()); + } + try { + RETURN_IF_ERROR(columns_guard.get_datatype_by_position(output_idx) + ->get_serde() + ->read_column_from_arrow(*columns[output_idx], + batch->column(c).get(), 0, rows, + _ctz)); + } catch (Exception& e) { + return Status::InternalError("Failed to convert from arrow to block: {}", + e.what()); + } + } + } + // Partition columns and other projected columns absent from the arrow batch + // are back-filled from split metadata / defaults. + return _fill_non_arrow_columns(block, rows, materialized_indices); +} + +Status PaimonRustTableReader::_fill_non_arrow_columns( + Block* block, size_t rows, const std::unordered_set<size_t>& materialized_indices) { + for (size_t idx = 0; idx < _projected_columns.size(); ++idx) { + if (materialized_indices.count(idx) != 0) { + continue; + } + const auto& column = _projected_columns[idx]; + VExprContextSPtr constant_expr; + if (const Field* value = find_partition_value(column, _partition_values); + column.is_partition_key && value != nullptr) { + // Partition values are split constants (same materialization the + // TableColumnMapper builds for native readers). + constant_expr = VExprContext::create_shared( + VLiteral::create_shared(column.type, *value)); + } else if (column.default_expr != nullptr) { + constant_expr = column.default_expr; + } else { + // The column is genuinely absent from the arrow batch. Schema + // evolution is handled by paimon-rust itself, so reaching here means + // an unexpected schema drift: fill defaults so the scan remains + // well-defined instead of failing the query. + LOG(WARNING) << "paimon-rust did not return projected column '" << column.name + << "'; filling with defaults"; + auto data = column.type->create_column(); + data->insert_many_defaults(rows); + block->replace_by_position(idx, std::move(data)); + continue; + } + ColumnPtr constant_column; + RETURN_IF_ERROR(_materialize_constant_column(constant_expr, column.type, column.name, rows, + &constant_column)); + block->replace_by_position(idx, std::move(constant_column)); + } + return Status::OK(); +} + +Status PaimonRustTableReader::_materialize_constant_column(const VExprContextSPtr& expr, + const DataTypePtr& type, + const std::string& name, size_t rows, + ColumnPtr* column) { + DORIS_CHECK(expr != nullptr); + DORIS_CHECK(column != nullptr); + RowDescriptor row_desc; + RETURN_IF_ERROR(expr->prepare(_runtime_state, row_desc)); + RETURN_IF_ERROR(expr->open(_runtime_state)); + // Constants evaluate per input row, so a rows-sized synthetic block yields a + // rows-sized result for both plain literals and default expressions. + Block eval_block; + eval_block.insert({type->create_column_const_with_default_value(rows), type, name}); + int result_column_id = -1; + RETURN_IF_ERROR(expr->execute(&eval_block, &result_column_id)); + DORIS_CHECK(result_column_id >= 0); + ColumnPtr result_column = eval_block.get_by_position(result_column_id).column; + if (result_column->size() == 1 && rows > 1) { + result_column = ColumnConst::create(std::move(result_column), rows); + } + *column = std::move(result_column); + return Status::OK(); +} + +Status PaimonRustTableReader::_decode_split_bytes(std::string* out) const { + if (!_current_range.__isset.table_format_params || + !_current_range.table_format_params.__isset.paimon_params || + !_current_range.table_format_params.paimon_params.__isset.paimon_split) { + return Status::InternalError("paimon-rust missing paimon_split in scan range"); + } + const auto& encoded_split = _current_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<std::string> PaimonRustTableReader::_resolve_table_path( + const TFileRangeDesc& range) 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<std::string> PaimonRustTableReader::_resolve_db_name( + const TFileRangeDesc& range) 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<std::string> PaimonRustTableReader::_resolve_table_name( + const TFileRangeDesc& range) 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<std::string> PaimonRustTableReader::_resolve_table_schema_json( + const TFileRangeDesc& range) const { + if (range.__isset.table_format_params && range.table_format_params.__isset.paimon_params && + range.table_format_params.paimon_params.__isset.paimon_table_schema_json && + !range.table_format_params.paimon_params.paimon_table_schema_json.empty()) { + return range.table_format_params.paimon_params.paimon_table_schema_json; + } + return std::nullopt; +} + +std::optional<std::string> PaimonRustTableReader::_resolve_branch( + const TFileRangeDesc& range) const { + // FE only sets paimon_branch when the branch is not `main` (matches + // upstream paimon commit 742da63: null-if-DEFAULT_MAIN_BRANCH). Unset here + // means main-branch semantics. + if (range.__isset.table_format_params && range.table_format_params.__isset.paimon_params && + range.table_format_params.paimon_params.__isset.paimon_branch && + !range.table_format_params.paimon_params.paimon_branch.empty()) { + return range.table_format_params.paimon_params.paimon_branch; + } + return std::nullopt; +} + +std::vector<std::string> PaimonRustTableReader::_build_read_columns() const { + std::vector<std::string> columns; + columns.reserve(_projected_columns.size()); + for (const auto& column : _projected_columns) { + if (column.is_partition_key) { + continue; + } + columns.emplace_back(column.name); + } + return columns; +} + +std::map<std::string, std::string> PaimonRustTableReader::_build_options() const { + std::map<std::string, std::string> options; + if (_scan_params && _scan_params->__isset.paimon_options && + !_scan_params->paimon_options.empty()) { + options.insert(_scan_params->paimon_options.begin(), _scan_params->paimon_options.end()); + } else if (_current_range.__isset.table_format_params && + _current_range.table_format_params.__isset.paimon_params && + _current_range.table_format_params.paimon_params.__isset.paimon_options) { + options.insert( + _current_range.table_format_params.paimon_params.paimon_options.begin(), + _current_range.table_format_params.paimon_params.paimon_options.end()); + } + + if (_scan_params && _scan_params->__isset.properties && !_scan_params->properties.empty()) { + for (const auto& kv : _scan_params->properties) { + options[kv.first] = kv.second; + } + } else if (_current_range.__isset.table_format_params && + _current_range.table_format_params.__isset.paimon_params && + _current_range.table_format_params.paimon_params.__isset.hadoop_conf) { + for (const auto& kv : + _current_range.table_format_params.paimon_params.hadoop_conf) { + 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; + } + }; + + // The pinned paimon-rust FileIO reads paimon-java's `s3.*` option family + // (io/storage_s3.rs normalizes the `fs.s3a.`/`s3a.`/`s3.` prefixes and the + // `s3.access.key`/`s3.path.style.access` aliases): s3.access-key, + // s3.secret-key, s3.session.token, s3.endpoint, s3.region and + // s3.path-style-access. `fs.s3a.*` keys therefore pass through natively, + // but the FE's storage-properties channel delivers the vended S3 config + // under the AWS_* / use_path_style aliases, which the crate does not read, + // and the OSS configs use their own fs.oss.* names — so remap both to the + // s3.* family. Without this the rust S3 FileIO builds with an empty region + // ("ConfigInvalid ... region is missing") and never connects. + copy_if_missing("AWS_ACCESS_KEY", "s3.access-key"); Review Comment: [P1] Preserve the configured S3 authentication mode The FE backend map can select anonymous access or emit `AWS_ROLE_ARN`/`AWS_EXTERNAL_ID`, but this bridge translates only static credentials and connection settings. The pinned Rust parser reads `s3.anonymous` and `s3.assumed.role.*`, so anonymous catalogs may consult/sign with the ambient chain and role-only catalogs never assume the requested role. Please map or explicitly reject each provider mode before selecting Rust, with anonymous and assume-role open tests. -- 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]
