github-actions[bot] commented on code in PR #66227:
URL: https://github.com/apache/doris/pull/66227#discussion_r4092336686
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonScanParams.java:
##########
@@ -403,6 +426,34 @@ public static Map<String, String> isolateSnapshotRead(long
snapshotId) {
return isolatedOptions;
}
+ /**
+ * Strips the time-travel planning selectors — and any derived scan mode
Paimon
+ * materialized for them — from a resolved schema's options, keeping every
+ * field and every other option intact. This is the transport-side inverse
of
+ * {@link #isolateSnapshotRead(long)}: a statement fence pins the data
snapshot by merging
+ * {@code scan.snapshot-id} into the schema options, but the pinned
paimon-rust reader
+ * re-resolves a transported selector inside {@code copy_with_time_travel}
and swaps the
+ * shipped fields for the pinned snapshot's older schema — a column added
after the last
+ * data commit then fails projection before per-file schema evolution can
fill it. The rust
+ * reader instead pins data through the serialized DataSplit, so these
selectors are
+ * planning-only state and must not be transported.
+ */
+ public static TableSchema withoutTimeTravelSelectors(TableSchema schema) {
+ if
(TIME_TRAVEL_SELECTOR_KEYS.stream().noneMatch(schema.options()::containsKey)) {
+ return schema;
+ }
+ Map<String, String> options = new HashMap<>(schema.options());
+ TIME_TRAVEL_SELECTOR_KEYS.forEach(options.keySet()::remove);
+ // Paimon's copyInternal may have materialized the derived scan mode
for the
+ // stripped selector; without the selector the rust ReadBuilder
validation
+ // ("from-snapshot requires one of scan.snapshot-id, ... to be set")
rejects
+ // the open. The reader instead pins data through the serialized
DataSplit.
+ if
(SELECTOR_DEPENDENT_SCAN_MODES.contains(options.get(CoreOptions.SCAN_MODE.key())))
{
Review Comment:
[P1] Strip selector-derived scan modes case-insensitively. Paimon accepts
enum option values case-insensitively but preserves the original string in
`TableSchema.options()`. A table can therefore persist
`scan.mode=FROM-SNAPSHOT`; explicit Doris snapshot/tag resolution merges its
selector into that table without clearing the inherited mode (and branch scans
retain the branch schema directly). Here the selector is removed, but this
case-sensitive lookup leaves the uppercase mode behind. The pinned Rust reader
recognizes it case-insensitively and then rejects the now-bare mode for lacking
its required selector, while JNI accepts the original option. Normalize the
mode before this membership check and cover persisted mixed-case time-travel
options.
##########
be/src/format_v2/table/paimon_rust_predicate_converter.cpp:
##########
@@ -0,0 +1,813 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "format_v2/table/paimon_rust_predicate_converter.h"
+
+#include <algorithm>
+#include <cctype>
+#include <memory>
+#include <utility>
+
+#include "common/logging.h"
+#include "core/column/column_const.h"
+#include "core/column/column_nullable.h"
+#include "core/data_type/data_type.h"
+#include "core/data_type/data_type_nullable.h"
+#include "core/field.h"
+#include "core/types.h"
+#include "core/value/decimalv2_value.h"
+#include "core/value/timestamptz_value.h"
+#include "core/value/vdatetime_value.h"
+#include "exprs/vcompound_pred.h"
+#include "exprs/vdirect_in_predicate.h"
+#include "exprs/vectorized_fn_call.h"
+#include "exprs/vexpr.h"
+#include "exprs/vin_predicate.h"
+#include "exprs/vliteral.h"
+#include "exprs/vslot_ref.h"
+
+namespace doris {
+
+namespace {
+// paimon_datum tags (see paimon.h / bindings/c/src/table.rs::datum_from_c).
+constexpr int32_t kTagBool = 0;
+constexpr int32_t kTagTinyInt = 1;
+constexpr int32_t kTagSmallInt = 2;
+constexpr int32_t kTagInt = 3;
+constexpr int32_t kTagLong = 4;
+constexpr int32_t kTagDouble = 6;
+constexpr int32_t kTagString = 7;
+constexpr int32_t kTagDate = 8;
+constexpr int32_t kTagTimestamp = 10;
+constexpr int32_t kTagDecimal = 12;
+constexpr int32_t kTagBytes = 13;
+
+// paimon decimal precision ceiling (paimon::Decimal::MAX_PRECISION).
+constexpr int32_t kPaimonDecimalMaxPrecision = 38;
+
+// RAII for an owned paimon_predicate*. and/or/not consume their inputs, so we
+// release() before handing pointers to them.
+struct predicate_deleter {
+ void operator()(paimon_predicate* p) const {
+ if (p) {
+ paimon_predicate_free(p);
+ }
+ }
+};
+using predicate_ptr = std::unique_ptr<paimon_predicate, predicate_deleter>;
+
+// RAII for an owned paimon_error*.
+struct error_deleter {
+ void operator()(paimon_error* p) const {
+ if (p) {
+ paimon_error_free(p);
+ }
+ }
+};
+using error_ptr = std::unique_ptr<paimon_error, error_deleter>;
+
+// Render a paimon_error into a string. Takes ownership of `err` via RAII so it
+// is freed on every return path. Safe to call with nullptr.
+std::string consume_predicate_error(paimon_error* err) {
+ error_ptr owned(err);
+ if (!owned) {
+ return "unknown error";
+ }
+ std::string msg;
+ if (owned->message.data != nullptr && owned->message.len > 0) {
+ msg.assign(reinterpret_cast<const char*>(owned->message.data),
owned->message.len);
+ }
+ return "code=" + std::to_string(owned->code) + ", msg=" + msg;
+}
+} // namespace
+
+PaimonRustPredicateConverter::PaimonRustPredicateConverter(
+ const std::vector<std::string>& column_names, const
std::vector<DataTypePtr>& column_types,
+ const paimon_table* table)
+ : _table(table) {
+ DORIS_CHECK(column_names.size() == column_types.size());
+ _columns_by_name.reserve(column_names.size());
+ for (size_t i = 0; i < column_names.size(); ++i) {
+ _columns_by_name.emplace(_normalize_name(column_names[i]),
+ std::make_pair(column_names[i],
column_types[i]));
+ }
+ // Paimon TIMESTAMP (wall clock) is stored as epoch-millis-of-the-wall-time
+ // and the DateTimeV2 serde decodes timezone-naive arrow values in UTC, so
+ // timestamp literals convert wall->epoch in UTC. utc_time_zone() needs no
+ // tzdata lookup, so the conversion cannot silently fall back to a
+ // machine-local zone.
+ _utc_tz = cctz::utc_time_zone();
+}
+
+paimon_predicate* PaimonRustPredicateConverter::build(const VExprContextSPtrs&
conjuncts) {
+ if (_table == nullptr) {
+ return nullptr;
+ }
+ predicate_ptr result;
+ for (const auto& conjunct : conjuncts) {
+ if (!conjunct || !conjunct->root()) {
+ continue;
+ }
+ auto root = conjunct->root();
+ if (root->is_rf_wrapper()) {
Review Comment:
[P1] Do not unwrap and push null-aware runtime filters as ordinary
predicates. `FileScannerV2` deliberately carries `_null_aware` into
`RuntimeFilterExpr`; for an `EQ_FOR_NULL` join whose build side contains NULL,
its residual execution restores NULL probe rows to true. This loop discards
that wrapper, and `VDirectInPredicate::get_slot_in_expr()` rebuilds only the
concrete set members as a normal `IN` predicate, so Rust can permanently prune
the NULL probes before the join sees them. The Lance pushdown path explicitly
declines `is_null_aware()` filters for this reason. Keep these filters residual
(or encode equivalent NULL semantics) and add a Rust-reader differential test
with NULLs on both sides of a null-safe join.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/source/PaimonScanNode.java:
##########
@@ -411,10 +598,254 @@ 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;
+ 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();
Review Comment:
[P1] Route deduplicate `ignore-delete=true` tables away from Rust. Java's
`DeduplicateMergeFunction` explicitly skips retract records when this valid
option is set (including old files that still contain them). The pinned Rust
reader does not pass table options to 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 this option until Rust
implements it and add a differential historical-file test.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/source/PaimonScanNode.java:
##########
@@ -411,10 +598,254 @@ 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;
+ 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
Review Comment:
[P1] Gate non-DV merge options that this Rust reader rejects. Java supports
`partial-update.remove-record-on-delete` and
`aggregation.remove-record-on-delete` (plus a wider field-option matrix), while
the pinned Rust `PartialUpdateConfig::validate_read_mode` /
`AggregationConfig::validate_runtime_mode` explicitly return `Unsupported` for
them. Because these fallback flags are only derived when deletion vectors are
enabled, an otherwise ordinary non-DV `DataSplit` is admitted here and fails
during Rust merge construction instead of succeeding through JNI. Derive
eligibility from the exact supported option matrix, or keep those tables on JNI.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/source/PaimonScanNode.java:
##########
@@ -411,10 +598,254 @@ 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;
+ 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);
+ }
+ }
+ }
+ // 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()));
+ // 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
+ && !dvMergeOnRead && !splitDvNotMaterialized
+ && !orcLtzSchema && !projectedVariant &&
schemeCapabilityVerified
+ && hdfsBackendVerified && paimonFileStoreTable != null;
Review Comment:
[P1] Keep nested schema-evolution splits off Rust until it maps nested field
IDs. Paimon supports adding a child inside a `ROW`, and JNI's
`SchemaEvolutionUtil` recursively remaps every ROW level by field ID and
NULL-fills the new child. The pinned Rust reader only builds an ID mapping for
top-level fields, then asks arrow-cast 58.4.0 to cast the old whole
`StructArray` to the current struct. When a target child is absent, Arrow falls
back to positional zipping; an old two-child struct cast to a three-child
target then errors in `StructArray::try_new` (and rename/reorder combinations
can bind the wrong same-typed child). This gate admits that historical split.
Add a recursive capability fallback or fix the Rust evolution and cover nested
ADD against JNI.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/source/PaimonScanNode.java:
##########
@@ -411,10 +598,254 @@ 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;
+ 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);
+ }
+ }
+ }
+ // 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()));
+ // 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());
Review Comment:
[P1] Include per-file external paths in this capability gate. Paimon's valid
`data-file.external-paths` feature stores an absolute location in each
`DataFileMeta`, and both Java and the serialized Rust split prefer that path
over the bucket path. The pinned Rust table, however, creates one `FileIO` from
`paimon_table`; its storage enum parses every subsequent file with that
warehouse-selected backend. Thus an admitted HDFS table with an `s3://`
external data file (or an S3 table with an `oss://` file) reaches the wrong
parser and fails, while JNI can read it. Verify every split file's external
scheme/auth shape or keep such splits on JNI.
--
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]
