github-actions[bot] commented on code in PR #68825: URL: https://github.com/apache/doris/pull/68825#discussion_r4272204393
########## fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonRustReaderSelector.java: ########## @@ -0,0 +1,298 @@ +// 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. + +package org.apache.doris.connector.paimon; + +import org.apache.doris.connector.spi.ConnectorSession; +import org.apache.doris.connector.spi.handle.ConnectorColumnHandle; + +import org.apache.commons.lang3.StringUtils; +import org.apache.paimon.CoreOptions; +import org.apache.paimon.io.DataFileMeta; +import org.apache.paimon.rest.RESTTokenFileIO; +import org.apache.paimon.schema.TableSchema; +import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.source.DataSplit; +import org.apache.paimon.types.ArrayType; +import org.apache.paimon.types.DataType; +import org.apache.paimon.types.DataTypeRoot; +import org.apache.paimon.types.MapType; +import org.apache.paimon.types.MultisetType; +import org.apache.paimon.types.RowType; + +import java.util.Arrays; +import java.util.HashSet; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.Optional; +import java.util.Set; + +/** + * Chooses the Rust reader only for table, split, storage and backend shapes it can read safely. + * Nested pruning and precision 7-9 timestamps still require JNI because the pinned Arrow ABI + * cannot preserve their full semantics. COUNT(*) metadata ranges bypass this selector entirely. + */ +final class PaimonRustReaderSelector { + static final String BACKEND_CAPABILITY = "paimon-rust-reader"; + private static final String ENABLE_PAIMON_RUST_READER = "enable_paimon_rust_reader"; + private static final String ENABLE_FILE_SCANNER_V2 = "enable_file_scanner_v2"; + private static final String FORCE_JNI_SCANNER = "force_jni_scanner"; + private static final String DELETION_VECTORS_MERGE_ON_READ = "deletion-vectors.merge-on-read"; + private static final Set<String> VERIFIED_LOCATION_SCHEMES = + new HashSet<>(Arrays.asList("s3", "s3a", "oss", "hdfs", "file")); + private static final Set<String> VERIFIED_HDFS_OPTION_KEYS = new HashSet<>(Arrays.asList( + "fs.defaultFS", + "hadoop.security.authentication", + "hdfs.security.authentication", + "ipc.client.fallback-to-simple-auth-allowed")); + + private final FileStoreTable table; + private final PaimonRustReaderCapabilities capabilities; + private final Map<String, String> backendStorageProperties; + private final boolean scanCompatible; + private final boolean partialUpdateOrAggregateWithDeletionVectors; + private final boolean deletionVectorsMergeOnRead; + + PaimonRustReaderSelector(ConnectorSession session, boolean backendsSupportRust, + FileStoreTable table, List<ConnectorColumnHandle> columns, + Map<String, String> backendStorageProperties, boolean usesFallbackRead, + boolean incrementalRead, boolean projectedVariant) { + this.table = table; + this.backendStorageProperties = backendStorageProperties; + CoreOptions options = table.coreOptions(); + CoreOptions.MergeEngine mergeEngine = options.mergeEngine(); + partialUpdateOrAggregateWithDeletionVectors = options.deletionVectorsEnabled() + && (mergeEngine == CoreOptions.MergeEngine.PARTIAL_UPDATE + || mergeEngine == CoreOptions.MergeEngine.AGGREGATE); + TableSchema schema = table.schema(); + String mergeOnRead = schema.options().get(DELETION_VECTORS_MERGE_ON_READ); + deletionVectorsMergeOnRead = "true".equalsIgnoreCase(mergeOnRead); + + String location = table.location().toString(); + boolean enabled = session != null && "true".equalsIgnoreCase( + session.getSessionProperties().get(ENABLE_PAIMON_RUST_READER)); + boolean fileScannerV2 = session != null && "true".equalsIgnoreCase( + session.getSessionProperties().get(ENABLE_FILE_SCANNER_V2)); + boolean forceJniScanner = session != null && "true".equalsIgnoreCase( + session.getSessionProperties().get(FORCE_JNI_SCANNER)); + boolean restTokenTable = table.fileIO() instanceof RESTTokenFileIO; + boolean unsupportedMergeOption = (mergeEngine == CoreOptions.MergeEngine.PARTIAL_UPDATE + || mergeEngine == CoreOptions.MergeEngine.AGGREGATE) + && hasRustUnsupportedMergeOption(schema.options(), mergeEngine); + boolean deduplicateIgnoreDelete = mergeEngine == CoreOptions.MergeEngine.DEDUPLICATE + && options.ignoreDelete(); + String scanMode = PaimonScanParams.withoutTimeTravelSelectors(schema) + .options().get(CoreOptions.SCAN_MODE.key()); + boolean scanModeSupported = scanMode == null || "default".equalsIgnoreCase(scanMode); + + scanCompatible = enabled && fileScannerV2 && !forceJniScanner + && backendsSupportRust + && !usesFallbackRead && !incrementalRead && !projectedVariant + && !options.queryAuthEnabled() && !restTokenTable + && !(partialUpdateOrAggregateWithDeletionVectors && deletionVectorsMergeOnRead) + && !deduplicateIgnoreDelete + && !unsupportedMergeOption && scanModeSupported + && isRustVerifiedLocationScheme(location) + && (!isHdfsLocationScheme(location) + || isRustVerifiedHdfsBackend(backendStorageProperties)) + && isCredentialProviderTranslatable(location, backendStorageProperties); + capabilities = new PaimonRustReaderCapabilities(table, columns); + } + + boolean canRead(DataSplit split) { + if (!scanCompatible || splitHasExternalFiles(split)) { + return false; + } + if (splitHasOrcFile(split) && table.schema().fields().stream() + .anyMatch(field -> containsTimestampLtz(field.type()))) { + return false; + } + if (partialUpdateOrAggregateWithDeletionVectors && !deletionVectorsMergeOnRead + && !isFullyMaterializedPkDvSplit(split)) { + return false; + } + return capabilities.canRead(split); + } + + static boolean isRustVerifiedLocationScheme(String location) { + if (location == null) { + return false; + } + int separator = location.indexOf("://"); Review Comment: [P2] Reject or qualify authority-less HDFS table locations. A filesystem catalog with warehouse=hdfs:///user/paimon and fs.defaultFS=hdfs://namenode:8020 is valid for Java, but Paimon Path.toString emits hdfs:/user/paimon/db.db/t. Because this helper only looks for ://, it admits that path as scheme-less and isHdfsLocationScheme also skips its checks. RustScanMetadata sends the path unchanged; the pinned Rust HDFS parser requires a host (and hdfs:// for file paths) and does not translate fs.defaultFS, so otherwise eligible scans fail when Rust is enabled. Parse schemes independently of authority, then fully qualify HDFS paths or keep these scans on JNI, and cover this catalog form in selector/differential tests. ########## be/src/format_v2/table/paimon_rust_table_reader.cpp: ########## @@ -0,0 +1,944 @@ +// 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 rust_handles { + +// --------------------------------------------------------------------------- +// 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 + +} // namespace rust_handles + +namespace { +constexpr const char* VALUE_KIND_FIELD = "_VALUE_KIND"; + +// 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) { + rust_handles::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 { + rust_handles::table_ptr table; + rust_handles::read_builder_ptr read_builder; + rust_handles::plan_ptr plan; + rust_handles::table_read_ptr table_read; + rust_handles::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); + // Open may dominate the scan or fail before get_block; include it in the parent timer. + SCOPED_TIMER(_rust_total_time); + 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(); + } + + if (!_pending_record_batch) { + paimon_result_next_batch next; + { + SCOPED_TIMER(_rust_read_batch_time); + next = paimon_record_batch_reader_next(_handles->reader.get()); Review Comment: [P1] Charge Rust read buffers to the query memory tracker. The buffers produced by this call remain Rust-owned, and the zero-copy import retains them in _pending_record_batch. In this build malloc/posix_memalign only forward to jemalloc; query tracking is explicit in Doris Allocator, which the Rust system allocator bypasses. Neither this reader nor the C binding reserves or charges those bytes, including Parquet row-group/decode buffers and Tokio worker allocations. Even the added wide-string fixture holds roughly 4 MiB in the Rust batch alongside its tracked Doris copy at a 32-row cap. Large row groups or concurrent scanners can therefore exceed query/workload-group limits while only output columns are counted. Add a query-owned accounting/reservation bridge with balanced release on EOF, error, abort and close; output slicing alone does not address this separate issue. ########## fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonRustReaderSelector.java: ########## @@ -0,0 +1,298 @@ +// 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. + +package org.apache.doris.connector.paimon; + +import org.apache.doris.connector.spi.ConnectorSession; +import org.apache.doris.connector.spi.handle.ConnectorColumnHandle; + +import org.apache.commons.lang3.StringUtils; +import org.apache.paimon.CoreOptions; +import org.apache.paimon.io.DataFileMeta; +import org.apache.paimon.rest.RESTTokenFileIO; +import org.apache.paimon.schema.TableSchema; +import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.source.DataSplit; +import org.apache.paimon.types.ArrayType; +import org.apache.paimon.types.DataType; +import org.apache.paimon.types.DataTypeRoot; +import org.apache.paimon.types.MapType; +import org.apache.paimon.types.MultisetType; +import org.apache.paimon.types.RowType; + +import java.util.Arrays; +import java.util.HashSet; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.Optional; +import java.util.Set; + +/** + * Chooses the Rust reader only for table, split, storage and backend shapes it can read safely. + * Nested pruning and precision 7-9 timestamps still require JNI because the pinned Arrow ABI + * cannot preserve their full semantics. COUNT(*) metadata ranges bypass this selector entirely. + */ +final class PaimonRustReaderSelector { + static final String BACKEND_CAPABILITY = "paimon-rust-reader"; + private static final String ENABLE_PAIMON_RUST_READER = "enable_paimon_rust_reader"; + private static final String ENABLE_FILE_SCANNER_V2 = "enable_file_scanner_v2"; + private static final String FORCE_JNI_SCANNER = "force_jni_scanner"; + private static final String DELETION_VECTORS_MERGE_ON_READ = "deletion-vectors.merge-on-read"; + private static final Set<String> VERIFIED_LOCATION_SCHEMES = + new HashSet<>(Arrays.asList("s3", "s3a", "oss", "hdfs", "file")); + private static final Set<String> VERIFIED_HDFS_OPTION_KEYS = new HashSet<>(Arrays.asList( + "fs.defaultFS", + "hadoop.security.authentication", + "hdfs.security.authentication", + "ipc.client.fallback-to-simple-auth-allowed")); + + private final FileStoreTable table; + private final PaimonRustReaderCapabilities capabilities; + private final Map<String, String> backendStorageProperties; + private final boolean scanCompatible; + private final boolean partialUpdateOrAggregateWithDeletionVectors; + private final boolean deletionVectorsMergeOnRead; + + PaimonRustReaderSelector(ConnectorSession session, boolean backendsSupportRust, + FileStoreTable table, List<ConnectorColumnHandle> columns, + Map<String, String> backendStorageProperties, boolean usesFallbackRead, + boolean incrementalRead, boolean projectedVariant) { + this.table = table; + this.backendStorageProperties = backendStorageProperties; + CoreOptions options = table.coreOptions(); + CoreOptions.MergeEngine mergeEngine = options.mergeEngine(); + partialUpdateOrAggregateWithDeletionVectors = options.deletionVectorsEnabled() + && (mergeEngine == CoreOptions.MergeEngine.PARTIAL_UPDATE + || mergeEngine == CoreOptions.MergeEngine.AGGREGATE); + TableSchema schema = table.schema(); + String mergeOnRead = schema.options().get(DELETION_VECTORS_MERGE_ON_READ); + deletionVectorsMergeOnRead = "true".equalsIgnoreCase(mergeOnRead); + + String location = table.location().toString(); + boolean enabled = session != null && "true".equalsIgnoreCase( + session.getSessionProperties().get(ENABLE_PAIMON_RUST_READER)); + boolean fileScannerV2 = session != null && "true".equalsIgnoreCase( + session.getSessionProperties().get(ENABLE_FILE_SCANNER_V2)); + boolean forceJniScanner = session != null && "true".equalsIgnoreCase( + session.getSessionProperties().get(FORCE_JNI_SCANNER)); + boolean restTokenTable = table.fileIO() instanceof RESTTokenFileIO; + boolean unsupportedMergeOption = (mergeEngine == CoreOptions.MergeEngine.PARTIAL_UPDATE + || mergeEngine == CoreOptions.MergeEngine.AGGREGATE) + && hasRustUnsupportedMergeOption(schema.options(), mergeEngine); + boolean deduplicateIgnoreDelete = mergeEngine == CoreOptions.MergeEngine.DEDUPLICATE + && options.ignoreDelete(); + String scanMode = PaimonScanParams.withoutTimeTravelSelectors(schema) + .options().get(CoreOptions.SCAN_MODE.key()); + boolean scanModeSupported = scanMode == null || "default".equalsIgnoreCase(scanMode); + + scanCompatible = enabled && fileScannerV2 && !forceJniScanner + && backendsSupportRust + && !usesFallbackRead && !incrementalRead && !projectedVariant + && !options.queryAuthEnabled() && !restTokenTable + && !(partialUpdateOrAggregateWithDeletionVectors && deletionVectorsMergeOnRead) + && !deduplicateIgnoreDelete + && !unsupportedMergeOption && scanModeSupported + && isRustVerifiedLocationScheme(location) + && (!isHdfsLocationScheme(location) + || isRustVerifiedHdfsBackend(backendStorageProperties)) + && isCredentialProviderTranslatable(location, backendStorageProperties); + capabilities = new PaimonRustReaderCapabilities(table, columns); + } + + boolean canRead(DataSplit split) { + if (!scanCompatible || splitHasExternalFiles(split)) { + return false; + } + if (splitHasOrcFile(split) && table.schema().fields().stream() Review Comment: [P2] Preserve the full range of low-precision ORC timestamps. This guard rejects only LTZ, so an append-only ORC table with current-schema TIMESTAMP(3) or TIMESTAMP(6) and a value such as 3000-01-01 still selects Rust. The pinned Paimon OrcFormatReader leaves orc-rust 0.8.0 at its default nanosecond precision; that decoder converts epoch nanoseconds to i64 and errors outside roughly 1677-2262, before any cast to the table's millisecond/microsecond unit. Enabling Rust therefore makes a valid scan fail although JNI can read it. This is a distinct remaining instance from the existing TIMESTAMP(7-9)/Parquet INT96 thread: the <=6 schema guard cannot cover it. Decode ORC at a suitable unit or conservatively route these timestamp schemas to JNI, including nested leaves, and add an out-of-range ORC parity case. The pinned [ORC builder contract](https://docs.rs/orc-rust/0.8.0/orc_rust/arrow_reader/struct.ArrowReaderBuilder.html#method.with_timestamp_precision) documents this default. ########## be/src/format_v2/table/paimon_rust_table_reader.cpp: ########## @@ -0,0 +1,944 @@ +// 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 rust_handles { + +// --------------------------------------------------------------------------- +// 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 + +} // namespace rust_handles + +namespace { +constexpr const char* VALUE_KIND_FIELD = "_VALUE_KIND"; + +// 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) { + rust_handles::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 { + rust_handles::table_ptr table; + rust_handles::read_builder_ptr read_builder; + rust_handles::plan_ptr plan; + rust_handles::table_read_ptr table_read; + rust_handles::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); + // Open may dominate the scan or fail before get_block; include it in the parent timer. + SCOPED_TIMER(_rust_total_time); + 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(); + } + + if (!_pending_record_batch) { + paimon_result_next_batch next; + { + SCOPED_TIMER(_rust_read_batch_time); + next = paimon_record_batch_reader_next(_handles->reader.get()); + } + if (next.error != nullptr) { Review Comment: [P2] Preserve missing-data-file status on the Rust path. Rust now precedes the native reader for eligible ordinary append-only splits. If a single-file Parquet split's data file disappears after planning, Rust's storage error is returned here as INTERNAL_ERROR. FileScannerV2 skips only NOT_FOUND, so ignore_not_found_file_in_external_table (true by default, and applicable to Paimon) stops working; the same split on the native route preserves ENOENT as NOT_FOUND and follows that policy. Preserve a distinguishable missing-data-file error through the Rust binding and map it to the scanner's expected status, without treating other IO/schema errors as missing files. Add native/Rust differential coverage with the file removed after planning and both ignore settings. ########## fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanPlanProvider.java: ########## @@ -897,6 +911,16 @@ && hasTrustworthyLimitAccounting(table) } } + boolean incrementalRead = pinnedOptions.containsKey("incremental-between") + || pinnedOptions.containsKey("incremental-between-timestamp"); + PaimonRustReaderSelector rustReaderSelector = table instanceof FileStoreTable + && !requiresMetadataColumns + ? new PaimonRustReaderSelector(session, backendsSupportRust, + (FileStoreTable) table, columns, backendStorageProperties(), Review Comment: [P2] Check and transport the effective Paimon storage configuration. This selector sees only the typed backend map, but PaimonCatalogFactory.applyStorageConfig applies raw fs.s3a.* and paimon.s3.* overrides after that map when constructing Java FileIO. For example, canonical s3.access_key/secret_key can name pair A while fs.s3a.access.key/secret.key override it with readable pair B. A non-raw append-only data-evolution split with INT columns previously uses JNI and retains B in its serialized FileIO; it now passes Rust's static-key gate using A. The Rust payload never carries the raw overlays and its schema-json FileIO constructor does not recover them, so the scan uses A and fails access where JNI succeeds. Preserve the effective option precedence in the Rust transport/eligibility check, or fall back for untranslatable overrides; add a differential catalog case. ########## be/src/format_v2/table/paimon_rust_table_reader.cpp: ########## @@ -0,0 +1,944 @@ +// 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 rust_handles { + +// --------------------------------------------------------------------------- +// 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 + +} // namespace rust_handles + +namespace { +constexpr const char* VALUE_KIND_FIELD = "_VALUE_KIND"; + +// 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) { + rust_handles::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 { + rust_handles::table_ptr table; + rust_handles::read_builder_ptr read_builder; + rust_handles::plan_ptr plan; + rust_handles::table_read_ptr table_read; + rust_handles::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); + // Open may dominate the scan or fail before get_block; include it in the parent timer. + SCOPED_TIMER(_rust_total_time); + 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(); + } + + if (!_pending_record_batch) { + 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); + arrow::Result<std::shared_ptr<arrow::RecordBatch>> import_result = + arrow::ImportRecordBatch(batch.array(), batch.schema()); + if (!import_result.ok()) { + return Status::InternalError("failed to import paimon-rust arrow batch: {}", + import_result.status().message()); + } + _pending_record_batch = std::move(import_result).ValueUnsafe(); + _pending_record_batch_offset = 0; + if (_pending_record_batch->num_rows() == 0) { + _pending_record_batch.reset(); + continue; + } + } + + const size_t batch_size = static_cast<size_t>( + _batch_size > 0 ? _batch_size : std::max(1, _runtime_state->batch_size())); + const size_t remaining = static_cast<size_t>(_pending_record_batch->num_rows()) - + _pending_record_batch_offset; + const size_t rows = std::min(batch_size, remaining); + RETURN_IF_ERROR(_fill_block_from_record_batch(_pending_record_batch, block, + _pending_record_batch_offset, rows)); + _pending_record_batch_offset += rows; + if (_pending_record_batch_offset == + static_cast<size_t>(_pending_record_batch->num_rows())) { + _pending_record_batch.reset(); + _pending_record_batch_offset = 0; + } + _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. + // Bound the Rust Arrow allocation itself: wide rows must respect the scanner's + // adaptive probe size before they are materialized into a Doris block. + const auto batch_size = std::to_string( + _batch_size > 0 ? _batch_size : std::max(1, _runtime_state->batch_size())); + const paimon_option batch_option {"read.batch-size", batch_size.c_str()}; + paimon_result_read_builder rb_res = + paimon_table_new_read_builder_with_options(_handles->table.get(), &batch_option, 1); + 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() { + _pending_record_batch.reset(); + _pending_record_batch_offset = 0; + 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(); + } + if (_scanner_profile != nullptr) { + COUNTER_UPDATE(_rust_predicates_input, _conjuncts.size()); + for (const auto& conjunct : _conjuncts) { + if (conjunct && conjunct->root() && conjunct->root()->is_rf_wrapper()) { + COUNTER_UPDATE(_rust_runtime_filters_input, 1); + } + } + } + VLOG_DEBUG << "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 (_scanner_profile != nullptr) { + COUNTER_UPDATE(_rust_predicates_converted, converter.converted_conjuncts()); + } + if (predicate == nullptr) { + VLOG_DEBUG << "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)); + } + // Count application only after the C API accepts the filter; reader timers + // alone cannot distinguish pushdown from Doris residual-only execution. + if (_scanner_profile != nullptr) { + COUNTER_UPDATE(_rust_predicates_applied, converter.converted_conjuncts()); + COUNTER_UPDATE(_rust_runtime_filters_applied, converter.converted_runtime_filters()); + } + VLOG_DEBUG << "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 start, size_t rows) { + SCOPED_TIMER(_rust_arrow_to_block_time); + DORIS_CHECK(batch != nullptr); + DORIS_CHECK(block != nullptr); + DORIS_CHECK_LE(start + rows, static_cast<size_t>(batch->num_rows())); + const auto arrow_start = static_cast<int64_t>(start); + const auto arrow_end = static_cast<int64_t>(start + rows); + 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], Review Comment: [P2] Handle sliced BOOLEAN arrays produced by data evolution. Eligible data-evolution.enabled append-only tables can produce sliced arrays inside Rust's column-wise merge. For overlapping partial-column files whose Parquet row groups have different boundaries (for example source A first emits 17 rows while a BOOLEAN source B emits 32), DataEvolutionReader emits 17 rows and then B.slice(17,15). Arrow 58.4 preserves the BOOLEAN bit offset in ArrayData and the C Data Interface, so this direct SerDe call fails with non-zero array offset is not supported under the default enable_arrow_input_validation=true. Keeping the original imported batch only avoids slices made by this C++ reader; it does not normalize Rust's incoming bit buffers. Normalize these BOOLEAN slices or keep affected data-evolution scans on JNI, and add a differential fixture with mismatched source batch boundaries. See [BOOLEAN export offset](https://github.com/apache/arrow-rs/blob/58.4.0/arrow-array/src/array/boolean_ array.rs#L745-L752). -- 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]
