This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new 9558285a5cd [fix](iceberg) Fail on missing snapshot files (#68143)
9558285a5cd is described below
commit 9558285a5cddc9d7902369cb22181c428712e301
Author: Gabriel <[email protected]>
AuthorDate: Fri Sep 18 13:44:33 2026 +0800
[fix](iceberg) Fail on missing snapshot files (#68143)
### What problem does this PR solve?
An Iceberg snapshot can reference a data file that has been removed from
storage. With the default
`ignore_not_found_file_in_external_table=true`, both file scanners can
skip the missing file and return a successful but incomplete result.
Exclude Iceberg ranges from the missing-file skip policy, regardless of
that setting, while preserving the existing behavior for other table
formats. Apply the shared policy at reader initialization, read
boundaries, and physical-split refinement on branch-4.1. This also
prevents a missing delete file reported as `NOT_FOUND` from being
swallowed by the scanner.
Add unit coverage for Iceberg Parquet/ORC/JNI and ordinary external
ranges with both setting values. Add an external regression that creates
multiple data files, deletes one actual object, and expects row scans to
fail with V1/V2 and both setting values. The regression disables
file/result caches and restores the BE configuration after testing.
### Release note
Iceberg scans now fail when a required snapshot file is missing instead
of silently skipping it, including when
`ignore_not_found_file_in_external_table` is enabled.
### Check List (For Author)
- Test: The shared-policy GoogleTests passed in a standalone build with
actual generated Thrift types; the Iceberg assertion failed with the
extracted original policy. Groovy compilation, clang-format 16, and
whitespace checks passed. The full BE UT runner was blocked during local
clucene/Zstd dependency configuration. The external regression has not
been executed locally.
- Behavior changed: Yes; missing Iceberg files are no longer ignored.
Other table formats retain their existing behavior.
- Does this need documentation: No new configuration or API; the
corrected behavior is described above.
---
be/src/exec/scan/file_scan_range_utils.h | 31 ++++++++
be/src/exec/scan/file_scanner.cpp | 13 ++--
be/src/exec/scan/file_scanner_v2.cpp | 15 +++-
be/test/exec/scan/file_scan_range_utils_test.cpp | 48 ++++++++++++
.../iceberg/test_iceberg_missing_data_file.groovy | 89 ++++++++++++++++++++++
5 files changed, 188 insertions(+), 8 deletions(-)
diff --git a/be/src/exec/scan/file_scan_range_utils.h
b/be/src/exec/scan/file_scan_range_utils.h
new file mode 100644
index 00000000000..cd4c06c9534
--- /dev/null
+++ b/be/src/exec/scan/file_scan_range_utils.h
@@ -0,0 +1,31 @@
+// 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.
+
+#pragma once
+
+#include "gen_cpp/PlanNodes_types.h"
+
+namespace doris {
+
+inline bool can_ignore_not_found_file(const TFileRangeDesc& range, bool
ignore_not_found) {
+ // Iceberg ranges belong to a committed snapshot, not a stale directory
listing. Skipping a
+ // missing data or delete file would silently return an incomplete or
incorrect snapshot.
+ return ignore_not_found && !(range.__isset.table_format_params &&
+ range.table_format_params.table_format_type
== "iceberg");
+}
+
+} // namespace doris
diff --git a/be/src/exec/scan/file_scanner.cpp
b/be/src/exec/scan/file_scanner.cpp
index d7c262dc77a..440bdb892c7 100644
--- a/be/src/exec/scan/file_scanner.cpp
+++ b/be/src/exec/scan/file_scanner.cpp
@@ -50,6 +50,7 @@
#include "exec/common/stringop_substring.h"
#include "exec/operator/file_scan_operator.h"
#include "exec/rowid_fetcher.h"
+#include "exec/scan/file_scan_range_utils.h"
#include "exec/scan/scan_node.h"
#include "exprs/aggregate/aggregate_function.h"
#include "exprs/function/function.h"
@@ -533,9 +534,11 @@ Status FileScanner::_get_block_wrapped(RuntimeState*
state, Block* block, bool*
if (_cur_reader == nullptr || _cur_reader_eof) {
// The file may not exist because the file list is got from meta
cache,
// And the file may already be removed from storage.
- // Just ignore not found files.
+ // Only formats without Iceberg snapshot guarantees may ignore
missing files.
Status st = _get_next_reader();
- if (st.is<ErrorCode::NOT_FOUND>() &&
config::ignore_not_found_file_in_external_table) {
+ if (st.is<ErrorCode::NOT_FOUND>() &&
+ can_ignore_not_found_file(_current_range,
+
config::ignore_not_found_file_in_external_table)) {
_cur_reader_eof = true;
COUNTER_UPDATE(_not_found_file_counter, 1);
continue;
@@ -1296,13 +1299,13 @@ Status FileScanner::_get_next_reader() {
COUNTER_UPDATE(_file_counter, 1);
// The FileScanner for external table may try to open not exist files,
// Because FE file cache for external table may out of date.
- // So, NOT_FOUND for FileScanner is not a fail case.
- // Will remove this after file reader refactor.
+ // Iceberg snapshot files must still fail the query if they are
missing.
if (init_status.is<END_OF_FILE>()) {
COUNTER_UPDATE(_empty_file_counter, 1);
continue;
} else if (init_status.is<ErrorCode::NOT_FOUND>()) {
- if (config::ignore_not_found_file_in_external_table) {
+ if (can_ignore_not_found_file(_current_range,
+
config::ignore_not_found_file_in_external_table)) {
COUNTER_UPDATE(_not_found_file_counter, 1);
continue;
}
diff --git a/be/src/exec/scan/file_scanner_v2.cpp
b/be/src/exec/scan/file_scanner_v2.cpp
index c0cdfaf7b78..221e1dbcd83 100644
--- a/be/src/exec/scan/file_scanner_v2.cpp
+++ b/be/src/exec/scan/file_scanner_v2.cpp
@@ -46,6 +46,7 @@
#include "exec/operator/scan_operator.h"
#include "exec/scan/access_path_parser.h"
#include "exec/scan/file_scan_io_context.h"
+#include "exec/scan/file_scan_range_utils.h"
#include "exprs/vexpr.h"
#include "exprs/vexpr_context.h"
#include "exprs/vruntimefilter_wrapper.h"
@@ -555,7 +556,10 @@ Status FileScannerV2::_get_block_impl(RuntimeState* state,
Block* block, bool* e
_table_reader->set_batch_size(_predict_reader_batch_rows());
}
const auto status = _table_reader->get_block(block, eof);
- if (_should_skip_not_found(status,
config::ignore_not_found_file_in_external_table)) {
+ if (_should_skip_not_found(
+ status,
+ can_ignore_not_found_file(
+ _current_range,
config::ignore_not_found_file_in_external_table))) {
RETURN_IF_ERROR(_table_reader->abort_split());
COUNTER_UPDATE(_not_found_file_counter, 1);
RETURN_IF_ERROR(_complete_current_split());
@@ -643,7 +647,10 @@ Status FileScannerV2::_prepare_next_split(bool* eos) {
RETURN_IF_ERROR(_generate_partition_values(_current_range,
&partition_values));
const auto status =
_prepare_table_reader_split(_current_range,
std::move(partition_values));
- if (_should_skip_not_found(status,
config::ignore_not_found_file_in_external_table)) {
+ if (_should_skip_not_found(
+ status,
+ can_ignore_not_found_file(_current_range,
+
config::ignore_not_found_file_in_external_table))) {
RETURN_IF_ERROR(_table_reader->abort_split());
COUNTER_UPDATE(_not_found_file_counter, 1);
RETURN_IF_ERROR(_complete_current_split());
@@ -672,7 +679,9 @@ Status FileScannerV2::_prepare_next_split(bool* eos) {
const auto split_status = _table_reader->build_physical_splits(
_current_split, &generated_splits, &was_split);
const auto ignored_split_status = _classify_ignored_split_status(
- split_status,
config::ignore_not_found_file_in_external_table,
+ split_status,
+ can_ignore_not_found_file(_current_range,
+
config::ignore_not_found_file_in_external_table),
_should_stop || _io_ctx->should_stop);
if (ignored_split_status == IgnoredSplitStatus::NOT_FOUND) {
RETURN_IF_ERROR(_table_reader->abort_split());
diff --git a/be/test/exec/scan/file_scan_range_utils_test.cpp
b/be/test/exec/scan/file_scan_range_utils_test.cpp
new file mode 100644
index 00000000000..bc8bbfaa465
--- /dev/null
+++ b/be/test/exec/scan/file_scan_range_utils_test.cpp
@@ -0,0 +1,48 @@
+// 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 "exec/scan/file_scan_range_utils.h"
+
+#include <gtest/gtest.h>
+
+namespace doris {
+
+TEST(FileScanRangeUtilsTest, IcebergFilesCannotBeIgnored) {
+ TFileRangeDesc range;
+ range.table_format_params.__set_table_format_type("iceberg");
+ range.__isset.table_format_params = true;
+ for (auto format : {TFileFormatType::FORMAT_PARQUET,
TFileFormatType::FORMAT_ORC,
+ TFileFormatType::FORMAT_JNI}) {
+ range.__set_format_type(format);
+ EXPECT_FALSE(can_ignore_not_found_file(range, true));
+ EXPECT_FALSE(can_ignore_not_found_file(range, false));
+ }
+}
+
+TEST(FileScanRangeUtilsTest, OtherFilesRespectIgnoreSetting) {
+ TFileRangeDesc range;
+ EXPECT_TRUE(can_ignore_not_found_file(range, true));
+ EXPECT_FALSE(can_ignore_not_found_file(range, false));
+ for (const auto* format : {"hive", "hudi", "paimon"}) {
+ range.table_format_params.__set_table_format_type(format);
+ range.__isset.table_format_params = true;
+ EXPECT_TRUE(can_ignore_not_found_file(range, true));
+ EXPECT_FALSE(can_ignore_not_found_file(range, false));
+ }
+}
+
+} // namespace doris
diff --git
a/regression-test/suites/external_table_p0/iceberg/test_iceberg_missing_data_file.groovy
b/regression-test/suites/external_table_p0/iceberg/test_iceberg_missing_data_file.groovy
new file mode 100644
index 00000000000..f71e1449cfb
--- /dev/null
+++
b/regression-test/suites/external_table_p0/iceberg/test_iceberg_missing_data_file.groovy
@@ -0,0 +1,89 @@
+// 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.
+
+import com.amazonaws.auth.AWSStaticCredentialsProvider
+import com.amazonaws.auth.BasicAWSCredentials
+import com.amazonaws.client.builder.AwsClientBuilder.EndpointConfiguration
+import com.amazonaws.services.s3.AmazonS3ClientBuilder
+
+suite("test_iceberg_missing_data_file",
+
"p0,external,iceberg,external_docker,external_docker_iceberg,nonConcurrent") {
+ if
(!"true".equalsIgnoreCase(context.config.otherConfigs.get("enableIcebergTest")))
{
+ logger.info("disable iceberg test")
+ return
+ }
+ String host = context.config.otherConfigs.get("externalEnvIp")
+ String restPort = context.config.otherConfigs.get("iceberg_rest_uri_port")
+ String minioPort = context.config.otherConfigs.get("iceberg_minio_port")
+ String endpoint = "http://${host}:${minioPort}";
+
+ sql "drop catalog if exists test_iceberg_missing_data_file"
+ sql """create catalog test_iceberg_missing_data_file properties (
+ "type" = "iceberg",
+ "iceberg.catalog.type" = "rest",
+ "uri" = "http://${host}:${restPort}";,
+ "s3.endpoint" = "${endpoint}",
+ "s3.access_key" = "admin",
+ "s3.secret_key" = "password",
+ "s3.region" = "us-east-1"
+ )"""
+ sql "switch test_iceberg_missing_data_file"
+ sql "create database if not exists missing_data_file_db"
+ sql "use missing_data_file_db"
+ sql "drop table if exists missing_data_file"
+ sql """create table missing_data_file (id int, payload string)
+ properties ("format-version" = "2", "write.format.default" =
"parquet")"""
+ sql "insert into missing_data_file values (1, 'first-file')"
+ sql "insert into missing_data_file values (2, 'second-file')"
+
+ // Inspect manifests only: reading table rows here could cache the object
that will be removed.
+ def files = sql "select file_path from missing_data_file\$files where
content = 0"
+ assertTrue(files.size() >= 2, "The snapshot must contain multiple data
files")
+ URI missingFile = new URI(files[0][0].toString())
+ // Hadoop-backed catalogs may expose the same S3 objects through s3a or
s3n URIs.
+ assertTrue(missingFile.scheme in ["s3", "s3a", "s3n"],
+ "Unexpected object storage URI: ${missingFile}")
+ String key = missingFile.path.substring(1)
+ def client = AmazonS3ClientBuilder.standard()
+ .withEndpointConfiguration(new EndpointConfiguration(endpoint,
"us-east-1"))
+ .withPathStyleAccessEnabled(true)
+ .withCredentials(new AWSStaticCredentialsProvider(
+ new BasicAWSCredentials("admin", "password")))
+ .build()
+ try {
+ // Remove only an object created by this suite, leaving the snapshot
metadata unchanged.
+ client.deleteObject(missingFile.host, key)
+ assertFalse(client.doesObjectExist(missingFile.host, key))
+ } finally {
+ client.shutdown()
+ }
+
+ sql "set enable_file_cache = false"
+ sql "set enable_sql_cache = false"
+ sql "set enable_query_cache = false"
+ for (boolean ignoreMissing : [true, false]) {
+ setBeConfigTemporary([ignore_not_found_file_in_external_table:
ignoreMissing]) {
+ for (boolean scannerV2 : [false, true]) {
+ sql "set enable_file_scanner_v2 = ${scannerV2}"
+ test {
+ sql "select id, payload from missing_data_file order by id"
+ exception "NOT_FOUND"
+ }
+ }
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
