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
commit 352e3a98e5c2ccec529c68d6fdf97d21fd95339c Author: Gabriel <[email protected]> AuthorDate: Tue Oct 13 09:24:54 2026 +0800 [fix](iceberg) Backport nested partition and default fixes to branch-4.1 (#66825) ### What problem does this PR solve? Backport the remaining applicable fixes from #66529 to the legacy Iceberg implementation on `branch-4.1`. The requested scope is limited to #66529 and #66750. | Source PR | Applicability on branch-4.1 | | --- | --- | | #66529 | Port nested partition-source handling, INSERT retry metadata reset, format V2 non-finite initial defaults and position-delete row defaults/requiredness. Adapt the changes to the existing synchronous writer and statement-pinned schema. | | #66750 | No code change needed: its lowercase system-table diagnostic regression belongs to the connector-provider implementation. The legacy Paimon scan paths already use `Paimon` in these diagnostics. Keep the branch's existing engine metadata and routing behavior. | Format V1 reader bug fixes and compatibility changes are outside this PR. ### Changes - Accept nested primitive partition sources in the pinned writer schema; resolve them by stable field IDs and pass the root slot plus child path through FE/BE routing. - Preserve parent NULLs in writer and UPDATE/MERGE exchange partition transforms. Bind nested identity predicates by their full schema path and reject static nested partition labels that cannot be validated against row contents. - Refresh planner-loaded MVCC metadata on INSERT retries while retaining caller-injected source snapshot fences across every attempt. A new execution still clears all snapshots. - Exclude Iceberg V3 lineage columns from positional field-ID binding so transformed and nested partitions retain partition-aware UPDATE/MERGE routing. - In the format V2 reader, materialize Iceberg `NaN`, `Infinity`, and `-Infinity` initial defaults without sending them through Doris's finite-only numeric parser. Prepare typed nested defaults and enforce required-field checks for position-delete row projections. - Include the source regression coverage and focused tests for nested writer metadata, field-ID routing, parent NULLs, static partition rejection, retry snapshot ownership, V3 routing, and default materialization. ### Validation - The new retry-snapshot and V3 routing tests each reproduced the reported failure before their corresponding fixes. - FE focused tests: 231 passed, with no failures, errors, or skipped tests. - ASAN BE focused tests: 257 passed from 27 suites, including nested partition routing, format V2 non-finite defaults, and position-delete row defaults. - FE Checkstyle: passed. - clang-format 16: all 16 affected C/C++ files passed, including the restored V1 files. - Full update-diff self-review and whitespace checks: passed. Full external-cluster regressions were not run locally. Existing regression output is retained from the earlier backport; it was not regenerated during this update. --- .../sink/writer/iceberg/viceberg_table_writer.cpp | 134 +++++++++++++++++--- .../sink/writer/iceberg/viceberg_table_writer.h | 7 ++ be/src/format/table/iceberg/schema.cpp | 29 ++++- be/src/format/table/iceberg/schema.h | 4 + .../transformer/iceberg_partition_function.cpp | 76 +++++++++++- .../transformer/iceberg_partition_function.h | 4 + .../iceberg_position_delete_sys_table_reader.cpp | 10 ++ be/src/format_v2/table/iceberg_reader.cpp | 37 +++++- be/src/format_v2/table/iceberg_reader.h | 4 + be/test/core/value/merge_partitioner_test.cpp | 91 ++++++++++++++ .../writer/iceberg/viceberg_table_writer_test.cpp | 83 ++++++++++++- be/test/format/table/iceberg/schema_test.cpp | 16 +++ ...eberg_position_delete_sys_table_reader_test.cpp | 29 +++++ be/test/format_v2/table/iceberg_reader_test.cpp | 33 +++++ .../datasource/iceberg/IcebergTransaction.java | 24 +++- .../iceberg/IcebergWriteSchemaContext.java | 7 +- .../org/apache/doris/nereids/StatementContext.java | 11 ++ .../glue/translator/PhysicalPlanTranslator.java | 3 +- .../nereids/properties/DistributionSpecMerge.java | 17 ++- .../commands/insert/InsertIntoTableCommand.java | 8 ++ .../plans/physical/PhysicalIcebergMergeSink.java | 92 +++++++++++++- .../org/apache/doris/planner/DataPartition.java | 10 ++ .../datasource/iceberg/IcebergTransactionTest.java | 14 +++ .../iceberg/IcebergWriteSchemaContextTest.java | 17 +++ .../plans/commands/insert/InsertUtilsTest.java | 47 +++++++ .../physical/PhysicalIcebergMergeSinkTest.java | 135 +++++++++++++++++++++ gensrc/thrift/Partitions.thrift | 2 + .../write/test_iceberg_write_complex_evolution.out | 12 +- .../test_iceberg_write_complex_evolution.groovy | 26 +++- 29 files changed, 930 insertions(+), 52 deletions(-) diff --git a/be/src/exec/sink/writer/iceberg/viceberg_table_writer.cpp b/be/src/exec/sink/writer/iceberg/viceberg_table_writer.cpp index 6b1f8318c33..71c5c1c931d 100644 --- a/be/src/exec/sink/writer/iceberg/viceberg_table_writer.cpp +++ b/be/src/exec/sink/writer/iceberg/viceberg_table_writer.cpp @@ -17,6 +17,7 @@ #include "exec/sink/writer/iceberg/viceberg_table_writer.h" +#include <algorithm> #include <boost/uuid/string_generator.hpp> #include "common/exception.h" @@ -25,8 +26,10 @@ #include "core/block/materialize_block.h" #include "core/column/column_const.h" #include "core/column/column_nullable.h" +#include "core/column/column_struct.h" #include "core/column/column_vector.h" #include "core/data_type/data_type_nullable.h" +#include "core/data_type/data_type_struct.h" #include "core/data_type_serde/data_type_serde.h" #include "exec/sink/writer/iceberg/iceberg_partition_path.h" #include "exec/sink/writer/iceberg/partition_transformers.h" @@ -123,31 +126,106 @@ std::vector<VIcebergTableWriter::IcebergPartitionColumn> VIcebergTableWriter::_to_iceberg_partition_columns() { std::vector<IcebergPartitionColumn> partition_columns; - std::unordered_map<int, int> id_to_column_idx; - id_to_column_idx.reserve(_schema->columns().size()); - for (int i = 0; i < _schema->columns().size(); i++) { - id_to_column_idx[_schema->columns()[i].field_id()] = i; - } for (const auto& partition_field : _partition_spec->fields()) { - auto column_idx_it = id_to_column_idx.find(partition_field.source_id()); - if (column_idx_it == id_to_column_idx.end()) { + const auto* field_path = _schema->find_field_path(partition_field.source_id()); + if (field_path == nullptr || field_path->empty()) { throw Exception( ErrorCode::INTERNAL_ERROR, "Iceberg partition field {} references source field {} outside writer schema", partition_field.field_id(), partition_field.source_id()); } - int column_idx = column_idx_it->second; + int column_idx = -1; + for (int i = 0; i < _schema->columns().size(); ++i) { + if (_schema->columns()[i].field_id() == field_path->front()->field_id()) { + column_idx = i; + break; + } + } + DORIS_CHECK(column_idx >= 0); + std::vector<size_t> child_indices; + iceberg::Type* iceberg_type = field_path->front()->field_type(); + DataTypePtr source_type = _vec_output_expr_ctxs[column_idx]->root()->data_type(); + for (size_t depth = 1; depth < field_path->size(); ++depth) { + if (!iceberg_type->is_struct_type()) { + throw Exception(ErrorCode::INTERNAL_ERROR, + "Iceberg partition source field {} has a non-struct ancestor", + partition_field.source_id()); + } + const auto& fields = iceberg_type->as_struct_type()->fields(); + auto child = std::find_if(fields.begin(), fields.end(), [&](const auto& candidate) { + return candidate.field_id() == (*field_path)[depth]->field_id(); + }); + DORIS_CHECK(child != fields.end()); + const size_t child_idx = std::distance(fields.begin(), child); + const auto* struct_type = + check_and_get_data_type<DataTypeStruct>(remove_nullable(source_type).get()); + if (struct_type == nullptr || child_idx >= struct_type->get_elements().size()) { + throw Exception( + ErrorCode::INTERNAL_ERROR, + "Iceberg nested partition source field {} does not match writer type", + partition_field.source_id()); + } + child_indices.push_back(child_idx); + iceberg_type = child->field_type(); + source_type = struct_type->get_element(child_idx); + } + if (!iceberg_type->is_primitive_type()) { + throw Exception(ErrorCode::INTERNAL_ERROR, + "Iceberg partition source field {} is not primitive", + partition_field.source_id()); + } std::unique_ptr<PartitionColumnTransform> partition_column_transform = - PartitionColumnTransforms::create( - partition_field, _vec_output_expr_ctxs[column_idx]->root()->data_type()); + PartitionColumnTransforms::create(partition_field, source_type); partition_columns.emplace_back( - partition_field, - _vec_output_expr_ctxs[column_idx]->root()->data_type()->get_primitive_type(), - column_idx, std::move(partition_column_transform)); + partition_field, remove_nullable(source_type)->get_primitive_type(), column_idx, + std::move(child_indices), std::move(partition_column_transform)); } return partition_columns; } +ColumnWithTypeAndName VIcebergTableWriter::_nested_partition_source( + const Block& block, const IcebergPartitionColumn& partition_column) const { + ColumnWithTypeAndName source = block.get_by_position(partition_column.source_idx()); + if (partition_column.child_indices().empty()) { + return source; + } + ColumnPtr column = source.column->convert_to_full_column_if_const(); + DataTypePtr type = source.type; + auto combined_nulls = ColumnUInt8::create(block.rows(), 0); + bool nullable = false; + auto unwrap_nullable = [&]() { + if (const auto* nullable_column = check_and_get_column<ColumnNullable>(column.get())) { + nullable = true; + const auto& nulls = nullable_column->get_null_map_data(); + auto& combined = combined_nulls->get_data(); + for (size_t row = 0; row < combined.size(); ++row) { + combined[row] |= nulls[row]; + } + column = nullable_column->get_nested_column_ptr(); + type = remove_nullable(type); + } + }; + for (size_t child_idx : partition_column.child_indices()) { + unwrap_nullable(); + const auto* struct_column = check_and_get_column<ColumnStruct>(column.get()); + const auto* struct_type = check_and_get_data_type<DataTypeStruct>(type.get()); + if (struct_column == nullptr || struct_type == nullptr || + child_idx >= struct_column->tuple_size()) { + throw Exception(ErrorCode::INTERNAL_ERROR, + "Iceberg nested partition source does not match writer block"); + } + column = struct_column->get_column_ptr(child_idx); + type = struct_type->get_element(child_idx); + } + // Parent NULL masks the leaf even when the nested storage column contains a materialized value. + unwrap_nullable(); + if (nullable) { + column = ColumnNullable::create(column, std::move(combined_nulls)); + type = make_nullable(type); + } + return {std::move(column), std::move(type), source.name}; +} + void VIcebergTableWriter::_init_static_partition_values() { auto& iceberg_sink = _t_sink.iceberg_table_sink; if (!iceberg_sink.__isset.static_partition_values || @@ -159,6 +237,29 @@ void VIcebergTableWriter::_init_static_partition_values() { _has_static_partition = true; const auto& static_values_map = iceberg_sink.static_partition_values; + if (_partition_spec == nullptr) { + throw Exception(ErrorCode::INVALID_ARGUMENT, + "Static Iceberg partitions require a partition specification"); + } + + for (const auto& partition_field : _partition_spec->fields()) { + if (!static_values_map.contains(partition_field.name())) { + continue; + } + const auto* field_path = _schema->find_field_path(partition_field.source_id()); + if (field_path == nullptr) { + throw Exception(ErrorCode::INTERNAL_ERROR, + "Iceberg partition field {} references a missing source field", + partition_field.field_id()); + } + if (field_path->size() > 1) { + // Static values bypass extraction from the row, so they cannot prove that a nested + // source leaf written to the file matches the partition label. + throw Exception(ErrorCode::INVALID_ARGUMENT, + "Static Iceberg partition field {} cannot use a nested source column", + partition_field.name()); + } + } size_t num_cols = _iceberg_partition_columns.size(); _partition_column_static_values.resize(num_cols); @@ -467,9 +568,12 @@ Status VIcebergTableWriter::_write_prepared_block(Block& output_block) { transformed_block.insert( {std::move(col), result_type, iceberg_partition_columns.field().name()}); } else { + Block source_block; + source_block.insert( + _nested_partition_source(output_block, iceberg_partition_columns)); transformed_block.insert( - iceberg_partition_columns.partition_column_transform().apply( - output_block, iceberg_partition_columns.source_idx())); + iceberg_partition_columns.partition_column_transform().apply(source_block, + 0)); } } for (int i = 0; i < output_block.rows(); ++i) { diff --git a/be/src/exec/sink/writer/iceberg/viceberg_table_writer.h b/be/src/exec/sink/writer/iceberg/viceberg_table_writer.h index c66cc1a74da..5baf351c72c 100644 --- a/be/src/exec/sink/writer/iceberg/viceberg_table_writer.h +++ b/be/src/exec/sink/writer/iceberg/viceberg_table_writer.h @@ -91,6 +91,7 @@ public: private: friend class IcebergTableSinkOperatorTest; FRIEND_TEST(VIcebergTableWriterTest, RejectMissingPartitionSource); + FRIEND_TEST(VIcebergTableWriterTest, ResolvesNestedPartitionSource); // The currently active partition writer (may be VIcebergPartitionWriter or VIcebergSortWriter). // Updated during write() to track which writer received the most recent data. @@ -102,10 +103,12 @@ private: public: IcebergPartitionColumn(const iceberg::PartitionField& field, const PrimitiveType& source_type, int source_idx, + std::vector<size_t> child_indices, std::unique_ptr<PartitionColumnTransform> partition_column_transform) : _field(field), _source_type(source_type), _source_idx(source_idx), + _child_indices(std::move(child_indices)), _partition_column_transform(std::move(partition_column_transform)) {} public: @@ -113,6 +116,7 @@ private: const PrimitiveType& source_type() const { return _source_type; } int source_idx() const { return _source_idx; } + const std::vector<size_t>& child_indices() const { return _child_indices; } const PartitionColumnTransform& partition_column_transform() const { return *_partition_column_transform; @@ -126,10 +130,13 @@ private: const iceberg::PartitionField& _field; PrimitiveType _source_type; int _source_idx; + std::vector<size_t> _child_indices; std::unique_ptr<PartitionColumnTransform> _partition_column_transform; }; std::vector<IcebergPartitionColumn> _to_iceberg_partition_columns(); + ColumnWithTypeAndName _nested_partition_source( + const Block& block, const IcebergPartitionColumn& partition_column) const; std::string _partition_to_writer_key(const doris::iceberg::StructLike& data); std::string _partition_to_path(const doris::iceberg::StructLike& data); diff --git a/be/src/format/table/iceberg/schema.cpp b/be/src/format/table/iceberg/schema.cpp index 5a1bdaf5c93..25e2d20919a 100644 --- a/be/src/format/table/iceberg/schema.cpp +++ b/be/src/format/table/iceberg/schema.cpp @@ -17,6 +17,8 @@ #include "format/table/iceberg/schema.h" +#include <functional> + namespace doris::iceberg { #include "common/compile_check_begin.h" @@ -25,10 +27,26 @@ const int Schema::DEFAULT_SCHEMA_ID = 0; Schema::Schema(int schema_id, std::vector<NestedField> columns) : _schema_id(schema_id), _root_struct(std::move(columns)) { - _id_to_field.reserve(_root_struct.fields().size()); + FieldPath path; + std::function<void(const NestedField&)> index_field = [&](const NestedField& field) { + path.push_back(&field); + _id_to_field[field.field_id()] = &field; + _id_to_field_path[field.field_id()] = path; + Type* type = field.field_type(); + if (type->is_struct_type()) { + for (const auto& child : type->as_struct_type()->fields()) { + index_field(child); + } + } else if (type->is_list_type()) { + index_field(type->as_list_type()->element_field()); + } else if (type->is_map_type()) { + index_field(type->as_map_type()->key_field()); + index_field(type->as_map_type()->value_field()); + } + path.pop_back(); + }; for (const auto& field : _root_struct.fields()) { - int field_id = field.field_id(); - _id_to_field[field_id] = &field; + index_field(field); } } Schema::Schema(std::vector<NestedField> columns) : Schema(DEFAULT_SCHEMA_ID, std::move(columns)) {} @@ -49,5 +67,10 @@ const NestedField* Schema::find_field(int id) const { return nullptr; } +const Schema::FieldPath* Schema::find_field_path(int id) const { + auto it = _id_to_field_path.find(id); + return it == _id_to_field_path.end() ? nullptr : &it->second; +} + #include "common/compile_check_end.h" } // namespace doris::iceberg diff --git a/be/src/format/table/iceberg/schema.h b/be/src/format/table/iceberg/schema.h index 29ea62510ee..70377a63ac0 100644 --- a/be/src/format/table/iceberg/schema.h +++ b/be/src/format/table/iceberg/schema.h @@ -27,6 +27,7 @@ class StructType; class Schema { public: + using FieldPath = std::vector<const NestedField*>; Schema(int schema_id, std::vector<NestedField> columns); Schema(std::vector<NestedField> columns); @@ -41,6 +42,8 @@ public: const NestedField* find_field(int id) const; + const FieldPath* find_field_path(int id) const; + private: static const char NEWLINE = '\n'; static const std::string ALL_COLUMNS; @@ -49,6 +52,7 @@ private: int _schema_id; StructType _root_struct; std::unordered_map<int, const NestedField*> _id_to_field; + std::unordered_map<int, FieldPath> _id_to_field_path; }; #include "common/compile_check_end.h" diff --git a/be/src/format/transformer/iceberg_partition_function.cpp b/be/src/format/transformer/iceberg_partition_function.cpp index 14eba85e28a..64062e7ee53 100644 --- a/be/src/format/transformer/iceberg_partition_function.cpp +++ b/be/src/format/transformer/iceberg_partition_function.cpp @@ -23,6 +23,8 @@ #include "core/column/column_const.h" #include "core/column/column_nullable.h" #include "core/column/column_struct.h" +#include "core/column/column_vector.h" +#include "core/data_type/data_type_nullable.h" #include "core/data_type/data_type_struct.h" #include "exec/sink/writer/iceberg/partition_transformers.h" #include "format/table/iceberg/partition_spec.h" @@ -88,6 +90,9 @@ Status IcebergInsertPartitionFunction::init(const std::vector<TExpr>& texprs) { insert_field.expr_ctx = std::move(ctx); insert_field.source_id = field.__isset.source_id ? field.source_id : 0; insert_field.name = field.__isset.name ? field.name : ""; + if (field.__isset.source_field_path) { + insert_field.source_field_path = field.source_field_path; + } _partition_fields.emplace_back(std::move(insert_field)); } } @@ -118,10 +123,22 @@ Status IcebergInsertPartitionFunction::open(RuntimeState* state) { RETURN_IF_ERROR(VExpr::open(field_ctxs, state)); for (auto& field : _partition_fields) { try { + DataTypePtr source_type = field.expr_ctx->root()->data_type(); + for (int32_t child_index : field.source_field_path) { + const auto* struct_type = check_and_get_data_type<DataTypeStruct>( + remove_nullable(source_type).get()); + if (child_index < 0 || struct_type == nullptr || + static_cast<size_t>(child_index) >= struct_type->get_elements().size()) { + throw Exception(ErrorCode::INTERNAL_ERROR, + "Iceberg nested merge partition source does not match " + "expression type"); + } + // Transform validation must use the leaf type that get_partitions() extracts. + source_type = struct_type->get_element(static_cast<size_t>(child_index)); + } doris::iceberg::PartitionField partition_field(field.source_id, 0, field.name, field.transform); - field.transformer = PartitionColumnTransforms::create( - partition_field, field.expr_ctx->root()->data_type()); + field.transformer = PartitionColumnTransforms::create(partition_field, source_type); } catch (const doris::Exception& e) { return Status::NotSupported("Unsupported Iceberg partition transform: {}", e.what()); @@ -168,12 +185,60 @@ Status IcebergInsertPartitionFunction::clone(RuntimeState* state, field.expr_ctx = dst_field_ctxs[i]; field.source_id = _partition_fields[i].source_id; field.name = _partition_fields[i].name; + field.source_field_path = _partition_fields[i].source_field_path; new_function->_partition_fields.emplace_back(std::move(field)); } } return Status::OK(); } +Status IcebergInsertPartitionFunction::_nested_partition_source( + size_t rows, const InsertPartitionField& field, ColumnWithTypeAndName* source) const { + if (field.source_field_path.empty()) { + return Status::OK(); + } + ColumnPtr column = source->column->convert_to_full_column_if_const(); + DataTypePtr type = source->type; + ColumnUInt8::MutablePtr combined_nulls; + bool nullable = false; + auto unwrap_nullable = [&]() { + if (const auto* nullable_column = check_and_get_column<ColumnNullable>(column.get())) { + nullable = true; + if (!combined_nulls) { + combined_nulls = ColumnUInt8::create(rows, 0); + } + const auto& nulls = nullable_column->get_null_map_data(); + auto& combined = combined_nulls->get_data(); + for (size_t row = 0; row < combined.size(); ++row) { + combined[row] |= nulls[row]; + } + column = nullable_column->get_nested_column_ptr(); + type = remove_nullable(type); + } + }; + for (int32_t child_index : field.source_field_path) { + unwrap_nullable(); + const auto* struct_column = check_and_get_column<ColumnStruct>(column.get()); + const auto* struct_type = check_and_get_data_type<DataTypeStruct>(type.get()); + if (child_index < 0 || struct_column == nullptr || struct_type == nullptr || + static_cast<size_t>(child_index) >= struct_column->tuple_size()) { + return Status::InternalError( + "Iceberg nested merge partition source does not match input block"); + } + column = struct_column->get_column_ptr(static_cast<size_t>(child_index)); + type = struct_type->get_element(static_cast<size_t>(child_index)); + } + // A nullable parent masks a materialized child value; exchange routing must match the writer's partition. + unwrap_nullable(); + if (nullable) { + column = ColumnNullable::create(column, std::move(combined_nulls)); + type = make_nullable(type); + } + std::string name = source->name; + *source = {std::move(column), std::move(type), std::move(name)}; + return Status::OK(); +} + Status IcebergInsertPartitionFunction::_compute_hashes_with_transform( Block* block, std::vector<HashValType>& partitions) const { const size_t rows = block->rows(); @@ -196,8 +261,13 @@ Status IcebergInsertPartitionFunction::_compute_hashes_with_transform( if (_partition_fields[i].transformer == nullptr) { return Status::InternalError("Merge partitioning transform is not initialized"); } + ColumnWithTypeAndName source = block->get_by_position(results[i]); + if (!_partition_fields[i].source_field_path.empty()) { + RETURN_IF_ERROR(_nested_partition_source(rows, _partition_fields[i], &source)); + } + Block source_block({source}); ColumnWithTypeAndName transformed = - _partition_fields[i].transformer->apply(*block, results[i]); + _partition_fields[i].transformer->apply(source_block, 0); const auto& [column, is_const] = unpack_if_const(transformed.column); if (is_const) { // A const column has the same value for all rows in this block, diff --git a/be/src/format/transformer/iceberg_partition_function.h b/be/src/format/transformer/iceberg_partition_function.h index d7698ed78b4..57f255991b0 100644 --- a/be/src/format/transformer/iceberg_partition_function.h +++ b/be/src/format/transformer/iceberg_partition_function.h @@ -23,6 +23,7 @@ #include <string> #include <vector> +#include "core/block/column_with_type_and_name.h" #include "exec/partitioner/partitioner.h" #include "exec/sink/writer/iceberg/partition_transformers.h" @@ -51,8 +52,11 @@ private: std::unique_ptr<PartitionColumnTransform> transformer; int32_t source_id = 0; std::string name; + std::vector<int32_t> source_field_path; }; + Status _nested_partition_source(size_t rows, const InsertPartitionField& field, + ColumnWithTypeAndName* source) const; Status _compute_hashes_with_transform(Block* block, std::vector<HashValType>& partitions) const; Status _compute_hashes_with_exprs(Block* block, std::vector<HashValType>& partitions) const; Status _clone_expr_ctxs(RuntimeState* state, const VExprContextSPtrs& src, diff --git a/be/src/format_v2/table/iceberg_position_delete_sys_table_reader.cpp b/be/src/format_v2/table/iceberg_position_delete_sys_table_reader.cpp index b2b37c1bf09..cd84a255e69 100644 --- a/be/src/format_v2/table/iceberg_position_delete_sys_table_reader.cpp +++ b/be/src/format_v2/table/iceberg_position_delete_sys_table_reader.cpp @@ -35,6 +35,7 @@ #include "core/types.h" #include "format/table/iceberg_delete_file_reader_helper.h" #include "format/table/parquet_utils.h" +#include "format_v2/table/iceberg_reader.h" #include "format_v2/table/iceberg_schema_utils.h" #include "runtime/descriptors.h" #include "runtime/runtime_state.h" @@ -147,6 +148,8 @@ protected: void configure_mapper_options(format::TableColumnMapperOptions* options) const override { options->enable_row_lineage_virtual_columns = true; + // Position-delete row projection must reject a physically absent required field exactly like data scans. + options->reject_missing_required_field = supports_iceberg_scan_semantics_v2(_scan_params); // Parquet may preserve a selected complex wrapper without its own ID; position-delete row // projection must use the same descendant-ID fallback as ordinary Iceberg data scans. options->allow_idless_complex_wrapper_projection = @@ -591,6 +594,13 @@ Status IcebergPositionDeleteSysTableV2Reader::_build_delete_file_projected_colum columns->push_back(*it); columns->back().type = column.type; set_iceberg_delete_field_id(&columns->back()); + // The copied row tree bypasses IcebergTableReader::annotate_projected_column, so prepare its + // typed nested defaults before the generic inner reader builds the column mapper. + if (supports_iceberg_scan_semantics_v2(_scan_params)) { + RETURN_IF_ERROR(prepare_iceberg_initial_default_exprs( + &columns->back(), + _runtime_state == nullptr ? nullptr : &_runtime_state->timezone_obj())); + } continue; } auto field = build_delete_file_column(column.name, column.type); diff --git a/be/src/format_v2/table/iceberg_reader.cpp b/be/src/format_v2/table/iceberg_reader.cpp index 3c1c098af93..0774c9655b4 100644 --- a/be/src/format_v2/table/iceberg_reader.cpp +++ b/be/src/format_v2/table/iceberg_reader.cpp @@ -24,6 +24,7 @@ #include <algorithm> #include <deque> +#include <limits> #include <memory> #include <sstream> #include <string_view> @@ -376,6 +377,28 @@ static Status build_v2_json_map_default(const format::ColumnDefinition& field, return Status::OK(); } +static bool parse_non_finite_default(doris::PrimitiveType type, std::string_view value, + Field* result) { + DORIS_CHECK(result != nullptr); + if (type != TYPE_FLOAT && type != TYPE_DOUBLE) { + return false; + } + double parsed; + if (value == "NaN") { + parsed = std::numeric_limits<double>::quiet_NaN(); + } else if (value == "Infinity") { + parsed = std::numeric_limits<double>::infinity(); + } else if (value == "-Infinity") { + parsed = -std::numeric_limits<double>::infinity(); + } else { + return false; + } + // Iceberg serializes non-finite defaults as strings, which generic Doris numeric parsers reject. + *result = type == TYPE_FLOAT ? Field::create_field<TYPE_FLOAT>(static_cast<float>(parsed)) + : Field::create_field<TYPE_DOUBLE>(parsed); + return true; +} + static Status build_v2_json_scalar_default(const format::ColumnDefinition& field, const DataTypePtr& value_type, const rapidjson::Value& json_value, @@ -416,6 +439,9 @@ static Status build_v2_json_scalar_default(const format::ColumnDefinition& field *result = Field::create_field<TYPE_STRING>(std::move(serialized_value)); return Status::OK(); } + if (parse_non_finite_default(primitive_type, serialized_value, result)) { + return Status::OK(); + } RETURN_IF_ERROR(doris::iceberg::detail::normalize_timestamp_for_doris(primitive_type, timezone, &serialized_value)); RETURN_IF_ERROR(value_type->get_serde()->from_fe_string(serialized_value, *result)); @@ -505,6 +531,9 @@ static Status build_v2_initial_default_field(const format::ColumnDefinition& fie } std::string serialized_value = *field.initial_default_value; + if (parse_non_finite_default(primitive_type, serialized_value, result)) { + return Status::OK(); + } RETURN_IF_ERROR(doris::iceberg::detail::normalize_timestamp_for_doris(primitive_type, timezone, &serialized_value)); RETURN_IF_ERROR(value_type->get_serde()->from_fe_string(serialized_value, *result)); @@ -526,8 +555,8 @@ static Status build_initial_default_literal(const format::ColumnDefinition& tabl return Status::OK(); } -static Status build_initial_default_exprs(format::ColumnDefinition* column, - const cctz::time_zone* timezone) { +Status prepare_iceberg_initial_default_exprs(format::ColumnDefinition* column, + const cctz::time_zone* timezone) { DORIS_CHECK(column != nullptr); if (column->initial_default_value.has_value()) { VExprSPtr literal; @@ -535,7 +564,7 @@ static Status build_initial_default_exprs(format::ColumnDefinition* column, column->default_expr = VExprContext::create_shared(std::move(literal)); } for (auto& child : column->children) { - RETURN_IF_ERROR(build_initial_default_exprs(&child, timezone)); + RETURN_IF_ERROR(prepare_iceberg_initial_default_exprs(&child, timezone)); } return Status::OK(); } @@ -850,7 +879,7 @@ Status IcebergTableReader::annotate_projected_column(const TFileScanSlotInfo& sl auto& schema_column = *context->schema_column; const cctz::time_zone* timezone = context->runtime_state == nullptr ? nullptr : &context->runtime_state->timezone_obj(); - RETURN_IF_ERROR(build_initial_default_exprs(&schema_column, timezone)); + RETURN_IF_ERROR(prepare_iceberg_initial_default_exprs(&schema_column, timezone)); column->initial_default_value = schema_column.initial_default_value; column->initial_default_value_is_base64 = schema_column.initial_default_value_is_base64; column->is_optional = schema_column.is_optional; diff --git a/be/src/format_v2/table/iceberg_reader.h b/be/src/format_v2/table/iceberg_reader.h index 21609b9866f..4000c173bff 100644 --- a/be/src/format_v2/table/iceberg_reader.h +++ b/be/src/format_v2/table/iceberg_reader.h @@ -42,6 +42,10 @@ struct FileSystemProperties; namespace doris::format::iceberg { +// Position-delete row projection also needs the typed defaults used by ordinary data scans. +Status prepare_iceberg_initial_default_exprs(format::ColumnDefinition* column, + const cctz::time_zone* timezone = nullptr); + // Iceberg table-level reader. // It reuses TableReader for split orchestration, dynamic partition pruning and table-block // finalization, while composing a FileReader for physical data-file reads instead of inheriting diff --git a/be/test/core/value/merge_partitioner_test.cpp b/be/test/core/value/merge_partitioner_test.cpp index 0836c6f8753..094c635a2a4 100644 --- a/be/test/core/value/merge_partitioner_test.cpp +++ b/be/test/core/value/merge_partitioner_test.cpp @@ -83,6 +83,41 @@ protected: return expr; } + TTypeDesc _nested_int_struct_type_desc() { + TTypeNode struct_node; + struct_node.__set_type(TTypeNodeType::STRUCT); + TStructField child; + child.__set_name("part"); + child.__set_contains_null(true); + struct_node.__set_struct_fields({child}); + + TTypeNode int_node; + int_node.__set_type(TTypeNodeType::SCALAR); + TScalarType scalar; + scalar.__set_type(TPrimitiveType::INT); + int_node.__set_scalar_type(scalar); + + TTypeDesc type_desc; + type_desc.__set_types({struct_node, int_node}); + type_desc.__set_is_nullable(true); + return type_desc; + } + + TExpr _make_nested_source_expr() { + TExprNode node; + node.__set_node_type(TExprNodeType::SLOT_REF); + node.__set_num_children(0); + TSlotRef slot_ref; + slot_ref.__set_slot_id(_nested_source_slot_id); + slot_ref.__set_tuple_id(_tuple_id); + node.__set_slot_ref(slot_ref); + node.__set_type(_nested_int_struct_type_desc()); + node.__set_is_nullable(true); + TExpr expr; + expr.nodes.emplace_back(std::move(node)); + return expr; + } + TMergePartitionInfo _make_base_merge_info(bool insert_random) { TMergePartitionInfo merge_info; merge_info.__set_operation_expr( @@ -180,6 +215,13 @@ protected: .column_name("delete_key") .column_pos(4) .build()); + TTypeDesc nested_type = _nested_int_struct_type_desc(); + tuple_builder.add_slot(TSlotDescriptorBuilder() + .set_slotType(nested_type) + .nullable(true) + .column_name("nested_source") + .column_pos(5) + .build()); tuple_builder.build(&dtb); TDescriptorTable thrift_tbl = dtb.desc_tbl(); @@ -205,11 +247,13 @@ protected: _row_id_slot_id = find_slot_id("row_id"); _insert_key_slot_id = find_slot_id("insert_key"); _delete_key_slot_id = find_slot_id("delete_key"); + _nested_source_slot_id = find_slot_id("nested_source"); ASSERT_GE(_operation_slot_id, 0); ASSERT_GE(_row_id_slot_id, 0); ASSERT_GE(_insert_key_slot_id, 0); ASSERT_GE(_delete_key_slot_id, 0); + ASSERT_GE(_nested_source_slot_id, 0); } ObjectPool _pool; @@ -220,6 +264,7 @@ protected: TSlotId _row_id_slot_id = -1; TSlotId _insert_key_slot_id = -1; TSlotId _delete_key_slot_id = -1; + TSlotId _nested_source_slot_id = -1; }; TEST_F(MergePartitionerTest, TestInsertDeleteUpdatePartitioning) { @@ -318,6 +363,52 @@ TEST_F(MergePartitionerTest, TestInsertPartitionFieldsIdentity) { ASSERT_TRUE(partitioner.close(&_state).ok()); } +TEST_F(MergePartitionerTest, TestNestedInsertPartitionFieldPreservesParentNulls) { + ScopedConfigValue<int32_t> max_partition_guard( + config::table_sink_partition_write_max_partition_nums_per_writer, 0); + + TMergePartitionInfo merge_info = _make_base_merge_info(false); + TIcebergPartitionField field; + field.__set_transform("bucket[8]"); + field.__set_source_expr(_make_nested_source_expr()); + field.__set_name("payload_part"); + field.__set_source_id(3); + field.__set_source_field_path({0}); + merge_info.__set_insert_partition_fields({field}); + + MergePartitioner partitioner(8, merge_info, false); + ASSERT_TRUE(partitioner.init({}).ok()); + ASSERT_TRUE(partitioner.prepare(&_state, *_row_desc).ok()); + ASSERT_TRUE(partitioner.open(&_state).ok()); + + Block block = _build_block({1, 1, 1, 1}, {"p1", "p2", "p3", "p4"}, {1, 2, 3, 4}, + {10, 11, 12, 13}, {"d1", "d2", "d3", "d4"}); + auto values = ColumnInt32::create(); + values->insert_value(9); + values->insert_value(9); + values->insert_value(7); + values->insert_value(8); + auto child_nulls = ColumnUInt8::create(4, 0); + ColumnPtr child = ColumnNullable::create(std::move(values), std::move(child_nulls)); + auto struct_column = ColumnStruct::create(Columns {std::move(child)}); + auto parent_nulls = ColumnUInt8::create(); + parent_nulls->get_data().assign({0, 0, 1, 1}); + DataTypePtr child_type = make_nullable(std::make_shared<DataTypeInt32>()); + DataTypePtr struct_type = + std::make_shared<DataTypeStruct>(DataTypes {child_type}, Strings {"part"}); + block.insert(ColumnWithTypeAndName( + ColumnNullable::create(std::move(struct_column), std::move(parent_nulls)), + make_nullable(struct_type), "nested_source")); + + ASSERT_TRUE(partitioner.do_partitioning(&_state, &block).ok()); + const auto& channel_ids = partitioner.get_channel_ids(); + ASSERT_EQ(4, channel_ids.size()); + EXPECT_EQ(channel_ids[0], channel_ids[1]); + EXPECT_EQ(channel_ids[2], channel_ids[3]); + + ASSERT_TRUE(partitioner.close(&_state).ok()); +} + TEST_F(MergePartitionerTest, TestInvalidTransformFallbacksToRandom) { ScopedConfigValue<int64_t> threshold_guard( config::table_sink_non_partition_write_scaling_data_processed_threshold, 0); diff --git a/be/test/exec/sink/writer/iceberg/viceberg_table_writer_test.cpp b/be/test/exec/sink/writer/iceberg/viceberg_table_writer_test.cpp index 849b55b2437..1bbbe5454af 100644 --- a/be/test/exec/sink/writer/iceberg/viceberg_table_writer_test.cpp +++ b/be/test/exec/sink/writer/iceberg/viceberg_table_writer_test.cpp @@ -22,11 +22,19 @@ #include <thrift/transport/TBufferTransports.h> #include "common/exception.h" +#include "core/block/block.h" +#include "core/column/column_nullable.h" +#include "core/column/column_struct.h" +#include "core/column/column_vector.h" #include "core/data_type/data_type_nullable.h" +#include "core/data_type/data_type_number.h" +#include "core/data_type/data_type_struct.h" #include "core/data_type/data_type_timestamptz.h" #include "core/data_type/data_type_varbinary.h" #include "exec/sink/writer/iceberg/partition_transformers.h" #include "exec/sink/writer/iceberg/vpartition_writer_base.h" +#include "exprs/vexpr_context.h" +#include "exprs/vslot_ref.h" #include "format/table/iceberg/partition_spec_parser.h" #include "format/table/iceberg/schema.h" #include "format/table/iceberg/types.h" @@ -74,8 +82,10 @@ TEST(VIcebergTableWriterTest, TimestampIdentityPreservesUtcCommitAndNull) { state.set_timezone("America/New_York"); writer._state = &state; writer._schema = schema; + writer._partition_spec = std::move(spec); + // Retain the real spec for static validation; top-level sources have no nested child path. writer._iceberg_partition_columns.emplace_back( - spec->fields()[0], TYPE_TIMESTAMPTZ, 0, + writer._partition_spec->fields()[0], TYPE_TIMESTAMPTZ, 0, std::vector<size_t> {}, std::make_unique<IdentityPartitionColumnTransform>(type)); auto column = ColumnTimeStampTz::create(); TimestampTzValue first; @@ -150,9 +160,10 @@ TEST(VIcebergTableWriterTest, StaticNullBinaryPartitionsKeepNullPathsAndCommitVa VExprContextSPtrs exprs; VIcebergTableWriter writer(data_sink, exprs); writer._schema = schema; + writer._partition_spec = std::move(spec); auto type = make_nullable(std::make_shared<DataTypeVarbinary>()); writer._iceberg_partition_columns.emplace_back( - spec->fields()[0], TYPE_VARBINARY, 0, + writer._partition_spec->fields()[0], TYPE_VARBINARY, 0, std::vector<size_t> {}, std::make_unique<IdentityPartitionColumnTransform>(type)); ASSERT_NO_THROW(writer._init_static_partition_values()); EXPECT_TRUE(writer._is_full_static_partition); @@ -172,7 +183,7 @@ TEST(VIcebergTableWriterTest, StaticNullBinaryPartitionsKeepNullPathsAndCommitVa // Partition columns retain a reference to their spec field throughout dispatch. iceberg::PartitionField dynamic_field(2, 1001, "id", "identity"); writer._iceberg_partition_columns.emplace_back( - dynamic_field, TYPE_INT, 1, + dynamic_field, TYPE_INT, 1, std::vector<size_t> {}, std::make_unique<IdentityPartitionColumnTransform>( std::make_shared<DataTypeInt32>())); ASSERT_NO_THROW(writer._init_static_partition_values()); @@ -214,8 +225,9 @@ TEST(VIcebergTableWriterTest, BinaryIdentityStaticAndDynamicRouting) { VExprContextSPtrs exprs; VIcebergTableWriter writer(data_sink, exprs); writer._schema = schema; + writer._partition_spec = std::move(spec); writer._iceberg_partition_columns.emplace_back( - spec->fields()[0], TYPE_VARBINARY, 0, + writer._partition_spec->fields()[0], TYPE_VARBINARY, 0, std::vector<size_t> {}, std::make_unique<IdentityPartitionColumnTransform>(type)); auto column = type->create_column(); const std::string bytes("\xDE\xAD", 2); @@ -239,7 +251,7 @@ TEST(VIcebergTableWriterTest, BinaryIdentityStaticAndDynamicRouting) { // Add a dynamic field to exercise hybrid routing with the same static binary value. iceberg::PartitionField dynamic_field(2, 1001, "id", "identity"); writer._iceberg_partition_columns.emplace_back( - dynamic_field, TYPE_INT, 1, + dynamic_field, TYPE_INT, 1, std::vector<size_t> {}, std::make_unique<IdentityPartitionColumnTransform>(std::make_shared<DataTypeInt32>())); writer._init_static_partition_values(); EXPECT_FALSE(writer._is_full_static_partition); @@ -288,8 +300,9 @@ TEST(VIcebergTableWriterTest, UuidStaticPartitionAcceptsCanonicalText) { VExprContextSPtrs exprs; VIcebergTableWriter writer(data_sink, exprs); writer._schema = schema; + writer._partition_spec = std::move(spec); writer._iceberg_partition_columns.emplace_back( - spec->fields()[0], TYPE_VARBINARY, 0, + writer._partition_spec->fields()[0], TYPE_VARBINARY, 0, std::vector<size_t> {}, std::make_unique<IdentityPartitionColumnTransform>( std::make_shared<DataTypeVarbinary>())); writer._init_static_partition_values(); @@ -300,4 +313,62 @@ TEST(VIcebergTableWriterTest, UuidStaticPartitionAcceptsCanonicalText) { EXPECT_THROW(writer._init_static_partition_values(), Exception); } +TEST(VIcebergTableWriterTest, ResolvesNestedPartitionSource) { + std::vector<iceberg::NestedField> children; + children.emplace_back(true, 2, "part", std::make_unique<iceberg::IntegerType>(), std::nullopt); + std::vector<iceberg::NestedField> columns; + columns.emplace_back(true, 1, "payload", + std::make_unique<iceberg::StructType>(std::move(children)), std::nullopt); + auto schema = std::make_shared<iceberg::Schema>(std::move(columns)); + const std::string spec_json = R"({"spec-id":1,"fields":[{"name":"part","transform":"identity",)" + R"("source-id":2,"field-id":1000}]})"; + auto child_type = make_nullable(std::make_shared<DataTypeInt32>()); + auto struct_type = make_nullable( + std::make_shared<DataTypeStruct>(DataTypes {child_type}, Strings {"part"})); + VExprContextSPtrs output_exprs { + VExprContext::create_shared(VSlotRef::create_shared(0, 0, -1, struct_type, "payload"))}; + + TIcebergTableSink iceberg_sink; + TDataSink data_sink; + data_sink.__set_iceberg_table_sink(iceberg_sink); + VIcebergTableWriter writer(data_sink, output_exprs); + writer._schema = schema; + writer._partition_spec = iceberg::PartitionSpecParser::from_json(schema, spec_json); + + auto partition_columns = writer._to_iceberg_partition_columns(); + ASSERT_EQ(partition_columns.size(), 1); + EXPECT_EQ(partition_columns[0].source_idx(), 0); + EXPECT_EQ(partition_columns[0].child_indices(), std::vector<size_t>({0})); + EXPECT_EQ(partition_columns[0].source_type(), TYPE_INT); + + auto child_data = ColumnInt32::create(); + child_data->insert_value(7); + child_data->insert_value(8); + auto child_nulls = ColumnUInt8::create(2, 0); + auto child_column = ColumnNullable::create(std::move(child_data), std::move(child_nulls)); + Columns children_columns {std::move(child_column)}; + auto struct_column = ColumnStruct::create(std::move(children_columns)); + auto parent_nulls = ColumnUInt8::create(2, 0); + parent_nulls->get_data()[1] = 1; + auto parent_column = ColumnNullable::create(std::move(struct_column), std::move(parent_nulls)); + Block block; + block.insert({std::move(parent_column), struct_type, "payload"}); + + auto source = writer._nested_partition_source(block, partition_columns[0]); + const auto* nullable_source = check_and_get_column<ColumnNullable>(source.column.get()); + ASSERT_NE(nullable_source, nullptr); + ASSERT_EQ(nullable_source->size(), 2); + EXPECT_EQ(nullable_source->get_null_map_data()[0], 0); + EXPECT_EQ(nullable_source->get_null_map_data()[1], 1); + const auto* source_data = + check_and_get_column<ColumnInt32>(nullable_source->get_nested_column_ptr().get()); + ASSERT_NE(source_data, nullptr); + EXPECT_EQ(source_data->get_data()[0], 7); + EXPECT_EQ(source_data->get_data()[1], 8); + + // A static label must not override the nested leaf actually stored in the row. + writer._t_sink.iceberg_table_sink.__set_static_partition_values({{"part", "9"}}); + EXPECT_THROW(writer._init_static_partition_values(), Exception); +} + } // namespace doris diff --git a/be/test/format/table/iceberg/schema_test.cpp b/be/test/format/table/iceberg/schema_test.cpp index bccf0f1f341..b6e56b38ced 100644 --- a/be/test/format/table/iceberg/schema_test.cpp +++ b/be/test/format/table/iceberg/schema_test.cpp @@ -66,5 +66,21 @@ TEST(SchemaTest, test_find_field) { EXPECT_EQ(found_field2->field_id(), 2); } +TEST(SchemaTest, FindNestedFieldPath) { + std::vector<NestedField> children; + children.emplace_back(true, 2, "part", std::make_unique<IntegerType>(), std::nullopt); + std::vector<NestedField> columns; + columns.emplace_back(true, 1, "payload", std::make_unique<StructType>(std::move(children)), + std::nullopt); + Schema schema(1, std::move(columns)); + + const auto* path = schema.find_field_path(2); + ASSERT_NE(path, nullptr); + ASSERT_EQ(path->size(), 2); + EXPECT_EQ((*path)[0]->field_id(), 1); + EXPECT_EQ((*path)[1]->field_id(), 2); + EXPECT_EQ(schema.find_type(2)->type_id(), TypeID::INTEGER); +} + } // namespace iceberg } // namespace doris diff --git a/be/test/format_v2/table/iceberg_position_delete_sys_table_reader_test.cpp b/be/test/format_v2/table/iceberg_position_delete_sys_table_reader_test.cpp index a219ba37f3d..8cf538eb0cf 100644 --- a/be/test/format_v2/table/iceberg_position_delete_sys_table_reader_test.cpp +++ b/be/test/format_v2/table/iceberg_position_delete_sys_table_reader_test.cpp @@ -19,6 +19,9 @@ #include <gtest/gtest.h> +#include "core/data_type/data_type_number.h" +#include "core/data_type/data_type_struct.h" +#include "format/table/iceberg_scan_semantics.h" #include "runtime/runtime_profile.h" #include "runtime/runtime_state.h" @@ -59,5 +62,31 @@ TEST(IcebergPositionDeleteSysTableV2ProfileTest, UsesDistinctProfileForNestedPos reader._position_reader_profile); } +TEST(IcebergPositionDeleteSysTableV2ProfileTest, PreparesNestedRowInitialDefaults) { + IcebergPositionDeleteSysTableV2Reader reader; + TFileScanRangeParams params; + params.__set_iceberg_scan_semantics_version(ICEBERG_SCAN_SEMANTICS_VERSION_2); + reader._scan_params = ¶ms; + const auto child_type = make_nullable(std::make_shared<DataTypeInt32>()); + const auto row_type = make_nullable( + std::make_shared<DataTypeStruct>(DataTypes {child_type}, Strings {"added"})); + ColumnDefinition row; + row.name = "row"; + row.type = row_type; + ColumnDefinition child; + child.name = "added"; + child.type = child_type; + child.initial_default_value = "7"; + row.children.push_back(std::move(child)); + reader._projected_columns = {row}; + reader._read_columns = {{"row", row_type}}; + + std::vector<ColumnDefinition> columns; + ASSERT_TRUE(reader._build_delete_file_projected_columns(&columns).ok()); + ASSERT_EQ(columns.size(), 1); + ASSERT_EQ(columns[0].children.size(), 1); + EXPECT_NE(columns[0].children[0].default_expr, nullptr); +} + } // namespace } // namespace doris::format::iceberg diff --git a/be/test/format_v2/table/iceberg_reader_test.cpp b/be/test/format_v2/table/iceberg_reader_test.cpp index 234531335af..dbf8109f40f 100644 --- a/be/test/format_v2/table/iceberg_reader_test.cpp +++ b/be/test/format_v2/table/iceberg_reader_test.cpp @@ -28,6 +28,7 @@ #include <array> #include <bit> #include <chrono> +#include <cmath> #include <cstring> #include <filesystem> #include <fstream> @@ -1553,6 +1554,38 @@ void apply_final_conjuncts(Block* block, const VExprContextSPtrs& conjuncts) { ASSERT_TRUE(status.ok()) << status; } +TEST(IcebergV2ReaderTest, PreparesIcebergNonFiniteInitialDefaults) { + struct Case { + DataTypePtr type; + std::string value; + bool nan; + bool negative; + }; + std::vector<Case> cases {{std::make_shared<DataTypeFloat32>(), "NaN", true, false}, + {std::make_shared<DataTypeFloat64>(), "Infinity", false, false}, + {std::make_shared<DataTypeFloat64>(), "-Infinity", false, true}}; + for (const auto& test_case : cases) { + ColumnDefinition column; + column.name = "value"; + column.type = test_case.type; + column.initial_default_value = test_case.value; + ASSERT_TRUE(iceberg::prepare_iceberg_initial_default_exprs(&column).ok()); + ASSERT_NE(column.default_expr, nullptr); + const auto* literal = dynamic_cast<const VLiteral*>(column.default_expr->root().get()); + ASSERT_NE(literal, nullptr); + Field value; + literal->get_column_ptr()->get(0, value); + const double number = test_case.type->get_primitive_type() == TYPE_FLOAT + ? value.get<TYPE_FLOAT>() + : value.get<TYPE_DOUBLE>(); + EXPECT_EQ(std::isnan(number), test_case.nan); + if (!test_case.nan) { + EXPECT_TRUE(std::isinf(number)); + EXPECT_EQ(std::signbit(number), test_case.negative); + } + } +} + TEST(IcebergV2ReaderTest, AnnotateBuildsTypedNestedInitialDefault) { const auto int_type = make_nullable(std::make_shared<DataTypeInt32>()); const auto struct_type = make_nullable( diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergTransaction.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergTransaction.java index 435d8af7d5d..444798b936a 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergTransaction.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergTransaction.java @@ -1019,14 +1019,18 @@ public class IcebergTransaction implements Transaction { if (sourceField == null) { return null; } + String sourceColumnName = identitySourceColumnName(schema, field.sourceId()); + if (sourceColumnName == null) { + return null; + } String valueStr = partitionValues.get(i); if ("null".equals(valueStr)) { valueStr = null; } Object value = IcebergUtils.parsePartitionValueFromString(valueStr, sourceField.type()); Expression predicate = value == null - ? Expressions.isNull(sourceField.name()) - : Expressions.equal(sourceField.name(), value); + ? Expressions.isNull(sourceColumnName) + : Expressions.equal(sourceColumnName, value); expression = expression == null ? predicate : Expressions.and(expression, predicate); } return expression; @@ -1232,14 +1236,18 @@ public class IcebergTransaction implements Transaction { throw new RuntimeException(String.format("Source field not found for partition field: %s", partitionColName)); } + String sourceColName = identitySourceColumnName(schema, field.sourceId()); + if (sourceColName == null) { + throw new RuntimeException(String.format( + "Source column path not found for partition field: %s", partitionColName)); + } // Convert partition value string to appropriate type Object partitionValue = IcebergUtils.parsePartitionValueFromString( partitionValueStr, sourceField.type()); - // Build equality expression using source field name (not partition field name) - // For identity partitions, Iceberg requires the source column name in expressions - String sourceColName = sourceField.name(); + // Build equality expression using the source column path (not partition field name). + // For identity partitions, Iceberg requires the source column in expressions. Expression eqExpr; if (partitionValue == null) { eqExpr = Expressions.isNull(sourceColName); @@ -1268,4 +1276,10 @@ public class IcebergTransaction implements Transaction { } return result; } + + private String identitySourceColumnName(Schema schema, int sourceId) { + // Nested identity predicates must bind to the full path when sibling structs reuse a leaf name. + return schema.findColumnName(sourceId); + } + } diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergWriteSchemaContext.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergWriteSchemaContext.java index 3437f26c4cf..8b0aebb2e0b 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergWriteSchemaContext.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergWriteSchemaContext.java @@ -307,13 +307,12 @@ public final class IcebergWriteSchemaContext { private static void validateWriterMetadataSources( Schema schema, PartitionSpec partitionSpec, SortOrder sortOrder, String tableName) { - Map<Integer, Types.NestedField> topLevelFields = schema.columns().stream() - .collect(ImmutableMap.toImmutableMap(Types.NestedField::fieldId, field -> field)); for (PartitionField field : partitionSpec.fields()) { - if (!topLevelFields.containsKey(field.sourceId())) { + // Iceberg permits a nested primitive partition source; its ID is schema-wide. + if (schema.findField(field.sourceId()) == null) { throw new AnalysisException("Iceberg partition field " + field.fieldId() + " references source field " + field.sourceId() - + " outside pinned top-level schema " + schema.schemaId() + + " outside pinned schema " + schema.schemaId() + " for table " + tableName); } } diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/StatementContext.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/StatementContext.java index fa6907965ff..7de912f1ef6 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/StatementContext.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/StatementContext.java @@ -282,6 +282,7 @@ public class StatementContext implements Closeable { private Backend groupCommitMergeBackend; private final Map<MvccTableInfo, MvccSnapshot> snapshots = Maps.newHashMap(); + private final Map<MvccTableInfo, MvccSnapshot> injectedSnapshots = Maps.newHashMap(); private final Map<MvccTableInfo, MvccSnapshot> latestSnapshots = Maps.newHashMap(); private final Map<MvccTableInfo, MvccSnapshot> latestSnapshotFences = Maps.newHashMap(); private final Map<MvccTableInfo, Map<String, String>> resolvedSnapshotScanParams = Maps.newHashMap(); @@ -1159,6 +1160,7 @@ public class StatementContext implements Closeable { * to one EXECUTE only; carrying it forward would make a later commit permanently invisible. */ public void resetMvccSnapshots() { + injectedSnapshots.clear(); snapshots.clear(); latestSnapshots.clear(); latestSnapshotFences.clear(); @@ -1173,6 +1175,14 @@ public class StatementContext implements Closeable { externalMetadataPreloadResult = null; } + public void resetMvccSnapshotsForRetry() { + // MTMV refresh partitions and rewrite inputs were selected against these task-owned fences. + // Only planner-loaded metadata may advance when the same execution retries planning. + Map<MvccTableInfo, MvccSnapshot> preservedSnapshots = Maps.newHashMap(injectedSnapshots); + resetMvccSnapshots(); + preservedSnapshots.forEach(this::setSnapshot); + } + /** * Obtain snapshot information of mvcc * @@ -1260,6 +1270,7 @@ public class StatementContext implements Closeable { * @param snapshot snapshot */ public void setSnapshot(MvccTableInfo mvccTableInfo, MvccSnapshot snapshot) { + injectedSnapshots.put(mvccTableInfo, snapshot); // Injected snapshots are execution fences (for example MTMV refresh); an unqualified // relation must reuse that fence instead of treating the latest-snapshot cache as empty. latestSnapshots.put(mvccTableInfo, snapshot); diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java index 59508d17793..98cc6ca4f9a 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java @@ -3307,7 +3307,8 @@ public class PhysicalPlanTranslator extends DefaultPlanVisitor<PlanFragment, Pla field.getTransform(), field.getParam(), field.getName(), - field.getSourceId())); + field.getSourceId(), + field.getSourceFieldPath())); } return new DataPartition(TPartitionType.MERGE_PARTITIONED, operationExpr, insertPartitionExprs, deletePartitionExprs, mergeSpec.isInsertRandom(), diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/DistributionSpecMerge.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/DistributionSpecMerge.java index c6ea4e37a8b..63935cb8f2d 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/DistributionSpecMerge.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/DistributionSpecMerge.java @@ -38,17 +38,25 @@ public class DistributionSpecMerge extends DistributionSpec { private final Integer param; private final String name; private final Integer sourceId; + private final ImmutableList<Integer> sourceFieldPath; /** * Create a partition field mapping for merge insert routing. */ public IcebergPartitionField(String transform, ExprId sourceExprId, Integer param, String name, Integer sourceId) { + this(transform, sourceExprId, param, name, sourceId, ImmutableList.of()); + } + + /** Create a partition field mapping whose source is nested below a top-level slot. */ + public IcebergPartitionField(String transform, ExprId sourceExprId, Integer param, + String name, Integer sourceId, List<Integer> sourceFieldPath) { this.transform = Objects.requireNonNull(transform, "transform should not be null"); this.sourceExprId = Objects.requireNonNull(sourceExprId, "sourceExprId should not be null"); this.param = param; this.name = name; this.sourceId = sourceId; + this.sourceFieldPath = ImmutableList.copyOf(sourceFieldPath); } public String getTransform() { @@ -71,6 +79,10 @@ public class DistributionSpecMerge extends DistributionSpec { return sourceId; } + public List<Integer> getSourceFieldPath() { + return sourceFieldPath; + } + @Override public boolean equals(Object o) { if (this == o) { @@ -84,12 +96,13 @@ public class DistributionSpecMerge extends DistributionSpec { && sourceExprId.equals(that.sourceExprId) && Objects.equals(param, that.param) && Objects.equals(name, that.name) - && Objects.equals(sourceId, that.sourceId); + && Objects.equals(sourceId, that.sourceId) + && sourceFieldPath.equals(that.sourceFieldPath); } @Override public int hashCode() { - return Objects.hash(transform, sourceExprId, param, name, sourceId); + return Objects.hash(transform, sourceExprId, param, name, sourceId, sourceFieldPath); } } diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoTableCommand.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoTableCommand.java index 20f21f5b8b2..97c58141256 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoTableCommand.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoTableCommand.java @@ -264,6 +264,7 @@ public class InsertIntoTableCommand extends Command implements NeedAuditEncrypti int retryTimes = 0; ctx.getStatementContext().setIsInsert(true); while (++retryTimes < Math.max(ctx.getSessionVariable().dmlPlanRetryTimes, 3)) { + prepareMvccSnapshotsForPlanAttempt(ctx.getStatementContext(), retryTimes); TableIf targetTableIf = getTargetTableIf(ctx, qualifiedTargetTableName); DatabaseIf<?> targetDatabase = getTargetDatabase(targetTableIf); // check auth @@ -341,6 +342,13 @@ public class InsertIntoTableCommand extends Command implements NeedAuditEncrypti throw new AnalysisException("Insert plan failed. Could not get target table lock."); } + static void prepareMvccSnapshotsForPlanAttempt(StatementContext statementContext, int attempt) { + if (attempt > 1) { + // Refresh planner metadata while retaining the source fences selected by the caller. + statementContext.resetMvccSnapshotsForRetry(); + } + } + /** * Hook method to determine if auth check is needed. * Subclasses can override this to skip auth check, e.g., WarmupSelectCommand. diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalIcebergMergeSink.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalIcebergMergeSink.java index 85748b4ce18..d20e948292d 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalIcebergMergeSink.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalIcebergMergeSink.java @@ -21,6 +21,7 @@ import org.apache.doris.catalog.Column; import org.apache.doris.datasource.iceberg.IcebergExternalDatabase; import org.apache.doris.datasource.iceberg.IcebergExternalTable; import org.apache.doris.datasource.iceberg.IcebergMergeOperation; +import org.apache.doris.datasource.iceberg.IcebergUtils; import org.apache.doris.datasource.iceberg.IcebergWriteSchemaContext; import org.apache.doris.nereids.memo.GroupExpression; import org.apache.doris.nereids.properties.DistributionSpecMerge; @@ -273,12 +274,13 @@ public class PhysicalIcebergMergeSink<CHILD_TYPE extends Plan> extends PhysicalB .orElse(targetIcebergTable.schema()); List<Column> partitionColumns = getRetainedPartitionColumns(partitionSpec, partitionSchema); Map<String, ExprId> columnExprIdMap = buildColumnExprIdMap(outputSlots, nameToExprId); + Map<Integer, ExprId> columnIdToExprId = buildColumnIdExprIdMap(outputSlots); boolean insertExprsOk = false; if (!partitionColumns.isEmpty()) { insertExprsOk = buildInsertPartitionExprIds(insertPartitionExprIds, partitionColumns, columnExprIdMap); } InsertPartitionFieldResult fieldResult = buildInsertPartitionFields( - insertPartitionFields, partitionSpec, partitionSchema, columnExprIdMap); + insertPartitionFields, partitionSpec, partitionSchema, columnIdToExprId); boolean insertFieldsOk = fieldResult.success; boolean hasNonIdentity = fieldResult.hasNonIdentity; if (insertFieldsOk) { @@ -344,7 +346,9 @@ public class PhysicalIcebergMergeSink<CHILD_TYPE extends Plan> extends PhysicalB if (IcebergMergeOperation.OPERATION_COLUMN.equalsIgnoreCase(name)) { continue; } - if (Column.ICEBERG_ROWID_COL.equalsIgnoreCase(name)) { + // V3 lineage slots have no visible schema column and must not shift positional binding. + if (Column.ICEBERG_ROWID_COL.equalsIgnoreCase(name) + || IcebergUtils.isIcebergRowLineageColumn(name)) { continue; } dataSlots.add(slot); @@ -356,7 +360,7 @@ public class PhysicalIcebergMergeSink<CHILD_TYPE extends Plan> extends PhysicalB List<DistributionSpecMerge.IcebergPartitionField> insertPartitionFields, PartitionSpec spec, Schema schema, - Map<String, ExprId> columnExprIdMap) { + Map<Integer, ExprId> columnIdToExprId) { if (spec == null || !spec.isPartitioned()) { return new InsertPartitionFieldResult(false, false, null); } @@ -376,15 +380,24 @@ public class PhysicalIcebergMergeSink<CHILD_TYPE extends Plan> extends PhysicalB insertPartitionFields.clear(); return new InsertPartitionFieldResult(false, hasNonIdentity, spec.specId()); } - ExprId exprId = columnExprIdMap.get(sourceField.name()); + Column rootColumn = findSourceRootColumn(cols, field.sourceId()); + // Nested routing must bind through stable Iceberg IDs; a name fallback can target a + // different field after rename/drop-and-add schema evolution. + ExprId exprId = rootColumn == null || rootColumn.getUniqueId() < 0 + ? null : columnIdToExprId.get(rootColumn.getUniqueId()); if (exprId == null) { insertPartitionFields.clear(); return new InsertPartitionFieldResult(false, hasNonIdentity, spec.specId()); } String transform = field.transform().toString(); Integer param = parseTransformParam(transform); + List<Integer> sourceFieldPath = resolveSourceFieldPath(rootColumn, field.sourceId()); + if (sourceFieldPath == null) { + insertPartitionFields.clear(); + return new InsertPartitionFieldResult(false, hasNonIdentity, spec.specId()); + } insertPartitionFields.add(new DistributionSpecMerge.IcebergPartitionField( - transform, exprId, param, field.name(), field.sourceId())); + transform, exprId, param, field.name(), field.sourceId(), sourceFieldPath)); } if (insertPartitionFields.isEmpty()) { return new InsertPartitionFieldResult(false, hasNonIdentity, spec.specId()); @@ -392,6 +405,75 @@ public class PhysicalIcebergMergeSink<CHILD_TYPE extends Plan> extends PhysicalB return new InsertPartitionFieldResult(true, hasNonIdentity, spec.specId()); } + private Map<Integer, ExprId> buildColumnIdExprIdMap(List<Slot> outputSlots) { + Map<Integer, ExprId> result = new java.util.HashMap<>(); + List<Column> visibleColumns = new ArrayList<>(); + for (Column column : cols) { + if (column.isVisible()) { + visibleColumns.add(column); + } + } + List<Slot> dataSlots = getDataSlots(outputSlots); + if (visibleColumns.size() != dataSlots.size()) { + return result; + } + for (int i = 0; i < visibleColumns.size(); i++) { + if (visibleColumns.get(i).getUniqueId() >= 0) { + result.put(visibleColumns.get(i).getUniqueId(), dataSlots.get(i).getExprId()); + } + } + return result; + } + + private static Column findSourceRootColumn(List<Column> columns, int sourceId) { + for (Column column : columns) { + if (column.getUniqueId() == sourceId || containsFieldId(column.getChildren(), sourceId)) { + return column; + } + } + return null; + } + + private static boolean containsFieldId(List<Column> columns, int sourceId) { + if (columns == null) { + return false; + } + for (Column column : columns) { + if (column.getUniqueId() == sourceId || containsFieldId(column.getChildren(), sourceId)) { + return true; + } + } + return false; + } + + private static List<Integer> resolveSourceFieldPath(Column root, int sourceId) { + if (root == null) { + return null; + } + if (root.getUniqueId() == sourceId) { + return ImmutableList.of(); + } + List<Integer> path = new ArrayList<>(); + return findSourceFieldPath(root.getChildren(), sourceId, path) + ? ImmutableList.copyOf(path) : null; + } + + private static boolean findSourceFieldPath(List<Column> columns, int sourceId, List<Integer> path) { + if (columns == null) { + return false; + } + for (int index = 0; index < columns.size(); index++) { + Column column = columns.get(index); + path.add(index); + if (column.getUniqueId() == sourceId + || findSourceFieldPath(column.getChildren(), sourceId, path)) { + return true; + } + path.remove(path.size() - 1); + } + return false; + } + private List<Column> getRetainedPartitionColumns(PartitionSpec partitionSpec, Schema schema) { Map<String, Column> columnsByName = new TreeMap<>(String.CASE_INSENSITIVE_ORDER); for (Column column : cols) { diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/DataPartition.java b/fe/fe-core/src/main/java/org/apache/doris/planner/DataPartition.java index 8e652821c76..9f20824a9c2 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/planner/DataPartition.java +++ b/fe/fe-core/src/main/java/org/apache/doris/planner/DataPartition.java @@ -205,14 +205,21 @@ public class DataPartition { private final Integer param; private final String name; private final Integer sourceId; + private final ImmutableList<Integer> sourceFieldPath; public IcebergPartitionField(Expr sourceExpr, String transform, Integer param, String name, Integer sourceId) { + this(sourceExpr, transform, param, name, sourceId, ImmutableList.of()); + } + + public IcebergPartitionField(Expr sourceExpr, String transform, Integer param, + String name, Integer sourceId, List<Integer> sourceFieldPath) { this.sourceExpr = Preconditions.checkNotNull(sourceExpr, "sourceExpr should not be null"); this.transform = Preconditions.checkNotNull(transform, "transform should not be null"); this.param = param; this.name = name; this.sourceId = sourceId; + this.sourceFieldPath = ImmutableList.copyOf(sourceFieldPath); } public TIcebergPartitionField toThrift() { @@ -228,6 +235,9 @@ public class DataPartition { if (sourceId != null) { field.setSourceId(sourceId); } + if (!sourceFieldPath.isEmpty()) { + field.setSourceFieldPath(sourceFieldPath); + } return field; } diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/IcebergTransactionTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/IcebergTransactionTest.java index 793e0653dd0..e9300c82d76 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/IcebergTransactionTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/IcebergTransactionTest.java @@ -49,6 +49,7 @@ import org.apache.iceberg.Table; import org.apache.iceberg.catalog.Namespace; import org.apache.iceberg.catalog.TableIdentifier; import org.apache.iceberg.exceptions.CommitFailedException; +import org.apache.iceberg.expressions.Binder; import org.apache.iceberg.expressions.Expression; import org.apache.iceberg.expressions.Expressions; import org.apache.iceberg.expressions.UnboundPredicate; @@ -1430,6 +1431,19 @@ public class IcebergTransactionTest { Collections.singletonMap("unknown", "2026-01-01"), spec, schema)); } + @Test + public void testStaticPartitionFilterBindsNestedIdentitySourceByFullPath() { + Schema schema = new Schema(Types.NestedField.required( + 1, "payload", Types.StructType.of( + Types.NestedField.required(2, "region", Types.StringType.get())))); + PartitionSpec spec = PartitionSpec.builderFor(schema).identity("payload.region").build(); + + Expression filter = getTxn().buildPartitionFilter( + Collections.singletonMap(spec.fields().get(0).name(), "us-east"), spec, schema); + + Assert.assertNotNull(Binder.bind(schema.asStruct(), filter, true)); + } + private DeleteFile buildDeletionVectorDeleteFile(String puffinPath, String referencedDataFile, long contentOffset, long contentLength) { return FileMetadata.deleteFileBuilder(PartitionSpec.unpartitioned()) diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/IcebergWriteSchemaContextTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/IcebergWriteSchemaContextTest.java index 3df06081bfd..f587a4bc411 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/IcebergWriteSchemaContextTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/IcebergWriteSchemaContextTest.java @@ -41,7 +41,9 @@ import org.apache.doris.qe.ConnectContext; import com.google.common.collect.ImmutableList; import com.google.common.collect.ImmutableMap; +import org.apache.iceberg.FileFormat; import org.apache.iceberg.HasTableOperations; +import org.apache.iceberg.MetricsConfig; import org.apache.iceberg.PartitionSpec; import org.apache.iceberg.Schema; import org.apache.iceberg.SchemaParser; @@ -75,6 +77,21 @@ import java.util.stream.Collectors; public class IcebergWriteSchemaContextTest { + @Test + public void testWriterMetadataAcceptsNestedPartitionSource() { + Schema schema = new Schema(Types.NestedField.optional(1, "payload", + Types.StructType.of(Types.NestedField.optional(2, "part", Types.IntegerType.get())))); + PartitionSpec spec = PartitionSpec.builderFor(schema).bucket("payload.part", 8).build(); + + IcebergWriteSchemaContext context = IcebergWriteSchemaContext.forSchema( + schema, 2, spec, SortOrder.unsorted(), FileFormat.PARQUET, + MetricsConfig.getDefault(), "zstd", "s3://bucket/table/data", + ImmutableMap.of(), true, true); + + Assertions.assertEquals(2, context.getPartitionSpec().fields().get(0).sourceId()); + Assertions.assertEquals(2, context.getColumns().get(0).getChildren().get(0).getUniqueId()); + } + @Test public void testReadAndWriteBindingsPreserveNestedTypesWithLegacyFlags() { Schema schema = new Schema( diff --git a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertUtilsTest.java b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertUtilsTest.java index d03ad56593d..e8617fa68f4 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertUtilsTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertUtilsTest.java @@ -22,6 +22,7 @@ import org.apache.doris.catalog.DatabaseIf; import org.apache.doris.catalog.PrimitiveType; import org.apache.doris.catalog.Type; import org.apache.doris.common.security.authentication.ExecutionAuthenticator; +import org.apache.doris.datasource.CatalogIf; import org.apache.doris.datasource.iceberg.IcebergExternalCatalog; import org.apache.doris.datasource.iceberg.IcebergExternalTable; import org.apache.doris.datasource.iceberg.IcebergMvccSnapshot; @@ -29,6 +30,8 @@ import org.apache.doris.datasource.iceberg.IcebergPartitionInfo; import org.apache.doris.datasource.iceberg.IcebergSnapshot; import org.apache.doris.datasource.iceberg.IcebergSnapshotCacheValue; import org.apache.doris.datasource.iceberg.IcebergWriteSchemaContext; +import org.apache.doris.datasource.mvcc.MvccSnapshot; +import org.apache.doris.datasource.mvcc.MvccTableInfo; import org.apache.doris.nereids.StatementContext; import org.apache.doris.nereids.analyzer.UnboundAlias; import org.apache.doris.nereids.analyzer.UnboundIcebergTableSink; @@ -151,6 +154,50 @@ public class InsertUtilsTest { .build(); } + @Test + public void testInsertPlanRetryPreservesInjectedSnapshotAndReloadsPlannerSnapshot() { + StatementContext statementContext = new StatementContext(); + IcebergExternalTable table = Mockito.mock(IcebergExternalTable.class); + DatabaseIf database = Mockito.mock(DatabaseIf.class); + CatalogIf catalog = Mockito.mock(CatalogIf.class); + Mockito.when(table.getName()).thenReturn("table"); + Mockito.when(table.getDatabase()).thenReturn(database); + Mockito.when(database.getFullName()).thenReturn("db"); + Mockito.when(database.getCatalog()).thenReturn(catalog); + Mockito.when(catalog.getName()).thenReturn("catalog"); + MvccSnapshot snapshot = Mockito.mock(MvccSnapshot.class); + MvccSnapshot newerSnapshot = Mockito.mock(MvccSnapshot.class); + Mockito.when(table.loadSnapshot(Optional.empty(), Optional.empty())).thenReturn(newerSnapshot); + statementContext.setSnapshot(new MvccTableInfo(table), snapshot); + + InsertIntoTableCommand.prepareMvccSnapshotsForPlanAttempt(statementContext, 1); + + Assertions.assertSame(snapshot, + statementContext.loadSnapshots(table, Optional.empty(), Optional.empty()) + .orElseThrow(AssertionError::new)); + Mockito.verify(table, Mockito.never()).loadSnapshot(Optional.empty(), Optional.empty()); + IcebergExternalTable plannerTable = Mockito.mock(IcebergExternalTable.class); + Mockito.when(plannerTable.getName()).thenReturn("planner_table"); + Mockito.when(plannerTable.getDatabase()).thenReturn(database); + Mockito.when(plannerTable.loadSnapshot(Optional.empty(), Optional.empty())) + .thenReturn(snapshot, newerSnapshot); + Assertions.assertSame(snapshot, statementContext.loadSnapshots( + plannerTable, Optional.empty(), Optional.empty()).orElseThrow(AssertionError::new)); + + // Target-schema retries must keep the source generation selected for the MTMV refresh. + for (int attempt = 2; attempt <= 3; attempt++) { + InsertIntoTableCommand.prepareMvccSnapshotsForPlanAttempt(statementContext, attempt); + Assertions.assertSame(snapshot, statementContext.loadSnapshots( + table, Optional.empty(), Optional.empty()).orElseThrow(AssertionError::new)); + Assertions.assertSame(newerSnapshot, statementContext.loadSnapshots( + plannerTable, Optional.empty(), Optional.empty()).orElseThrow(AssertionError::new)); + } + Mockito.verify(table, Mockito.never()).loadSnapshot(Optional.empty(), Optional.empty()); + statementContext.resetMvccSnapshots(); + Assertions.assertSame(newerSnapshot, statementContext.loadSnapshots( + table, Optional.empty(), Optional.empty()).orElseThrow(AssertionError::new)); + } + private String generateString(int length) { return generateString(length, "X"); } diff --git a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/physical/PhysicalIcebergMergeSinkTest.java b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/physical/PhysicalIcebergMergeSinkTest.java new file mode 100644 index 00000000000..64de644c983 --- /dev/null +++ b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/physical/PhysicalIcebergMergeSinkTest.java @@ -0,0 +1,135 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package org.apache.doris.nereids.trees.plans.physical; + +import org.apache.doris.catalog.Column; +import org.apache.doris.catalog.Type; +import org.apache.doris.datasource.iceberg.IcebergExternalDatabase; +import org.apache.doris.datasource.iceberg.IcebergExternalTable; +import org.apache.doris.datasource.iceberg.IcebergMergeOperation; +import org.apache.doris.datasource.iceberg.IcebergWriteSchemaContext; +import org.apache.doris.nereids.properties.DistributionSpecMerge; +import org.apache.doris.nereids.properties.LogicalProperties; +import org.apache.doris.nereids.trees.expressions.ExprId; +import org.apache.doris.nereids.trees.expressions.NamedExpression; +import org.apache.doris.nereids.trees.expressions.SlotReference; +import org.apache.doris.nereids.trees.plans.Plan; +import org.apache.doris.nereids.trees.plans.RelationId; +import org.apache.doris.nereids.trees.plans.commands.delete.DeleteCommandContext; +import org.apache.doris.nereids.types.DataType; +import org.apache.doris.nereids.types.IntegerType; +import org.apache.doris.qe.ConnectContext; + +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.Schema; +import org.apache.iceberg.Table; +import org.apache.iceberg.types.Types; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; +import org.mockito.Mockito; + +import java.lang.reflect.Method; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.Optional; + +public class PhysicalIcebergMergeSinkTest { + @Test + public void nestedPartitionRoutesThroughRootIdAfterRename() throws Exception { + Schema schema = new Schema(Types.NestedField.optional(1, "payload", + Types.StructType.of(Types.NestedField.optional(2, "part", Types.IntegerType.get())))); + PartitionSpec spec = PartitionSpec.builderFor(schema).bucket("payload.part", 8).build(); + Column root = Mockito.mock(Column.class); + Column leaf = new Column("renamed_part", Type.INT); + leaf.setUniqueId(2); + Mockito.when(root.getUniqueId()).thenReturn(1); + Mockito.when(root.getChildren()).thenReturn(new ArrayList<>(Collections.singletonList(leaf))); + PhysicalIcebergMergeSink<Plan> sink = new PhysicalIcebergMergeSink<>( + Mockito.mock(IcebergExternalDatabase.class), Mockito.mock(IcebergExternalTable.class), + Mockito.mock(Table.class), Collections.singletonList(root), Collections.emptyList(), + new DeleteCommandContext(), true, false, Optional.empty(), + Mockito.mock(LogicalProperties.class), + new PhysicalEmptyRelation(new RelationId(1), Collections.emptyList(), null)); + Method build = PhysicalIcebergMergeSink.class.getDeclaredMethod("buildInsertPartitionFields", + List.class, PartitionSpec.class, Schema.class, Map.class); + build.setAccessible(true); + ExprId rootExprId = new ExprId(42); + List<DistributionSpecMerge.IcebergPartitionField> fields = new ArrayList<>(); + + build.invoke(sink, fields, spec, schema, Collections.singletonMap(1, rootExprId)); + + Assertions.assertEquals(1, fields.size()); + Assertions.assertEquals(rootExprId, fields.get(0).getSourceExprId()); + Assertions.assertEquals(2, fields.get(0).getSourceId()); + Assertions.assertEquals(Collections.singletonList(0), fields.get(0).getSourceFieldPath()); + Assertions.assertEquals(8, fields.get(0).getParam()); + + // A missing stable root ID must disable field-based routing to prevent a wrong partition. + fields.clear(); + Mockito.when(root.getUniqueId()).thenReturn(-1); + build.invoke(sink, fields, spec, schema, Collections.singletonMap(1, rootExprId)); + Assertions.assertTrue(fields.isEmpty()); + } + + @Test + public void v3LineageSlotsDoNotDisablePartitionRouting() { + Schema schema = new Schema( + Types.NestedField.optional(1, "payload", Types.StructType.of( + Types.NestedField.optional(2, "part", Types.IntegerType.get()))), + Types.NestedField.optional(3, "id", Types.IntegerType.get())); + List<Column> columns = IcebergWriteSchemaContext.forSchema(schema, 3, true, true).getMergeColumns(); + List<NamedExpression> output = new ArrayList<>(); + for (Column column : columns) { + output.add(new SlotReference(column.getName(), DataType.fromCatalogType(column.getType()))); + } + output.add(new SlotReference(IcebergMergeOperation.OPERATION_COLUMN, IntegerType.INSTANCE)); + PhysicalEmptyRelation child = new PhysicalEmptyRelation(new RelationId(2), output, null); + ConnectContext previous = ConnectContext.get(); + ConnectContext context = new ConnectContext(); + context.getSessionVariable().enableIcebergMergePartitioning = true; + context.setThreadLocalInfo(); + try { + for (PartitionSpec spec : new PartitionSpec[] { + PartitionSpec.builderFor(schema).bucket("id", 8).build(), + PartitionSpec.builderFor(schema).bucket("payload.part", 8).build(), + PartitionSpec.builderFor(schema).identity("payload.part").build()}) { + Table table = Mockito.mock(Table.class); + Mockito.when(table.schema()).thenReturn(schema); + Mockito.when(table.spec()).thenReturn(spec); + PhysicalIcebergMergeSink<PhysicalEmptyRelation> sink = new PhysicalIcebergMergeSink<>( + Mockito.mock(IcebergExternalDatabase.class), Mockito.mock(IcebergExternalTable.class), + table, columns, output, new DeleteCommandContext(), true, false, + Optional.empty(), null, child); + DistributionSpecMerge distribution = (DistributionSpecMerge) + sink.getRequirePhysicalProperties().getDistributionSpec(); + Assertions.assertFalse(distribution.isInsertRandom()); + Assertions.assertEquals(1, distribution.getInsertPartitionFields().size()); + Assertions.assertEquals(spec.fields().get(0).sourceId(), + distribution.getInsertPartitionFields().get(0).getSourceId()); + } + } finally { + if (previous == null) { + ConnectContext.remove(); + } else { + previous.setThreadLocalInfo(); + } + } + } +} diff --git a/gensrc/thrift/Partitions.thrift b/gensrc/thrift/Partitions.thrift index 73863696115..a49a4c9d1b8 100644 --- a/gensrc/thrift/Partitions.thrift +++ b/gensrc/thrift/Partitions.thrift @@ -99,6 +99,8 @@ struct TIcebergPartitionField { 3: required Exprs.TExpr source_expr 4: optional string name 5: optional i32 source_id + // Zero-based STRUCT child indexes below source_expr; empty/unset means a top-level source. + 6: optional list<i32> source_field_path } struct TMergePartitionInfo { diff --git a/regression-test/data/external_table_p0/iceberg/write/test_iceberg_write_complex_evolution.out b/regression-test/data/external_table_p0/iceberg/write/test_iceberg_write_complex_evolution.out index 76bede82d3f..673d6c879cb 100644 --- a/regression-test/data/external_table_p0/iceberg/write/test_iceberg_write_complex_evolution.out +++ b/regression-test/data/external_table_p0/iceberg/write/test_iceberg_write_complex_evolution.out @@ -3,8 +3,9 @@ 1 A [1, null, 3] {"x":10, "null-value":null} {"metric":10, "label":"old-a", "nested":{"count":1, "comment":null, "score":null}, "tags":null, "attributes":null} 2 N \N {"x":null} {"metric":20, "label":null, "nested":{"count":null, "comment":"old-null", "score":null}, "tags":null, "attributes":null} 3 B [] {} \N -4 A1 [4000000000, null] {"large":5000000000, "null-value":null} {"metric":6000000000, "label":"new-a", "nested":{"count":7000000000, "comment":"nested-new", "score":7.5}, "tags":["x", null, "z"], "attributes":{"a":8000000000, "b":null}} +4 A2 [4000000000, null] {"large":5000000000, "null-value":null} {"metric":6000000000, "label":"new-a", "nested":{"count":7000000000, "comment":"nested-new", "score":7.5}, "tags":["x", null, "z"], "attributes":{"a":8000000000, "b":null}} 5 N2 [null] \N {"metric":50, "label":null, "nested":{"count":5, "comment":null, "score":null}, "tags":null, "attributes":{"null-value":null}} +6 Z4 \N \N \N -- !complex_children -- 1 10 1 \N \N \N @@ -12,19 +13,24 @@ 3 \N \N \N \N \N 4 6000000000 7000000000 7.5 ["x", null, "z"] {"a":8000000000, "b":null} 5 50 5 \N \N {"null-value":null} +6 \N \N \N \N \N -- !complex_nulls -- 1 2 3 5 +6 -- !complex_partition_specs -- 0 3 -2 2 +3 5 + +-- !complex_nested_partition_pruning -- +4 +6 -- !complex_base_tag -- 1 [1, null, 3] {"x":10, "null-value":null} 10 old-a 1 \N 2 \N {"x":null} 20 \N \N old-null 3 [] {} \N \N \N \N - diff --git a/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_complex_evolution.groovy b/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_complex_evolution.groovy index e5ad9e7c6ed..3460561577b 100644 --- a/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_complex_evolution.groovy +++ b/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_complex_evolution.groovy @@ -116,6 +116,15 @@ suite("test_iceberg_write_complex_evolution", """ sql """alter table complex_evolution add partition key bucket(8, id) as id_bucket""" sql """alter table complex_evolution add partition key truncate(1, group_key) as group_prefix""" + // Iceberg permits a nested primitive source. Create it through Spark to verify Doris can plan and + // physically partition the following INSERT by the schema-wide nested field id. Invalidate Spark's + // cached table first so its commit requirement sees the partition ids assigned by the Doris DDLs. + spark_iceberg """refresh table demo.${dbName}.complex_evolution""" + spark_iceberg """ + alter table demo.${dbName}.complex_evolution + add partition field bucket(4, payload.nested.count) + """ + sql """refresh table complex_evolution""" sql """ insert into complex_evolution values @@ -135,7 +144,15 @@ suite("test_iceberg_write_complex_evolution", struct(cast(5 as bigint), null, null), null, map('null-value', null) - )) + )), + (6, 'Z3', null, null, null) + """ + + // Route an UPDATE insert image by the nested source and preserve the parent-NULL partition value. + sql """ + update complex_evolution + set group_key = case id when 4 then 'A2' else 'Z4' end + where id in (4, 6) """ // W02-S03: Current schema reads both old and new files without moving old child values. @@ -166,6 +183,13 @@ suite("test_iceberg_write_complex_evolution", group by spec_id order by spec_id """ + order_qt_complex_nested_partition_pruning """ + select id + from complex_evolution + where payload.nested.count = cast(7000000000 as bigint) + or (id = 6 and payload.nested.count is null) + order by id + """ assertSparkMatchesDoris() // W02-S04: A pre-evolution tag binds the old files to their historical complex schema. --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
