github-actions[bot] commented on code in PR #66227: URL: https://github.com/apache/doris/pull/66227#discussion_r4110145552
########## 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()); Review Comment: [P1] Make the blocking Rust read cancellable. This checks should_stop only before entering paimon_record_batch_reader_next, while the pinned C function synchronously block_on(stream.next()) and exposes no cancellation token or polling handle. If S3/HDFS stalls here, try_stop cannot reach the reader or _close_split_reader, so a cancelled query keeps its scanner thread and remote I/O alive until the request times out. Propagate a Doris cancellation token into the Rust stream/FileIO (or add an interruptible poll API), and cover cancellation while next is blocked. ########## fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/source/PaimonScanNode.java: ########## @@ -411,10 +681,298 @@ private void setPaimonParams(TFileRangeDesc rangeDesc, PaimonSplit paimonSplit) String fileFormat = getFileFormat(paimonSplit.getPathString()); if (split != null) { + // use jni reader / paimon-cpp reader / paimon-rust reader rangeDesc.setFormatType(TFileFormatType.FORMAT_JNI); - // A logical DataSplit may span multiple files, so keep it intact for the JNI reader. - fileDesc.setReaderType(TPaimonReaderType.PAIMON_JNI); - fileDesc.setPaimonSplit(PaimonUtil.encodeObjectToString(split)); + // paimon-cpp and paimon-rust both consume Paimon native binary serialization, + // which only supports DataSplit. Any other split type falls back to JNI. + boolean nativeSplit = split instanceof DataSplit; + // Fallback-read splits stay on JNI: FallbackDataSplit extends + // DataSplit, so the instanceof above passes, but its serializer + // appends an isFallback byte after the ordinary split that the + // pinned rust decoder rejects outright ("trailing bytes after + // DataSplit" — it requires full-buffer consumption), and even a + // permissive decode would still lack the second table identity + // needed to honor the fallback-side discriminator. Both sides of a + // FallbackReadFileStoreTable wrap their splits, so the table + // wrapper is gated as a whole (any split from it routes to JNI) + // until the rust ABI represents both sides; the FallbackSplit + // interface also catches a wrapper split regardless of how the + // table was resolved here. + boolean fallbackRead = split instanceof FallbackReadFileStoreTable.FallbackSplit + || processedTable instanceof FallbackReadFileStoreTable; + // Serialize the same effective table that planning and the JNI reader use. + // Relation options such as t@options('read.batch-size'='1') are applied by + // getProcessedTable() (doInitialize caches it in processedTable), and the + // rust reader derives its read batch size from the schema options — the raw + // cached table would silently drop the override. Copies, delegates and + // fallback wrappers of getProcessedTable() are still FileStoreTable, so the + // instanceof gate keeps its semantics. + Table paimonTable = processedTable; + FileStoreTable paimonFileStoreTable = + paimonTable instanceof FileStoreTable ? (FileStoreTable) paimonTable : null; + // query-auth.enabled tables stay on JNI: when catalog authorization + // succeeds with no row filter or column mask, Paimon still leaves an + // ordinary DataSplit (restricted results use QueryAuthSplit and are + // already handled by the nativeSplit gate above), so this table shape + // passes the compound gate — but the shipped schema keeps + // query-auth.enabled=true and the pinned rust ReadBuilder rejects + // every such table (its CoreOptions::ensure_read_authorized fails + // closed because the client cannot enforce the row filter / column + // masking), turning a valid authorized scan into a BE-open failure. + // Until the authorization result can be transported and enforced by + // the rust ABI, these tables route to JNI. + boolean queryAuthTable = false; + // Partial-update / aggregation tables with deletion vectors only pass + // the rust reader in the fully materialized shape: the pinned rust + // read_pk rejects merge-engine=partial-update/aggregation with + // deletion-vectors.merge-on-read=true outright, and otherwise requires + // every split to be compacted and known free of retract rows + // (DataSplit::is_fully_materialized_pk_dv). Their ordinary DataSplits + // sail through the compound gate above, so without this check a valid + // Java/JNI scan reaches BE and the rust open fails. Deduplicate stays + // rust-eligible: its read_pk routes uncompacted splits to the KV + // reader, which applies the attached per-file DVs. merge-on-read=true + // is a table option, so the whole table routes to JNI; + // non-materialized splits are gated per split below. + boolean puAggDeletionVectors = false; + boolean dvMergeOnRead = false; + boolean deduplicateIgnoreDelete = false; + boolean rustUnsupportedMergeOption = false; + if (paimonFileStoreTable != null) { + CoreOptions resolvedCoreOptions = paimonFileStoreTable.coreOptions(); + // Null-safe: a table handle whose CoreOptions is not resolved + // (e.g. some wrapper shapes) stays rust-eligible rather than + // failing the scan here — the rust open itself rejects such a + // table if the option is really set. + if (resolvedCoreOptions != null) { + queryAuthTable = resolvedCoreOptions.queryAuthEnabled(); + CoreOptions.MergeEngine mergeEngine = resolvedCoreOptions.mergeEngine(); + if (resolvedCoreOptions.deletionVectorsEnabled() + && (mergeEngine == CoreOptions.MergeEngine.PARTIAL_UPDATE + || mergeEngine == CoreOptions.MergeEngine.AGGREGATE)) { + puAggDeletionVectors = true; + // The merge-engine and deletion-vectors.enabled checks + // above resolve through the Java CoreOptions accessors, + // which the table builds from this same schema options + // map — the one the BE rust reader deserializes from + // the shipped schema JSON — so they cannot diverge from + // what BE sees. merge-on-read has no Java accessor in + // paimon 1.4, so it is read raw from the map, with the + // rust parsing semantics (any case-insensitive "true" + // is on, default false). + TableSchema dvSchema = paimonFileStoreTable.schema(); + Map<String, String> dvOptions = dvSchema == null ? null : dvSchema.options(); + String mergeOnRead = dvOptions == null + ? null : dvOptions.get(DELETION_VECTORS_MERGE_ON_READ); + dvMergeOnRead = "true".equalsIgnoreCase(mergeOnRead); + } + // deduplicate.ignore-delete=true tables stay on JNI: + // Java's DeduplicateMergeFunction skips retract records + // when the option is set — including old, uncompacted + // files that still contain them — but the pinned rust + // read_pk does not pass table options into its + // deduplicate merge: it picks the latest row and omits + // the key when that row is DELETE/UPDATE_BEFORE. An + // uncompacted insert followed by a delete therefore + // returns the insert through JNI but silently disappears + // through rust. Gate the option until the rust merge + // implements it. + if (mergeEngine == CoreOptions.MergeEngine.DEDUPLICATE + && resolvedCoreOptions.ignoreDelete()) { + deduplicateIgnoreDelete = true; + } + // Non-DV merge options the pinned rust read rejects: Java + // supports partial-update.remove-record-on-delete / + // aggregation.remove-record-on-delete and the wider + // per-field retract matrix, but the rust + // PartialUpdateConfig / AggregationConfig validations + // return Unsupported for them — and the DV-derived gates + // above only cover deletion-vector tables, so an ordinary + // non-DV DataSplit with one of these options would pass the + // compound gate and fail during the rust merge + // construction. Mirror the exact rust key matrix (presence, + // not values) against the same schema options map BE + // deserializes. + if (mergeEngine == CoreOptions.MergeEngine.PARTIAL_UPDATE + || mergeEngine == CoreOptions.MergeEngine.AGGREGATE) { + TableSchema mergeSchema = paimonFileStoreTable.schema(); + Map<String, String> mergeOptions = + mergeSchema == null ? null : mergeSchema.options(); + rustUnsupportedMergeOption = mergeOptions != null + && hasRustUnsupportedMergeOption(mergeOptions, mergeEngine); + } + } + } + // paimon-rust additionally requires (a) FileScannerV2: the V1 FileScanner + // explicitly rejects PAIMON_RUST, so with enable_file_scanner_v2 disabled + // the split falls back to JNI instead of encoding a rust request that the + // selected scanner cannot consume, and (b) a FileStoreTable: BE opens the + // table via paimon_table_from_schema_json, which needs the resolved + // TableSchema that only FileStoreTable exposes via schema(). If the table + // is not a FileStoreTable (e.g. a sys table backed by DataSplit), we cannot + // ship a schema JSON, so fall back to CPP / JNI rather than sending an + // incomplete PAIMON_RUST request that BE would reject. + // + // The paimon-rust S3 bridge maps static credentials, anonymous + // access (AWS_CREDENTIALS_PROVIDER_TYPE=ANONYMOUS -> s3.anonymous) + // and assume-role (AWS_ROLE_ARN / AWS_EXTERNAL_ID -> + // s3.assumed.role.*), but the remaining credential-provider modes + // are ambient JVM provider chains (ENV, SYSTEM_PROPERTIES, + // WEB_IDENTITY, CONTAINER, INSTANCE_PROFILE) with no paimon-rust + // equivalent — rust would silently sign with whatever the ambient + // chain resolves to. Gate those modes away from the rust reader + // here so the configured provider is honored via the JNI path. + boolean providerModeTranslatable = true; + String providerType = backendStorageProperties == null + ? null : backendStorageProperties.get("AWS_CREDENTIALS_PROVIDER_TYPE"); + if (providerType != null) { + String mode = providerType.trim().toUpperCase(Locale.ROOT); + providerModeTranslatable = mode.equals("DEFAULT") + || mode.equals("ANONYMOUS"); + // The rust OSS FileIO parser (oss:// warehouses) has no + // skip-signature switch, so an anonymous OSS catalog cannot be + // served by the rust reader either — fall back to JNI. + if (mode.equals("ANONYMOUS")) { + String location = source.getTableLocation(); + if (location != null && location.startsWith("oss://")) { + providerModeTranslatable = false; + } + } + } + // Incremental scans (binlog / changelog / delta / diff) must stay + // on the JNI path: this wire format carries only an ordinary + // DataSplit and the rust reader invokes TableRead::to_arrow, but + // paimon 1.4 marks incremental splits as streaming (which the + // pinned rust deserializer rejects), diff requires a separate + // IncrementalPlan instead of an ordinary plan, and ordinary + // primary-key reads can merge versions rather than return the + // changes — until the C ABI transports the mode and plan, the + // rust reader cannot express any of these. + TableScanParams incrementalParams = getScanParams(); + boolean isIncremental = incrementalParams != null && incrementalParams.incrementalRead(); + // ORC TIMESTAMP_WITH_LOCAL_TIME_ZONE schemas stay on JNI: the pinned + // paimon-rust ORC decoder materializes LTZ instants shifted by the + // writer timezone (an upstream crate limitation), so a logical ORC + // DataSplit that selects rust (e.g. with force_jni_scanner=true or + // when raw conversion is unavailable) returns a different instant + // than JNI — applying the session timezone in BE cannot repair an + // epoch already shifted during decode. Two bypasses are covered: + // (a) the format must come from EVERY member file — paimon allows + // per-level file.format, so one DataSplit can mix Parquet and ORC + // files and the split path's suffix (the first file) would hide + // the ORC members; (b) the LTZ search must recurse into nested + // types — an LTZ under MAP/ARRAY/ROW reaches the same shifted ORC + // decode through the container's field materialization. Parquet + // files with any LTZ, and ORC without any recursive LTZ, stay + // rust-eligible. nativeSplit only guards the cast — non-DataSplit + // splits already route to JNI. + boolean orcLtzSchema = paimonFileStoreTable != null + && nativeSplit + && splitHasOrcFile((DataSplit) split) + && paimonFileStoreTable.schema().fields().stream() + .anyMatch(field -> containsTimestampLtz(field.type())); + // data-file.external-paths splits stay on JNI (see + // splitHasExternalFiles): the rust table's single FileIO cannot + // serve an external file's backend. nativeSplit only guards the + // cast — non-DataSplit splits already route to JNI. + boolean externalFileSplit = nativeSplit && splitHasExternalFiles((DataSplit) split); + // Projected VARIANT columns stay on JNI: the rust leaf feeds its + // Arrow arrays to the slot serdes, and DataTypeVariantV2SerDe:: + // read_column_from_arrow unconditionally returns + // NOT_IMPLEMENTED_ERROR — a nested Variant (ARRAY / MAP / STRUCT + // containing one) reaches the same decoder through the container + // serdes. desc carries only the slots this query projects, so a + // table whose VARIANT column is not projected still scans on + // rust. Gate until the rust leaf has a Variant Arrow decoder. + boolean projectedVariant = desc.getSlots().stream() + .anyMatch(slot -> PaimonUtil.containsVariant(slot.getType())); + // Scheme capability gate: the pinned paimon-rust storage + // dispatcher (io/storage.rs) selects the FileIO parser from the + // table location's URI scheme, and libpaimon_c.a compiles in + // separate COS, OBS, GCS and Azdls parsers besides the OSS and S3 + // ones. Doris normalizes every object store's credentials into + // the AWS_* / use_path_style aliases (see the *Properties storage + // classes), which the BE rust bridge translates only into the + // fs.oss.* and s3.* key families — a cosn:// / obs:// / gs:// / + // abfs:// warehouse would reach its scheme's parser without the + // key family it reads (fs.cosn.userinfo.*, fs.obs.*, gcs.*, + // azure.*) and fail the open instead of using JNI. Only the + // schemes whose property translation is implemented and + // open-tested (s3 / s3a / oss, via RUST_VERIFIED_LOCATION_SCHEMES) + // plus the credential-free hdfs and local-filesystem parsers stay + // rust-eligible; every other scheme falls back to JNI. A null + // location also routes to JNI: the rust path needs the + // paimon_table that only a real location can provide (BE rejects + // a split without it). + boolean schemeCapabilityVerified = isRustVerifiedLocationScheme(source.getTableLocation()); + // An hdfs:// location is scheme-verified only together with the credential-free + // backend shape: the backend storage properties that ship to BE also carry an + // HDFS catalog's authentication (kerberos principal / keytab, proxy user, HA + // nameservice config), none of which the pinned rust HDFS parser reads — the + // scan would open as the BE process's ambient identity instead of the + // catalog's configured one and fail the access JNI honors. See + // isRustVerifiedHdfsBackend. + boolean hdfsBackendVerified = !isHdfsLocationScheme(source.getTableLocation()) + || isRustVerifiedHdfsBackend(backendStorageProperties); + // With merge-on-read=true the whole table already routes to JNI (dvMergeOnRead); + // for the remaining partial-update/aggregation DV tables, a split that is + // not fully materialized (uncompacted level-0 data, or retractions not + // known to be applied — even a split with no deletion file attached yet) + // fails the rust is_fully_materialized_pk_dv guard, so it falls back per + // split instead of turning into a BE-open failure. nativeSplit and + // !fallbackRead only guard the cast — those splits already route to JNI. + boolean splitDvNotMaterialized = puAggDeletionVectors && !dvMergeOnRead + && nativeSplit && !fallbackRead + && !isFullyMaterializedPkDvSplit((DataSplit) split); + boolean canUseRust = sessionVariable.isEnablePaimonRustReader() + && sessionVariable.enableFileScannerV2 && nativeSplit && !fallbackRead + && !isIncremental && providerModeTranslatable && !queryAuthTable Review Comment: [P1] Keep renewable REST-token tables on JNI. doInitialize snapshots RESTTokenFileIO.validToken().token() into backendStorageProperties, discarding expireAtMillis and the REST refresh context, but this gate treats those temporary S3/OSS credentials like static ones. The Rust table then reuses one option map and has no refresh callback. Paimon 1.4.2's JNI RESTTokenFileIO checks expiry before each file operation and obtains a replacement token, so a queued or long scan that crosses the token TTL now starts on Rust and later fails authentication. Gate RESTTokenFileIO tables to JNI until the Rust ABI can refresh and atomically update credentials, and cover a scan across a short token lifetime. ########## 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); Review Comment: [P2] Publish Rust physical I/O through the shared accounting path. The Rust table owns its FileIO outside Doris, and after a batch this records only rows; read_bytes/read_calls/read_time_ns and FileCacheStatistics stay zero. FileScannerV2 derives ScanBytes, query ResourceContext usage, FileRead counters, local/remote attribution, and global metrics solely from those structures, so a large S3/HDFS Rust scan is reported as consuming zero bytes. Expose monotonic per-stream I/O/source metrics (or a Doris callback adapter) and flush them at batch/error/EOF boundaries. ########## fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/source/PaimonScanNode.java: ########## @@ -333,6 +357,252 @@ private void setScanLevelPaimonOptions() { } } + /** + * Whether the table location's URI scheme is served by a paimon-rust FileIO parser whose + * property translation the FE/BE bridge implements (the s3.* / fs.oss.* key families for + * s3 / s3a / oss) or that needs no credentials at all (hdfs hadoop conf and local + * filesystem paths). Every other scheme the pinned crate dispatches to its own parser + * (cosn / obs / gs / abfs and friends) must fall back to the JNI reader because Doris + * delivers those credentials only as AWS_* aliases that those parsers do not read. + * A null location cannot be verified (and cannot ship paimon_table either), so it is + * not rust-eligible. + * + * <p>The scheme must also appear in the exact lowercase form the pinned crate consumes. + * URI schemes are case-insensitive, but the crate lowercases only its storage + * dispatch: its object-store path extraction strips a lowercase {@code s3://} prefix + * from the original string, and the hdfs / file helpers likewise match only lowercase + * prefixes. A {@code S3://} or {@code Hdfs://} warehouse also produces DataSplit file + * paths in that original casing (serialized by the paimon SDK before the FE sees + * them), so the mixed-case shape fails the rust open beyond the transported location. + * Those valid URI variants must therefore route to JNI, whose Java stack is + * case-insensitive everywhere. + */ + @VisibleForTesting + static boolean isRustVerifiedLocationScheme(String location) { + if (location == null) { + return false; + } + int sep = location.indexOf("://"); + if (sep <= 0) { + // No URI scheme: a plain local path reads through the crate's local-filesystem + // parser, which needs no credentials. + return true; + } + return RUST_VERIFIED_LOCATION_SCHEMES.contains(location.substring(0, sep)); + } + + // Whether the location is an hdfs:// table (the only HDFS-family scheme in + // RUST_VERIFIED_LOCATION_SCHEMES; viewfs / jfs never pass it). + private static boolean isHdfsLocationScheme(String location) { + if (location == null) { + return false; + } + int sep = location.indexOf("://"); + return sep > 0 && "hdfs".equalsIgnoreCase(location.substring(0, sep)); + } + + // Whether any member file of the split is ORC. Paimon allows per-level + // file.format, so one DataSplit can mix Parquet and ORC files; the split + // path's suffix (the first file) cannot speak for the whole split, and the + // shifted ORC LTZ decode applies to whichever ORC members rust reads. + @VisibleForTesting + static boolean splitHasOrcFile(DataSplit dataSplit) { + if (dataSplit == null) { + return false; + } + for (DataFileMeta fileMeta : dataSplit.dataFiles()) { + String format = fileMeta.fileFormat(); + if (format != null && "orc".equalsIgnoreCase(format)) { + return true; + } + } + return false; + } + + // Mirrors the pinned paimon-rust read-mode option matrix + // (PartialUpdateConfig::read_unsupported_option_keys and + // AggregationConfig's runtime-unsupported keys). Both validate option-key + // PRESENCE, not values, so a table carrying the key with an off value is + // still rejected by the rust merge construction while Java reads it — the + // gate must mirror presence exactly. + // + // Partial-update reads support basic mode, sequence groups and field + // aggregation; unsupported keys are the remove-record-on-delete family, + // per-field ignore-delete / ignore-retract / distinct / nested-key / + // count-limit options. + private static boolean isRustUnsupportedPartialUpdateReadOption(String key) { + return (key.endsWith(".ignore-delete") + && !"ignore-delete".equals(key) + && !"partial-update.ignore-delete".equals(key)) + || "partial-update.remove-record-on-delete".equals(key) + || "partial-update.remove-record-on-sequence-group".equals(key) + || hasFieldOptionSuffix(key, ".ignore-retract") + || hasFieldOptionSuffix(key, ".distinct") + || hasFieldOptionSuffix(key, ".nested-key") + || hasFieldOptionSuffix(key, ".count-limit"); + } + + // Aggregation reads support the per-field aggregate-function / + // list-agg-delimiter / default-aggregate-function matrix; unsupported keys + // are the remove-record-on-delete family, every ignore-delete spelling + // (including the bare one), and per-field sequence-group / ignore-retract / + // distinct / nested-key / count-limit options. + private static boolean isRustUnsupportedAggregationRuntimeOption(String key) { + return "ignore-delete".equals(key) + || key.endsWith(".ignore-delete") + || "aggregation.remove-record-on-delete".equals(key) + || hasFieldOptionSuffix(key, ".sequence-group") + || hasFieldOptionSuffix(key, ".ignore-retract") + || hasFieldOptionSuffix(key, ".distinct") + || hasFieldOptionSuffix(key, ".nested-key") + || hasFieldOptionSuffix(key, ".count-limit"); + } + + private static boolean hasFieldOptionSuffix(String key, String suffix) { + return key.startsWith("fields.") && key.endsWith(suffix); + } + + // Whether the schema options carry any option key the pinned paimon-rust + // read rejects for this merge engine. + @VisibleForTesting + static boolean hasRustUnsupportedMergeOption(Map<String, String> options, + CoreOptions.MergeEngine mergeEngine) { + boolean partialUpdate = mergeEngine == CoreOptions.MergeEngine.PARTIAL_UPDATE; + for (String key : options.keySet()) { + if (key == null) { + continue; + } + if (partialUpdate ? isRustUnsupportedPartialUpdateReadOption(key) + : isRustUnsupportedAggregationRuntimeOption(key)) { + return true; + } + } + return false; + } + + // Whether any member file of the split carries a data-file.external-paths + // location. Paimon can store an absolute location in each DataFileMeta, and + // both Java and the serialized rust split prefer it over the bucket path — + // but the pinned rust table builds ONE FileIO from paimon_table, whose + // storage enum parses every file with that warehouse-selected backend: an + // admitted hdfs table with an s3:// external file (or an s3 table with an + // oss:// file) reaches the wrong parser and fails the open, while JNI + // reads it. The shipped options describe only the warehouse, so any + // external file keeps the split on JNI. + @VisibleForTesting + static boolean splitHasExternalFiles(DataSplit dataSplit) { + if (dataSplit == null) { + return false; + } + for (DataFileMeta fileMeta : dataSplit.dataFiles()) { + if (fileMeta.externalPath().isPresent()) { + return true; + } + } + return false; + } + + // Recursively whether this paimon type, or any member of it, is + // TIMESTAMP_WITH_LOCAL_TIME_ZONE: an LTZ nested under MAP/ARRAY/ROW + // reaches the same shifted ORC decode through the container's field + // materialization. + @VisibleForTesting + static boolean containsTimestampLtz(DataType type) { + if (type == null) { + return false; + } + if (type.getTypeRoot() == DataTypeRoot.TIMESTAMP_WITH_LOCAL_TIME_ZONE) { + return true; + } + if (type instanceof ArrayType) { + return containsTimestampLtz(((ArrayType) type).getElementType()); + } + if (type instanceof MapType) { + MapType mapType = (MapType) type; + return containsTimestampLtz(mapType.getKeyType()) + || containsTimestampLtz(mapType.getValueType()); + } + if (type instanceof RowType) { + return ((RowType) type).getFields().stream() + .anyMatch(field -> containsTimestampLtz(field.type())); + } + return false; + } + + /** + * Whether an HDFS catalog's shipped backend properties carry no identity the + * pinned paimon-rust HDFS parser would silently drop. The parser reads only + * the hdfs.name-node / hdfs.enable-append keys: no kerberos, no proxy user, + * no hadoop HA resolution. A Kerberized catalog (or one with hadoop.username, + * or with HA nameservice config) would pass the location scheme gate, open + * as the BE process's ambient identity and fail — or silently miss — the + * catalog's configured access, while the same query works through JNI. Only + * the open-tested credential-free shape stays rust-eligible: simple (or + * unset) authentication, no principal / keytab / proxy user, no HA + * nameservice resolution. Generic fs.* / dfs.* tunables still pass through + * untouched — fs.defaultFS always ships and carries no identity. + */ + @VisibleForTesting + static boolean isRustVerifiedHdfsBackend(Map<String, String> backendStorageProperties) { + if (backendStorageProperties == null) { + return true; + } + // Authentication type: "simple" (or unset) authenticates as the same + // ambient OS user on both readers; kerberos (or anything else) needs the + // channel the rust parser does not read. + for (String authKey : new String[] {"hadoop.security.authentication", + "hdfs.security.authentication"}) { + String value = backendStorageProperties.get(authKey); + if (value != null && !"simple".equalsIgnoreCase(value.trim())) { + return false; + } + } + // Kerberos identity and the proxy user: any of these configured means + // the open must carry a specific identity, which only JNI can honor. + for (String identityKey : new String[] {"hadoop.kerberos.principal", + "hadoop.kerberos.keytab", "hadoop.username"}) { + String value = backendStorageProperties.get(identityKey); + if (value != null && !value.trim().isEmpty()) { + return false; + } + } + // HA nameservice resolution: the rust parser receives no dfs.* config, + // so a nameservice-authority location (dfs.nameservices / dfs.ha.*) + // cannot be resolved; the proven shape is a single name-node URI. + String nameServices = backendStorageProperties.get("dfs.nameservices"); + if (nameServices != null && !nameServices.trim().isEmpty()) { + return false; + } + for (Map.Entry<String, String> entry : backendStorageProperties.entrySet()) { + if (entry.getKey() != null && entry.getKey().startsWith("dfs.ha.") + && entry.getValue() != null && !entry.getValue().trim().isEmpty()) { + return false; + } + } + return true; Review Comment: [P1] Keep HDFS client-option catalogs on JNI. This returns true for a simple-auth catalog carrying required dfs.* settings, but the pinned Rust storage_hdfs parser copies only hdfs.name-node/enable-append and leaves HdfsNativeConfig.options empty. For example, dfs.client.use.datanode.hostname=true is transported by HdfsProperties and is valid without Kerberos; hdfs-native otherwise defaults false and connects to the DataNode's advertised IP instead of its hostname, which commonly fails behind containers/NAT while JNI succeeds. The existing auth/HA exclusions do not catch this shape. Until Rust forwards the client option map, reject nontrivial dfs.*/hadoop.* settings (or use a proven-safe allowlist) and cover this simple-auth routing case. -- 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]
