github-actions[bot] commented on code in PR #66227:
URL: https://github.com/apache/doris/pull/66227#discussion_r4100817272
##########
thirdparty/build-thirdparty.sh:
##########
@@ -2103,15 +2186,15 @@ build_lance_c() {
local cargo_bin="${LANCE_C_CARGO:-${CARGO:-cargo}}"
if ! command -v "${cargo_bin}" >/dev/null 2>&1; then
- echo "cargo is required to build lance-c. Install Rust 1.91.0 or set
LANCE_C_CARGO."
+ echo "cargo is required to build lance-c. Install Rust 1.94.0 or set
LANCE_C_CARGO."
exit 1
fi
if [[ ! -x "${TP_INSTALL_DIR}/bin/protoc" ]]; then
echo "protoc is required to build lance-c. Build protobuf first."
exit 1
fi
- local required_rust_version="1.91.0"
+ local required_rust_version="1.94.0"
Review Comment:
[P1] Install the toolchain that the new minimum actually requires. Both
builders now set required_rust_version to 1.94.0, but the rustup guard still
succeeds when only 1.91.0 is installed; it then skips rustup toolchain install
and forces RUSTUP_TOOLCHAIN=1.94.0, so cargo fails with a missing toolchain
before producing the archive. Use the required version in the presence check
(or probe the selected cargo/rustc) in both functions, and cover an
old-toolchain-only rustup host.
##########
be/src/format_v2/table/paimon_rust_table_reader.cpp:
##########
@@ -0,0 +1,885 @@
+// 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");
+ }
+ // 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();
+}
+
+#ifdef BE_TEST
+std::string PaimonRustTableReader::TEST_format_options(
+ const std::map<std::string, std::string>& options) {
+ return format_options(options);
+}
+
+std::map<std::string, std::string> PaimonRustTableReader::TEST_build_options(
+ TFileScanRangeParams* scan_params, const TFileRangeDesc& range) {
+ TFileScanRangeParams* previous_params = _scan_params;
+ TFileRangeDesc previous_range = _current_range;
+ _scan_params = scan_params;
+ _current_range = range;
+ std::map<std::string, std::string> options = _build_options();
+ _scan_params = previous_params;
+ _current_range = std::move(previous_range);
+ return options;
+}
+#endif
+
+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() {
Review Comment:
[P2] Reapply late runtime filters to the Rust reader. _apply_predicate runs
only while prepare_split builds the read builder/Arrow stream (lines 455-490),
but FileScannerV2 can call refresh_conjuncts after that. PaimonRustTableReader
inherits TableReader::refresh_conjuncts, which returns after updating
_conjuncts because this direct path leaves _data_reader.reader null, so the
already-open Rust stream never sees the new filter; only the scanner residual
runs. Rebuild/reopen the split (or add an explicit refresh-capable Rust
boundary) so late filters still prune source IO.
--
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]
