This is an automated email from the ASF dual-hosted git repository.
Gabriel39 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 335df1dc014 [fix](iceberg) Harden schema evolution and nested
partition writes (#66529)
335df1dc014 is described below
commit 335df1dc0144ae3f807a5bf092ef800cba1ab92d
Author: Gabriel <[email protected]>
AuthorDate: Fri Aug 7 17:34:39 2026 +0800
[fix](iceberg) Harden schema evolution and nested partition writes (#66529)
## Proposed changes
- make Iceberg compatibility gates conservative and bounded while
pinning column handles to historical schemas
- reset connector metadata across INSERT retries and align V1/V2
defaults, required-field checks, and position-delete row projection
- support primitive partition sources nested in structs, including
nullable-parent propagation and regression coverage
- cache the compact equality-delete field-ID projection by immutable
table snapshot even when the optional full manifest cache is disabled
- resolve nested partition sources through the top-level Nereids slot ID
and fail closed when stable Iceberg IDs are unavailable
## Compatibility-gate trade-offs
- Equality-delete fencing remains conservative across all delete
manifests in the selected snapshot, including partition-pruned scans.
The snapshot-scoped field-ID cache removes repeated manifest walks
without weakening correctness; initial-load failures still fail closed
and remain retryable.
- Requiredness fencing intentionally uses bounded schema-history
inspection rather than an O(snapshot-count) ancestry walk because
snapshot schema IDs are optional. Once a projected requiredness hazard
exists, every non-empty selected snapshot is fenced. This can reduce
rolling-upgrade availability but cannot create a correctness false
negative.
## Testing
- `mvn -pl fe-core,fe-connector/fe-connector-iceberg -am
-Dtest=PhysicalExternalRowLevelMergeSinkTest,IcebergManifestCacheTest,IcebergScanPlanProviderTest
-Dsurefire.failIfNoSpecifiedTests=false test`
- `mvn -pl fe-connector/fe-connector-iceberg -am
-Dtest=IcebergScanPlanProviderTest,IcebergConnectorMetadataTest,IcebergWritePlanProviderTest
-Dsurefire.failIfNoSpecifiedTests=false test`
- `mvn -pl fe-core -am
-Dtest=ConnectorStatementScopeTest,InsertIntoTableCommandTest
-Dsurefire.failIfNoSpecifiedTests=false test`
- `./run-be-ut.sh --run
--filter=SchemaTest.*:VIcebergTableWriterTest.*:IcebergReaderTest.v1_materializes_non_finite_initial_defaults:IcebergV2ReaderTest.PreparesIcebergNonFiniteInitialDefaults:IcebergPositionDeleteSysTableV2ProfileTest.*`
- FE Checkstyle for all affected modules
- clang-format 16 check for all changed C/C++ files
---
.../sink/writer/iceberg/viceberg_table_writer.cpp | 112 ++++++++++--
.../sink/writer/iceberg/viceberg_table_writer.h | 7 +
be/src/format/table/iceberg/schema.cpp | 29 ++-
be/src/format/table/iceberg/schema.h | 4 +
be/src/format/table/iceberg_default_value.h | 31 +++-
.../transformer/iceberg_partition_function.cpp | 75 +++++++-
.../transformer/iceberg_partition_function.h | 4 +
.../iceberg_position_delete_sys_table_reader.cpp | 6 +
be/src/format_v2/table/iceberg_reader.cpp | 15 +-
be/src/format_v2/table/iceberg_reader.h | 3 +
be/test/core/value/merge_partitioner_test.cpp | 91 ++++++++++
.../writer/iceberg/viceberg_table_writer_test.cpp | 63 +++++++
.../format/table/iceberg/iceberg_reader_test.cpp | 17 ++
be/test/format/table/iceberg/schema_test.cpp | 31 ++++
...eberg_position_delete_sys_table_reader_test.cpp | 25 +++
be/test/format_v2/table/iceberg_reader_test.cpp | 33 ++++
.../iceberg/IcebergConnectorMetadata.java | 26 ++-
.../connector/iceberg/IcebergManifestCache.java | 54 +++++-
.../connector/iceberg/IcebergScanPlanProvider.java | 196 +++++++++------------
.../iceberg/IcebergWritePlanProvider.java | 34 +++-
.../iceberg/IcebergWriteSchemaContext.java | 7 +-
.../iceberg/IcebergConnectorMetadataTest.java | 36 ++++
.../iceberg/IcebergManifestCacheTest.java | 55 ++++++
.../iceberg/IcebergScanPlanProviderTest.java | 135 ++++++++++++--
.../iceberg/IcebergWritePlanProviderTest.java | 34 ++++
.../datasource/scan/PluginDrivenScanNode.java | 8 +-
.../glue/translator/PhysicalPlanTranslator.java | 3 +-
.../nereids/properties/DistributionSpecMerge.java | 17 +-
.../commands/insert/InsertIntoTableCommand.java | 3 +
.../PhysicalExternalRowLevelMergeSink.java | 99 ++++++++++-
.../org/apache/doris/planner/DataPartition.java | 10 ++
.../PluginDrivenScanNodeColumnPruningTest.java | 15 ++
.../PhysicalExternalRowLevelMergeSinkTest.java | 85 +++++++++
gensrc/thrift/Partitions.thrift | 2 +
.../write/test_iceberg_write_complex_evolution.out | 12 +-
.../test_iceberg_write_complex_evolution.groovy | 26 ++-
36 files changed, 1229 insertions(+), 174 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 e3f8ed645ed..3519252d7ca 100644
--- a/be/src/exec/sink/writer/iceberg/viceberg_table_writer.cpp
+++ b/be/src/exec/sink/writer/iceberg/viceberg_table_writer.cpp
@@ -17,14 +17,18 @@
#include "exec/sink/writer/iceberg/viceberg_table_writer.h"
+#include <algorithm>
+
#include "common/exception.h"
#include "core/block/block.h"
#include "core/block/column_with_type_and_name.h"
#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"
@@ -121,31 +125,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 ||
@@ -358,9 +437,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 2cb83f73ed0..20d9562c35e 100644
--- a/be/src/exec/sink/writer/iceberg/viceberg_table_writer.h
+++ b/be/src/exec/sink/writer/iceberg/viceberg_table_writer.h
@@ -83,6 +83,7 @@ public:
private:
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.
@@ -93,10 +94,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:
@@ -104,6 +107,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;
@@ -117,10 +121,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_path(const doris::iceberg::StructLike& data);
std::string _escape(const std::string& path);
diff --git a/be/src/format/table/iceberg/schema.cpp
b/be/src/format/table/iceberg/schema.cpp
index 76dac166327..a05456627bf 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 {
const std::string Schema::ALL_COLUMNS = "*";
@@ -24,10 +26,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)) {}
@@ -48,4 +66,9 @@ 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;
+}
+
} // namespace doris::iceberg
diff --git a/be/src/format/table/iceberg/schema.h
b/be/src/format/table/iceberg/schema.h
index 0273a4450da..5781b86b3fa 100644
--- a/be/src/format/table/iceberg/schema.h
+++ b/be/src/format/table/iceberg/schema.h
@@ -26,6 +26,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);
@@ -40,6 +41,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;
@@ -48,6 +51,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;
};
} // namespace doris::iceberg
diff --git a/be/src/format/table/iceberg_default_value.h
b/be/src/format/table/iceberg_default_value.h
index ae75924336f..5fe1834e3bc 100644
--- a/be/src/format/table/iceberg_default_value.h
+++ b/be/src/format/table/iceberg_default_value.h
@@ -24,6 +24,7 @@
#include <cstddef>
#include <deque>
+#include <limits>
#include <string>
#include <string_view>
#include <unordered_map>
@@ -46,6 +47,28 @@ namespace doris::iceberg {
namespace detail {
+inline 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;
+}
+
inline const schema::external::TField* get_field_ptr(const
schema::external::TFieldPtr& field_ptr) {
if (!field_ptr.__isset.field_ptr || field_ptr.field_ptr == nullptr) {
return nullptr;
@@ -135,7 +158,7 @@ inline std::string json_scalar_text(const rapidjson::Value&
value) {
return {buffer.GetString(), buffer.GetSize()};
}
-inline void normalize_timestamp_for_doris(PrimitiveType primitive_type,
std::string* value) {
+inline void normalize_timestamp_for_doris(doris::PrimitiveType primitive_type,
std::string* value) {
if (primitive_type != TYPE_DATETIME && primitive_type != TYPE_DATETIMEV2 &&
primitive_type != TYPE_TIMESTAMPTZ) {
return;
@@ -344,6 +367,9 @@ inline Status build_json_scalar_default(const
schema::external::TField& field,
return Status::OK();
}
normalize_timestamp_for_doris(primitive_type, &serialized_value);
+ if (parse_non_finite_default(primitive_type, serialized_value, result)) {
+ return Status::OK();
+ }
RETURN_IF_ERROR(value_type->get_serde()->from_fe_string(serialized_value,
*result));
return Status::OK();
}
@@ -422,6 +448,9 @@ inline Status build_initial_default_field(const
schema::external::TField& field,
return Status::OK();
}
+ if (parse_non_finite_default(primitive_type, field.initial_default_value,
result)) {
+ return Status::OK();
+ }
RETURN_IF_ERROR(value_type->get_serde()->from_fe_string(field.initial_default_value,
*result));
return Status::OK();
}
diff --git a/be/src/format/transformer/iceberg_partition_function.cpp
b/be/src/format/transformer/iceberg_partition_function.cpp
index 3030fed3d60..4d134b062f3 100644
--- a/be/src/format/transformer/iceberg_partition_function.cpp
+++ b/be/src/format/transformer/iceberg_partition_function.cpp
@@ -24,6 +24,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,21 @@ 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");
+ }
+ 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) {
LOG(WARNING) << "Merge partitioning fallback to RR: " <<
e.what();
_fallback_to_random = true;
@@ -173,6 +189,7 @@ 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));
}
}
@@ -180,6 +197,53 @@ Status IcebergInsertPartitionFunction::clone(RuntimeState*
state,
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();
@@ -202,8 +266,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 d2c1a25724b..0ab36c91a0e 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"
@@ -52,8 +53,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..94f489747e9 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,9 @@ 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.
+
RETURN_IF_ERROR(prepare_iceberg_initial_default_exprs(&columns->back()));
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 097e6bab111..37ed4c00e28 100644
--- a/be/src/format_v2/table/iceberg_reader.cpp
+++ b/be/src/format_v2/table/iceberg_reader.cpp
@@ -47,6 +47,7 @@
#include "exprs/vliteral.h"
#include "exprs/vslot_ref.h"
#include "format/table/deletion_vector_reader.h"
+#include "format/table/iceberg_default_value.h"
#include "format_v2/expr/cast.h"
#include "format_v2/expr/equality_delete_predicate.h"
#include "format_v2/orc/orc_reader.h"
@@ -357,6 +358,10 @@ static Status build_v2_json_scalar_default(const
format::ColumnDefinition& field
return Status::OK();
}
normalize_iceberg_json_timestamp(primitive_type, &serialized_value);
+ if (doris::iceberg::detail::parse_non_finite_default(primitive_type,
serialized_value,
+ result)) {
+ return Status::OK();
+ }
RETURN_IF_ERROR(value_type->get_serde()->from_fe_string(serialized_value,
*result));
return Status::OK();
}
@@ -432,6 +437,10 @@ static Status build_v2_initial_default_field(const
format::ColumnDefinition& fie
return Status::OK();
}
+ if (doris::iceberg::detail::parse_non_finite_default(primitive_type,
+
*field.initial_default_value, result)) {
+ return Status::OK();
+ }
RETURN_IF_ERROR(value_type->get_serde()->from_fe_string(*field.initial_default_value,
*result));
return Status::OK();
}
@@ -451,7 +460,7 @@ static Status build_initial_default_literal(const
format::ColumnDefinition& tabl
return Status::OK();
}
-static Status build_initial_default_exprs(format::ColumnDefinition* column) {
+Status prepare_iceberg_initial_default_exprs(format::ColumnDefinition* column)
{
DORIS_CHECK(column != nullptr);
if (column->initial_default_value.has_value()) {
VExprSPtr literal;
@@ -459,7 +468,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));
+ RETURN_IF_ERROR(prepare_iceberg_initial_default_exprs(&child));
}
return Status::OK();
}
@@ -771,7 +780,7 @@ Status IcebergTableReader::annotate_projected_column(const
TFileScanSlotInfo& sl
}
auto& schema_column = *context->schema_column;
- RETURN_IF_ERROR(build_initial_default_exprs(&schema_column));
+ RETURN_IF_ERROR(prepare_iceberg_initial_default_exprs(&schema_column));
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 2768e4cd3e8..2b29d26ee3a 100644
--- a/be/src/format_v2/table/iceberg_reader.h
+++ b/be/src/format_v2/table/iceberg_reader.h
@@ -42,6 +42,8 @@ struct FileSystemProperties;
namespace doris::format::iceberg {
+Status prepare_iceberg_initial_default_exprs(format::ColumnDefinition* column);
+
// 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
@@ -75,6 +77,7 @@ public:
protected:
void configure_mapper_options(format::TableColumnMapperOptions* options)
const override {
options->enable_row_lineage_virtual_columns = true;
+ options->reject_missing_required_field =
supports_iceberg_scan_semantics_v2(_scan_params);
options->allow_idless_complex_wrapper_projection =
supports_iceberg_scan_semantics_v1(_scan_params) && _format ==
FileFormat::PARQUET;
}
diff --git a/be/test/core/value/merge_partitioner_test.cpp
b/be/test/core/value/merge_partitioner_test.cpp
index f2682cf657a..43c494832d0 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();
@@ -204,11 +246,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;
@@ -219,6 +263,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) {
@@ -317,6 +362,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("identity");
+ 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 3800ef4f86e..06066de0fd7 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
@@ -20,6 +20,15 @@
#include <gtest/gtest.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 "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"
@@ -50,4 +59,58 @@ TEST(VIcebergTableWriterTest, RejectMissingPartitionSource) {
}
}
+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, nullptr, nullptr);
+ 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);
+}
+
} // namespace doris
diff --git a/be/test/format/table/iceberg/iceberg_reader_test.cpp
b/be/test/format/table/iceberg/iceberg_reader_test.cpp
index a212de6adfc..ef1c469ca23 100644
--- a/be/test/format/table/iceberg/iceberg_reader_test.cpp
+++ b/be/test/format/table/iceberg/iceberg_reader_test.cpp
@@ -28,6 +28,7 @@
#include <parquet/arrow/writer.h>
#include <array>
+#include <cmath>
#include <filesystem>
#include <fstream>
#include <iostream>
@@ -1796,6 +1797,22 @@ TEST_F(IcebergReaderTest,
initial_default_rejects_invalid_nullability) {
EXPECT_TRUE(value.is_null());
}
+TEST_F(IcebergReaderTest, v1_materializes_non_finite_initial_defaults) {
+ schema::external::TField field;
+ field.__set_name("value");
+ field.__set_id(1);
+ field.__set_is_optional(false);
+ field.__set_initial_default_value("NaN");
+
+ ColumnPtr column;
+ ASSERT_TRUE(iceberg::create_initial_default_column(field,
std::make_shared<DataTypeFloat32>(),
+ &column)
+ .ok());
+ Field value;
+ column->get(0, value);
+ EXPECT_TRUE(std::isnan(value.get<TYPE_FLOAT>()));
+}
+
// GTest assertion macros inflate clang-tidy's cognitive-complexity score.
// NOLINTNEXTLINE(readability-function-cognitive-complexity)
TEST_F(IcebergReaderTest,
v1_reuses_prepared_complex_initial_default_across_block_types) {
diff --git a/be/test/format/table/iceberg/schema_test.cpp
b/be/test/format/table/iceberg/schema_test.cpp
index bccf0f1f341..91c9ae778b3 100644
--- a/be/test/format/table/iceberg/schema_test.cpp
+++ b/be/test/format/table/iceberg/schema_test.cpp
@@ -19,6 +19,10 @@
#include <gtest/gtest.h>
+#include <cmath>
+
+#include "format/table/iceberg_default_value.h"
+
namespace doris {
namespace iceberg {
@@ -66,5 +70,32 @@ 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);
+}
+
+TEST(SchemaTest, ParsesIcebergNonFiniteDefaults) {
+ Field value;
+ EXPECT_TRUE(detail::parse_non_finite_default(TYPE_FLOAT, "NaN", &value));
+ EXPECT_TRUE(std::isnan(value.get<TYPE_FLOAT>()));
+ EXPECT_TRUE(detail::parse_non_finite_default(TYPE_DOUBLE, "Infinity",
&value));
+ EXPECT_TRUE(std::isinf(value.get<TYPE_DOUBLE>()));
+ EXPECT_GT(value.get<TYPE_DOUBLE>(), 0);
+ EXPECT_TRUE(detail::parse_non_finite_default(TYPE_DOUBLE, "-Infinity",
&value));
+ EXPECT_LT(value.get<TYPE_DOUBLE>(), 0);
+}
+
} // 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..6db534fa06d 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,8 @@
#include <gtest/gtest.h>
+#include "core/data_type/data_type_number.h"
+#include "core/data_type/data_type_struct.h"
#include "runtime/runtime_profile.h"
#include "runtime/runtime_state.h"
@@ -59,5 +61,28 @@ TEST(IcebergPositionDeleteSysTableV2ProfileTest,
UsesDistinctProfileForNestedPos
reader._position_reader_profile);
}
+TEST(IcebergPositionDeleteSysTableV2ProfileTest,
PreparesNestedRowInitialDefaults) {
+ IcebergPositionDeleteSysTableV2Reader reader;
+ 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 806cc7e729a..5c78f6b19ef 100644
--- a/be/test/format_v2/table/iceberg_reader_test.cpp
+++ b/be/test/format_v2/table/iceberg_reader_test.cpp
@@ -27,6 +27,7 @@
#include <algorithm>
#include <array>
#include <chrono>
+#include <cmath>
#include <cstring>
#include <filesystem>
#include <fstream>
@@ -1475,6 +1476,38 @@ TEST(IcebergV2ReaderTest,
AnnotateBuildsTypedNestedInitialDefault) {
EXPECT_EQ(value.get<TYPE_INT>(), 7);
}
+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, AnnotateBuildsComplexInitialDefaults) {
const auto required_int_type = std::make_shared<DataTypeInt32>();
const auto optional_string_type =
make_nullable(std::make_shared<DataTypeString>());
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadata.java
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadata.java
index a75da020914..0e6425a8f07 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadata.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadata.java
@@ -709,7 +709,31 @@ public class IcebergConnectorMetadata implements
ConnectorMetadata {
// metadata-table columns (t$snapshots -> committed_at/...) so the
generic scan node can look up
// its pruned sys-table slots by name; a data handle resolves the base
table's columns.
Table table = iceHandle.isSystemTable() ? loadSysTable(session,
iceHandle) : loadTable(session, iceHandle);
- List<Types.NestedField> fields = table.schema().columns();
+ return buildColumnHandles(table.schema());
+ }
+
+ @Override
+ public Map<String, ConnectorColumnHandle> getColumnHandles(
+ ConnectorSession session, ConnectorTableHandle handle,
+ ConnectorMvccSnapshot snapshot) {
+ IcebergTableHandle iceHandle = (IcebergTableHandle) handle;
+ if (iceHandle.isSystemTable() || snapshot == null ||
snapshot.getSchemaId() < 0) {
+ return getColumnHandles(session, handle);
+ }
+ Table table = loadTable(session, iceHandle);
+ Schema schema = table.currentSnapshot() == null
+ ? table.schema() : table.schemas().get((int)
snapshot.getSchemaId());
+ // Keep the handle-schema fallback identical to getTableSchema so
slots and handles cannot diverge.
+ return buildColumnHandles(schema == null ? table.schema() : schema);
+ }
+
+ @Override
+ public boolean supportsColumnHandleSnapshotPin(ConnectorSession session) {
+ return true;
+ }
+
+ private static Map<String, ConnectorColumnHandle>
buildColumnHandles(Schema schema) {
+ List<Types.NestedField> fields = schema.columns();
Map<String, ConnectorColumnHandle> handles = new
LinkedHashMap<>(fields.size());
for (Types.NestedField field : fields) {
String name = field.name();
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergManifestCache.java
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergManifestCache.java
index f3c46732fb2..7790d43062b 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergManifestCache.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergManifestCache.java
@@ -30,12 +30,17 @@ import org.apache.iceberg.Table;
import java.io.IOException;
import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.TimeUnit;
import java.util.function.LongSupplier;
+import java.util.function.Supplier;
/**
* Per-catalog cache of an iceberg manifest's parsed files, keyed by {@link
IcebergManifestEntryKey}
@@ -46,8 +51,9 @@ import java.util.function.LongSupplier;
* <p>Consumed by {@link IcebergScanPlanProvider}'s manifest-level planning
path (gated by
* {@code meta.cache.iceberg.manifest.enable}, default off — the default scan
path is the iceberg SDK
* {@code planFiles()}). The external enable-gate lives in the scan provider
(which decides whether to take the
- * manifest-planning path at all); this cache is unconditionally on when
consulted. Within one catalog the same
- * manifest file is parsed once and shared across queries (and across tables
that reference it).
+ * full manifest-planning path); the compact equality-delete field-id
projection is always reused per immutable
+ * snapshot. Within one catalog the same manifest file is parsed once and
shared across queries (and across
+ * tables that reference it).
*
* <p><b>No TTL; capacity-bounded; cleared on REFRESH CATALOG.</b> This
mirrors the legacy entry's
* {@code contextualOnly(CacheSpec.of(false, CACHE_NO_TTL, 100_000))} default
spec: a manifest's content is
@@ -68,6 +74,35 @@ final class IcebergManifestCache {
private static final long DEFAULT_STATS_TTL_SECONDS = 300L;
private final MetaCacheEntry<IcebergManifestEntryKey, ManifestCacheValue>
entry;
+ private final MetaCacheEntry<SnapshotKey, Set<Integer>>
equalityDeleteFieldIds;
+
+ /** Immutable snapshot key for the compact equality-delete field-id
projection. */
+ private static final class SnapshotKey {
+ private final String tableLocation;
+ private final long snapshotId;
+
+ private SnapshotKey(String tableLocation, long snapshotId) {
+ this.tableLocation = tableLocation;
+ this.snapshotId = snapshotId;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (!(o instanceof SnapshotKey)) {
+ return false;
+ }
+ SnapshotKey that = (SnapshotKey) o;
+ return snapshotId == that.snapshotId &&
Objects.equals(tableLocation, that.tableLocation);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(tableLocation, snapshotId);
+ }
+ }
// Per-scan manifest-cache access tally, keyed by the statement's stable
queryId
// (ConnectorSession.getQueryId()), so VERBOSE EXPLAIN can report THIS
scan's hits/misses/failures (the
@@ -106,10 +141,24 @@ final class IcebergManifestCache {
CacheSpec spec = CacheSpec.of(true, CacheSpec.CACHE_NO_TTL,
Math.max(1, maxSize));
this.entry = new MetaCacheEntry<>("iceberg-manifest", null, spec,
ForkJoinPool.commonPool(), false, true, 0L, true);
+ this.equalityDeleteFieldIds = new
MetaCacheEntry<>("iceberg-equality-delete-field-ids", null, spec,
+ ForkJoinPool.commonPool(), false, true, 0L, true);
this.statsTtlNanos = TimeUnit.SECONDS.toNanos(Math.max(1L,
statsTtlSeconds));
this.nanoClock = nanoClock;
}
+ /**
+ * Returns the equality-delete field ids for one immutable snapshot.
Unlike the optional full manifest-file
+ * cache, this compact projection is always reused so scan properties do
not re-read every delete manifest on
+ * every query. Loader failures are deliberately not cached, preserving
retry after transient storage errors.
+ */
+ Set<Integer> getOrLoadEqualityDeleteFieldIds(
+ String tableLocation, long snapshotId, Supplier<Set<Integer>>
loader) {
+ SnapshotKey key = new SnapshotKey(tableLocation, snapshotId);
+ return equalityDeleteFieldIds.get(key,
+ ignored -> Collections.unmodifiableSet(new
HashSet<>(loader.get())));
+ }
+
/**
* Returns the parsed files for {@code manifest}, loading (and reading
from storage) only on a miss. The
* loader runs OUTSIDE Caffeine's compute lock (manual miss-load;
single-flight per key), so a same-key
@@ -223,6 +272,7 @@ final class IcebergManifestCache {
*/
void invalidateAll() {
entry.invalidateAll();
+ equalityDeleteFieldIds.invalidateAll();
statsByQuery.clear();
}
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java
index a38af307a3d..09f8fa44615 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java
@@ -55,6 +55,8 @@ import org.apache.iceberg.FileScanTask;
import org.apache.iceberg.HasTableOperations;
import org.apache.iceberg.ManifestContent;
import org.apache.iceberg.ManifestFile;
+import org.apache.iceberg.ManifestFiles;
+import org.apache.iceberg.ManifestReader;
import org.apache.iceberg.MetadataColumns;
import org.apache.iceberg.MetadataTableType;
import org.apache.iceberg.MetadataTableUtils;
@@ -66,7 +68,6 @@ import org.apache.iceberg.ScanTask;
import org.apache.iceberg.Schema;
import org.apache.iceberg.SchemaParser;
import org.apache.iceberg.Snapshot;
-import org.apache.iceberg.SnapshotSummary;
import org.apache.iceberg.SplittableScanTask;
import org.apache.iceberg.Table;
import org.apache.iceberg.TableOperations;
@@ -100,10 +101,8 @@ import java.io.Closeable;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.time.ZoneId;
-import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Collections;
-import java.util.Deque;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
@@ -230,8 +229,9 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
// null in offline unit tests via the 2-arg ctor, in which case
resolveTable resolves directly.
private final ConnectorContext context;
// T08: per-catalog manifest cache, owned by the long-lived
IcebergConnector and injected via getScanPlanProvider.
- // Nullable — null via the 2-/3-arg ctors (offline tests, default-disabled
gate); when null the gate is
- // forced off and planScan uses the SDK splitFiles path.
+ // Its compact equality-delete field-id projection is used regardless of
the full-cache feature gate. Nullable
+ // via the 2-/3-arg ctors (offline tests); when null the projection is
loaded directly, the full-cache gate is
+ // forced off, and planScan uses the SDK splitFiles path.
private final IcebergManifestCache manifestCache;
// PERF-01: cross-query RAW-table cache shared with the metadata layer,
owned by the long-lived
// IcebergConnector and injected via getScanPlanProvider. Nullable — null
via the offline-test ctors and
@@ -1601,11 +1601,13 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
boolean systemTable = iceHandle.isSystemTable();
Schema scanSchema = null;
TableScan exactScan = null;
+ Set<Integer> applicableEqualityDeleteFieldIds = Collections.emptySet();
boolean hasApplicableEqualityDeletes = false;
if (!systemTable) {
scanSchema = pinnedSchema(table, iceHandle);
exactScan = buildScan(table, iceHandle, filter, session);
- hasApplicableEqualityDeletes =
hasApplicableEqualityDeletes(exactScan);
+ applicableEqualityDeleteFieldIds =
cachedApplicableEqualityDeleteFieldIds(table, exactScan);
+ hasApplicableEqualityDeletes =
!applicableEqualityDeleteFieldIds.isEmpty();
Optional<Map<Integer, List<String>>> nameMapping =
IcebergSchemaUtils.extractNameMapping(table);
if (requiresCurrentScanSemantics(
table, exactScan, scanSchema, columns,
hasApplicableEqualityDeletes, nameMapping)) {
@@ -1701,7 +1703,7 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
// every branch so the default and current field type match BE's
read.
if (hasApplicableEqualityDeletes) {
List<NestedField> equalityFields =
schemaForPotentialEqualityDeletes(
- table, exactScan, scanSchema);
+ table, scanSchema, applicableEqualityDeleteFieldIds);
dict = IcebergSchemaUtils.encodeEqualitySchemaEvolutionProp(
table, equalityFields, appendRowLineage,
enableVarbinary, enableTimestampTz);
@@ -1725,6 +1727,14 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
}
props.put(SCHEMA_EVOLUTION_PROP, dict);
} else if (isPositionDeletesSysTable(iceHandle)) {
+ // The native position-delete row reader depends on the current
nested-default and requiredness
+ // semantics, so rolling upgrades must not route this system-table
scan to an older backend.
+ // Metadata-only projections never materialize `row`, so fencing
those scans needlessly reduces
+ // rolling-upgrade availability without preserving a reader
invariant.
+ if (requestsColumn(columns, "row")) {
+
props.put(ScanNodePropertyKeys.REQUIRED_CURRENT_BACKEND_SEMANTICS,
+ "Current Iceberg position delete semantics");
+ }
// [D-065] narrowed: $position_deletes is the ONE system table BE
reads with a NATIVE reader, so
// the "schema rides inside the serialized FileScanTask" rationale
above does not hold for it — no
// FileScanTask is serialized on this path. Both native readers
resolve the `row` column through
@@ -1801,47 +1811,77 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
return names;
}
+ private static boolean requestsColumn(List<ConnectorColumnHandle> columns,
String requestedName) {
+ if (columns == null || columns.isEmpty()) {
+ return true;
+ }
+ for (ConnectorColumnHandle column : columns) {
+ if (requestedName.equalsIgnoreCase(((IcebergColumnHandle)
column).getName())) {
+ return true;
+ }
+ }
+ return false;
+ }
+
@VisibleForTesting
- static boolean hasApplicableEqualityDeletes(TableScan scan) {
+ static Set<Integer> applicableEqualityDeleteFieldIds(Table table,
TableScan scan) {
Snapshot snapshot = scan.snapshot();
- if (snapshot == null
- || "0".equals(snapshot.summary().get(TOTAL_EQUALITY_DELETES)))
{
- return false;
+ if (snapshot == null ||
"0".equals(snapshot.summary().get(TOTAL_EQUALITY_DELETES))) {
+ return Collections.emptySet();
}
- // planFiles binds delete files to the exact filtered data-file tasks
after partition and sequence
- // pruning. A snapshot summary of zero returns above without planning;
a positive or missing summary
- // needs this exact proof. Iterate whole-file tasks lazily and stop at
the first equality delete: this
- // keeps memory O(1), does not create or retain byte-split tasks, and
avoids snapshot-wide delete
- // counters forcing new-BE-only semantics when no dispatched task can
consume an equality delete.
- try (CloseableIterable<FileScanTask> tasks = scan.planFiles()) {
- for (FileScanTask task : tasks) {
- for (DeleteFile delete : task.deletes()) {
- if (delete.content() == FileContent.EQUALITY_DELETES) {
- return true;
+ Set<Integer> fieldIds = new HashSet<>();
+ for (ManifestFile manifest : snapshot.deleteManifests(table.io())) {
+ if (!manifest.hasAddedFiles() && !manifest.hasExistingFiles()) {
+ continue;
+ }
+ try (ManifestReader<DeleteFile> reader =
ManifestFiles.readDeleteManifest(
+ manifest, table.io(), table.specs())) {
+ for (DeleteFile deleteFile : reader) {
+ if (deleteFile.content() == FileContent.EQUALITY_DELETES) {
+ fieldIds.addAll(deleteFile.equalityFieldIds());
}
}
+ } catch (IOException e) {
+ throw new DorisConnectorException(
+ "Failed to read iceberg delete manifest " +
manifest.path() + ": " + e.getMessage(), e);
}
- } catch (IOException e) {
- throw new DorisConnectorException(
- "Failed to inspect applicable Iceberg equality deletes: "
+ e.getMessage(), e);
}
- return false;
+ return fieldIds;
+ }
+
+ private Set<Integer> cachedApplicableEqualityDeleteFieldIds(Table table,
TableScan scan) {
+ Snapshot snapshot = scan.snapshot();
+ if (snapshot == null ||
"0".equals(snapshot.summary().get(TOTAL_EQUALITY_DELETES))) {
+ return Collections.emptySet();
+ }
+ if (manifestCache == null) {
+ return applicableEqualityDeleteFieldIds(table, scan);
+ }
+ // Snapshot contents are immutable, so this compact projection can be
shared even when the optional
+ // full manifest cache is disabled; a loader failure must still escape
and remain retryable.
+ return manifestCache.getOrLoadEqualityDeleteFieldIds(
+ table.location(), snapshot.snapshotId(), () ->
applicableEqualityDeleteFieldIds(table, scan));
}
/**
- * Build a schema carrier that can resolve any equality key reachable
before the selected schema without
- * enumerating data files, manifests, or byte-split tasks. Its retained
state is bounded by table schema
- * history rather than scan cardinality. At execution time BE looks fields
up by the exact IDs on each
- * {@link FileScanTask#deletes()}; unrelated carrier fields never
participate in delete matching.
+ * Build a schema carrier that can resolve the field IDs referenced by
live equality delete files. Reading
+ * delete manifests does not enumerate data files or byte-split tasks, and
retained state is bounded by the
+ * number of equality keys rather than the table's entire schema history.
*
- * <p>The selected snapshot lineage wins when a field was renamed. The
metadata schema list, in its actual
- * chronology up to the selected schema (schema IDs are identifiers, not a
sequence), fills schema-only
- * changes and expired ancestors. Current fields remain first, so a
dropped/re-added name still resolves the
- * projected current field by name while a historical equality key
resolves by its stable field ID.</p>
+ * <p>The metadata schema list is searched in reverse chronology up to the
selected schema (schema IDs are
+ * identifiers, not a sequence), so the latest definition of each stable
field ID wins. Current fields remain
+ * first, allowing a dropped/re-added name to resolve the projected field
by name while a live historical
+ * equality key resolves by its stable field ID.</p>
*/
@VisibleForTesting
static List<NestedField> schemaForPotentialEqualityDeletes(
Table table, TableScan scan, Schema scanSchema) {
+ return schemaForPotentialEqualityDeletes(
+ table, scanSchema, applicableEqualityDeleteFieldIds(table,
scan));
+ }
+
+ private static List<NestedField> schemaForPotentialEqualityDeletes(
+ Table table, Schema scanSchema, Set<Integer> equalityFieldIds) {
List<Schema> metadataSchemas = metadataSchemaHistory(table);
int selectedSchemaIndex = -1;
for (int index = 0; index < metadataSchemas.size(); index++) {
@@ -1851,45 +1891,19 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
}
int lastRelevantIndex = selectedSchemaIndex >= 0
? selectedSchemaIndex : metadataSchemas.size() - 1;
- Set<Integer> missing = new HashSet<>();
- for (int index = 0; index <= lastRelevantIndex; index++) {
- Schema schema = metadataSchemas.get(index);
- for (NestedField field :
TypeUtil.indexById(schema.asStruct()).values()) {
- if (field.type().isPrimitiveType()) {
- missing.add(field.fieldId());
- }
- }
- }
+ Set<Integer> missing = new HashSet<>(equalityFieldIds);
missing.removeAll(TypeUtil.indexById(scanSchema.asStruct()).keySet());
if (missing.isEmpty()) {
return scanSchema.columns();
}
List<NestedField> fields = new ArrayList<>(scanSchema.columns());
- Map<Integer, Schema> schemasById = table.schemas();
- Snapshot snapshot = scan.snapshot();
- while (snapshot != null && !missing.isEmpty()) {
- Integer schemaId = snapshot.schemaId();
- if (schemaId != null) {
- Schema historicalSchema = schemasById.get(schemaId);
- if (historicalSchema == null) {
- throw new IllegalStateException(
- "Iceberg snapshot schema " + schemaId + " is
absent from table metadata");
- }
- addHistoricalEqualityFields(fields, missing, historicalSchema);
- }
- if (missing.isEmpty()) {
- break;
- }
- Long parentId = snapshot.parentId();
- snapshot = parentId == null ? null : table.snapshot(parentId);
- }
for (int index = lastRelevantIndex; index >= 0 && !missing.isEmpty();
index--) {
addHistoricalEqualityFields(fields, missing,
metadataSchemas.get(index));
}
if (!missing.isEmpty()) {
throw new IllegalStateException(
- "Iceberg historical primitive fields are absent from
schema history: " + missing);
+ "Iceberg equality-delete fields are absent from schema
history: " + missing);
}
return fields;
}
@@ -2081,62 +2095,22 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
static boolean selectedHistoryRequiresMissingRequiredFieldRejection(
Table table, Schema scanSchema, Set<Integer> projectedFieldIds,
Snapshot selectedSnapshot) {
- Map<Integer, Schema> schemasById = table.schemas();
- Set<Integer> relevantSchemaIds =
schemaIdsRequiringMissingRequiredFieldRejection(
- scanSchema, projectedFieldIds, schemasById.values());
- if (relevantSchemaIds.isEmpty()) {
+ if (!schemaHistoryRequiresMissingRequiredFieldRejection(
+ scanSchema, projectedFieldIds, table.schemas().values())) {
return false;
}
- Deque<Snapshot> snapshots = new ArrayDeque<>();
- if (selectedSnapshot != null) {
- snapshots.add(selectedSnapshot);
- }
- Set<Long> visitedSnapshotIds = new HashSet<>();
- while (!snapshots.isEmpty()) {
- Snapshot snapshot = snapshots.removeFirst();
- if (!visitedSnapshotIds.add(snapshot.snapshotId())) {
- continue;
- }
- Integer schemaId = snapshot.schemaId();
- if (schemaId != null) {
- Schema historical = schemasById.get(schemaId);
- if (historical == null) {
- throw new IllegalStateException(
- "Iceberg snapshot schema " + schemaId + " is
absent from table metadata");
- }
- if (relevantSchemaIds.contains(schemaId)) {
- return true;
- }
- }
- Long parentId = snapshot.parentId();
- if (parentId != null) {
- Snapshot parent = table.snapshot(parentId);
- if (parent == null) {
- return true;
- }
- snapshots.addLast(parent);
- }
- String sourceSnapshotId =
-
snapshot.summary().get(SnapshotSummary.SOURCE_SNAPSHOT_ID_PROP);
- if (sourceSnapshotId != null) {
- Snapshot source =
table.snapshot(Long.parseLong(sourceSnapshotId));
- if (source == null) {
- return true;
- }
- snapshots.addLast(source);
- }
- }
- return false;
+ // Snapshot schema IDs are optional and proving ancestry is O(snapshot
count). Once schema history
+ // exposes a requiredness hazard, conservatively fence every non-empty
selected snapshot.
+ return selectedSnapshot != null;
}
- private static Set<Integer>
schemaIdsRequiringMissingRequiredFieldRejection(
+ private static boolean schemaHistoryRequiresMissingRequiredFieldRejection(
Schema scanSchema, Set<Integer> projectedFieldIds,
Iterable<Schema> historicalSchemas) {
Map<Integer, NestedField> currentFields =
TypeUtil.indexById(scanSchema.asStruct());
Map<Integer, Integer> parentById =
TypeUtil.indexParents(scanSchema.asStruct());
Set<Integer> collectionWrapperIds = new HashSet<>();
collectCollectionWrapperFieldIds(scanSchema.asStruct(),
collectionWrapperIds);
- Set<Integer> schemaIds = new HashSet<>();
for (Schema historicalSchema : historicalSchemas) {
Map<Integer, NestedField> historicalFields =
TypeUtil.indexById(historicalSchema.asStruct());
@@ -2149,8 +2123,7 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
NestedField historicalField = historicalFields.get(fieldId);
if (historicalField != null) {
if (historicalField.isOptional()) {
- schemaIds.add(historicalSchema.schemaId());
- break;
+ return true;
}
continue;
}
@@ -2164,12 +2137,11 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
if (!collectionWrapperIds.contains(highestMissing.fieldId())
&& highestMissing.isRequired()
&& highestMissing.initialDefault() == null) {
- schemaIds.add(historicalSchema.schemaId());
- break;
+ return true;
}
}
}
- return schemaIds;
+ return false;
}
private static void collectCollectionWrapperFieldIds(Type type,
Set<Integer> result) {
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergWritePlanProvider.java
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergWritePlanProvider.java
index 02bb47ac08c..16b6ea50c19 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergWritePlanProvider.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergWritePlanProvider.java
@@ -61,6 +61,7 @@ import org.apache.iceberg.SortField;
import org.apache.iceberg.SortOrder;
import org.apache.iceberg.Table;
import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.types.Type;
import org.apache.iceberg.types.Types.NestedField;
import org.apache.iceberg.util.LocationUtil;
@@ -534,18 +535,39 @@ public class IcebergWritePlanProvider implements
ConnectorWritePlanProvider {
? active.get().getSchema() : table.schema();
List<ConnectorWritePartitionField> fields = new ArrayList<>();
for (PartitionField field : spec.fields()) {
- // sourceColumnName mirrors the legacy
schema.findField(field.sourceId()).name() lookup the engine
- // used to map a partition field back to a bound output expr id.
transform/param mirror
- // field.transform().toString() + parseTransformParam (kept
connector-side so fe-core never parses).
- NestedField sourceField = schema.findField(field.sourceId());
- String sourceColumnName = sourceField == null ? null :
sourceField.name();
+ // Bind a nested source to its top-level slot. fe-core recovers
child indexes from the stable Iceberg
+ // field ids already carried by the Doris Column tree, so the
public connector SPI stays unchanged.
+ String sourceColumnName = findPartitionSourceColumnName(schema,
field.sourceId());
String transform = field.transform().toString();
fields.add(new ConnectorWritePartitionField(
- transform, parseTransformParam(transform),
sourceColumnName, field.name(), field.sourceId()));
+ transform, parseTransformParam(transform),
sourceColumnName,
+ field.name(), field.sourceId()));
}
return new ConnectorWritePartitionSpec(spec.specId(), fields);
}
+ private static String findPartitionSourceColumnName(Schema schema, int
sourceId) {
+ List<NestedField> columns = schema.columns();
+ for (NestedField column : columns) {
+ if (column.fieldId() == sourceId ||
containsStructField(column.type(), sourceId)) {
+ return column.name();
+ }
+ }
+ return null;
+ }
+
+ private static boolean containsStructField(Type type, int sourceId) {
+ if (!type.isStructType()) {
+ return false;
+ }
+ for (NestedField field : type.asStructType().fields()) {
+ if (field.fieldId() == sourceId ||
containsStructField(field.type(), sourceId)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
@Override
public List<ConnectorColumn> getSyntheticWriteColumns(ConnectorSession
session,
ConnectorTableHandle tableHandle) {
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergWriteSchemaContext.java
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergWriteSchemaContext.java
index ae16a1ee4fb..b04c2fa4216 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergWriteSchemaContext.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergWriteSchemaContext.java
@@ -202,13 +202,12 @@ 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 field as a partition source;
field IDs are schema-wide.
+ if (schema.findField(field.sourceId()) == null) {
throw new DorisConnectorException("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-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadataTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadataTest.java
index 5e98e90fe0a..61f275f3593 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadataTest.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadataTest.java
@@ -25,13 +25,17 @@ import
org.apache.doris.connector.spi.DorisConnectorException;
import org.apache.doris.connector.spi.handle.ConnectorColumnHandle;
import org.apache.doris.connector.spi.handle.ConnectorTableHandle;
import org.apache.doris.connector.spi.handle.WriteOperation;
+import org.apache.doris.connector.spi.mvcc.ConnectorMvccSnapshot;
+import org.apache.iceberg.DataFiles;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.RowLevelOperationMode;
import org.apache.iceberg.Schema;
import org.apache.iceberg.SortOrder;
import org.apache.iceberg.TableProperties;
import org.apache.iceberg.catalog.Namespace;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.inmemory.InMemoryCatalog;
import org.apache.iceberg.types.Types;
import org.apache.iceberg.view.ImmutableSQLViewRepresentation;
import org.apache.iceberg.view.ImmutableViewVersion;
@@ -1313,6 +1317,38 @@ public class IcebergConnectorMetadataTest {
"getColumnHandles must load the table via the seam using the
handle coordinates");
}
+ @Test
+ public void getColumnHandlesUsesPinnedHistoricalSchema() {
+ RecordingIcebergCatalogOps ops = new RecordingIcebergCatalogOps();
+ Schema oldSchema = new Schema(
+ Types.NestedField.required(7, "old_name",
Types.IntegerType.get()),
+ Types.NestedField.optional(9, "survivor",
Types.StringType.get()));
+ InMemoryCatalog catalog = new InMemoryCatalog();
+ catalog.initialize("test", Collections.emptyMap());
+ catalog.createNamespace(Namespace.of("db1"));
+ org.apache.iceberg.Table table = catalog.createTable(
+ TableIdentifier.of("db1", "t1"), oldSchema,
PartitionSpec.unpartitioned());
+
table.newAppend().appendFile(DataFiles.builder(PartitionSpec.unpartitioned())
+ .withPath("s3://bucket/db1/t1/old.parquet")
+ .withFileSizeInBytes(1).withRecordCount(1).build()).commit();
+ Schema historicalSchema = table.schema();
+ table.updateSchema().renameColumn("old_name", "new_name").commit();
+ ops.table = table;
+
+ ConnectorMvccSnapshot pin = ConnectorMvccSnapshot.builder()
+ .snapshotId(11L).schemaId(historicalSchema.schemaId()).build();
+ IcebergConnectorMetadata metadata = metadataWith(ops);
+ Map<String, ConnectorColumnHandle> handles = metadata.getColumnHandles(
+ null, new IcebergTableHandle("db1", "t1"), pin);
+
+ Assertions.assertTrue(metadata.supportsColumnHandleSnapshotPin(null));
+ Assertions.assertTrue(handles.containsKey("old_name"));
+ Assertions.assertTrue(handles.containsKey("survivor"));
+ Assertions.assertFalse(handles.containsKey("new_name"));
+
Assertions.assertEquals(historicalSchema.findField("old_name").fieldId(),
+ ((IcebergColumnHandle) handles.get("old_name")).getFieldId());
+ }
+
// ---------------------------------------------------------------------
// P6.3-T03: write transaction wiring (gate-closed / dormant)
// ---------------------------------------------------------------------
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergManifestCacheTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergManifestCacheTest.java
index 28fb5019dec..920e5ae0c6d 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergManifestCacheTest.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergManifestCacheTest.java
@@ -32,6 +32,8 @@ import org.junit.jupiter.api.Test;
import java.util.Collections;
import java.util.List;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicInteger;
/**
* Unit tests for {@link IcebergManifestCache} (T08). Uses a real {@link
InMemoryCatalog} table so the cache is
@@ -78,6 +80,49 @@ public class IcebergManifestCacheTest {
Assertions.assertEquals(1, cache.size());
}
+ @Test
+ public void equalityDeleteFieldIdsLoadOncePerSnapshot() {
+ IcebergManifestCache cache = new IcebergManifestCache();
+ AtomicInteger loads = new AtomicInteger();
+
+ Set<Integer> first =
cache.getOrLoadEqualityDeleteFieldIds("/warehouse/db/t", 10L, () -> {
+ loads.incrementAndGet();
+ return Collections.singleton(7);
+ });
+ Set<Integer> second =
cache.getOrLoadEqualityDeleteFieldIds("/warehouse/db/t", 10L, () -> {
+ loads.incrementAndGet();
+ return Collections.singleton(8);
+ });
+ Set<Integer> nextSnapshot =
cache.getOrLoadEqualityDeleteFieldIds("/warehouse/db/t", 11L, () -> {
+ loads.incrementAndGet();
+ return Collections.singleton(9);
+ });
+
+ Assertions.assertEquals(Collections.singleton(7), first);
+ Assertions.assertEquals(first, second, "the same immutable snapshot
must reuse its field-id set");
+ Assertions.assertEquals(Collections.singleton(9), nextSnapshot);
+ Assertions.assertEquals(2, loads.get(), "only one manifest walk is
allowed per snapshot");
+ }
+
+ @Test
+ public void equalityDeleteFieldIdFailureIsNotCached() {
+ IcebergManifestCache cache = new IcebergManifestCache();
+ AtomicInteger loads = new AtomicInteger();
+
+ Assertions.assertThrows(IllegalStateException.class,
+ () -> cache.getOrLoadEqualityDeleteFieldIds("/warehouse/db/t",
10L, () -> {
+ loads.incrementAndGet();
+ throw new IllegalStateException("transient manifest
failure");
+ }));
+ Set<Integer> retry =
cache.getOrLoadEqualityDeleteFieldIds("/warehouse/db/t", 10L, () -> {
+ loads.incrementAndGet();
+ return Collections.singleton(7);
+ });
+
+ Assertions.assertEquals(Collections.singleton(7), retry);
+ Assertions.assertEquals(2, loads.get(), "a failed manifest walk must
be retried, not memoized");
+ }
+
@Test
public void capacityOverflowFlushesWholesale() {
Table table = tableWithTwoDataFiles();
@@ -97,11 +142,21 @@ public class IcebergManifestCacheTest {
ManifestFile manifest =
table.currentSnapshot().dataManifests(table.io()).get(0);
IcebergManifestCache cache = new IcebergManifestCache();
cache.getManifestCacheValue(manifest, table);
+ AtomicInteger equalityLoads = new AtomicInteger();
+ cache.getOrLoadEqualityDeleteFieldIds(table.location(),
table.currentSnapshot().snapshotId(), () -> {
+ equalityLoads.incrementAndGet();
+ return Collections.singleton(1);
+ });
Assertions.assertEquals(1, cache.size());
// REFRESH CATALOG hook (H-5): invalidateAll drops every cached
manifest (legacy catalog-wide
// group.invalidateAll parity). MUTATION: a no-op invalidateAll ->
size stays 1 -> red.
cache.invalidateAll();
Assertions.assertEquals(0, cache.size());
+ cache.getOrLoadEqualityDeleteFieldIds(table.location(),
table.currentSnapshot().snapshotId(), () -> {
+ equalityLoads.incrementAndGet();
+ return Collections.singleton(1);
+ });
+ Assertions.assertEquals(2, equalityLoads.get(), "catalog refresh must
also clear the snapshot projection");
}
@Test
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergScanPlanProviderTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergScanPlanProviderTest.java
index 55f6c087f08..26fa913062a 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergScanPlanProviderTest.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergScanPlanProviderTest.java
@@ -67,6 +67,7 @@ import org.apache.iceberg.io.StorageCredential;
import org.apache.iceberg.io.SupportsStorageCredentials;
import org.apache.iceberg.types.Conversions;
import org.apache.iceberg.types.Types;
+import org.apache.iceberg.types.Types.NestedField;
import org.apache.iceberg.util.SerializationUtil;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
@@ -814,6 +815,31 @@ public class IcebergScanPlanProviderTest {
"the unprojected equality-delete key column must be
force-included (#65502), got " + top);
}
+ @Test
+ public void
getScanNodePropertiesCachesEqualityDeleteFieldIdsWithManifestCacheDisabled() {
+ Schema schema = new Schema(
+ Types.NestedField.required(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "value",
Types.StringType.get()));
+ Table table = createTable("cached_eq_ids", schema,
PartitionSpec.unpartitioned(),
+ Collections.singletonMap(TableProperties.FORMAT_VERSION, "2"));
+ table.newAppend().appendFile(dataFile(table.spec(),
+ "s3://b/db/cached_eq_ids/f1.parquet", 1024, null,
null)).commit();
+ table.newRowDelta().addDeletes(equalityDeleteFile(
+ "s3://b/db/cached_eq_ids/eq.parquet", FileFormat.PARQUET,
1)).commit();
+ IcebergManifestCache cache = new IcebergManifestCache();
+ // Keep construction behind the shared helper so catalog-property API
migrations do not break this test.
+ IcebergScanPlanProvider provider =
manifestProvider(Collections.emptyMap(), table, cache);
+
+ provider.getScanNodeProperties(null, new IcebergTableHandle("db1",
"cached_eq_ids"),
+ Collections.singletonList(new IcebergColumnHandle("value",
2)), Optional.empty());
+
+ Set<Integer> cached = cache.getOrLoadEqualityDeleteFieldIds(
+ table.location(), table.currentSnapshot().snapshotId(), () -> {
+ throw new AssertionError("the provider must populate the
snapshot-scoped projection cache");
+ });
+ Assertions.assertEquals(Collections.singleton(1), cached);
+ }
+
@Test
public void equalityCarrierAllowsUnrelatedDropAndReaddNames() throws
Exception {
Schema schema = new Schema(
@@ -833,8 +859,6 @@ public class IcebergScanPlanProviderTest {
"s3://b/db/drop_readd/eq.parquet", FileFormat.PARQUET,
1))
.commit();
- int oldTopLevelId = table.schema().findField("same_name").fieldId();
- int oldNestedId =
table.schema().findField("payload.same_name").fieldId();
table.updateSchema().deleteColumn("same_name").deleteColumn("payload.same_name").commit();
table.updateSchema()
.addColumn("same_name", Types.IntegerType.get())
@@ -863,7 +887,8 @@ public class IcebergScanPlanProviderTest {
payload = field;
}
}
- Assertions.assertEquals(Arrays.asList(currentTopLevelId,
oldTopLevelId), sameNameIds);
+ Assertions.assertEquals(Collections.singletonList(currentTopLevelId),
sameNameIds,
+ "unrelated historical fields must not inflate the
equality-delete schema carrier");
Assertions.assertNotNull(payload);
List<Integer> nestedSameNameIds = new ArrayList<>();
for (TFieldPtr field :
payload.getFieldPtr().getNestedField().getStructField().getFields()) {
@@ -871,11 +896,60 @@ public class IcebergScanPlanProviderTest {
nestedSameNameIds.add(field.getFieldPtr().getId());
}
}
- Assertions.assertEquals(Arrays.asList(currentNestedId, oldNestedId),
nestedSameNameIds);
+ Assertions.assertEquals(Collections.singletonList(currentNestedId),
nestedSameNameIds,
+ "only field IDs referenced by live equality deletes may be
retained");
+ }
+
+ @Test
+ public void
equalityCarrierResolvesDroppedLiveEqualityKeyFromSchemaHistory() throws
Exception {
+ Schema schema = new Schema(
+ Types.NestedField.required(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "old_key",
Types.StringType.get()));
+ Table table = createTable("dropped_eq_key", schema,
PartitionSpec.unpartitioned(),
+ Collections.singletonMap(TableProperties.FORMAT_VERSION, "2"));
+ table.newAppend().appendFile(dataFile(table.spec(),
+ "s3://b/db/dropped_eq_key/f1.parquet", 1024, null,
null)).commit();
+ table.newRowDelta().addDeletes(equalityDeleteFile(
+ "s3://b/db/dropped_eq_key/eq.parquet", FileFormat.PARQUET,
2)).commit();
+ table.updateSchema().deleteColumn("old_key").commit();
+
+ IcebergScanPlanProvider provider = providerOver(table);
+ Map<String, String> props = provider.getScanNodeProperties(
+ null, new IcebergTableHandle("db1", "dropped_eq_key"),
+ Collections.singletonList(new IcebergColumnHandle("id", 1)),
Optional.empty());
+ TFileScanRangeParams params = new TFileScanRangeParams();
+ provider.populateScanLevelParams(params, props);
+
+ List<TFieldPtr> fields =
params.getHistorySchemaInfo().get(0).getRootField().getFields();
+ Assertions.assertTrue(fields.stream().anyMatch(field ->
field.getFieldPtr().getId() == 2),
+ "a dropped field still referenced by a live equality delete
must remain resolvable by ID");
}
@Test
- public void
partitionPrunedEqualityDeleteDoesNotRequireCurrentBackendSemantics() {
+ public void equalityCarrierSizeDoesNotGrowWithUnrelatedSchemaHistory()
throws Exception {
+ Schema schema = new Schema(
+ Types.NestedField.required(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "churn",
Types.IntegerType.get()));
+ Table table = createTable("bounded_eq_carrier", schema,
PartitionSpec.unpartitioned(),
+ Collections.singletonMap(TableProperties.FORMAT_VERSION, "2"));
+ table.newAppend().appendFile(dataFile(table.spec(),
+ "s3://b/db/bounded_eq_carrier/f1.parquet", 1024, null,
null)).commit();
+ table.newRowDelta().addDeletes(equalityDeleteFile(
+ "s3://b/db/bounded_eq_carrier/eq.parquet", FileFormat.PARQUET,
1)).commit();
+ for (int index = 0; index < 64; index++) {
+ table.updateSchema().deleteColumn("churn").commit();
+ table.updateSchema().addColumn("churn",
Types.IntegerType.get()).commit();
+ }
+
+ List<NestedField> carrier =
IcebergScanPlanProvider.schemaForPotentialEqualityDeletes(
+ table, table.newScan(), table.schema());
+ Assertions.assertEquals(2, carrier.size(),
+ "unrelated schema churn must not increase FE heap or schema
RPC size");
+ Assertions.assertEquals(table.schema().columns(), carrier);
+ }
+
+ @Test
+ public void
partitionPrunedEqualityDeleteConservativelyRequiresCurrentBackendSemantics() {
PartitionSpec spec =
PartitionSpec.builderFor(PART_SCHEMA).identity("p").build();
Table table = createTable("partition_pruned_eqdel", PART_SCHEMA, spec,
Collections.singletonMap(TableProperties.FORMAT_VERSION, "2"));
@@ -901,9 +975,9 @@ public class IcebergScanPlanProviderTest {
Map<String, String> prunedProps = provider.getScanNodeProperties(
null, new IcebergTableHandle("db1", "partition_pruned_eqdel"),
columns, Optional.of(eqInt("p", 1)));
- Assertions.assertFalse(prunedProps.containsKey(
+ Assertions.assertTrue(prunedProps.containsKey(
ScanNodePropertyKeys.REQUIRED_CURRENT_BACKEND_SEMANTICS),
- "an equality delete in a pruned partition must not gate the
selected tasks");
+ "delete-manifest inspection must stay conservative without
enumerating data tasks");
Map<String, String> applicableProps = provider.getScanNodeProperties(
null, new IcebergTableHandle("db1", "partition_pruned_eqdel"),
@@ -914,7 +988,7 @@ public class IcebergScanPlanProviderTest {
}
@Test
- public void
sequencePrunedEqualityDeleteDoesNotRequireCurrentBackendSemantics() {
+ public void
sequencePrunedEqualityDeleteConservativelyRequiresCurrentBackendSemantics() {
Table table = createTable("sequence_pruned_eqdel", SCHEMA,
PartitionSpec.unpartitioned(),
Collections.singletonMap(TableProperties.FORMAT_VERSION, "2"));
DataFile oldFile = dataFile(table.spec(),
@@ -935,9 +1009,9 @@ public class IcebergScanPlanProviderTest {
Map<String, String> props = provider.getScanNodeProperties(
null, new IcebergTableHandle("db1", "sequence_pruned_eqdel"),
Collections.singletonList(new IcebergColumnHandle("id", 1)),
Optional.empty());
- Assertions.assertFalse(props.containsKey(
+ Assertions.assertTrue(props.containsKey(
ScanNodePropertyKeys.REQUIRED_CURRENT_BACKEND_SEMANTICS),
- "an older equality delete must not gate a later-sequence
replacement data file");
+ "snapshot metadata must gate conservatively without
synchronously replanning all files");
}
@Test
@@ -1032,6 +1106,28 @@ public class IcebergScanPlanProviderTest {
"the equality carrier must not walk snapshots when no
historical field is missing");
}
+ @Test
+ public void
schemaHistoryPlanningStaysBoundedWhenHistoricalRequirednessNeedsTheFence() {
+ Schema oldSchema = new Schema(
+ Types.NestedField.optional(1, "id", Types.IntegerType.get()));
+ Table table = createTable("requiredness_history", oldSchema,
PartitionSpec.unpartitioned());
+ table.newAppend().appendFile(dataFile(table.spec(),
+ "s3://b/db/requiredness_history/old.parquet", 128, null,
null)).commit();
+
table.updateSchema().allowIncompatibleChanges().requireColumn("id").commit();
+ table.newAppend().appendFile(dataFile(table.spec(),
+ "s3://b/db/requiredness_history/new.parquet", 128, null,
null)).commit();
+
+ FakeIcebergTable countingTable = new FakeIcebergTable(
+ table.name(), table.schema(), table.spec(), table.location(),
table.properties());
+ countingTable.setScanTable(table);
+ Assertions.assertTrue(
+
IcebergScanPlanProvider.selectedHistoryRequiresMissingRequiredFieldRejection(
+ countingTable, table.schema(),
Collections.singleton(1),
+ table.currentSnapshot()));
+ Assertions.assertEquals(0, countingTable.getSnapshotLookupCount(),
+ "the upgrade fence must be conservative and bounded by schema
history, not snapshot count");
+ }
+
@Test
public void
getScanNodePropertiesEmitsSchemaEvolutionDictForPartitionedTableToo() {
// The dict is emitted alongside path_partition_keys (it is
unconditional, like legacy
@@ -1361,10 +1457,29 @@ public class IcebergScanPlanProviderTest {
Assertions.assertTrue(props.containsKey("iceberg.schema_evolution"),
"position_deletes reads natively and needs the field-id dict
to resolve `row`");
+ Assertions.assertFalse(props.containsKey(
+
ScanNodePropertyKeys.REQUIRED_CURRENT_BACKEND_SEMANTICS),
+ "metadata-only projections do not depend on
position_deletes.row semantics");
Assertions.assertFalse(props.containsKey("path_partition_keys"),
"a metadata table is still not base-spec partitioned -> no
path_partition_keys");
}
+ @Test
+ public void
getScanNodePropertiesForPositionDeletesRowRequiresCurrentBackendSemantics() {
+ Table table = tableWithPositionDelete(
+ positionDeleteFile("s3://b/db/t1/pos.parquet",
FileFormat.PARQUET, null, null));
+ // The helper follows the active catalog-property wrapper while this
test stays focused on scan semantics.
+ IcebergScanPlanProvider provider = providerOver(table);
+
+ Map<String, String> props = provider.getScanNodeProperties(
+ null, IcebergTableHandle.forSystemTable("db1", "t1",
"position_deletes", -1L, null, -1L),
+ Collections.singletonList(new IcebergColumnHandle("row", 3)),
Optional.empty());
+
+ Assertions.assertTrue(props.containsKey(
+
ScanNodePropertyKeys.REQUIRED_CURRENT_BACKEND_SEMANTICS),
+ "projecting position_deletes.row depends on current
nested-default semantics");
+ }
+
@Test
public void
getScanNodePropertiesForPositionDeletesLoadsTheBaseTableOnlyOnce() {
// WHY: the dict branch needs the METADATA table, and the obvious way
to get one is resolveSysTable().
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWritePlanProviderTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWritePlanProviderTest.java
index bcfaf94a932..4ab51cbe4e7 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWritePlanProviderTest.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWritePlanProviderTest.java
@@ -351,6 +351,20 @@ public class IcebergWritePlanProviderTest {
Assertions.assertFalse(plan.getDataSink().getIcebergTableSink().getSchemaJson().contains("renamed_name"));
}
+ @Test
+ public void writeSchemaAllowsNestedPrimitivePartitionSource() {
+ InMemoryCatalog catalog = freshCatalog();
+ Schema schema = new Schema(
+ Types.NestedField.required(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "payload", Types.StructType.of(
+ Types.NestedField.optional(3, "part",
Types.IntegerType.get()))));
+ Table table = catalog.createTable(TableIdentifier.of("db1",
"nested_partition"), schema,
+
PartitionSpec.builderFor(schema).identity("payload.part").build());
+
+ Assertions.assertDoesNotThrow(() -> IcebergWriteSchemaContext.create(
+ table, "db1.nested_partition", Optional.empty(), false,
false));
+ }
+
@Test
public void branchWriteRejectsCurrentRequiredFieldAbsentWithoutDefault() {
InMemoryCatalog catalog = freshCatalog();
@@ -1292,6 +1306,26 @@ public class IcebergWritePlanProviderTest {
Assertions.assertEquals(table.schema().findField("id").fieldId(),
f.getSourceId());
}
+ @Test
+ public void getWritePartitioningRoutesNestedSourceByItsTopLevelColumn() {
+ InMemoryCatalog catalog = freshCatalog();
+ Schema schema = new Schema(
+ Types.NestedField.required(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "payload", Types.StructType.of(
+ Types.NestedField.optional(3, "part",
Types.IntegerType.get()))));
+ Table table = catalog.createTable(TableIdentifier.of("db1",
"nested_partition"), schema,
+ PartitionSpec.builderFor(schema).bucket("payload.part",
8).build());
+
+ ConnectorWritePartitionField field = providerFor(table,
contextWithStorage())
+ .getWritePartitioning(sessionFor(table, contextWithStorage()),
+ new IcebergTableHandle("db1", "nested_partition"))
+ .getFields().get(0);
+
+ Assertions.assertEquals("payload", field.getSourceColumnName(),
+ "merge exchange must route a nested source through its bound
top-level struct slot");
+ Assertions.assertEquals(Integer.valueOf(3), field.getSourceId());
+ }
+
// ───────────────────── getSyntheticWriteColumns (connector declares the
row-id STRUCT, ③ C3b-core) ─────────────────────
//
// WHY: post-flip the iceberg DML hidden column
__DORIS_ICEBERG_ROWID_COL__ that legacy
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/PluginDrivenScanNode.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/PluginDrivenScanNode.java
index 51f45809c27..7710b46224c 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/PluginDrivenScanNode.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/PluginDrivenScanNode.java
@@ -2200,7 +2200,7 @@ public class PluginDrivenScanNode extends
FileQueryScanNode {
ConnectorColumnHandle ch = allHandles.get(name);
if (ch != null) {
selected.add(withProjectedFieldIds(ch, slot));
- } else if (pinnedNames.contains(name)) {
+ } else if (requiresPinnedColumnHandle(slot.getColumn(),
pinnedNames)) {
throw new UserException("Column '" + name + "' of table "
+ getTargetTable().getName() + " resolves in the
pinned time-travel schema"
+ " but has no connector column handle; refusing
to silently drop it"
@@ -2211,6 +2211,12 @@ public class PluginDrivenScanNode extends
FileQueryScanNode {
return selected;
}
+ static boolean requiresPinnedColumnHandle(Column column, Set<String>
pinnedNames) {
+ // A connector-reserved passthrough column is generated by the scan
provider rather than resolved
+ // from the physical table schema, so it may legitimately be absent
from the column-handle map.
+ return pinnedNames.contains(column.getName()) &&
!column.isReservedPassthrough();
+ }
+
static ConnectorColumnHandle withProjectedFieldIds(
ConnectorColumnHandle handle, SlotDescriptor slot) {
Set<Integer> projectedFieldIds = new HashSet<>();
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 099d6a1edb3..09c8bc24628 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
@@ -3388,7 +3388,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 f1c567b2515..6e722aba19d 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 MergePartitionField(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 MergePartitionField(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 b40366aa3b9..7c7a2c5003a 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
@@ -259,6 +259,9 @@ public class InsertIntoTableCommand extends Command
implements NeedAuditEncrypti
int retryTimes = 0;
ctx.getStatementContext().setIsInsert(true);
while (++retryTimes <
Math.max(ctx.getSessionVariable().dmlPlanRetryTimes, 3)) {
+ // Each internal attempt must repin connector metadata; retaining
the previous writer schema can
+ // plan defaults and partition fields against the table version
that triggered the retry.
+ ctx.getStatementContext().resetConnectorStatementScope();
TableIf targetTableIf = getTargetTableIf(ctx,
qualifiedTargetTableName);
DatabaseIf<?> targetDatabase = getTargetDatabase(targetTableIf);
// check auth
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalExternalRowLevelMergeSink.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalExternalRowLevelMergeSink.java
index 292cf34fd1d..95de03f2e03 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalExternalRowLevelMergeSink.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalExternalRowLevelMergeSink.java
@@ -369,7 +369,8 @@ public class PhysicalExternalRowLevelMergeSink<CHILD_TYPE
extends Plan>
return new InsertPartitionFieldResult(false, false, null);
}
ConnectorWritePartitionSpec spec =
writePlanProvider.getWritePartitioning(session, handle);
- return reconstructPartitionFields(insertPartitionFields, spec,
columnExprIdMap, columnIdToExprId);
+ return reconstructPartitionFields(
+ insertPartitionFields, spec, columnExprIdMap,
columnIdToExprId, cols);
}
/**
@@ -397,7 +398,7 @@ public class PhysicalExternalRowLevelMergeSink<CHILD_TYPE
extends Plan>
ConnectorWritePartitionSpec spec,
Map<String, ExprId> columnExprIdMap) {
return reconstructPartitionFields(insertPartitionFields, spec,
columnExprIdMap,
- java.util.Collections.emptyMap());
+ java.util.Collections.emptyMap(), null);
}
static InsertPartitionFieldResult reconstructPartitionFields(
@@ -405,6 +406,25 @@ public class PhysicalExternalRowLevelMergeSink<CHILD_TYPE
extends Plan>
ConnectorWritePartitionSpec spec,
Map<String, ExprId> columnExprIdMap,
Map<Integer, ExprId> columnIdToExprId) {
+ return reconstructPartitionFields(
+ insertPartitionFields, spec, columnExprIdMap,
columnIdToExprId, null);
+ }
+
+ static InsertPartitionFieldResult reconstructPartitionFields(
+ List<DistributionSpecMerge.MergePartitionField>
insertPartitionFields,
+ ConnectorWritePartitionSpec spec,
+ Map<String, ExprId> columnExprIdMap,
+ List<Column> tableColumns) {
+ return reconstructPartitionFields(insertPartitionFields, spec,
columnExprIdMap,
+ java.util.Collections.emptyMap(), tableColumns);
+ }
+
+ static InsertPartitionFieldResult reconstructPartitionFields(
+ List<DistributionSpecMerge.MergePartitionField>
insertPartitionFields,
+ ConnectorWritePartitionSpec spec,
+ Map<String, ExprId> columnExprIdMap,
+ Map<Integer, ExprId> columnIdToExprId,
+ List<Column> tableColumns) {
if (spec == null) {
return new InsertPartitionFieldResult(false, false, null);
}
@@ -424,16 +444,32 @@ public class PhysicalExternalRowLevelMergeSink<CHILD_TYPE
extends Plan>
}
// Prefer the stable source field id carried by the bind-time
schema. A same-name replacement
// must not inherit the old output expression after concurrent
Iceberg schema evolution.
- ExprId exprId = columnIdToExprId.isEmpty()
- ? columnExprIdMap.get(sourceColumnName)
- : columnIdToExprId.get(field.getSourceId());
+ Column sourceColumn = findSourceColumn(tableColumns,
sourceColumnName);
+ ExprId exprId;
+ if (columnIdToExprId.isEmpty()) {
+ exprId = columnExprIdMap.get(sourceColumnName);
+ } else if (sourceColumn == null) {
+ // The id-only test seam has no column tree, so preserve its
exact top-level-id lookup.
+ exprId = tableColumns == null ?
columnIdToExprId.get(field.getSourceId()) : null;
+ } else {
+ // A nested Iceberg source id identifies a child, but the
Nereids slot and its ExprId belong to
+ // the top-level struct. Resolve the slot by its root id and
use sourceFieldPath for the child.
+ exprId = sourceColumn.getUniqueId() < 0
+ ? null :
columnIdToExprId.get(sourceColumn.getUniqueId());
+ }
if (exprId == null) {
insertPartitionFields.clear();
return new InsertPartitionFieldResult(false, hasNonIdentity,
spec.getSpecId());
}
+ List<Integer> sourceFieldPath = resolveSourceFieldPath(
+ tableColumns, sourceColumnName, field.getSourceId());
+ if (sourceFieldPath == null) {
+ insertPartitionFields.clear();
+ return new InsertPartitionFieldResult(false, hasNonIdentity,
spec.getSpecId());
+ }
insertPartitionFields.add(new
DistributionSpecMerge.MergePartitionField(
field.getTransform(), exprId, field.getTransformParam(),
- field.getFieldName(), field.getSourceId()));
+ field.getFieldName(), field.getSourceId(),
sourceFieldPath));
}
if (insertPartitionFields.isEmpty()) {
return new InsertPartitionFieldResult(false, hasNonIdentity,
spec.getSpecId());
@@ -441,6 +477,57 @@ public class PhysicalExternalRowLevelMergeSink<CHILD_TYPE
extends Plan>
return new InsertPartitionFieldResult(true, hasNonIdentity,
spec.getSpecId());
}
+ private static List<Integer> resolveSourceFieldPath(
+ List<Column> tableColumns, String sourceColumnName, int sourceId) {
+ if (tableColumns == null) {
+ return ImmutableList.of();
+ }
+ Column sourceColumn = findSourceColumn(tableColumns, sourceColumnName);
+ if (sourceColumn == null) {
+ return null;
+ }
+ if (sourceColumn.getUniqueId() < 0) {
+ // Without the root id an empty path cannot distinguish a
top-level source from an unstamped child.
+ return null;
+ }
+ if (sourceColumn.getUniqueId() == sourceId) {
+ return ImmutableList.of();
+ }
+ List<Integer> path = new ArrayList<>();
+ // Iceberg field ids are stable across rename/evolution; resolving by
id avoids ambiguous dotted names
+ // and keeps exchange routing on the same nested value used by the
writer.
+ return findSourceFieldPath(sourceColumn.getChildren(), sourceId, path)
+ ? ImmutableList.copyOf(path) : null;
+ }
+
+ private static Column findSourceColumn(List<Column> tableColumns, String
sourceColumnName) {
+ if (tableColumns == null) {
+ return null;
+ }
+ for (Column column : tableColumns) {
+ if (column.getName().equalsIgnoreCase(sourceColumnName)) {
+ return column;
+ }
+ }
+ return 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;
+ }
+
// Package-private (not private) so the same-package parity test can
assert on the reconstructed
// result of {@link #reconstructPartitionFields} directly, without driving
the full distribution.
static class InsertPartitionFieldResult {
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 e156fa4336f..0ef85f8ee67 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
@@ -161,14 +161,21 @@ public class DataPartition {
private final Integer param;
private final String name;
private final Integer sourceId;
+ private final ImmutableList<Integer> sourceFieldPath;
public MergePartitionField(Expr sourceExpr, String transform, Integer
param,
String name, Integer sourceId) {
+ this(sourceExpr, transform, param, name, sourceId,
ImmutableList.of());
+ }
+
+ public MergePartitionField(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() {
@@ -184,6 +191,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/scan/PluginDrivenScanNodeColumnPruningTest.java
b/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/PluginDrivenScanNodeColumnPruningTest.java
index 4f7c8cbe9ca..9eccf6102ed 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/PluginDrivenScanNodeColumnPruningTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/PluginDrivenScanNodeColumnPruningTest.java
@@ -31,6 +31,7 @@ import org.junit.jupiter.api.Test;
import org.mockito.Mockito;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
@@ -162,6 +163,20 @@ public class PluginDrivenScanNodeColumnPruningTest {
Assertions.assertSame(all.get("c1"), selected.get(0));
}
+ @Test
+ public void
testPinnedHandleGuardExemptsConnectorReservedPassthroughColumn() {
+ // Iceberg v3 row-lineage columns are part of the bound Doris schema
but are generated scan slots,
+ // not physical Iceberg schema columns. Requiring a connector handle
for one rejects every pinned
+ // v3 scan before the scan provider can append the generated field to
its schema dictionary.
+ Column rowId = new Column("_row_id", PrimitiveType.BIGINT);
+ rowId.setReservedPassthrough(true);
+
+ Assertions.assertFalse(PluginDrivenScanNode.requiresPinnedColumnHandle(
+ rowId, Collections.singleton("_row_id")));
+ Assertions.assertTrue(PluginDrivenScanNode.requiresPinnedColumnHandle(
+ new Column("physical_col", PrimitiveType.INT),
Collections.singleton("physical_col")));
+ }
+
@Test
public void testEmptyTupleProjectsNothing() {
// A tuple with no slots projects nothing — the ONLY input that makes
the jdbc connector fall back to
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/physical/PhysicalExternalRowLevelMergeSinkTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/physical/PhysicalExternalRowLevelMergeSinkTest.java
index 793f14a3a83..970e80d380d 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/physical/PhysicalExternalRowLevelMergeSinkTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/physical/PhysicalExternalRowLevelMergeSinkTest.java
@@ -19,6 +19,9 @@ package org.apache.doris.nereids.trees.plans.physical;
import org.apache.doris.catalog.Column;
import org.apache.doris.catalog.PrimitiveType;
+import org.apache.doris.catalog.StructField;
+import org.apache.doris.catalog.StructType;
+import org.apache.doris.catalog.Type;
import org.apache.doris.common.jmockit.Deencapsulation;
import org.apache.doris.connector.spi.Connector;
import org.apache.doris.connector.spi.ConnectorMetadata;
@@ -132,6 +135,87 @@ public class PhysicalExternalRowLevelMergeSinkTest {
"transform param/name/sourceId must be carried verbatim from
the connector field");
}
+ @Test
+ public void reconstructCarriesNestedSourcePathWithoutRandomFallback() {
+ ExprId payload = exprId("payload");
+ List<MergePartitionField> out = new ArrayList<>();
+ ConnectorWritePartitionField nested = new ConnectorWritePartitionField(
+ "bucket[8]", 8, "payload", "payload_part_bucket", 3);
+ Column payloadColumn = new Column("payload", new StructType(
+ new StructField("part", Type.INT)));
+ payloadColumn.setUniqueId(2);
+ payloadColumn.getChildren().get(0).setUniqueId(3);
+
+ InsertPartitionFieldResult result =
PhysicalExternalRowLevelMergeSink.reconstructPartitionFields(
+ out, spec(4, nested), map("payload", payload),
ImmutableList.of(payloadColumn));
+
+ Assertions.assertTrue(result.success);
+ Assertions.assertEquals(ImmutableList.of(
+ new MergePartitionField("bucket[8]", payload, 8,
"payload_part_bucket", 3,
+ ImmutableList.of(0))), out);
+ }
+
+ @Test
+ public void reconstructNestedSourceUsesTopLevelIdMapFromProductionPath() {
+ ExprId payload = exprId("payload");
+ List<MergePartitionField> out = new ArrayList<>();
+ ConnectorWritePartitionField nested = new ConnectorWritePartitionField(
+ "bucket[8]", 8, "payload", "payload_part_bucket", 3);
+ Column payloadColumn = new Column("payload", new StructType(
+ new StructField("part", Type.INT)));
+ payloadColumn.setUniqueId(2);
+ payloadColumn.getChildren().get(0).setUniqueId(3);
+
+ InsertPartitionFieldResult result =
PhysicalExternalRowLevelMergeSink.reconstructPartitionFields(
+ out, spec(4, nested), map("payload", payload),
+ java.util.Collections.singletonMap(2, payload),
ImmutableList.of(payloadColumn));
+
+ // Production maps expressions by top-level column id, while Iceberg
partition sources carry the
+ // nested child id. The root id must select the slot and the child id
must select the path within it.
+ Assertions.assertTrue(result.success);
+ Assertions.assertEquals(ImmutableList.of(
+ new MergePartitionField("bucket[8]", payload, 8,
"payload_part_bucket", 3,
+ ImmutableList.of(0))), out);
+ }
+
+ @Test
+ public void reconstructUnstampedNestedSourceHardFails() {
+ ExprId payload = exprId("payload");
+ List<MergePartitionField> out = new ArrayList<>();
+ ConnectorWritePartitionField nested = new ConnectorWritePartitionField(
+ "bucket[8]", 8, "payload", "payload_part_bucket", 3);
+ Column payloadColumn = new Column("payload", new StructType(
+ new StructField("part", Type.INT)));
+
+ InsertPartitionFieldResult result =
PhysicalExternalRowLevelMergeSink.reconstructPartitionFields(
+ out, spec(4, nested), map("payload", payload),
ImmutableList.of(payloadColumn));
+
+ // An unstamped tree cannot prove whether the requested id is
top-level or nested. Treating it as an
+ // empty path would hash the whole struct and silently diverge from
the writer's partition source.
+ Assertions.assertFalse(result.success);
+ Assertions.assertTrue(out.isEmpty());
+ }
+
+ @Test
+ public void reconstructMissingNestedSourceIdHardFails() {
+ Column payloadColumn = new Column("payload", new StructType(
+ new StructField("part", Type.INT)));
+ payloadColumn.setUniqueId(2);
+ payloadColumn.getChildren().get(0).setUniqueId(3);
+ List<MergePartitionField> out = new ArrayList<>();
+
+ InsertPartitionFieldResult result =
PhysicalExternalRowLevelMergeSink.reconstructPartitionFields(
+ out,
+ spec(4, new ConnectorWritePartitionField(
+ "bucket[8]", 8, "payload", "payload_part_bucket", 99)),
+ map("payload", exprId("payload")),
+ ImmutableList.of(payloadColumn));
+
+ Assertions.assertFalse(result.success,
+ "an evolved schema must not hash the whole struct when the
nested source id disappeared");
+ Assertions.assertTrue(out.isEmpty());
+ }
+
@Test
public void reconstructNullSourceColumnNameHardFailsAndClears() {
// PARITY-1a: a null source-column-name field hard-fails the whole
spec; the already-added prior
@@ -220,6 +304,7 @@ public class PhysicalExternalRowLevelMergeSinkTest {
// must flow into the DistributionSpecMerge: one identity partition
column 'id' resolved to the
// child's id slot, insertRandom=false, spec id carried.
Column id = new Column("id", PrimitiveType.INT);
+ id.setUniqueId(1);
SlotReference idSlot = new SlotReference("id", IntegerType.INSTANCE);
SlotReference opSlot = new
SlotReference(MergeOperation.OPERATION_COLUMN, IntegerType.INSTANCE);
SlotReference rowidSlot = new SlotReference(Column.ICEBERG_ROWID_COL,
IntegerType.INSTANCE);
diff --git a/gensrc/thrift/Partitions.thrift b/gensrc/thrift/Partitions.thrift
index 19ab0a17dc7..da172fac735 100644
--- a/gensrc/thrift/Partitions.thrift
+++ b/gensrc/thrift/Partitions.thrift
@@ -183,6 +183,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]
