Gabriel39 commented on code in PR #66227: URL: https://github.com/apache/doris/pull/66227#discussion_r4122166606
########## be/src/format_v2/table/paimon_rust_table_reader.cpp: ########## @@ -0,0 +1,912 @@ +// 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 <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/assert_cast.h" +#include "core/block/block.h" +#include "core/block/column_with_type_and_name.h" +#include "core/column/column_const.h" +#include "core/data_type/data_type_nullable.h" +#include "core/data_type/data_type_string.h" +#include "exprs/vexpr_context.h" +#include "exprs/vliteral.h" +#include "format_v2/column_mapper.h" +#include "format_v2/table/paimon_rust_predicate_converter.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 option KEYS for diagnostics. Values are never rendered: +// credential keys arrive under many spellings and cases (AWS_SECRET_KEY, +// AWS_TOKEN, fs.oss.accessKeySecret, s3.secret-key, ...), and a key-name +// blocklist that misses one alias leaks the value into the INFO log, so +// only the key names are printed at all. +std::string format_options(const std::map<std::string, std::string>& options) { + std::string out; + for (const auto& kv : options) { + if (!out.empty()) { + out += ", "; + } + out += kv.first; + } + 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"); + _rust_predicates_input = ADD_CHILD_COUNTER(_scanner_profile, "RustPredicatesInput", + TUnit::UNIT, "PaimonRustReader"); + _rust_predicates_converted = ADD_CHILD_COUNTER( + _scanner_profile, "RustPredicatesConverted", TUnit::UNIT, "PaimonRustReader"); + _rust_predicates_applied = ADD_CHILD_COUNTER(_scanner_profile, "RustPredicatesApplied", + TUnit::UNIT, "PaimonRustReader"); + _rust_runtime_filters_input = ADD_CHILD_COUNTER( + _scanner_profile, "RustRuntimeFiltersInput", TUnit::UNIT, "PaimonRustReader"); + _rust_runtime_filters_applied = ADD_CHILD_COUNTER( + _scanner_profile, "RustRuntimeFiltersApplied", TUnit::UNIT, "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) { Review Comment: Addressed the missing regression coverage in 6df36266b92. After tracing the FE-to-BE path, I could not reproduce the claimed loss of existing footer pushdown: - `PaimonScanNode.getSplits()` prefers native raw-file splits before logical JNI/Rust dispatch. Enabling Rust does not change this precedence. Those Parquet/ORC splits still use `PaimonReader` and the base file-aggregate contract. - Rust replaces the logical JNI reader, whose `get_block()` also reads rows for COUNT(col)/MINMAX. A logical DataSplit can need merge/delete semantics; independent file statistics cannot generally replace that stream. The Rust leaf has no base-class FileReader or column mapper, so invoking `_try_materialize_aggregate_pushdown_rows()` there is not valid. Falling back to JNI would still consume rows. - The existing table-level COUNT(*) shortcut remains separate and unchanged. Added an FE test covering COUNT(col)/MINMAX with Rust enabled and both Parquet/ORC raw files; it passes against the existing implementation. Added a SQL/profile differential across Rust enabled/disabled and aggregate pushdown enabled/disabled: expected results include NULLs, footer paths must report `RawRowsRead=0`, and full-scan controls must report positive reads. Forced logical JNI/Rust paths also compare aggregate results. Added a comment documenting the native-first routing invariant. Validation: 78 PaimonScanNodeTest tests passed, FE reactor Checkstyle passed, and the modified Groovy suite parsed successfully. The external SQL/profile regression was added but has not been executed locally; CI validation is pending. No reader behavior change was needed based on the reachable paths above. -- 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]
