This is an automated email from the ASF dual-hosted git repository.
Gabriel39 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 00273504b2f [fix](iceberg) Read ID-less Parquet files by positional
field ids (#68962)
00273504b2f is described below
commit 00273504b2fdf50fd6443ed8fac04a03d1a573c8
Author: daidai <[email protected]>
AuthorDate: Thu Oct 15 11:13:51 2026 +0800
[fix](iceberg) Read ID-less Parquet files by positional field ids (#68962)
### What problem does this PR solve?
Issue Number: N/A (tracked as DORIS-29685)
Related PR: N/A
Problem Summary:
A Parquet file without Iceberg field ids (`add_files` / migrated file)
was read as an all-NULL row, while Iceberg/Spark reads it by position
(`ParquetSchemaUtil.addFallbackIds`,
`iceberg.netflix.unsafe-parquet-id-fallback.enabled`). Fix: when a
split's file has no field id at all, the plan uses v2 scan semantics and
the table exposes no usable name mapping, stamp ordinal ids 1..N on the
file's top-level fields so the existing `BY_FIELD_ID` path resolves
them. Nested children stay NULL (same as Iceberg); ORC untouched.
### Release note
None
### Check List (For Author)
- Test
- [x] Regression test
- [x] Unit Test
- Behavior changed:
- [x] Yes.
- Does this need documentation?
- [x] No.
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label
---
be/src/format_v2/column_mapper.cpp | 26 ++
be/src/format_v2/table/iceberg_reader.cpp | 48 +++
be/src/format_v2/table/iceberg_reader.h | 27 ++
be/src/format_v2/table_reader.h | 7 +-
be/test/format_v2/column_mapper_test.cpp | 32 ++
be/test/format_v2/table/iceberg_reader_test.cpp | 424 ++++++++++++++++++++-
...ceberg_migrated_nested_without_name_mapping.out | 7 +-
.../iceberg/test_gen_iceberg_by_api.groovy | 18 +-
...erg_migrated_nested_without_name_mapping.groovy | 16 +-
9 files changed, 592 insertions(+), 13 deletions(-)
diff --git a/be/src/format_v2/column_mapper.cpp
b/be/src/format_v2/column_mapper.cpp
index c991b2fd940..0420f765b2e 100644
--- a/be/src/format_v2/column_mapper.cpp
+++ b/be/src/format_v2/column_mapper.cpp
@@ -1246,6 +1246,10 @@ static bool rewrite_in_struct_literal_predicate(
return true;
}
+// Defined with the other mapping-shape helpers below; needed when rewriting
table expressions to
+// file-local expressions (a slot with unmapped children cannot stand in for
the table value).
+static bool mapping_slot_has_unmapped_child(const ColumnMapping& mapping);
+
static VExprSPtr rewrite_struct_or_slot_expr_to_file_expr(
const VExprSPtr& expr,
const std::map<GlobalIndex, FileSlotRewriteInfo>& global_to_file_slot,
@@ -1288,6 +1292,15 @@ static VExprSPtr
rewrite_struct_or_slot_expr_to_file_expr(
return expr;
}
const auto& rewrite_info = rewrite_it->second;
+ DORIS_CHECK(rewrite_info.root_mapping != nullptr);
+ if (mapping_slot_has_unmapped_child(*rewrite_info.root_mapping)) {
+ // Final materialization substitutes a default/NULL for every child
the projection left
+ // unmapped, so the file slot carries only a partial value: any
file-local evaluation under
+ // this slot (whole-root casts included) would use a different value
than the user sees.
+ // Keep such predicates above TableReader instead of filtering on the
physical row.
+ *can_localize = false;
+ return expr;
+ }
auto file_slot = create_file_slot_ref(*slot_ref, rewrite_info,
rewrite_context);
if (rewrite_info.file_type->equals(*rewrite_info.table_type)) {
return file_slot;
@@ -1500,6 +1513,16 @@ static bool needs_complex_rematerialize(const
ColumnMapping& mapping) {
return false;
}
+// The physical slot is only a partial carrier of the table value when the
projection left at least
+// one descendant without a file mapping: final materialization substitutes a
default/NULL for that
+// child, so a predicate rooted at this slot must not be evaluated on the file
value (DORIS-29685).
+// Scalar type promotion and struct renaming keep their existing localization
paths.
+static bool mapping_slot_has_unmapped_child(const ColumnMapping& mapping) {
+ return std::ranges::any_of(mapping.child_mappings, [](const ColumnMapping&
child) {
+ return !child.file_local_id.has_value() ||
mapping_slot_has_unmapped_child(child);
+ });
+}
+
static bool mapping_can_use_file_column_directly(const ColumnMapping& mapping)
{
if (mapping.table_type == nullptr || mapping.file_type == nullptr) {
return false;
@@ -1512,6 +1535,9 @@ static bool mapping_can_use_file_column_directly(const
ColumnMapping& mapping) {
if (!mapping.table_type->equals(*mapping.file_type) &&
!same_timestamptz_with_different_scale) {
return false;
}
+ if (mapping_slot_has_unmapped_child(mapping)) {
+ return false;
+ }
return !needs_complex_rematerialize(mapping);
}
diff --git a/be/src/format_v2/table/iceberg_reader.cpp
b/be/src/format_v2/table/iceberg_reader.cpp
index 1ac44f41766..67e35b110c5 100644
--- a/be/src/format_v2/table/iceberg_reader.cpp
+++ b/be/src/format_v2/table/iceberg_reader.cpp
@@ -199,6 +199,50 @@ Status IcebergTableReader::validate_variant_file_mappings(
file_format_name(format));
}
+Status IcebergTableReader::annotate_file_schema(
+ std::vector<format::ColumnDefinition>* file_schema) {
+ if (file_schema == nullptr || file_schema->empty()) {
+ return Status::OK();
+ }
+ // Iceberg's positional fallback is a Parquet-reader-only compatibility
path. Its ORC/Avro
+ // readers instead derive a name mapping from the expected schema for
id-less files, which is
+ // exactly the current-name resurrection that v2 scan semantics
deliberately rejects (#65784),
+ // so ORC keeps its existing behavior and stays out of scope here.
+ if (_format != FileFormat::PARQUET) {
+ return Status::OK();
+ }
+ // Iceberg's `hasIds` contract is existential: a single real field id
keeps the file on the id
+ // path and, exactly as in Iceberg, skips both the name mapping and the
positional fallback.
+ if (schema_has_any_field_id(*file_schema)) {
+ return Status::OK();
+ }
+ // An old-FE plan keeps the legacy name fallback (rolling upgrade), and an
explicit table-level
+ // name mapping stays authoritative even when it does not cover this file.
+ if (!supports_iceberg_scan_semantics_v2(_scan_params) ||
+ _scan_has_any_authoritative_name_mapping()) {
+ return Status::OK();
+ }
+ // ParquetSchemaUtil.addFallbackIds: ordinals start at 1, are counted over
the physical
+ // top-level fields in file order, and are not propagated to nested
fields. Reader-synthesized
+ // carriers (row number / global row id) are not file fields and must not
consume an ordinal.
+ int32_t ordinal = 0;
+ for (auto& field : *file_schema) {
+ if (field.column_type != format::ColumnType::DATA_COLUMN) {
+ continue;
+ }
+ ++ordinal;
+ // ID-less Parquet fields carry their physical name as a STRING
identifier
+ // (ParquetReader::_fill_column_definition), so only a real integer
field id may skip
+ // stamping; overwriting the string identifier is what lets
FieldIdMatcher resolve the file.
+ if (field.has_identifier_field_id()) {
+ continue;
+ }
+ field.identifier = Field::create_field<TYPE_INT>(ordinal);
+ }
+ _positional_id_fallback = true;
+ return Status::OK();
+}
+
Status IcebergTableReader::validate_file_mapping(const
format::TableColumnMapper& mapper) const {
const bool metadata_only_count = _push_down_agg_type ==
TPushAggOp::type::COUNT &&
_push_down_count_columns.has_value() &&
@@ -1053,6 +1097,9 @@ Status IcebergTableReader::prepare_split(const
format::SplitReadOptions& options
_row_lineage_columns = {};
_iceberg_params.reset();
_delete_predicates_initialized = false;
+ // Per-file diagnostic: the reader instance is reused across splits,
so a previous ID-less
+ // file must not leave this set for the next (possibly ID-bearing)
split.
+ _positional_id_fallback = false;
_position_delete_rows_storage.clear();
_equality_delete_filters.clear();
_split_cache = options.cache;
@@ -1145,6 +1192,7 @@ std::string IcebergTableReader::debug_string() const {
<< ", deletion_vector_file_count=" << deletion_vector_file_count
<< ", position_delete_rows_storage_count=" <<
_position_delete_rows_storage.size()
<< ", equality_delete_filter_count=" << _equality_delete_filters.size()
+ << ", positional_id_fallback=" << _positional_id_fallback
<< ", equality_delete_filters=" << equality_filters.str() << "}";
return out.str();
}
diff --git a/be/src/format_v2/table/iceberg_reader.h
b/be/src/format_v2/table/iceberg_reader.h
index f20207d6e2c..1a4292d0994 100644
--- a/be/src/format_v2/table/iceberg_reader.h
+++ b/be/src/format_v2/table/iceberg_reader.h
@@ -89,6 +89,29 @@ protected:
Status validate_file_mapping(const format::TableColumnMapper& mapper)
const override;
+ // Iceberg's Parquet reader resolves a data file that carries no field ids
at all through the
+ // table's `schema.name-mapping.default`; when that mapping is absent it
falls back to
+ // positional ids (ParquetSchemaUtil.addFallbackIds +
pruneColumnsFallback, gated by
+ // `iceberg.netflix.unsafe-parquet-id-fallback.enabled`, default true
through 1.12 and
+ // scheduled for removal in 2.0). Mirror that decision for Parquet by
stamping ordinal ids
+ // (1..N) on the file's top-level fields, so the shared BY_FIELD_ID path
resolves those files
+ // instead of silently materializing NULL. Only top-level fields are
stamped: Iceberg assigns
+ // no ids to nested fields, so struct children of an id-less file stay
unreadable without a
+ // name mapping (or a rewrite that persists field ids).
+ Status annotate_file_schema(std::vector<format::ColumnDefinition>*
file_schema) override;
+
+ // Iceberg's positional fallback matches an ID-less Parquet file by
physical ordinals, so the
+ // same table column can bind a different file column when the table's
+ // `schema.name-mapping.default` is added or removed without touching the
data file. The
+ // condition cache key covers the file identity and the table-level
predicate digest only, so it
+ // cannot tell those bindings apart; keep such splits out of the cache
(neither read nor
+ // written) instead of reusing a verdict produced by the other binding.
+ bool _should_enable_condition_cache(
+ const format::FileScanRequest& file_request) const override {
+ return !_positional_id_fallback &&
+
format::TableReader::_should_enable_condition_cache(file_request);
+ }
+
void configure_mapper_options(format::TableColumnMapperOptions* options)
const override {
options->enable_row_lineage_virtual_columns = true;
options->enable_iceberg_metadata_virtual_columns = true;
@@ -205,6 +228,10 @@ private:
size_t _row_position_block_position = 0;
std::optional<TIcebergFileDesc> _iceberg_params;
bool _delete_predicates_initialized = false;
+ // Set when this split's Parquet file carried no field ids and the table
exposed no usable name
+ // mapping, so the reader stamped Iceberg's positional fallback ids on the
file's top-level
+ // fields. Observability only; the decision itself lives in
annotate_file_schema().
+ bool _positional_id_fallback = false;
format::DeleteRows _position_delete_rows_storage;
struct EqualityDeleteFilter {
std::vector<int> field_ids;
diff --git a/be/src/format_v2/table_reader.h b/be/src/format_v2/table_reader.h
index 7aa4400320f..800d254de83 100644
--- a/be/src/format_v2/table_reader.h
+++ b/be/src/format_v2/table_reader.h
@@ -555,7 +555,12 @@ protected:
Status _open_local_filter_exprs(const FileScanRequest& file_request);
Status _init_reader_condition_cache(const FileScanRequest& file_request);
void _finalize_reader_condition_cache();
- bool _should_enable_condition_cache(const FileScanRequest& file_request)
const;
+ // The condition cache key covers the file identity plus the table-level
predicate/session
+ // digest only. A table format whose file-column binding can change
without changing that digest
+ // (Iceberg's positional fallback resolves an ID-less file by physical
ordinals) must override
+ // this to false: a stale all-false granule would silently drop matching
rows, because the
+ // cached verdict was produced by a different physical binding.
+ virtual bool _should_enable_condition_cache(const FileScanRequest&
file_request) const;
Status _evaluate_constant_filters(bool* can_filter_all) {
DORIS_CHECK(can_filter_all != nullptr);
diff --git a/be/test/format_v2/column_mapper_test.cpp
b/be/test/format_v2/column_mapper_test.cpp
index cf6e3cc4688..249c6683673 100644
--- a/be/test/format_v2/column_mapper_test.cpp
+++ b/be/test/format_v2/column_mapper_test.cpp
@@ -2172,6 +2172,38 @@ TEST(ColumnMapperLocalizeFiltersTest,
VisibleLocalFilterAddsPredicateColumnAndCo
EXPECT_TRUE(localized_slot->data_type()->equals(*int_type));
}
+// DORIS-29685: an ID-less Parquet file is matched by positional ids, so a
struct root can be
+// mapped while its children stay unmatched. Final materialization then
substitutes NULL for those
+// children, which means the physical slot no longer carries the table value:
a whole-root predicate
+// such as LIKE(CAST(s AS STRING), '%null%') must stay above TableReader
instead of being evaluated
+// on the raw file slot (which would still see the child's physical value and
drop the row).
+TEST(ColumnMapperLocalizeFiltersTest,
PartiallyMappedStructRootPredicateStaysAboveFileReader) {
+ auto table_struct = struct_col("s", 1, {field_id_col("inner", 2, i32())});
+ // Positional fallback stamped the root only; the child keeps its physical
name identifier.
+ auto file_struct = struct_col("s", 1, {name_col("inner", i32(), 0)}, 0);
+
+ TableColumnMapper mapper({.mode = TableColumnMappingMode::BY_FIELD_ID});
+ ASSERT_TRUE(mapper.create_mapping({table_struct}, {}, {file_struct}).ok());
+
+ ASSERT_EQ(mapper.mappings().size(), 1);
+ ASSERT_TRUE(mapper.mappings()[0].file_local_id.has_value());
+ ASSERT_EQ(mapper.mappings()[0].child_mappings.size(), 1);
+
EXPECT_FALSE(mapper.mappings()[0].child_mappings[0].file_local_id.has_value());
+ EXPECT_FALSE(mapper.mappings()[0].is_trivial);
+
+ auto cast_to_string = cast_expr(table_slot(11, 0, table_struct.type, "s"),
str());
+ TableFilter filter {
+ .conjunct = VExprContext::create_shared(like_expr(cast_to_string,
"%null%")),
+ .global_indices = {GlobalIndex(0)}};
+
+ FileScanRequest request;
+ ASSERT_TRUE(mapper.localize_filters({filter}, &request).ok());
+
+ // No file-local copy of the predicate: the row must be filtered on the
materialized
+ // {"inner": null} value, not on the physical {"inner": <file value>}.
+ EXPECT_TRUE(request.conjuncts.empty());
+}
+
TEST(ColumnMapperLocalizeFiltersTest, VarbinaryFilterStaysAboveFileReader) {
const auto binary_type = varbinary();
const auto table_column = name_col("partition_key", binary_type);
diff --git a/be/test/format_v2/table/iceberg_reader_test.cpp
b/be/test/format_v2/table/iceberg_reader_test.cpp
index 8ee7788a40e..9a657dfbab6 100644
--- a/be/test/format_v2/table/iceberg_reader_test.cpp
+++ b/be/test/format_v2/table/iceberg_reader_test.cpp
@@ -644,6 +644,55 @@ void write_two_int_parquet_file(const std::string&
file_path, const std::string&
builder.build()));
}
+// Parquet files written by non-Iceberg writers carry no `field_id` at all.
Iceberg reads them by
+// file ordinal through ParquetSchemaUtil.addFallbackIds/pruneColumnsFallback
whenever the table has
+// no `schema.name-mapping.default`; these helpers produce exactly that
physical layout.
+void write_idless_int_columns_parquet_file(const std::string& file_path,
+ const std::vector<std::string>&
names,
+ const
std::vector<std::vector<int32_t>>& column_values) {
+ ASSERT_EQ(names.size(), column_values.size());
+ std::vector<std::shared_ptr<arrow::Field>> fields;
+ std::vector<std::shared_ptr<arrow::Array>> arrays;
+ for (size_t idx = 0; idx < names.size(); ++idx) {
+ fields.push_back(arrow::field(names[idx], arrow::int32(), false));
+ arrays.push_back(build_int32_array(column_values[idx]));
+ }
+ auto table = arrow::Table::Make(arrow::schema(fields), arrays);
+
+ auto file_result = arrow::io::FileOutputStream::Open(file_path);
+ ASSERT_TRUE(file_result.ok()) << file_result.status();
+ std::shared_ptr<arrow::io::FileOutputStream> out = *file_result;
+ ::parquet::WriterProperties::Builder builder;
+ builder.version(::parquet::ParquetVersion::PARQUET_2_6);
+ builder.data_page_version(::parquet::ParquetDataPageVersion::V2);
+ builder.compression(::parquet::Compression::UNCOMPRESSED);
+ const auto rows = column_values.empty() ? static_cast<int64_t>(0)
+ :
static_cast<int64_t>(column_values[0].size());
+ PARQUET_THROW_NOT_OK(::parquet::arrow::WriteTable(*table,
arrow::default_memory_pool(), out,
+ rows, builder.build()));
+}
+
+// A fully id-less nested file: `s STRUCT<inner INT>` with no `field_id` at
any level.
+void write_idless_struct_parquet_file(const std::string& file_path,
+ const std::vector<int32_t>& values) {
+ auto inner_field = arrow::field("inner", arrow::int32(), false);
+ auto struct_result = arrow::StructArray::Make({build_int32_array(values)},
{inner_field});
+ ASSERT_TRUE(struct_result.ok()) << struct_result.status();
+ auto struct_field = arrow::field("s", arrow::struct_({inner_field}),
false);
+ auto table = arrow::Table::Make(arrow::schema({struct_field}),
{*struct_result});
+
+ auto file_result = arrow::io::FileOutputStream::Open(file_path);
+ ASSERT_TRUE(file_result.ok()) << file_result.status();
+ std::shared_ptr<arrow::io::FileOutputStream> out = *file_result;
+ ::parquet::WriterProperties::Builder builder;
+ builder.version(::parquet::ParquetVersion::PARQUET_2_6);
+ builder.data_page_version(::parquet::ParquetDataPageVersion::V2);
+ builder.compression(::parquet::Compression::UNCOMPRESSED);
+ PARQUET_THROW_NOT_OK(::parquet::arrow::WriteTable(*table,
arrow::default_memory_pool(), out,
+
static_cast<int64_t>(values.size()),
+ builder.build()));
+}
+
void write_two_int_orc_file(const std::string& file_path, const std::string&
first_name,
const std::vector<int64_t>& first_values,
std::optional<int32_t> first_field_id, const
std::string& second_name,
@@ -1327,6 +1376,15 @@ void expect_int32_column_values(const IColumn& column,
}
}
+void expect_nullable_int32_column_is_null(const IColumn& column, size_t rows) {
+ const auto full_column = column.convert_to_full_column_if_const();
+ const auto& nullable_column = assert_cast<const
ColumnNullable&>(*full_column);
+ ASSERT_EQ(nullable_column.size(), rows);
+ for (size_t row = 0; row < rows; ++row) {
+ EXPECT_EQ(nullable_column.get_null_map_data()[row], 1);
+ }
+}
+
void expect_string_column_values(const IColumn& column,
const std::vector<std::string>&
expected_values) {
const auto full_column = column.convert_to_full_column_if_const();
@@ -2952,7 +3010,10 @@ TEST(IcebergV2ReaderTest,
IcebergLegacyPlanKeepsAllFieldIdsMappingRule) {
TableColumnMappingMode::BY_NAME);
}
-TEST(IcebergV2ReaderTest, IcebergV2IdlessFileRequiresAuthoritativeNameMapping)
{
+// Mapping-mode contract for an id-less file under v2 semantics: without a
table-level name mapping
+// the file stays on the BY_FIELD_ID path (DORIS-29685 stamps Iceberg's
positional fallback ids on
+// its top-level fields), and an authoritative table-level mapping switches
the split to BY_NAME.
+TEST(IcebergV2ReaderTest, IcebergV2IdlessFileMappingModeSelection) {
IcebergTableReaderMappingModeTestHelper reader;
TFileScanRangeParams scan_params;
scan_params.__set_iceberg_scan_semantics_version(ICEBERG_SCAN_SEMANTICS_VERSION_2);
@@ -2971,6 +3032,367 @@ TEST(IcebergV2ReaderTest,
IcebergV2IdlessFileRequiresAuthoritativeNameMapping) {
TableColumnMappingMode::BY_NAME);
}
+// Exposes the condition-cache decision of the current split so a test can
assert that the
+// positional fallback keeps its splits out of the cache.
+class IcebergConditionCacheProbeReader final : public
doris::format::iceberg::IcebergTableReader {
+public:
+ void TEST_force_condition_cache_digest(uint64_t digest) {
+ _condition_cache_digest = digest;
+ _condition_cache_digest_covers_current_split = true;
+ }
+ bool TEST_condition_cache_enabled_for_current_split() const {
+ DORIS_CHECK(_file_scan_request != nullptr);
+ return _should_enable_condition_cache(*_file_scan_request);
+ }
+};
+
+// DORIS-29685: the condition cache key covers the file identity plus the
table-level
+// predicate/session digest only, so it cannot tell two physical bindings of
the same table column
+// apart. Iceberg's positional fallback can change that binding when
+// `schema.name-mapping.default` is added or removed without touching the data
file, so those
+// splits must stay out of the condition cache entirely (neither read nor
written).
+TEST(IcebergV2ReaderTest, PositionalIdFallbackSplitsDisableConditionCache) {
+ const auto run_case = [](bool idless_file, int32_t table_field_id) {
+ const auto test_dir = std::filesystem::temp_directory_path() /
+ (idless_file ?
"doris_iceberg_cache_fallback_test"
+ :
"doris_iceberg_cache_field_id_test");
+ std::filesystem::remove_all(test_dir);
+ std::filesystem::create_directories(test_dir);
+ const auto file_path = (test_dir / "split.parquet").string();
+ if (idless_file) {
+ write_idless_int_columns_parquet_file(file_path, {"legacy_id"},
{{7}});
+ } else {
+ write_single_int_parquet_file(file_path, "id", {7}, 0);
+ }
+
+ std::vector<ColumnDefinition> projected_columns;
+ projected_columns.push_back(
+ make_table_column(table_field_id, "id",
std::make_shared<DataTypeInt32>()));
+
+ RuntimeProfile profile("test_profile");
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ auto scan_params = make_local_parquet_scan_params();
+
scan_params.__set_iceberg_scan_semantics_version(ICEBERG_SCAN_SEMANTICS_VERSION_2);
+ io::FileReaderStats file_reader_stats;
+ io::FileCacheStatistics file_cache_stats;
+ auto io_ctx = make_io_context(&file_reader_stats, &file_cache_stats);
+ ShardedKVCache cache(1);
+ IcebergConditionCacheProbeReader reader;
+ ASSERT_TRUE(reader.init({.projected_columns = projected_columns,
+ .conjuncts = {prepared_conjunct(
+ &state,
table_int32_greater_than_expr(0, 0, 0))},
+ .format = FileFormat::PARQUET,
+ .scan_params = &scan_params,
+ .io_ctx = io_ctx,
+ .runtime_state = &state,
+ .scanner_profile = &profile})
+ .ok());
+
+ auto split_options = build_split_options(file_path);
+ split_options.cache = &cache;
+ split_options.current_range.__set_table_format_params(
+ make_iceberg_table_format_desc(file_path, {}));
+ ASSERT_TRUE(reader.prepare_split(split_options).ok());
+
+ Block block = build_table_block(projected_columns);
+ bool eos = false;
+ ASSERT_TRUE(reader.get_block(&block, &eos).ok());
+ ASSERT_FALSE(eos);
+ // Force a digest so the base decision is not disabled for unrelated
reasons: without the
+ // override an ID-less split would be cacheable here.
+ reader.TEST_force_condition_cache_digest(0x5f5e100);
+ EXPECT_EQ(reader.TEST_condition_cache_enabled_for_current_split(),
!idless_file);
+
+ ASSERT_TRUE(reader.close().ok());
+ std::filesystem::remove_all(test_dir);
+ };
+
+ // A file carrying real field ids keeps the table column bound to the same
physical column, so
+ // the condition cache stays available.
+ run_case(false, 0);
+ // The positional fallback binds by physical ordinal, which the cache key
cannot represent.
+ run_case(true, 1);
+}
+
+// DORIS-29685: an add_files split (no field ids) cherry-picked back into an
evolved table read as a
+// whole all-NULL row because the table no longer carried a usable name
mapping. Iceberg's Parquet
+// reader binds such files by position (`ParquetSchemaUtil.addFallbackIds` +
`pruneColumnsFallback`,
+// `iceberg.netflix.unsafe-parquet-id-fallback.enabled` defaults to true), so
Doris must do the same
+// instead of silently materializing NULL. The physical names deliberately
differ from the table
+// names: the fallback is "file ordinal k == Iceberg field id k", never name
based.
+TEST(IcebergV2ReaderTest,
IcebergIdlessParquetWithoutNameMappingUsesPositionalIds) {
+ const auto test_dir =
+ std::filesystem::temp_directory_path() /
"doris_iceberg_positional_id_fallback_test";
+ std::filesystem::remove_all(test_dir);
+ std::filesystem::create_directories(test_dir);
+ const auto file_path = (test_dir / "split.parquet").string();
+ write_idless_int_columns_parquet_file(file_path, {"legacy_id",
"legacy_value"},
+ {{1, 2, 3}, {10, 20, 30}});
+
+ std::vector<ColumnDefinition> projected_columns;
+ projected_columns.push_back(make_table_column(1, "id",
std::make_shared<DataTypeInt32>()));
+ projected_columns.push_back(make_table_column(2, "value",
std::make_shared<DataTypeInt32>()));
+
+ RuntimeProfile profile("test_profile");
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ auto scan_params = make_local_parquet_scan_params();
+
scan_params.__set_iceberg_scan_semantics_version(ICEBERG_SCAN_SEMANTICS_VERSION_2);
+ io::FileReaderStats file_reader_stats;
+ io::FileCacheStatistics file_cache_stats;
+ auto io_ctx = make_io_context(&file_reader_stats, &file_cache_stats);
+ ShardedKVCache cache(1);
+ doris::format::iceberg::IcebergTableReader reader;
+ init_iceberg_reader(&reader, projected_columns, &scan_params, io_ctx,
&state, &profile);
+
+ auto split_options = build_split_options(file_path);
+ split_options.cache = &cache;
+ split_options.current_range.__set_table_format_params(
+ make_iceberg_table_format_desc(file_path, {}));
+ ASSERT_TRUE(reader.prepare_split(split_options).ok());
+
+ Block block = build_table_block(projected_columns);
+ bool eos = false;
+ ASSERT_TRUE(reader.get_block(&block, &eos).ok());
+ ASSERT_FALSE(eos);
+ EXPECT_NE(reader.debug_string().find("positional_id_fallback=1"),
std::string::npos);
+ ASSERT_EQ(block.rows(), 3);
+ expect_int32_column_values(*block.get_by_position(0).column, {1, 2, 3});
+ expect_int32_column_values(*block.get_by_position(1).column, {10, 20, 30});
+
+ ASSERT_TRUE(reader.close().ok());
+ std::filesystem::remove_all(test_dir);
+}
+
+// Iceberg's fallback matches a table field with id k to the k-th physical
top-level field, so a
+// dropped (or re-added) field does not shift the remaining bindings the way a
table-position rule
+// would. Table ids 1 and 3 must therefore read file ordinals 1 and 3.
+TEST(IcebergV2ReaderTest,
IcebergPositionalFallbackMatchesFieldIdNotColumnPosition) {
+ const auto test_dir =
+ std::filesystem::temp_directory_path() /
"doris_iceberg_positional_id_field_id_test";
+ std::filesystem::remove_all(test_dir);
+ std::filesystem::create_directories(test_dir);
+ const auto file_path = (test_dir / "split.parquet").string();
+ write_idless_int_columns_parquet_file(file_path, {"first", "second",
"third"},
+ {{1, 2}, {10, 20}, {100, 200}});
+
+ std::vector<ColumnDefinition> projected_columns;
+ projected_columns.push_back(make_table_column(1, "first",
std::make_shared<DataTypeInt32>()));
+ projected_columns.push_back(make_table_column(3, "third",
std::make_shared<DataTypeInt32>()));
+
+ RuntimeProfile profile("test_profile");
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ auto scan_params = make_local_parquet_scan_params();
+
scan_params.__set_iceberg_scan_semantics_version(ICEBERG_SCAN_SEMANTICS_VERSION_2);
+ io::FileReaderStats file_reader_stats;
+ io::FileCacheStatistics file_cache_stats;
+ auto io_ctx = make_io_context(&file_reader_stats, &file_cache_stats);
+ ShardedKVCache cache(1);
+ doris::format::iceberg::IcebergTableReader reader;
+ init_iceberg_reader(&reader, projected_columns, &scan_params, io_ctx,
&state, &profile);
+
+ auto split_options = build_split_options(file_path);
+ split_options.cache = &cache;
+ split_options.current_range.__set_table_format_params(
+ make_iceberg_table_format_desc(file_path, {}));
+ ASSERT_TRUE(reader.prepare_split(split_options).ok());
+
+ Block block = build_table_block(projected_columns);
+ bool eos = false;
+ ASSERT_TRUE(reader.get_block(&block, &eos).ok());
+ ASSERT_FALSE(eos);
+ ASSERT_EQ(block.rows(), 2);
+ expect_int32_column_values(*block.get_by_position(0).column, {1, 2});
+ expect_int32_column_values(*block.get_by_position(1).column, {100, 200});
+
+ ASSERT_TRUE(reader.close().ok());
+ std::filesystem::remove_all(test_dir);
+}
+
+// The positional fallback only applies while the table exposes no usable name
mapping. When an
+// authoritative mapping is present, Iceberg applies it and prunes by id: the
covered alias binds,
+// and a field whose mapping says "absent from this file" stays NULL even
though a physical column
+// exists at that ordinal.
+TEST(IcebergV2ReaderTest,
IcebergIdlessParquetKeepsAuthoritativeNameMappingPath) {
+ const auto test_dir = std::filesystem::temp_directory_path() /
+
"doris_iceberg_positional_id_mapping_precedence_test";
+ std::filesystem::remove_all(test_dir);
+ std::filesystem::create_directories(test_dir);
+ const auto file_path = (test_dir / "split.parquet").string();
+ write_idless_int_columns_parquet_file(file_path, {"legacy_id",
"legacy_value"},
+ {{1, 2}, {10, 20}});
+
+ auto mapped_column = make_table_column(1, "id",
std::make_shared<DataTypeInt32>());
+ mapped_column.has_name_mapping = true;
+ mapped_column.name_mapping = {"legacy_id"};
+ auto unmapped_column = make_table_column(2, "value",
std::make_shared<DataTypeInt32>());
+ unmapped_column.has_name_mapping = true;
+ std::vector<ColumnDefinition> projected_columns {mapped_column,
unmapped_column};
+
+ RuntimeProfile profile("test_profile");
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ auto scan_params = make_local_parquet_scan_params();
+
scan_params.__set_iceberg_scan_semantics_version(ICEBERG_SCAN_SEMANTICS_VERSION_2);
+ io::FileReaderStats file_reader_stats;
+ io::FileCacheStatistics file_cache_stats;
+ auto io_ctx = make_io_context(&file_reader_stats, &file_cache_stats);
+ ShardedKVCache cache(1);
+ doris::format::iceberg::IcebergTableReader reader;
+ init_iceberg_reader(&reader, projected_columns, &scan_params, io_ctx,
&state, &profile);
+
+ auto split_options = build_split_options(file_path);
+ split_options.cache = &cache;
+ split_options.current_range.__set_table_format_params(
+ make_iceberg_table_format_desc(file_path, {}));
+ ASSERT_TRUE(reader.prepare_split(split_options).ok());
+
+ Block block = build_table_block(projected_columns);
+ bool eos = false;
+ ASSERT_TRUE(reader.get_block(&block, &eos).ok());
+ ASSERT_FALSE(eos);
+ EXPECT_NE(reader.debug_string().find("positional_id_fallback=0"),
std::string::npos);
+ ASSERT_EQ(block.rows(), 2);
+ expect_int32_column_values(*block.get_by_position(0).column, {1, 2});
+ expect_nullable_int32_column_is_null(*block.get_by_position(1).column, 2);
+
+ ASSERT_TRUE(reader.close().ok());
+ std::filesystem::remove_all(test_dir);
+}
+
+// An old-FE plan (scan semantics version 1) keeps the historical name
fallback for id-less files;
+// the positional fallback must stay opt-in for version 2 so a rolling upgrade
cannot change it.
+TEST(IcebergV2ReaderTest, IcebergLegacyPlanKeepsNameFallbackForIdlessParquet) {
+ const auto test_dir =
+ std::filesystem::temp_directory_path() /
"doris_iceberg_positional_id_legacy_fe_test";
+ std::filesystem::remove_all(test_dir);
+ std::filesystem::create_directories(test_dir);
+ const auto file_path = (test_dir / "split.parquet").string();
+ write_idless_int_columns_parquet_file(file_path, {"id", "value"}, {{1, 2},
{10, 20}});
+
+ std::vector<ColumnDefinition> projected_columns;
+ projected_columns.push_back(make_table_column(1, "id",
std::make_shared<DataTypeInt32>()));
+ projected_columns.push_back(make_table_column(2, "value",
std::make_shared<DataTypeInt32>()));
+
+ RuntimeProfile profile("test_profile");
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ auto scan_params = make_local_parquet_scan_params();
+ io::FileReaderStats file_reader_stats;
+ io::FileCacheStatistics file_cache_stats;
+ auto io_ctx = make_io_context(&file_reader_stats, &file_cache_stats);
+ ShardedKVCache cache(1);
+ doris::format::iceberg::IcebergTableReader reader;
+ init_iceberg_reader(&reader, projected_columns, &scan_params, io_ctx,
&state, &profile);
+
+ auto split_options = build_split_options(file_path);
+ split_options.cache = &cache;
+ split_options.current_range.__set_table_format_params(
+ make_iceberg_table_format_desc(file_path, {}));
+ ASSERT_TRUE(reader.prepare_split(split_options).ok());
+
+ Block block = build_table_block(projected_columns);
+ bool eos = false;
+ ASSERT_TRUE(reader.get_block(&block, &eos).ok());
+ ASSERT_FALSE(eos);
+ EXPECT_NE(reader.debug_string().find("positional_id_fallback=0"),
std::string::npos);
+ ASSERT_EQ(block.rows(), 2);
+ expect_int32_column_values(*block.get_by_position(0).column, {1, 2});
+ expect_int32_column_values(*block.get_by_position(1).column, {10, 20});
+
+ ASSERT_TRUE(reader.close().ok());
+ std::filesystem::remove_all(test_dir);
+}
+
+// Iceberg stamps ordinal ids on top-level fields only, so an id-less nested
file still materializes
+// the struct root by position while its children stay NULL until a name
mapping (or a rewrite that
+// persists field ids) resolves them. Doris must not invent nested positions
either.
+TEST(IcebergV2ReaderTest, IcebergIdlessParquetStructChildStaysNull) {
+ const auto test_dir =
+ std::filesystem::temp_directory_path() /
"doris_iceberg_positional_id_nested_test";
+ std::filesystem::remove_all(test_dir);
+ std::filesystem::create_directories(test_dir);
+ const auto file_path = (test_dir / "split.parquet").string();
+ write_idless_struct_parquet_file(file_path, {7});
+
+ const auto int_type = std::make_shared<DataTypeInt32>();
+ auto inner_type = std::make_shared<DataTypeStruct>(DataTypes {int_type},
Strings {"inner"});
+ auto outer = make_table_column(1, "s", inner_type);
+ outer.children = {make_table_column(2, "inner", int_type)};
+ std::vector<ColumnDefinition> projected_columns {outer};
+
+ RuntimeProfile profile("test_profile");
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ auto scan_params = make_local_parquet_scan_params();
+
scan_params.__set_iceberg_scan_semantics_version(ICEBERG_SCAN_SEMANTICS_VERSION_2);
+ io::FileReaderStats file_reader_stats;
+ io::FileCacheStatistics file_cache_stats;
+ auto io_ctx = make_io_context(&file_reader_stats, &file_cache_stats);
+ ShardedKVCache cache(1);
+ doris::format::iceberg::IcebergTableReader reader;
+ init_iceberg_reader(&reader, projected_columns, &scan_params, io_ctx,
&state, &profile);
+
+ auto split_options = build_split_options(file_path);
+ split_options.cache = &cache;
+ split_options.current_range.__set_table_format_params(
+ make_iceberg_table_format_desc(file_path, {}));
+ ASSERT_TRUE(reader.prepare_split(split_options).ok());
+
+ Block block = build_table_block(projected_columns);
+ bool eos = false;
+ ASSERT_TRUE(reader.get_block(&block, &eos).ok());
+ ASSERT_FALSE(eos);
+ EXPECT_NE(reader.debug_string().find("positional_id_fallback=1"),
std::string::npos);
+ ASSERT_EQ(block.rows(), 1);
+ const auto& struct_column =
+ assert_cast<const
ColumnStruct&>(expect_not_null_table_column(block, 0));
+ expect_nullable_int32_column_is_null(struct_column.get_column(0), 1);
+
+ ASSERT_TRUE(reader.close().ok());
+ std::filesystem::remove_all(test_dir);
+}
+
+// Iceberg's hasIds contract is existential: one real field id keeps the file
on the id path, so
+// neither the name mapping nor the positional fallback may rescue the
remaining id-less fields.
+TEST(IcebergV2ReaderTest, IcebergPartiallyIdedParquetKeepsIdOnlyMatching) {
+ const auto test_dir =
+ std::filesystem::temp_directory_path() /
"doris_iceberg_positional_id_partial_ids_test";
+ std::filesystem::remove_all(test_dir);
+ std::filesystem::create_directories(test_dir);
+ const auto file_path = (test_dir / "split.parquet").string();
+ write_two_int_parquet_file(file_path, "id", {1, 2}, 0, "extra", {9, 9},
std::nullopt);
+
+ std::vector<ColumnDefinition> projected_columns;
+ projected_columns.push_back(make_table_column(0, "id",
std::make_shared<DataTypeInt32>()));
+ projected_columns.push_back(make_table_column(1, "extra",
std::make_shared<DataTypeInt32>()));
+
+ RuntimeProfile profile("test_profile");
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ auto scan_params = make_local_parquet_scan_params();
+
scan_params.__set_iceberg_scan_semantics_version(ICEBERG_SCAN_SEMANTICS_VERSION_2);
+ io::FileReaderStats file_reader_stats;
+ io::FileCacheStatistics file_cache_stats;
+ auto io_ctx = make_io_context(&file_reader_stats, &file_cache_stats);
+ ShardedKVCache cache(1);
+ doris::format::iceberg::IcebergTableReader reader;
+ init_iceberg_reader(&reader, projected_columns, &scan_params, io_ctx,
&state, &profile);
+
+ auto split_options = build_split_options(file_path);
+ split_options.cache = &cache;
+ split_options.current_range.__set_table_format_params(
+ make_iceberg_table_format_desc(file_path, {}));
+ ASSERT_TRUE(reader.prepare_split(split_options).ok());
+
+ Block block = build_table_block(projected_columns);
+ bool eos = false;
+ ASSERT_TRUE(reader.get_block(&block, &eos).ok());
+ ASSERT_FALSE(eos);
+ EXPECT_NE(reader.debug_string().find("positional_id_fallback=0"),
std::string::npos);
+ ASSERT_EQ(block.rows(), 2);
+ expect_int32_column_values(*block.get_by_position(0).column, {1, 2});
+ expect_nullable_int32_column_is_null(*block.get_by_position(1).column, 2);
+
+ ASSERT_TRUE(reader.close().ok());
+ std::filesystem::remove_all(test_dir);
+}
+
TEST(IcebergV2ReaderTest, VariantFormatGateUsesPhysicalFileMappings) {
ColumnMapping missing_variant;
missing_variant.table_type =
make_nullable(std::make_shared<DataTypeVariantV2>());
diff --git
a/regression-test/data/external_table_p0/iceberg/test_iceberg_migrated_nested_without_name_mapping.out
b/regression-test/data/external_table_p0/iceberg/test_iceberg_migrated_nested_without_name_mapping.out
index 9aea1c252d0..df25da67760 100644
---
a/regression-test/data/external_table_p0/iceberg/test_iceberg_migrated_nested_without_name_mapping.out
+++
b/regression-test/data/external_table_p0/iceberg/test_iceberg_migrated_nested_without_name_mapping.out
@@ -1,3 +1,6 @@
-- This file is automatically generated. You should know what you did if you
want to edit this
--- !idless_nested_fields_are_missing --
-\N \N \N
+-- !idless_top_level_field_uses_positional_id --
+1 \N \N
+
+-- !idless_top_level_field_filter_uses_positional_id --
+1
diff --git
a/regression-test/suites/external_table_p0/iceberg/test_gen_iceberg_by_api.groovy
b/regression-test/suites/external_table_p0/iceberg/test_gen_iceberg_by_api.groovy
index 0227dbf95eb..9ae1107a1d9 100644
---
a/regression-test/suites/external_table_p0/iceberg/test_gen_iceberg_by_api.groovy
+++
b/regression-test/suites/external_table_p0/iceberg/test_gen_iceberg_by_api.groovy
@@ -48,12 +48,18 @@ suite("test_gen_iceberg_by_api", "p0,external") {
def q01 = {
qt_q01 """ select * from multi_partition2 order by val """
- test {
- sql """ select count(*) from table_with_append_file where MAN_ID
is not null """
- // This fixture has no field IDs or authoritative name mapping, so
its required
- // columns must be treated as missing instead of being matched by
their current names.
- exception "Missing required field: MAN_ID"
- }
+ // This fixture has no field IDs and no authoritative name mapping.
Iceberg's Parquet reader
+ // binds such a file by physical ordinals
(ParquetSchemaUtil.addFallbackIds +
+ // pruneColumnsFallback), so a required column is no longer reported
as missing and the
+ // appended record must stay readable. Exact values of ID-less
fixtures are pinned by
+ // test_iceberg_migrated_nested_without_name_mapping.
+ def appendedRows = sql """ select count(*) from table_with_append_file
"""
+ assertTrue(appendedRows.size() == 1 &&
appendedRows[0][0].toString().toInteger() >= 1,
+ "ID-less append fixture must stay readable, got
${appendedRows}")
+
+ // The positional binding resolves MAN_ID to one of the file's
physical columns; the query
+ // must no longer fail with "Missing required field: MAN_ID".
+ sql """ select count(*) from table_with_append_file where MAN_ID is
not null """
}
q01()
diff --git
a/regression-test/suites/external_table_p0/iceberg/test_iceberg_migrated_nested_without_name_mapping.groovy
b/regression-test/suites/external_table_p0/iceberg/test_iceberg_migrated_nested_without_name_mapping.groovy
index 8f1ceec0906..546d98d8f68 100644
---
a/regression-test/suites/external_table_p0/iceberg/test_iceberg_migrated_nested_without_name_mapping.groovy
+++
b/regression-test/suites/external_table_p0/iceberg/test_iceberg_migrated_nested_without_name_mapping.groovy
@@ -43,13 +43,23 @@ suite("test_iceberg_migrated_nested_without_name_mapping",
"p0,external") {
sql """use multi_catalog"""
sql """set enable_fallback_to_original_planner=false"""
- // The physical Parquet fields predate migration and have no Iceberg IDs.
Once the explicit
- // default name mapping is removed, neither the root nor nested values may
match by current name.
- qt_idless_nested_fields_are_missing """
+ // The physical Parquet fields predate migration and carry no Iceberg IDs,
and this registered
+ // metadata has no `schema.name-mapping.default`. Iceberg's Parquet reader
binds such a file by
+ // position (ParquetSchemaUtil.addFallbackIds + pruneColumnsFallback, see
DORIS-29685): the k-th
+ // physical top-level field is field id k, so `id` (field id 1) must read
its real value instead
+ // of NULL. Fallback ids are stamped on top-level fields only, so the
nested children stay NULL
+ // until a name mapping is restored.
+ qt_idless_top_level_field_uses_positional_id """
SELECT id, nested_struct.value_a, nested_struct.value_b
FROM migrated_nested_without_name_mapping
ORDER BY id
"""
+ // The same positional binding must feed predicate/row-group filtering,
otherwise the row would
+ // be readable but unfilterable.
+ qt_idless_top_level_field_filter_uses_positional_id """
+ SELECT count(*) FROM migrated_nested_without_name_mapping WHERE id = 1
+ """
+
sql """drop catalog if exists ${catalog_name}"""
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
