github-actions[bot] commented on code in PR #66032:
URL: https://github.com/apache/doris/pull/66032#discussion_r3649385086
##########
be/src/format_v2/table/paimon_reader.cpp:
##########
@@ -199,7 +178,10 @@ Status
PaimonHybridReader::_ensure_current_split_reader(const format::SplitReadO
} else {
format::FileFormat file_format;
RETURN_IF_ERROR(_to_file_format(options.current_range, &file_format));
- DCHECK(options.current_split_format == file_format);
+ // Old FE plans encoded a native file as FORMAT_JNI without
paimon_split and carried the
+ // physical format only in paimon_params.file_format.
+ DCHECK(options.current_split_format == file_format ||
Review Comment:
[P1] Preserve the Paimon split kind and resolved native format
This native branch receives two `FORMAT_JNI` compatibility shapes.
`PAIMON_CPP` carries a complete serialized `DataSplit`, but the raw child never
consumes it and its synthetic path names only the first file. A legacy range
with no `reader_type`/`paimon_split` is a valid raw file, but `prepare_split()`
forwards `current_split_format == JNI`, overwrites the resolved Parquet/ORC
format, and fails to open it. Please keep encoded CPP splits on V1 until V2 has
an equivalent child, and pass the resolved physical format when preparing
legacy raw splits.
##########
be/src/format_v2/column_mapper.cpp:
##########
@@ -2332,28 +2300,17 @@ Status TableColumnMapper::localize_filters(const
std::vector<TableFilter>& table
// This keeps expression localization independent from filter iteration
order.
filter_mappings = _filter_visible_mappings();
const auto global_to_file_slot =
build_file_slot_rewrite_map(filter_mappings, _filter_entries);
- for (size_t filter_index = 0; filter_index < table_filters.size();
++filter_index) {
- const auto& table_filter = table_filters[filter_index];
+ for (const auto& table_filter : table_filters) {
if (table_filter.conjunct != nullptr && table_filter.conjunct->root()
!= nullptr) {
const auto root = table_filter.conjunct->root();
const auto impl = root->get_impl();
const auto predicate = impl != nullptr ? impl : root;
- if (!table_filter.can_localize || !predicate->is_deterministic() ||
+ if (!predicate->is_deterministic() ||
Review Comment:
[P1] Keep truncation-sensitive predicates above materialization
This gate no longer excludes columns whose physical CHAR/VARCHAR value must
be shortened to the table width. The localized clone runs in the FileReader
before `finalize_chunk()` performs that truncation, so it can irreversibly drop
rows: with physical `"abcd"`, table `VARCHAR(3)`, and `value = 'abc'`, the
local comparison rejects the row before finalization would produce `"abc"` for
Scanner's authoritative predicate. Please restore the width-change guard (and
its deleted regression), or make the localized expression observe the exact
bounded table value.
##########
be/src/exec/operator/file_scan_operator.cpp:
##########
@@ -118,12 +118,8 @@ bool FileScanLocalState::_should_use_file_scanner_v2(const
TQueryOptions& query_
const bool is_transactional_hive =
scan_params.__isset.table_format_params &&
scan_params.table_format_params.table_format_type ==
"transactional_hive";
- // JNI reader selection is stored per split, but this scan-level selector
cannot inspect the
- // split yet. Older FEs may omit both the scan-level Paimon marker and
split-level reader_type,
- // so keep JNI scans on V1 until scanner selection can distinguish every
compatibility shape.
return query_options.__isset.enable_file_scanner_v2 &&
query_options.enable_file_scanner_v2 &&
Review Comment:
[P1] Keep unsupported JNI sources on FileScanner V1
This scan-wide selector now sends every query-side `FORMAT_JNI` plan to V2
before a split/table format is visible. LakeSoul always emits `FORMAT_JNI`, but
`FileScannerV2::is_supported_jni_table_format()` rejects `table_format_type ==
"lakesoul"` and there is no per-range fallback; the old scanner's
`LakeSoulJniReader` is therefore never reached. Because FileScanner V2 is
enabled by default, existing LakeSoul scans now fail `NotSupported`. Please
retain the JNI exclusion until all sources are supported, or make
selection/fallback range-aware and include LakeSoul.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/FileQueryScanNode.java:
##########
@@ -733,6 +747,29 @@ public TableScanParams getScanParams() {
return this.scanParams;
}
+ /**
+ * Return metadata pinned for this scan relation.
+ */
+ protected Optional<MvccSnapshot> getRelationSnapshot() {
+ if (relationSnapshotInitialized) {
+ return relationSnapshot;
+ }
+ relationSnapshotInitialized = true;
+ TableIf targetTable = desc.getTable();
+ if (!(targetTable instanceof MvccTable)) {
+ return Optional.empty();
+ }
+ if (tableSnapshot != null || scanParams != null) {
+ // A statement can scan several versions of one table, so
execution must reconstruct
+ // the snapshot from this scan node's own qualifiers rather than
the table-only map.
+ relationSnapshot = Optional.of(((MvccTable)
targetTable).loadSnapshot(
Review Comment:
[P1] Carry the already pinned snapshot instead of resolving the ref again
Logical binding has already resolved this branch/tag and captured its
columns, but the physical plan carries only the mutable qualifier, so this call
resolves it again. If the branch advances meanwhile, tuple slots can describe
head A while schema mappings/defaults and scan files come from head B,
defeating the statement's MVCC pin. Please carry the resolved `MvccSnapshot`
through `LogicalFileScan`/`PhysicalFileScan` and reuse it here.
##########
be/test/format_v2/table_reader_test.cpp:
##########
@@ -2020,13 +1756,8 @@ TEST(TableReaderTest,
SlotlessConjunctDisablesAggregatePushdown) {
// presence still prevents the fake aggregate count (3) from replacing the
two physical rows.
ASSERT_NE(fake_state->last_request, nullptr);
EXPECT_TRUE(fake_state->last_request->conjuncts.empty());
- // The two physical rows are then filtered at the table boundary, where
slotless predicates are
- // evaluated exactly even though they cannot be localized to a file column.
- EXPECT_EQ(block.rows(), 0);
- EXPECT_FALSE(eos);
+ EXPECT_EQ(block.rows(), 2);
EXPECT_TRUE(predicate_executed);
Review Comment:
[P1] Update the direct-TableReader expectations for scanner-owned predicates
After this patch TableReader no longer executes residual conjuncts, and this
non-deterministic predicate is deliberately not localized
(`last_request->conjuncts` is empty). Its flag can therefore never become true
in this direct `TableReader::get_block()` test. The same stale `EXPECT_TRUE`
remains at line 1466. Nearby `UnsafePredicateStaysOnScannerPath` already
expects false under the new ownership model, so these assertions will fail the
BE unit target. Change both to false, or exercise final filtering through
Scanner if execution is what the test intends.
##########
regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_nullable_truncate_negative.groovy:
##########
@@ -0,0 +1,78 @@
+// 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.
+
+suite("test_iceberg_write_nullable_truncate_negative",
+ "p0,external,iceberg,external_docker,external_docker_iceberg") {
+ String enabled = context.config.otherConfigs.get("enableIcebergTest")
+ if (enabled == null || !enabled.equalsIgnoreCase("true")) {
+ logger.info("disable iceberg test")
+ return
+ }
+
+ // This opt-in switch isolates a BE-fatal negative scenario from the
shared P0 cluster.
+ // Enable it only in a cluster whose BE processes can be restarted after
the suite.
+ String crashTestEnabled =
context.config.otherConfigs.get("enableIcebergCrashTest")
+ if (crashTestEnabled == null ||
!crashTestEnabled.equalsIgnoreCase("true")) {
+ logger.info("skip isolated Iceberg crash regression")
+ return
+ }
+
+ String restPort = context.config.otherConfigs.get("iceberg_rest_uri_port")
+ String minioPort = context.config.otherConfigs.get("iceberg_minio_port")
+ String externalEnvIp = context.config.otherConfigs.get("externalEnvIp")
+ String catalogName = "test_iceberg_write_nullable_truncate_negative"
+ String dbName = "iceberg_write_nullable_truncate_negative_db"
+
+ sql """drop catalog if exists ${catalogName}"""
+ sql """
+ create catalog ${catalogName} properties (
+ "type" = "iceberg",
+ "iceberg.catalog.type" = "rest",
+ "uri" = "http://${externalEnvIp}:${restPort}";,
+ "s3.access_key" = "admin",
+ "s3.secret_key" = "password",
+ "s3.endpoint" = "http://${externalEnvIp}:${minioPort}";,
+ "s3.region" = "us-east-1"
+ )
+ """
+ sql """switch ${catalogName}"""
+ sql """drop database if exists ${dbName} force"""
+ sql """create database ${dbName}"""
+ sql """use ${dbName}"""
+
+ sql """
+ create table nullable_truncate (
+ id int not null,
+ zone string
+ )
+ partition by list (zone) ()
+ properties ("format-version" = "2")
+ """
+ sql """insert into nullable_truncate values (1, 'CN'), (2, null)"""
+
+ // Negative scenario: evolve to a truncate transform whose source remains
nullable,
+ // then write both non-NULL and NULL partition values through Doris.
+ sql """
+ alter table nullable_truncate
+ add partition key truncate(2, zone) as zone_prefix
+ """
+ sql """insert into nullable_truncate values (3, 'US-east'), (4, null)"""
+
+ order_qt_nullable_truncate_rows """
Review Comment:
[P2] Add result oracles for the opt-in negative suites
This `order_qt` tag has no corresponding `.out` file, so a normal
verification run throws `Missing outputFile` once `enableIcebergCrashTest`
reaches it. The same omission exists for `order_qt_merge_truncate_after_fix`
and `order_qt_duplicate_source_atomic_state` in the other two new opt-in
suites. Please add the three golden outputs or use explicit assertions so these
tests can actually validate a fixed cluster.
##########
be/src/format_v2/table_reader.cpp:
##########
@@ -739,152 +736,35 @@ Status TableReader::init(TableReadOptions&& options) {
}
_system_properties = create_system_properties(_scan_params);
_mapper_options.mode = TableColumnMappingMode::BY_NAME;
- return _replace_conjuncts(options.conjuncts);
-}
-
-Status TableReader::_prepare_conjunct(const VExprContextSPtr& source,
VExprContextSPtr* prepared) {
- DORIS_CHECK(source != nullptr);
- DORIS_CHECK(source->root() != nullptr);
- DORIS_CHECK(prepared != nullptr);
- VExprSPtr root;
- RETURN_IF_ERROR(clone_table_expr_tree(source->root(), &root));
- auto conjunct = VExprContext::create_shared(std::move(root));
- RETURN_IF_ERROR(conjunct->prepare(_runtime_state, RowDescriptor {}));
- RETURN_IF_ERROR(conjunct->open(_runtime_state));
- *prepared = std::move(conjunct);
- return Status::OK();
-}
-
-Status TableReader::_replace_conjuncts(const VExprContextSPtrs& conjuncts) {
- VExprContextSPtrs prepared;
- prepared.reserve(conjuncts.size());
- for (const auto& source : conjuncts) {
- VExprContextSPtr conjunct;
- RETURN_IF_ERROR(_prepare_conjunct(source, &conjunct));
- prepared.push_back(std::move(conjunct));
- }
- _conjuncts = std::move(prepared);
- return Status::OK();
-}
-
-Status TableReader::append_conjuncts_with_ownership(const VExprContextSPtrs&
conjuncts,
- size_t
table_reader_owned_conjunct_count) {
- DORIS_CHECK(!_appended_table_reader_owned_conjunct_count.has_value());
- _appended_table_reader_owned_conjunct_count =
table_reader_owned_conjunct_count;
- auto status = append_conjuncts(conjuncts);
- _appended_table_reader_owned_conjunct_count.reset();
- return status;
-}
-
-Status TableReader::append_conjuncts(const VExprContextSPtrs& conjuncts) {
- const size_t owned_count =
-
_appended_table_reader_owned_conjunct_count.value_or(conjuncts.size());
- DORIS_CHECK_LE(owned_count, conjuncts.size());
- // Once Scanner owns a suffix, later predicates cannot be inserted into
the TableReader-owned
- // prefix without reordering them ahead of that stateful/error-preserving
barrier.
- DORIS_CHECK(owned_count == 0 || _table_reader_owned_conjunct_count ==
_conjuncts.size());
- for (size_t conjunct_index = 0; conjunct_index < conjuncts.size();
++conjunct_index) {
- const auto& source = conjuncts[conjunct_index];
- VExprContextSPtr conjunct;
- RETURN_IF_ERROR(_prepare_conjunct(source, &conjunct));
- _conjuncts.push_back(conjunct);
- if (conjunct_index < owned_count) {
- ++_table_reader_owned_conjunct_count;
- }
- if (_current_task != nullptr && conjunct_index < owned_count) {
- // The active reader has already fixed its localized predicate
set. Appended runtime
- // filters must remain residual until the next split rebuilds its
FileScanRequest.
- _remaining_conjuncts.push_back(std::move(conjunct));
- }
- }
+ _conjuncts = std::move(options.conjuncts);
return Status::OK();
}
Status TableReader::_build_table_filters_from_conjuncts() {
_table_filters.clear();
_constant_pruning_safe_filter_count = 0;
bool in_safe_prefix = true;
- for (size_t conjunct_index = 0; conjunct_index < _conjuncts.size();
++conjunct_index) {
- const auto& conjunct = _conjuncts[conjunct_index];
+ for (const auto& conjunct : _conjuncts) {
DORIS_CHECK(conjunct != nullptr);
DORIS_CHECK(conjunct->root() != nullptr);
// `_table_filters` omits expressions without slot references, but
such an expression still
// occupies a position in the row-level conjunct order. Record how
many localized filters
// precede the first unsafe original conjunct so constant pruning
cannot jump over a
- // slotless non-deterministic/error-preserving barrier. An unsafe
predicate is either kept
- // on TableReader's post-materialization path by a standalone caller
or carried only for
- // analysis when FileScannerV2 owns the ordered suffix.
- if (in_safe_prefix && !is_safe_to_pre_execute(conjunct)) {
+ // slotless non-deterministic/error-preserving barrier. Unsafe
predicates remain solely on
+ // Scanner's original row-level path because localizing a clone would
execute their state
+ // twice with independent state.
+ if (in_safe_prefix && !_is_safe_to_pre_execute(conjunct)) {
Review Comment:
[P1] Preserve the unsafe-conjunct barrier for localization
`in_safe_prefix` now limits only constant pruning. Because the patch removes
the bit that carried this decision into each `TableFilter`,
`localize_filters()` can push a later deterministic predicate below an earlier
stateful or error-preserving conjunct. The FileReader may then discard a row
before Scanner reaches the original ordered vector, suppressing the earlier
invocation or error. Please propagate the safe-prefix decision to localization
and keep the first unsafe conjunct and the entire suffix off the file-local
path.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/StatementContext.java:
##########
@@ -934,7 +934,12 @@ public void loadSnapshots(TableIf specificTable,
Optional<TableSnapshot> tableSn
Optional<TableScanParams> scanParams) {
if (specificTable instanceof MvccTable) {
MvccTableInfo mvccTableInfo = new MvccTableInfo(specificTable);
- if (!snapshots.containsKey(mvccTableInfo)) {
+ if (tableSnapshot.isPresent() || scanParams.isPresent()) {
+ // Explicit time-travel relations must pin their own metadata
even when another
+ // relation for the same table was already bound in this
statement.
+ snapshots.put(mvccTableInfo,
+ ((MvccTable)
specificTable).loadSnapshot(tableSnapshot, scanParams));
+ } else if (!snapshots.containsKey(mvccTableInfo)) {
Review Comment:
[P1] Pin the snapshot per relation, including the current relation
This is still a single snapshot slot keyed only by catalog/database/table.
If a historical relation binds first, a later unqualified current relation hits
this `containsKey` branch and captures the historical schema too. In the
reverse order, the historical relation overwrites the map after current slots
were captured, and the current scan node later consumes that historical
snapshot. Thus a current+historical self-join is wrong in either order; the
added tests qualify both sides and do not cover it. Please carry the resolved
snapshot per relation through planning.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalTable.java:
##########
@@ -178,6 +178,10 @@ public List<Column> getFullSchema() {
return schemaCacheValue.map(SchemaCacheValue::getSchema).orElse(null);
}
+ public List<Column> getFullSchema(Optional<MvccSnapshot> snapshot) {
Review Comment:
[P1] Honor the relation snapshot for HMS-backed Iceberg
`HMSExternalTable` with `DLAType.ICEBERG` is an MVCC Iceberg scan, but it
does not override this overload, so the supplied relation snapshot is discarded
and its no-argument schema lookup falls back to the table-keyed context.
`LogicalFileScan.captureRelationSchema()` also excludes this subtype. Two
qualified HMS-Iceberg relations can therefore bind against whichever snapshot
remains in that shared map. Please add the snapshot-aware HMS Iceberg
implementation and make relation-schema capture capability-based.
--
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]
