This is an automated email from the ASF dual-hosted git repository.
Gabriel39 pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new a17fe99ac4a [fix](iceberg) Backport nested partition and default fixes
to branch-4.1 (#66825)
a17fe99ac4a is described below
commit a17fe99ac4a08bff7869da96eacabc367314ea94
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 0a182fee34e..10be865bc73 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
@@ -278,6 +278,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();
@@ -1146,6 +1147,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();
@@ -1160,6 +1162,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
*
@@ -1247,6 +1257,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 be66a89526c..527442b6363 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
@@ -3427,7 +3427,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]
