This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.2 by this push:
new de40eaee9e1 branch-4.2: [fix](iceberg) Read ID-less Parquet files by
positional field ids (#68962) (#69839)
de40eaee9e1 is described below
commit de40eaee9e132f580ae00deb023fdb46d24ae885
Author: daidai <[email protected]>
AuthorDate: Thu Nov 5 16:48:27 2026 +0800
branch-4.2: [fix](iceberg) Read ID-less Parquet files by positional field
ids (#68962) (#69839)
pick #68962
---
be/src/format_v2/column_mapper.cpp | 86 +++-
be/src/format_v2/table/iceberg_reader.cpp | 33 ++
be/src/format_v2/table/iceberg_reader.h | 20 +-
be/src/format_v2/table_reader.h | 7 +-
be/test/format_v2/column_mapper_test.cpp | 246 ++++++++++
be/test/format_v2/table/iceberg_reader_test.cpp | 495 ++++++++++++++++++++-
...ceberg_migrated_nested_without_name_mapping.out | 7 +-
.../iceberg/test_gen_iceberg_by_api.groovy | 16 +-
...erg_migrated_nested_without_name_mapping.groovy | 18 +-
9 files changed, 893 insertions(+), 35 deletions(-)
diff --git a/be/src/format_v2/column_mapper.cpp
b/be/src/format_v2/column_mapper.cpp
index 5839ececd7c..baa84ad5c7c 100644
--- a/be/src/format_v2/column_mapper.cpp
+++ b/be/src/format_v2/column_mapper.cpp
@@ -526,6 +526,10 @@ std::string TableColumnMapper::debug_string() const {
return out.str();
}
+// Defined with the mapping-shape helpers below. Whole-root shortcuts must use
the same
+// missing-child contract as the generic slot rewrite.
+static bool mapping_slot_has_unmapped_child(const ColumnMapping& mapping);
+
static const FileSlotRewriteInfo* find_slot_rewrite_info(
const VExprSPtr& expr,
const std::map<GlobalIndex, FileSlotRewriteInfo>& global_to_file_slot,
@@ -535,7 +539,7 @@ static const FileSlotRewriteInfo* find_slot_rewrite_info(
}
VExprSPtr slot_expr = expr;
const bool input_is_cast = is_cast_expr(expr) && expr->get_num_children()
== 1;
- if (is_cast_expr(expr) && expr->get_num_children() == 1) {
+ if (input_is_cast) {
slot_expr = expr->children()[0];
}
if (!slot_expr->is_slot_ref()) {
@@ -549,6 +553,13 @@ static const FileSlotRewriteInfo* find_slot_rewrite_info(
if (input_is_cast &&
!expr->data_type()->equals(*rewrite_it->second.table_type)) {
return nullptr;
}
+ DORIS_CHECK(rewrite_it->second.root_mapping != nullptr);
+ if (mapping_slot_has_unmapped_child(*rewrite_it->second.root_mapping)) {
+ // A file slot with missing descendants is not the finalized table
value. Let the generic
+ // slot rewrite retain the predicate above TableReader, including
binary/IN and cast
+ // shortcuts that would otherwise return before visiting the slot.
+ return nullptr;
+ }
if (slot_ref != nullptr) {
*slot_ref = candidate_slot_ref;
}
@@ -965,6 +976,22 @@ static const ColumnMapping*
find_projected_child_mapping(const ColumnMapping& ma
return child_it == mapping.child_mappings.end() ? nullptr : &*child_it;
}
+static bool nested_result_has_unmapped_child(const ColumnMapping& mapping,
+ const LocalColumnIndex& path) {
+ if (path.children.empty()) {
+ return mapping_slot_has_unmapped_child(mapping);
+ }
+ DORIS_CHECK(path.children.size() == 1);
+ const auto& child_path = path.children.front();
+ const auto* child = find_projected_child_mapping(mapping,
child_path.local_id());
+ if (child == nullptr) {
+ // Schema-only descriptors can resolve a path without child mappings.
They carry no
+ // missing-child metadata; the existing type/nullability checks still
apply to that path.
+ return false;
+ }
+ return nested_result_has_unmapped_child(*child, child_path);
+}
+
static bool projected_mapping_allows_file_filtering(const ColumnMapping&
mapping,
const LocalColumnIndex*
projection) {
if (!can_filter_before_table_nullability_alignment(mapping.file_type,
mapping.table_type)) {
@@ -1051,6 +1078,13 @@ static bool rewrite_struct_element_path_to_file_expr(
&rewrite_it->second.scan_projection)) {
return false;
}
+ if (nested_result_has_unmapped_child(*rewrite_it->second.root_mapping,
+ resolved.file_projection)) {
+ // A selected complex value must include its default/NULL-filled
descendants before any
+ // wrapper consumes it. Missing siblings outside the selected result
do not affect a mapped
+ // scalar leaf, so check the result subtree rather than the whole root.
+ return false;
+ }
for (size_t idx = 0; idx < struct_element_chain.size(); ++idx) {
// Accessor results become nullable for missing ARRAY indices and NULL
parents. Compare the
// file child with the declared table child instead, or that
execution-only wrapper can
@@ -1285,6 +1319,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;
@@ -1347,25 +1390,19 @@ static VExprSPtr rewrite_table_expr_to_file_expr(
// inserted Cast(slot). Keep that rewrite idempotent: rewrite the cast
child from table slot to
// the current split's file slot, and drop the cast when the current split
no longer needs it.
if (is_cast_expr(expr) && expr->get_num_children() == 1) {
- const auto& child = expr->children()[0];
- if (child->is_slot_ref()) {
- const auto* slot_ref = assert_cast<const VSlotRef*>(child.get());
- const auto rewrite_it =
global_to_file_slot.find(slot_ref_global_index(*slot_ref));
- if (rewrite_it != global_to_file_slot.end() &&
- expr->data_type()->equals(*rewrite_it->second.table_type)) {
- auto rewritten_child =
- create_file_slot_ref(*slot_ref, rewrite_it->second,
rewrite_context);
- if
(rewrite_it->second.file_type->equals(*rewrite_it->second.table_type)) {
- return rewritten_child;
- }
- if (needs_complex_file_slot_cast(rewrite_it->second.file_type,
-
rewrite_it->second.table_type)) {
- *can_localize = false;
- return expr;
- }
- expr->set_children({std::move(rewritten_child)});
+ const VSlotRef* slot_ref = nullptr;
+ const auto* rewrite_info = find_slot_rewrite_info(expr,
global_to_file_slot, &slot_ref);
+ if (rewrite_info != nullptr) {
+ auto rewritten_child = create_file_slot_ref(*slot_ref,
*rewrite_info, rewrite_context);
+ if (rewrite_info->file_type->equals(*rewrite_info->table_type)) {
+ return rewritten_child;
+ }
+ if (needs_complex_file_slot_cast(rewrite_info->file_type,
rewrite_info->table_type)) {
+ *can_localize = false;
return expr;
}
+ expr->set_children({std::move(rewritten_child)});
+ return expr;
}
}
@@ -1497,6 +1534,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;
@@ -1509,6 +1556,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 03de2fa308c..2d999089462 100644
--- a/be/src/format_v2/table/iceberg_reader.cpp
+++ b/be/src/format_v2/table/iceberg_reader.cpp
@@ -138,6 +138,37 @@ bool
IcebergTableReader::_scan_has_any_authoritative_name_mapping() const {
return false;
}
+Status IcebergTableReader::annotate_file_schema(
+ std::vector<format::ColumnDefinition>* file_schema) {
+ DORIS_CHECK(file_schema != nullptr);
+ if (file_schema->empty() || _format != FileFormat::PARQUET) {
+ return Status::OK();
+ }
+ // Iceberg's hasIds is existential: even a single nested ID prevents
positional fallback.
+ if (schema_has_any_field_id(*file_schema)) {
+ return Status::OK();
+ }
+ // An explicit name mapping remains authoritative even when it does not
cover this file.
+ // Legacy FE plans keep their historical name fallback during rolling
upgrades.
+ if (!supports_iceberg_scan_semantics_v2(_scan_params) ||
+ _scan_has_any_authoritative_name_mapping()) {
+ return Status::OK();
+ }
+ // ParquetSchemaUtil.addFallbackIds numbers physical top-level fields
only. Reader-generated
+ // row-position/global-row-ID columns do not consume ordinals, and nested
IDs are not invented.
+ int32_t ordinal = 0;
+ for (auto& field : *file_schema) {
+ if (field.column_type != format::ColumnType::DATA_COLUMN) {
+ continue;
+ }
+ // ID-less Parquet fields have STRING identifiers containing their
physical names.
+ // Replace those identifiers so the shared BY_FIELD_ID projection and
predicates agree.
+ field.identifier = Field::create_field<TYPE_INT>(++ordinal);
+ }
+ _positional_id_fallback = true;
+ return Status::OK();
+}
+
template <typename T>
static std::string join_values_for_debug(const std::vector<T>& values) {
std::ostringstream out;
@@ -999,6 +1030,7 @@ Status IcebergTableReader::prepare_split(const
format::SplitReadOptions& options
_row_lineage_columns = {};
_iceberg_params.reset();
_delete_predicates_initialized = false;
+ _positional_id_fallback = false;
_position_delete_rows_storage.clear();
_equality_delete_filters.clear();
_split_cache = options.cache;
@@ -1091,6 +1123,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 68279112493..3aecd776fc1 100644
--- a/be/src/format_v2/table/iceberg_reader.h
+++ b/be/src/format_v2/table/iceberg_reader.h
@@ -75,8 +75,8 @@ public:
}
if (!_data_reader.file_schema.empty() &&
supports_iceberg_scan_semantics_v2(_scan_params) &&
!_scan_has_any_authoritative_name_mapping()) {
- // ID-less migrated files are name-readable only while Iceberg's
explicit default name
- // mapping exists; current names must not resurrect file fields
after it is removed.
+ // Parquet annotation supplies positional IDs. Other ID-less files
remain unreadable
+ // without an explicit default name mapping; current names must
not resurrect fields.
return format::TableColumnMappingMode::BY_FIELD_ID;
}
return format::TableColumnMappingMode::BY_NAME;
@@ -91,6 +91,20 @@ protected:
// Iceberg UUID uses the same 16-byte STRING/VARBINARY carrier in data,
defaults and deletes.
bool preserve_binary_uuid() const override { return true; }
+ // Match Iceberg's ParquetSchemaUtil.addFallbackIds: when a Parquet file
has neither field IDs
+ // nor an authoritative name mapping, annotate only its top-level fields
with IDs 1..N.
+ // Nested children keep their original identifiers and remain missing
without a name mapping.
+ // Legacy FE plans retain name matching, and ORC retains its existing
behavior.
+ Status annotate_file_schema(std::vector<format::ColumnDefinition>*
file_schema) override;
+
+ // Changing the name mapping can change a positional binding without
changing the file or
+ // table-predicate digest. Such splits must neither read nor publish
cached filter results.
+ 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->allow_idless_complex_wrapper_projection =
@@ -203,6 +217,8 @@ private:
size_t _row_position_block_position = 0;
std::optional<TIcebergFileDesc> _iceberg_params;
bool _delete_predicates_initialized = false;
+ // Per-split positional binding state, used by condition-cache eligibility
and debug_string().
+ 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 1f8a6d9b29a..7ecf39b6f18 100644
--- a/be/src/format_v2/table_reader.h
+++ b/be/src/format_v2/table_reader.h
@@ -578,7 +578,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 4d80ab209ac..fa3b183b2fb 100644
--- a/be/test/format_v2/column_mapper_test.cpp
+++ b/be/test/format_v2/column_mapper_test.cpp
@@ -40,6 +40,7 @@
#include "core/data_type/data_type_timestamptz.h"
#include "core/data_type/data_type_varbinary.h"
#include "core/data_type/data_type_variant_v2.h"
+#include "exprs/runtime_filter_expr.h"
#include "exprs/vectorized_fn_call.h"
#include "exprs/vexpr.h"
#include "exprs/vexpr_context.h"
@@ -656,6 +657,73 @@ VExprSPtr create_in_predicate() {
return VInPredicate::create_shared(node);
}
+std::vector<std::pair<std::string, VExprSPtr>> whole_struct_filter_exprs(const
DataTypePtr& type,
+ const
Field& inner_value) {
+ const auto slot = [&] { return table_slot(11, 0, type, "s"); };
+ const auto value = [&] {
+ return literal(type, Field::create_field<TYPE_STRUCT>(Struct
{inner_value}));
+ };
+ const auto in = [&](const VExprSPtr& probe) {
+ auto predicate = create_in_predicate();
+ predicate->add_child(probe);
+ predicate->add_child(value());
+ return predicate;
+ };
+ return {
+ {"binary", binary_predicate(TExprOpcode::EQ, slot(), value())},
+ {"null_safe_binary", binary_predicate(TExprOpcode::EQ_FOR_NULL,
slot(), value())},
+ {"reverse_binary", binary_predicate(TExprOpcode::EQ, value(),
slot())},
+ {"in", in(slot())},
+ {"cast_binary", binary_predicate(TExprOpcode::EQ,
cast_expr(slot(), type), value())},
+ {"cast_in", in(cast_expr(slot(), type))},
+ {"same_type_cast", like_expr(cast_expr(cast_expr(slot(), type),
str()),
+ inner_value.is_null() ? "%null%" :
"%0%")},
+ };
+}
+
+VExprSPtr runtime_filter_wrapper_expr(VExprSPtr impl) {
+ TExprNode node;
+ node.__set_node_type(TExprNodeType::SLOT_REF);
+ node.__set_type(u8()->to_thrift());
+ node.__set_num_children(1);
+ return RuntimeFilterExpr::create_shared(node, std::move(impl), 0.0, false,
/*filter_id=*/1);
+}
+
+void expect_residual_whole_struct_filter(const FileScanRequest& request) {
+ EXPECT_TRUE(request.conjuncts.empty());
+ EXPECT_EQ(request.metadata_pruning_safe_conjunct_count, 0);
+ EXPECT_EQ(request.constant_pruning_safe_table_filter_count, 0);
+ EXPECT_TRUE(request.predicate_columns.empty());
+ ASSERT_EQ(request.non_predicate_columns.size(), 1);
+ EXPECT_EQ(request.non_predicate_columns[0].column_id(), LocalColumnId(7));
+ ASSERT_EQ(request.residual_predicate_columns.size(), 1);
+ EXPECT_EQ(request.residual_predicate_columns[0], LocalColumnId(7));
+}
+
+void expect_localized_whole_struct_filter(const FileScanRequest& request) {
+ EXPECT_EQ(request.conjuncts.size(), 1);
+ EXPECT_TRUE(request.residual_predicate_columns.empty());
+ ASSERT_EQ(request.predicate_columns.size(), 1);
+ EXPECT_EQ(request.predicate_columns[0].column_id(), LocalColumnId(7));
+}
+
+void check_struct_filter(const ColumnDefinition& table_struct, const
ColumnDefinition& file_struct,
+ const VExprSPtr& predicate, bool runtime_filter, bool
should_localize) {
+ TableColumnMapper mapper({.mode = TableColumnMappingMode::BY_FIELD_ID});
+ ASSERT_TRUE(mapper.create_mapping({table_struct}, {}, {file_struct}).ok());
+ auto root = runtime_filter ? runtime_filter_wrapper_expr(predicate) :
predicate;
+ TableFilter filter {.conjunct =
VExprContext::create_shared(std::move(root)),
+ .global_indices = {GlobalIndex(0)}};
+ FileScanRequest request;
+ ASSERT_TRUE(mapper.create_scan_request({filter}, {table_struct},
&request).ok());
+
+ if (should_localize) {
+ expect_localized_whole_struct_filter(request);
+ } else {
+ expect_residual_whole_struct_filter(request);
+ }
+}
+
// ----------------------------------------------------------------------
// L0 schema projection helper tests.
// These tests isolate LocalColumnIndex projection semantics before
@@ -2103,6 +2171,184 @@ 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());
+}
+
+// A physical s.inner=5 must not discard a row whose logical s.inner is a
missing child's
+// default (0) or NULL. Exercise literal shortcuts, split-local same-type
casts, and RF wrappers.
+TEST(ColumnMapperLocalizeFiltersTest,
PartiallyMappedWholeStructShortcutsStayAboveFileReader) {
+ const auto child_type = make_nullable(i32());
+ for (const bool has_default : {true, false}) {
+ auto child = field_id_col("inner", 2, child_type);
+ const auto zero = Field::create_field<TYPE_INT>(0);
+ if (has_default) {
+ child.default_expr =
VExprContext::create_shared(literal(child_type, zero));
+ }
+ const auto table_struct = struct_col("s", 1, {child});
+ const auto file_struct = struct_col("s", 1, {name_col("inner",
child_type, 0)}, 7);
+ for (const auto& [name, predicate] :
+ whole_struct_filter_exprs(table_struct.type, has_default ? zero :
Field {})) {
+ for (const bool runtime_filter : {false, true}) {
+ SCOPED_TRACE(::testing::Message() << name << " default=" <<
has_default
+ << " runtime_filter=" <<
runtime_filter);
+ check_struct_filter(table_struct, file_struct, predicate,
runtime_filter,
+ /*should_localize=*/false);
+ }
+ }
+ }
+}
+
+TEST(ColumnMapperLocalizeFiltersTest,
FullyMappedWholeStructShortcutsRemainLocalized) {
+ const auto child = field_id_col("inner", 2, make_nullable(i32()));
+ const auto table_struct = struct_col("s", 1, {child});
+ const auto file_struct = struct_col("s", 1, {field_id_col("inner", 2,
child.type, 0)}, 7);
+ for (const auto& [name, predicate] :
+ whole_struct_filter_exprs(table_struct.type,
Field::create_field<TYPE_INT>(0))) {
+ for (const bool runtime_filter : {false, true}) {
+ SCOPED_TRACE(::testing::Message() << name << " runtime_filter=" <<
runtime_filter);
+ check_struct_filter(table_struct, file_struct, predicate,
runtime_filter,
+ /*should_localize=*/true);
+ }
+ }
+}
+
+TEST(ColumnMapperLocalizeFiltersTest,
MappedStructLeafWithMissingSiblingRemainsLocalized) {
+ auto missing = field_id_col("missing", 3, i32());
+ missing.default_expr =
+ VExprContext::create_shared(literal(i32(),
Field::create_field<TYPE_INT>(0)));
+ const auto table_struct = struct_col("s", 1, {field_id_col("kept", 2,
i32()), missing});
+ const auto file_struct = struct_col("s", 1, {field_id_col("kept", 2,
i32(), 0)}, 7);
+ auto leaf = struct_element(table_slot(11, 0, table_struct.type, "s"),
i32(), "kept");
+ TableFilter filter {.conjunct = VExprContext::create_shared(int_gt(leaf,
0)),
+ .global_indices = {GlobalIndex(0)}};
+ TableColumnMapper mapper({.mode = TableColumnMappingMode::BY_FIELD_ID});
+ ASSERT_TRUE(mapper.create_mapping({table_struct}, {}, {file_struct}).ok());
+
ASSERT_FALSE(mapper.mappings()[0].child_mappings[1].file_local_id.has_value());
+
+ FileScanRequest request;
+ ASSERT_TRUE(mapper.create_scan_request({filter}, {table_struct},
&request).ok());
+
+ EXPECT_EQ(request.conjuncts.size(), 1);
+ EXPECT_TRUE(request.residual_predicate_columns.empty());
+}
+
+TEST(ColumnMapperLocalizeFiltersTest,
PartiallyMappedNestedStructResultStaysAboveFileReader) {
+ for (const bool has_default : {true, false}) {
+ auto inner = field_id_col("inner", 3, make_nullable(i32()));
+ if (has_default) {
+ inner.default_expr = VExprContext::create_shared(
+ literal(inner.type, Field::create_field<TYPE_INT>(0)));
+ }
+ const auto nested = struct_col("a", 2, {inner});
+ const auto table_struct = struct_col("s", 1, {nested});
+ const auto file_nested = struct_col("a", 2, {name_col("inner",
inner.type, 0)}, 0);
+ const auto file_struct = struct_col("s", 1, {file_nested}, 7);
+ for (const bool use_element_at : {false, true}) {
+ for (const bool same_type_cast : {false, true}) {
+ for (const bool runtime_filter : {false, true}) {
+ SCOPED_TRACE(::testing::Message()
+ << "default=" << has_default << "
element_at=" << use_element_at
+ << " same_type_cast=" << same_type_cast
+ << " runtime_filter=" << runtime_filter);
+ auto slot = table_slot(11, 0, table_struct.type, "s");
+ auto selected = use_element_at ? element_at(slot,
nested.type, "a")
+ : struct_element(slot,
nested.type, "a");
+ if (same_type_cast) {
+ selected = cast_expr(selected, nested.type);
+ }
+ auto predicate =
+ like_expr(cast_expr(selected, str()), has_default
? "%0%" : "%null%");
+ check_struct_filter(table_struct, file_struct, predicate,
runtime_filter,
+ /*should_localize=*/false);
+ }
+ }
+ }
+ }
+}
+
+TEST(ColumnMapperLocalizeFiltersTest,
FullyMappedNestedStructResultRemainsLocalized) {
+ const auto inner = field_id_col("inner", 3, make_nullable(i32()));
+ const auto nested = struct_col("a", 2, {inner});
+ const auto table_struct = struct_col("s", 1, {nested});
+ const auto file_nested = struct_col("a", 2, {field_id_col("inner", 3,
inner.type, 0)}, 0);
+ const auto file_struct = struct_col("s", 1, {file_nested}, 7);
+ for (const bool use_element_at : {false, true}) {
+ for (const bool same_type_cast : {false, true}) {
+ for (const bool runtime_filter : {false, true}) {
+ SCOPED_TRACE(::testing::Message() << "element_at=" <<
use_element_at
+ << " same_type_cast=" <<
same_type_cast
+ << " runtime_filter=" <<
runtime_filter);
+ auto slot = table_slot(11, 0, table_struct.type, "s");
+ auto selected = use_element_at ? element_at(slot, nested.type,
"a")
+ : struct_element(slot,
nested.type, "a");
+ if (same_type_cast) {
+ selected = cast_expr(selected, nested.type);
+ }
+ auto predicate = like_expr(cast_expr(selected, str()), "%0%");
+ check_struct_filter(table_struct, file_struct, predicate,
runtime_filter,
+ /*should_localize=*/true);
+ }
+ }
+ }
+}
+
+TEST(ColumnMapperLocalizeFiltersTest,
MappedNestedStructLeafWithMissingSiblingRemainsLocalized) {
+ auto missing = field_id_col("missing", 4, i32());
+ missing.default_expr =
+ VExprContext::create_shared(literal(i32(),
Field::create_field<TYPE_INT>(0)));
+ const auto nested = struct_col("a", 2, {field_id_col("kept", 3, i32()),
missing});
+ const auto table_struct = struct_col("s", 1, {nested});
+ const auto file_nested = struct_col("a", 2, {field_id_col("kept", 3,
i32(), 0)}, 0);
+ const auto file_struct = struct_col("s", 1, {file_nested}, 7);
+ for (const bool runtime_filter : {false, true}) {
+ auto selected = element_at(table_slot(11, 0, table_struct.type, "s"),
nested.type, "a");
+ auto leaf = element_at(selected, i32(), "kept");
+ check_struct_filter(table_struct, file_struct, int_gt(leaf, 0),
runtime_filter,
+ /*should_localize=*/true);
+ }
+}
+
+TEST(ColumnMapperLocalizeFiltersTest,
SchemaOnlyNestedStructResultRemainsLocalized) {
+ const auto inner = field_id_col("inner", 3, make_nullable(i32()));
+ const auto nested = struct_col("a", 2, {inner});
+ auto table_struct = struct_col("s", 1, {nested});
+ table_struct.children.clear();
+ const auto file_nested = struct_col("a", 2, {field_id_col("inner", 3,
inner.type, 0)}, 0);
+ const auto file_struct = struct_col("s", 1, {file_nested}, 7);
+ auto selected = element_at(table_slot(11, 0, table_struct.type, "s"),
nested.type, "a");
+ check_struct_filter(table_struct, file_struct,
like_expr(cast_expr(selected, str()), "%0%"),
+ /*runtime_filter=*/false, /*should_localize=*/true);
+}
+
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 df941721cdf..a2bf5ee031d 100644
--- a/be/test/format_v2/table/iceberg_reader_test.cpp
+++ b/be/test/format_v2/table/iceberg_reader_test.cpp
@@ -248,6 +248,13 @@ private:
class IcebergTableReaderMappingModeTestHelper final
: public doris::format::iceberg::IcebergTableReader {
public:
+ Status annotate_schema(std::vector<ColumnDefinition>* file_schema,
+ TFileScanRangeParams* scan_params, FileFormat
format) {
+ _scan_params = scan_params;
+ _format = format;
+ return annotate_file_schema(file_schema);
+ }
+
TableColumnMappingMode mapping_mode_for_schema(
std::vector<ColumnDefinition> file_schema, TFileScanRangeParams*
scan_params = nullptr,
std::vector<ColumnDefinition> projected_columns = {}) {
@@ -650,6 +657,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,
@@ -1322,6 +1378,15 @@ void expect_iceberg_rowid_column_values(const IColumn&
column, const std::string
}
}
+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_int32_column_values(const IColumn& column,
const std::vector<int32_t>& expected_values) {
const auto full_column = column.convert_to_full_column_if_const();
@@ -3400,7 +3465,8 @@ TEST(IcebergV2ReaderTest,
IcebergLegacyPlanKeepsAllFieldIdsMappingRule) {
TableColumnMappingMode::BY_NAME);
}
-TEST(IcebergV2ReaderTest, IcebergV2IdlessFileRequiresAuthoritativeNameMapping)
{
+// Parquet positional annotation keeps ID-less files on the field-ID path
without a name mapping.
+TEST(IcebergV2ReaderTest, IcebergV2IdlessFileMappingModeSelection) {
IcebergTableReaderMappingModeTestHelper reader;
TFileScanRangeParams scan_params;
scan_params.__set_iceberg_scan_semantics_version(ICEBERG_SCAN_SEMANTICS_VERSION_2);
@@ -3419,6 +3485,433 @@ TEST(IcebergV2ReaderTest,
IcebergV2IdlessFileRequiresAuthoritativeNameMapping) {
TableColumnMappingMode::BY_NAME);
}
+TEST(IcebergV2ReaderTest,
IcebergPositionalIdsSkipVirtualColumnsAndNestedFields) {
+ IcebergTableReaderMappingModeTestHelper reader;
+ TFileScanRangeParams scan_params;
+
scan_params.__set_iceberg_scan_semantics_version(ICEBERG_SCAN_SEMANTICS_VERSION_2);
+ auto first = make_file_column(1, "first",
std::make_shared<DataTypeInt32>());
+ first.identifier = Field::create_field<TYPE_STRING>("first");
+ auto child = first;
+ child.local_id = 3;
+ auto second = first;
+ second.local_id = 2;
+ second.name = "second";
+ second.identifier = Field::create_field<TYPE_STRING>("second");
+ second.type = std::make_shared<DataTypeStruct>(DataTypes {child.type},
Strings {child.name});
+ second.children = {child};
+ std::vector<ColumnDefinition> file_schema
{global_rowid_column_definition(), first,
+
row_position_column_definition(), second};
+
+ ASSERT_TRUE(reader.annotate_schema(&file_schema, &scan_params,
FileFormat::PARQUET).ok());
+ EXPECT_EQ(file_schema[0].column_type, ColumnType::GLOBAL_ROWID);
+ EXPECT_EQ(file_schema[1].get_identifier_field_id(), 1);
+ EXPECT_EQ(file_schema[1].local_id, 1);
+ EXPECT_EQ(file_schema[2].column_type, ColumnType::ROW_NUMBER);
+ EXPECT_EQ(file_schema[3].get_identifier_field_id(), 2);
+ EXPECT_EQ(file_schema[3].local_id, 2);
+ EXPECT_FALSE(file_schema[3].children[0].has_identifier_field_id());
+}
+
+TEST(IcebergV2ReaderTest, IcebergPositionalIdsExcludeOrcAndPartiallyIdedFiles)
{
+ IcebergTableReaderMappingModeTestHelper reader;
+ TFileScanRangeParams scan_params;
+
scan_params.__set_iceberg_scan_semantics_version(ICEBERG_SCAN_SEMANTICS_VERSION_2);
+ auto root = make_file_column(1, "root", std::make_shared<DataTypeInt32>());
+ root.identifier = Field::create_field<TYPE_STRING>("root");
+ std::vector<ColumnDefinition> file_schema {root};
+ ASSERT_TRUE(reader.annotate_schema(&file_schema, &scan_params,
FileFormat::ORC).ok());
+ EXPECT_FALSE(file_schema[0].has_identifier_field_id());
+
+ // A real ID anywhere in the physical tree prevents fallback for the
entire file.
+ file_schema[0].children = {make_file_column(2, "child",
std::make_shared<DataTypeInt32>())};
+ file_schema[0].type = std::make_shared<DataTypeStruct>(
+ DataTypes {file_schema[0].children[0].type}, Strings {"child"});
+ ASSERT_TRUE(reader.annotate_schema(&file_schema, &scan_params,
FileFormat::PARQUET).ok());
+ EXPECT_FALSE(file_schema[0].has_identifier_field_id());
+ EXPECT_EQ(file_schema[0].children[0].get_identifier_field_id(), 2);
+}
+
+// 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});
+
+ // A reused reader must clear the positional binding state before the next
ID-bearing split.
+ block.clear_column_data();
+ ASSERT_TRUE(reader.get_block(&block, &eos).ok());
+ ASSERT_TRUE(eos);
+ const auto next_file_path = (test_dir / "with-ids.parquet").string();
+ write_two_int_parquet_file(next_file_path, "id", {7}, 1, "value", {70}, 2);
+ split_options = build_split_options(next_file_path);
+ split_options.cache = &cache;
+ split_options.current_range.__set_table_format_params(
+ make_iceberg_table_format_desc(next_file_path, {}));
+ ASSERT_TRUE(reader.prepare_split(split_options).ok());
+ block = build_table_block(projected_columns);
+ 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(), 1);
+ expect_int32_column_values(*block.get_by_position(0).column, {7});
+ expect_int32_column_values(*block.get_by_position(1).column, {70});
+
+ 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,
IcebergTableReaderDoesNotPushDownAggregateWithPositionDelete) {
const auto test_dir =
std::filesystem::temp_directory_path() /
"doris_iceberg_aggregate_position_delete_test";
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 00d14b30cae..638b1f5058d 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,16 @@ suite("test_gen_iceberg_by_api",
"p0,external,doris,external_docker,external_doc
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.
+ sql """ select count(*) from table_with_append_file """
+
+ // 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..5f5d96e0a7a 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
@@ -41,15 +41,23 @@ suite("test_iceberg_migrated_nested_without_name_mapping",
"p0,external") {
"""
sql """switch ${catalog_name}"""
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
"""
- sql """drop catalog if exists ${catalog_name}"""
+ // The same positional binding must feed predicate/row-group filtering,
otherwise the row would
+ // be readable but unfilterable.
+ order_qt_idless_top_level_field_filter_uses_positional_id """
+ SELECT count(*) FROM migrated_nested_without_name_mapping WHERE id = 1
+ """
+
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
