This is an automated email from the ASF dual-hosted git repository.
morningman 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 1c9b51c0c75 [test](iceberg) Port the branch-4.1 Iceberg V3 default
tests and reserve TIcebergFileDesc field 13 (#69394)
1c9b51c0c75 is described below
commit 1c9b51c0c759a9c9135593f7ce376025178407a8
Author: Mingyu Chen (Rayner) <[email protected]>
AuthorDate: Fri Oct 23 10:11:07 2026 +0100
[test](iceberg) Port the branch-4.1 Iceberg V3 default tests and reserve
TIcebergFileDesc field 13 (#69394)
### What problem does this PR solve?
Issue Number: None
Related PR: #66538 #65851
Problem Summary:
**In short.** This ports the Iceberg V3 default-value tests of
branch-4.1 #66538 to master: 3 BE unit tests, 37 FE unit tests and one
regression check, all for behavior master already has. It also reserves
field id 13 of `TIcebergFileDesc`, which branch-4.1 and branch-4.2 use
for `equality_delete_schema`. No product code changes.
**Background.** Iceberg V3 initial / write defaults were implemented
twice: by #65851 on master and by #66538 on branch-4.1, which branch-4.2
inherits. The 4.1 tests target fe-core classes (`IcebergScanNode`,
`IcebergWriteSchemaContext`, `IcebergTransaction`, …) and V1 reader
internals that master no longer has. On master the Iceberg FE lives in
the `fe-connector-iceberg` plugin and the V1 reader in
`iceberg_reader_mixin.h`. So most of the 4.1 tests never reached master,
even where master implements the same behavior.
**The problem, and what it cost.** A scenario-by-scenario comparison of
the two test suites found that master implements most of the 4.1-tested
behavior with no test of its own. The untested behavior:
| Area | Untested on master before this PR |
|---|---|
| Equality-delete schema carrier | a dropped key keeps its latest name,
optionality and initial default; a dropped nested key comes back inside
its parent; a dropped nanosecond key stays `TIMESTAMP_NS`; only
referenced historical fields enter the carrier |
| Rolling-upgrade fences | missing-required-field and
nested-initial-default fences follow the projection; collection wrappers
do not trigger them; schema-only targets and cherry-picked ancestry are
covered; name-mapping alias collisions |
| Write commit fences | schema skew, a recreated table and current-spec
replacement are rejected before a transaction opens; a required column,
a spec drift to unpartitioned or a table replacement that lands while
the commit stages its snapshot is rejected at the final metadata swap |
| Pinned writer metadata | INSERT commit, merge sink (including the
unpartitioned delete spec id) and file conversion use the
statement-pinned spec, sort order and format, not the live table |
| Write defaults | typed SQL literal per primitive type; maps and NULL
elements in complex defaults; `DEFAULT` resolves by field id, not by a
reused name; UPDATE and matched MERGE do not inject write defaults |
| Wire compatibility | `TIcebergFileDesc` ends at field 12 on master,
while 4.1 / 4.2 send `13: optional ExternalTableSchema.TSchema
equality_delete_schema`; nothing stops master from reusing 13 for an
unrelated field |
Without these tests a regression in any of these paths would go
unnoticed, and a reused field 13 would misdecode 4.x plans.
**How this PR fixes it.** It rewrites the 4.1 scenarios against master's
APIs. Plugin tests use the existing Mockito-free fakes
(`InMemoryCatalog`, `Proxy` over `TableOperations`). The commit-fence
tests hook `newSnapshotId` on the staging operations, so the concurrent
change lands after the commit preflight and before the metadata swap.
The BE tests are copied unchanged from 4.1 (the helper signatures match
on master). `PlanNodes.thrift` gets a comment reserving field 13.
`test_iceberg_write_default` gains a `qt_` check that
`information_schema.columns.COLUMN_DEFAULT` stays NULL for a column
whose default is a connector-only write default. The behavior
differences found in the same comparison are fixed in separate PRs (for
example #69369).
**Results.** All new tests pass on master. To show they check something,
the fence and pinned-metadata tests were re-run with the guarded
condition removed from the product code; each failed.
| Suite | Result |
|---|---|
| `fe-connector-iceberg` (all tests) | 1656 run, 0 failures, 0 errors, 5
skipped; checkstyle clean |
| fe-core `ConnectorWriteSchemaUtilsTest`,
`ExternalRowLevelUpdatePlanBuilderTest`,
`ExternalRowLevelMergePlanBuilderTest` | 13 run, 0 failures |
| BE
`IcebergV2ReaderTest.IcebergProjectedStructDoesNotReuseReaddedEqualityKeyByName`,
`ParquetReadLinesTest.iceberg_row_id_fetch_materializes_readded_missing_column`,
`VOrcTransformerTest.SkipsInvalidBinaryChildrenHiddenByNullableCollections`
| pass |
| Regression `external_table_p0/iceberg/test_iceberg_write_default` |
pass |
Co-authored-by: daidai <[email protected]>
---
be/test/format/parquet/parquet_read_lines.cpp | 147 ++++++++
.../format/transformer/vorc_transformer_test.cpp | 76 ++++
be/test/format_v2/table/iceberg_reader_test.cpp | 72 ++++
.../iceberg/IcebergConnectorMetadataTest.java | 35 ++
.../iceberg/IcebergConnectorTransactionTest.java | 278 +++++++++++++++
.../iceberg/IcebergScanPlanProviderTest.java | 387 +++++++++++++++++++++
.../connector/iceberg/IcebergSchemaUtilsTest.java | 50 +++
.../iceberg/IcebergWritePlanProviderTest.java | 306 ++++++++++++++++
.../connector/iceberg/IcebergWriterHelperTest.java | 47 +++
.../commands/ConnectorWriteSchemaUtilsTest.java | 109 ++++++
.../ExternalRowLevelMergePlanBuilderTest.java | 55 +++
.../ExternalRowLevelUpdatePlanBuilderTest.java | 38 ++
gensrc/thrift/PlanNodes.thrift | 3 +
.../iceberg/test_iceberg_write_default.out | 4 +
.../iceberg/test_iceberg_write_default.groovy | 9 +
15 files changed, 1616 insertions(+)
diff --git a/be/test/format/parquet/parquet_read_lines.cpp
b/be/test/format/parquet/parquet_read_lines.cpp
index 835ae04bdf1..2769b94531c 100644
--- a/be/test/format/parquet/parquet_read_lines.cpp
+++ b/be/test/format/parquet/parquet_read_lines.cpp
@@ -42,6 +42,7 @@
#include "exprs/vexpr_context.h"
#include "format/orc/vorc_reader.h"
#include "format/parquet/vparquet_reader.h"
+#include "format/table/iceberg_scan_semantics.h"
#include "gtest/gtest_pred_impl.h"
#include "io/fs/local_file_system.h"
#include "orc/sargs/SearchArgument.hh"
@@ -322,4 +323,150 @@ TEST_F(ParquetReadLinesTest, test4) {
read_parquet_lines(numeric_types, types, read_lines, block_dump);
}
+TEST_F(ParquetReadLinesTest,
iceberg_row_id_fetch_materializes_readded_missing_column) {
+ TDescriptorTable thrift_desc;
+ TTableDescriptor table_desc;
+ table_desc.__set_id(0);
+ table_desc.__set_tableType(TTableType::OLAP_TABLE);
+ table_desc.__set_numCols(0);
+ table_desc.__set_numClusteringCols(0);
+ thrift_desc.tableDescriptors.push_back(table_desc);
+ thrift_desc.__isset.tableDescriptors = true;
+
+ const std::vector<std::tuple<std::string, int32_t, TPrimitiveType::type>>
columns {
+ {"new_name", 2, TPrimitiveType::STRING},
+ {"data", 3, TPrimitiveType::STRING},
+ {"id", 4, TPrimitiveType::INT},
+ };
+ for (size_t index = 0; index < columns.size(); ++index) {
+ const auto& [name, field_id, primitive_type] = columns[index];
+ TTypeNode type_node;
+ type_node.__set_type(TTypeNodeType::SCALAR);
+ TScalarType scalar_type;
+ scalar_type.__set_type(primitive_type);
+ type_node.__set_scalar_type(scalar_type);
+ TTypeDesc type;
+ type.types.push_back(type_node);
+
+ TSlotDescriptor slot;
+ slot.__set_id(cast_set<int32_t>(index));
+ slot.__set_parent(0);
+ slot.__set_slotType(type);
+ slot.__set_columnPos(cast_set<int32_t>(index));
+ slot.__set_byteOffset(0);
+ slot.__set_nullIndicatorByte(0);
+ slot.__set_nullIndicatorBit(cast_set<int32_t>(index));
+ slot.__set_colName(name);
+ slot.__set_slotIdx(cast_set<int32_t>(index));
+ slot.__set_isMaterialized(true);
+ slot.__set_col_unique_id(field_id);
+ thrift_desc.slotDescriptors.push_back(slot);
+ }
+ thrift_desc.__isset.slotDescriptors = true;
+
+ TTupleDescriptor tuple;
+ tuple.__set_id(0);
+ tuple.__set_byteSize(16);
+ tuple.__set_numNullBytes(1);
+ tuple.__set_tableId(0);
+ tuple.__isset.tableId = true;
+ thrift_desc.tupleDescriptors.push_back(tuple);
+
+ ObjectPool object_pool;
+ DescriptorTbl* desc_tbl = nullptr;
+ ASSERT_TRUE(DescriptorTbl::create(&object_pool, thrift_desc,
&desc_tbl).ok());
+ auto* tuple_desc =
const_cast<TupleDescriptor*>(desc_tbl->get_tuple_descriptor(0));
+ ASSERT_NE(tuple_desc, nullptr);
+
+ const auto make_field = [](const std::string& name, int32_t field_id,
+ TPrimitiveType::type primitive_type, bool
optional) {
+ auto field = std::make_shared<schema::external::TField>();
+ field->__set_name(name);
+ field->__set_id(field_id);
+ field->__set_is_optional(optional);
+ TColumnType type;
+ type.__set_type(primitive_type);
+ field->__set_type(type);
+ schema::external::TFieldPtr field_ptr;
+ field_ptr.__set_field_ptr(std::move(field));
+ return field_ptr;
+ };
+ schema::external::TStructField root;
+ root.__set_fields({make_field("new_new_id", 1, TPrimitiveType::INT, false),
+ make_field("new_name", 2, TPrimitiveType::STRING, true),
+ make_field("data", 3, TPrimitiveType::STRING, false),
+ make_field("id", 4, TPrimitiveType::INT, true)});
+ schema::external::TSchema schema;
+ schema.__set_schema_id(4);
+ schema.__set_root_field(root);
+
+ TFileScanRangeParams scan_params;
+ scan_params.__set_file_type(TFileType::FILE_LOCAL);
+ scan_params.__set_format_type(TFileFormatType::FORMAT_PARQUET);
+
scan_params.__set_num_of_columns_from_file(cast_set<int32_t>(columns.size()));
+ scan_params.__set_current_schema_id(4);
+ scan_params.__set_history_schema_info({schema});
+
scan_params.__set_iceberg_scan_semantics_version(ICEBERG_SCAN_SEMANTICS_VERSION_2);
+ for (size_t index = 0; index < columns.size(); ++index) {
+ const auto* slot = tuple_desc->slots()[index];
+ TFileScanSlotInfo slot_info;
+ slot_info.__set_slot_id(slot->id());
+ slot_info.__set_is_file_slot(true);
+ scan_params.required_slots.push_back(slot_info);
+ scan_params.default_value_of_src_slot.emplace(slot->id(), TExpr {});
+ scan_params.column_idxs.push_back(cast_set<int32_t>(index + 1));
+ scan_params.slot_name_to_schema_pos.emplace(slot->col_name(),
cast_set<int32_t>(index + 1));
+ }
+ scan_params.__isset.required_slots = true;
+ scan_params.__isset.default_value_of_src_slot = true;
+ scan_params.__isset.column_idxs = true;
+ scan_params.__isset.slot_name_to_schema_pos = true;
+
+ const std::string path =
+
"./docker/thirdparties/docker-compose/iceberg/scripts/preinstalled_data/iceberg/"
+ "equality_delete_par_1/data/"
+ "00000-0-bd4d0a30-cdf6-48d7-933d-91e860870eb9-00001.parquet";
+ io::FileReaderSPtr file_reader;
+ ASSERT_TRUE(io::global_local_filesystem()->open_file(path,
&file_reader).ok());
+ TFileRangeDesc range;
+ range.__set_path(path);
+ range.__set_start_offset(0);
+ range.__set_size(file_reader->size());
+ range.__set_file_size(file_reader->size());
+ range.__set_format_type(TFileFormatType::FORMAT_PARQUET);
+ TTableFormatFileDesc table_format;
+ table_format.__set_table_format_type("iceberg");
+ table_format.__set_iceberg_params(TIcebergFileDesc {});
+ range.__set_table_format_params(table_format);
+
+ RuntimeState runtime_state {TQueryOptions(), TQueryGlobals()};
+ runtime_state.set_desc_tbl(desc_tbl);
+ std::unordered_map<std::string, int> colname_to_slot_id;
+ Block block;
+ for (const auto* slot : tuple_desc->slots()) {
+ colname_to_slot_id.emplace(slot->col_name(), slot->id());
+ block.insert({slot->get_empty_mutable_column(), slot->type(),
slot->col_name()});
+ }
+ RuntimeProfile profile("ExternalRowIDFetcher");
+ auto scanner = FileScanner::create_unique(&runtime_state, &profile,
&scan_params,
+ &colname_to_slot_id, tuple_desc);
+ ASSERT_TRUE(scanner->prepare_for_read_lines(range).ok());
+ ExternalFileMappingInfo external_info(0, range, false);
+ int64_t init_reader_ms = 0;
+ int64_t get_block_ms = 0;
+ const auto status = scanner->read_lines_from_range(range, {0}, &block,
external_info,
+ &init_reader_ms,
&get_block_ms);
+ ASSERT_TRUE(status.ok()) << status;
+ ASSERT_EQ(block.rows(), 1);
+ EXPECT_EQ(block.get_by_position(block.get_position_by_name("new_name"))
+ .column->get_data_at(0)
+ .to_string(),
+ "bob");
+ EXPECT_EQ(block.get_by_position(block.get_position_by_name("data"))
+ .column->get_data_at(0)
+ .to_string(),
+ "e");
+
EXPECT_TRUE(block.get_by_position(block.get_position_by_name("id")).column->is_null_at(0));
+}
+
} // namespace doris
diff --git a/be/test/format/transformer/vorc_transformer_test.cpp
b/be/test/format/transformer/vorc_transformer_test.cpp
index 5eeeef7fa45..a4af867cd33 100644
--- a/be/test/format/transformer/vorc_transformer_test.cpp
+++ b/be/test/format/transformer/vorc_transformer_test.cpp
@@ -631,7 +631,83 @@ TEST_F(VOrcTransformerTest,
SkipsInvalidBinaryChildrenHiddenByNullableCollection
block.insert({std::move(nullable_array), nullable_array_type,
"uuid_array"});
block.insert({std::move(nullable_map), nullable_map_type, "binary_map"});
ASSERT_TRUE(transformer.write(block).ok());
+
+ auto make_visible_block = [&](const std::string& array_uuid, const
std::string& map_uuid,
+ const std::string& map_fixed) {
+ auto visible_array_elements = ColumnString::create();
+ visible_array_elements->insert_data(array_uuid.data(),
array_uuid.size());
+ auto visible_array_offsets = ColumnArray::ColumnOffsets::create();
+ visible_array_offsets->get_data().push_back(1);
+ auto visible_array = ColumnNullable::create(
+
ColumnArray::create(ColumnNullable::create(std::move(visible_array_elements),
+
ColumnUInt8::create(1, 0)),
+ std::move(visible_array_offsets)),
+ ColumnUInt8::create(1, 0));
+
+ auto visible_map_keys = ColumnString::create();
+ visible_map_keys->insert_data(map_uuid.data(), map_uuid.size());
+ auto visible_map_values = ColumnString::create();
+ visible_map_values->insert_data(map_fixed.data(), map_fixed.size());
+ auto visible_map_offsets = ColumnArray::ColumnOffsets::create();
+ visible_map_offsets->get_data().push_back(1);
+ auto visible_map = ColumnNullable::create(
+ ColumnMap::create(std::move(visible_map_keys),
std::move(visible_map_values),
+ std::move(visible_map_offsets)),
+ ColumnUInt8::create(1, 0));
+
+ Block visible_block;
+ visible_block.insert({std::move(visible_array), nullable_array_type,
"uuid_array"});
+ visible_block.insert({std::move(visible_map), nullable_map_type,
"binary_map"});
+ return visible_block;
+ };
+
+ const std::string valid_uuid = "00112233-4455-6677-8899-aabbccddeeff";
+ Block invalid_uuid_block = make_visible_block("invalid-visible-uuid",
valid_uuid, "ABCD");
+ Status invalid_uuid = transformer.write(invalid_uuid_block);
+ ASSERT_FALSE(invalid_uuid.ok());
+ EXPECT_NE(invalid_uuid.to_string().find("Invalid UUID string length"),
std::string::npos);
+
+ Block invalid_fixed_block = make_visible_block(valid_uuid, valid_uuid,
"BAD");
+ Status invalid_fixed = transformer.write(invalid_fixed_block);
+ ASSERT_FALSE(invalid_fixed.ok());
+ EXPECT_NE(invalid_fixed.to_string().find("FIXED[4]"), std::string::npos);
ASSERT_TRUE(transformer.close().ok());
+
+ io::FileReaderSPtr file_reader;
+ ASSERT_TRUE(_fs->open_file(_file_path, &file_reader).ok());
+ auto input_stream = std::make_unique<ORCFileInputStream>(
+ _file_path, file_reader, nullptr, nullptr, 8L * 1024L * 1024L, 1L
* 1024L * 1024L);
+ auto reader = orc::createReader(std::move(input_stream),
orc::ReaderOptions());
+ auto row_reader = reader->createRowReader();
+ auto row_batch = row_reader->createRowBatch(2);
+ ASSERT_TRUE(row_reader->next(*row_batch));
+ const auto& root = assert_cast<const orc::StructVectorBatch&>(*row_batch);
+ ASSERT_EQ(root.numElements, 2);
+ const auto& array_batch = assert_cast<const
orc::ListVectorBatch&>(*root.fields[0]);
+ const auto& array_values = assert_cast<const
orc::StringVectorBatch&>(*array_batch.elements);
+ const auto& map_batch = assert_cast<const
orc::MapVectorBatch&>(*root.fields[1]);
+ const auto& map_keys_batch = assert_cast<const
orc::StringVectorBatch&>(*map_batch.keys);
+ const auto& map_values_batch = assert_cast<const
orc::StringVectorBatch&>(*map_batch.elements);
+ ASSERT_TRUE(array_batch.hasNulls);
+ EXPECT_FALSE(array_batch.notNull[0]);
+ EXPECT_TRUE(array_batch.notNull[1]);
+ EXPECT_EQ(array_batch.offsets[0], 0);
+ EXPECT_EQ(array_batch.offsets[1], 0);
+ EXPECT_EQ(array_batch.offsets[2], 1);
+ ASSERT_TRUE(map_batch.hasNulls);
+ EXPECT_FALSE(map_batch.notNull[0]);
+ EXPECT_TRUE(map_batch.notNull[1]);
+ EXPECT_EQ(map_batch.offsets[0], 0);
+ EXPECT_EQ(map_batch.offsets[1], 0);
+ EXPECT_EQ(map_batch.offsets[2], 1);
+
+ const std::array<uint8_t, 16> expected_uuid = {0x00, 0x11, 0x22, 0x33,
0x44, 0x55, 0x66, 0x77,
+ 0x88, 0x99, 0xaa, 0xbb,
0xcc, 0xdd, 0xee, 0xff};
+ ASSERT_EQ(array_values.length[0], expected_uuid.size());
+ EXPECT_EQ(0, std::memcmp(array_values.data[0], expected_uuid.data(),
expected_uuid.size()));
+ ASSERT_EQ(map_keys_batch.length[0], expected_uuid.size());
+ EXPECT_EQ(0, std::memcmp(map_keys_batch.data[0], expected_uuid.data(),
expected_uuid.size()));
+ EXPECT_EQ(std::string_view(map_values_batch.data[0],
map_values_batch.length[0]), "ABCD");
}
TEST_F(VOrcTransformerTest,
DetectsOnlyFieldsThatNeedIcebergBinaryNormalization) {
diff --git a/be/test/format_v2/table/iceberg_reader_test.cpp
b/be/test/format_v2/table/iceberg_reader_test.cpp
index adcecd1c3ad..d905b8cb726 100644
--- a/be/test/format_v2/table/iceberg_reader_test.cpp
+++ b/be/test/format_v2/table/iceberg_reader_test.cpp
@@ -4440,6 +4440,78 @@ TEST(IcebergV2ReaderTest,
IcebergMissingWholeStructEqualityKeyMaterializesDefaul
}
}
+TEST(IcebergV2ReaderTest,
IcebergProjectedStructDoesNotReuseReaddedEqualityKeyByName) {
+ const auto run_case = [](FileFormat file_format) {
+ const bool is_parquet = file_format == FileFormat::PARQUET;
+ const std::string format_name = is_parquet ? "parquet" : "orc";
+ const auto test_dir = std::filesystem::temp_directory_path() /
+ ("doris_v2_missing_reused_equality_key_" +
format_name);
+ std::filesystem::remove_all(test_dir);
+ std::filesystem::create_directories(test_dir);
+ const auto file_path = (test_dir / ("split." + format_name)).string();
+ const auto delete_file_path = (test_dir / ("equality-delete." +
format_name)).string();
+ if (is_parquet) {
+ write_nested_equality_parquet_file(file_path, {1, 2, 3}, {7, 8, 9},
+ {false, false, false}, true,
"k", 4);
+ write_nested_equality_parquet_file(delete_file_path, {}, {7},
{false}, true, "k", 3);
+ } else {
+ write_nested_equality_orc_file(file_path, {1, 2, 3}, {7, 8, 9},
{false, false, false},
+ "k", 4);
+ write_nested_equality_orc_file(delete_file_path, {}, {7}, {false},
"k", 3);
+ }
+
+ auto payload = external_struct_schema_field(
+ "payload", 1,
+ {external_schema_field("k", 4, {}, std::nullopt,
+
external_primitive_type(TPrimitiveType::INT), false, true),
+ external_schema_field("k", 3, {}, std::nullopt,
+
external_primitive_type(TPrimitiveType::INT), false, true)},
+ true);
+ std::vector<schema::external::TSchema> history = {
+ external_schema(-1, {external_schema_field("id", 0),
std::move(payload)})};
+
+ const auto int_type = std::make_shared<DataTypeInt32>();
+ std::vector<ColumnDefinition> projected_columns;
+ projected_columns.push_back(make_table_column(0, "id", int_type));
+ auto current_key = make_table_column(4, "k", int_type);
+ auto payload_type = std::make_shared<DataTypeStruct>(DataTypes
{int_type}, Strings {"k"});
+ auto projected_payload = make_table_column(1, "payload", payload_type);
+ projected_payload.children = {std::move(current_key)};
+ projected_columns.push_back(std::move(projected_payload));
+ auto scan_params = make_local_scan_params(file_format);
+
scan_params.__set_iceberg_scan_semantics_version(ICEBERG_SCAN_SEMANTICS_VERSION_2);
+ scan_params.__set_current_schema_id(-1);
+ scan_params.__set_history_schema_info(std::move(history));
+
+ RuntimeProfile profile("test_profile");
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ io::FileReaderStats file_reader_stats;
+ io::FileCacheStatistics file_cache_stats;
+ auto io_ctx = make_io_context(&file_reader_stats, &file_cache_stats);
+ ShardedKVCache cache(1);
+ doris::format::iceberg::IcebergTableReader reader;
+ init_iceberg_reader(&reader, projected_columns, &scan_params, io_ctx,
&state, &profile,
+ file_format);
+
+ auto split_options = build_split_options(file_path);
+ split_options.cache = &cache;
+ split_options.current_split_format = file_format;
+ const auto thrift_file_format =
+ is_parquet ? TFileFormatType::FORMAT_PARQUET :
TFileFormatType::FORMAT_ORC;
+
split_options.current_range.__set_table_format_params(make_iceberg_table_format_desc(
+ file_path,
+ {make_iceberg_equality_delete_file(delete_file_path, {3},
thrift_file_format)}, 3));
+ ASSERT_TRUE(reader.prepare_split(split_options).ok());
+ EXPECT_EQ(read_iceberg_ids(&reader, projected_columns),
std::vector<int32_t>({1, 2, 3}));
+ ASSERT_TRUE(reader.close().ok());
+ std::filesystem::remove_all(test_dir);
+ };
+
+ for (const auto file_format : {FileFormat::PARQUET, FileFormat::ORC}) {
+ run_case(file_format);
+ }
+}
+
// A schema carrier may contain a dropped and a re-added child with the same
name. Keep the
// Parquet/ORC setup identical so both readers prove that equality-key
reconstruction follows the
// exact field-id path instead of attaching both same-name children to one
synthetic struct slot.
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 4b28492c4ff..877ecfb65e1 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
@@ -48,6 +48,7 @@ import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import java.io.FileNotFoundException;
+import java.nio.ByteBuffer;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
@@ -56,6 +57,7 @@ import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
+import java.util.UUID;
/**
* Characterization tests for {@link IcebergConnectorMetadata}, pinning the
read-path behavior after
@@ -907,6 +909,39 @@ public class IcebergConnectorMetadataTest {
Assertions.assertTrue(name.isSourceNullable());
}
+ @Test
+ public void getTableSchemaLeavesIcebergDefaultsOutOfCatalogColumns() {
+ // Initial defaults reach BE through the scan dictionary and write
defaults are pinned per write, so the
+ // cached catalog column carries neither: DESCRIBE and SHOW CREATE
must not expose a possibly stale value.
+ RecordingIcebergCatalogOps ops = new RecordingIcebergCatalogOps();
+ Schema defaultsSchema = new Schema(
+
Types.NestedField.optional("added_column").withId(1).ofType(Types.IntegerType.get())
+ .withInitialDefault(7).withWriteDefault(9).build(),
+ Types.NestedField.optional("added_timestamp").withId(2)
+ .ofType(Types.TimestampType.withoutZone())
+ .withInitialDefault(1_704_067_200_123_456L).build(),
+ Types.NestedField.optional("added_timestamptz").withId(3)
+ .ofType(Types.TimestampType.withZone())
+ .withInitialDefault(1_704_067_200_123_456L).build(),
+
Types.NestedField.optional("added_uuid").withId(4).ofType(Types.UUIDType.get())
+
.withInitialDefault(UUID.fromString("00000000-0000-0000-0000-000000000000")).build(),
+
Types.NestedField.optional("added_binary").withId(5).ofType(Types.BinaryType.get())
+ .withInitialDefault(ByteBuffer.wrap(new byte[] {0, 1,
2, (byte) 0xFF})).build(),
+
Types.NestedField.optional("added_fixed").withId(6).ofType(Types.FixedType.ofLength(4))
+ .withInitialDefault(ByteBuffer.wrap(new byte[] {3, 2,
1, 0})).build());
+ ops.table = new FakeIcebergTable(
+ "t1", defaultsSchema, PartitionSpec.unpartitioned(),
+ "s3://bucket/db1/t1",
Collections.singletonMap("format-version", "3"));
+
+ List<ConnectorColumn> columns =
+ metadataWith(ops).getTableSchema(null, new
IcebergTableHandle("db1", "t1")).getColumns();
+
+ for (ConnectorColumn column : columns.subList(0,
defaultsSchema.columns().size())) {
+ Assertions.assertNull(column.getDefaultValue(), column.getName());
+ Assertions.assertNull(column.getDefaultValueSql(),
column.getName());
+ }
+ }
+
@Test
public void getTableSchemaCopiesTablePropertiesAndLocation() {
RecordingIcebergCatalogOps ops = new RecordingIcebergCatalogOps();
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorTransactionTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorTransactionTest.java
index c0b42a634dc..acc8400b2bc 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorTransactionTest.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorTransactionTest.java
@@ -48,6 +48,7 @@ import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
import org.apache.iceberg.Snapshot;
import org.apache.iceberg.Table;
+import org.apache.iceberg.TableOperations;
import org.apache.iceberg.Transaction;
import org.apache.iceberg.catalog.Namespace;
import org.apache.iceberg.catalog.TableIdentifier;
@@ -67,6 +68,7 @@ import org.junit.jupiter.api.Test;
import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.nio.ByteBuffer;
import java.util.ArrayList;
@@ -78,8 +80,10 @@ import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
+import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
@@ -284,6 +288,59 @@ public class IcebergConnectorTransactionTest {
return catalog.loadTable(id).currentSnapshot();
}
+ /** Writer metadata pinned from {@code table} at bind time, as the write
planner hands it to beginWrite. */
+ private static IcebergWriteContext pinnedCtx(WriteOperation operation,
Table table) {
+ return new IcebergWriteContext(operation, operation ==
WriteOperation.OVERWRITE, Collections.emptyMap(),
+ Optional.empty(), -1L, false,
+ IcebergWriteSchemaContext.create(table, table.name(),
Optional.empty(), false, false));
+ }
+
+ /** {@code table} behind operations that count metadata refreshes; opening
an SDK transaction refreshes once. */
+ private static Table countingRefreshes(Table table, AtomicInteger
refreshes) {
+ TableOperations operations = ((HasTableOperations) table).operations();
+ return new BaseTable((TableOperations)
Proxy.newProxyInstance(TableOperations.class.getClassLoader(),
+ new Class<?>[] {TableOperations.class}, (proxy, method, args)
-> {
+ if ("refresh".equals(method.getName())) {
+ refreshes.incrementAndGet();
+ }
+ return invokeUnwrapped(operations, method, args);
+ }), table.name());
+ }
+
+ /**
+ * A write table over {@code operations} that runs {@code
concurrentChange} once, while the SDK transaction stages
+ * its snapshot: after the commit-time preflight, before the final
metadata swap replays the staged update. The
+ * change may repoint {@code operations}, as a catalog that resolves the
name to a replacement table does.
+ */
+ private static Table stagingTable(String name,
AtomicReference<TableOperations> operations,
+ Runnable concurrentChange) {
+ return new BaseTable(stagingOperations(operations, concurrentChange,
new AtomicBoolean()), name);
+ }
+
+ private static TableOperations
stagingOperations(AtomicReference<TableOperations> operations,
+ Runnable concurrentChange, AtomicBoolean changed) {
+ return (TableOperations)
Proxy.newProxyInstance(TableOperations.class.getClassLoader(),
+ new Class<?>[] {TableOperations.class}, (proxy, method, args)
-> {
+ if ("temp".equals(method.getName())) {
+ // An SDK transaction stages its pending updates on
temporary operations.
+ TableOperations temp = (TableOperations)
invokeUnwrapped(operations.get(), method, args);
+ return stagingOperations(new AtomicReference<>(temp),
concurrentChange, changed);
+ }
+ if ("newSnapshotId".equals(method.getName()) &&
changed.compareAndSet(false, true)) {
+ concurrentChange.run();
+ }
+ return invokeUnwrapped(operations.get(), method, args);
+ });
+ }
+
+ private static Object invokeUnwrapped(Object target, Method method,
Object[] args) throws Throwable {
+ try {
+ return method.invoke(target, args);
+ } catch (InvocationTargetException e) {
+ throw e.getCause();
+ }
+ }
+
// ─────────────────── addCommitData: 14-field TBinaryProtocol round-trip
(T03, regression) ───────────────────
@Test
@@ -662,6 +719,43 @@ public class IcebergConnectorTransactionTest {
"the append must land on branch b1, not on main");
}
+ @Test
+ public void insertCommitUsesStatementPinnedWriterMetadata() {
+ // Files written against the pinned writer metadata are committed with
that spec, sort order and format,
+ // even though a spec, sort-order and format change the fences admit
landed before beginWrite.
+ InMemoryCatalog catalog = freshCatalog();
+ TableIdentifier id = TableIdentifier.of("db1", "t1");
+ Schema schema = new Schema(
+ Types.NestedField.required(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "age", Types.IntegerType.get()));
+ Table table = catalog.createTable(id, schema,
PartitionSpec.builderFor(schema).identity("id").build(),
+ props("format-version", "2", "write.format.default", "orc"));
+ table.replaceSortOrder().asc("age").commit();
+ int pinnedSpecId = table.spec().specId();
+ int pinnedSortOrderId = table.sortOrder().orderId();
+ IcebergWriteContext ctx = pinnedCtx(WriteOperation.INSERT, table);
+ table.updateSpec().removeField("id").commit();
+ table.replaceSortOrder().desc("id").commit();
+ table.updateProperties().set("write.format.default",
"parquet").commit();
+ IcebergConnectorTransaction txn = txnFor(opsReturning(table), new
RecordingConnectorContext());
+
+ txn.beginWrite(SESSION, "db1", "t1", ctx);
+ txn.addCommitData(commitBytes(
+ dataFileItem("s3://b/db1/t1/id=7/f1.orc", 1L, 1024L,
Collections.singletonList("7"))));
+ txn.commit();
+
+ Table committed = catalog.loadTable(id);
+ List<DataFile> added = new ArrayList<>();
+
committed.currentSnapshot().addedDataFiles(committed.io()).forEach(added::add);
+ Assertions.assertEquals(1, added.size());
+ DataFile file = added.get(0);
+ Assertions.assertNotEquals(pinnedSpecId, committed.spec().specId(),
"the live spec must have moved on");
+ Assertions.assertEquals(pinnedSpecId, file.specId());
+ Assertions.assertEquals(Integer.valueOf(7), file.partition().get(0,
Integer.class));
+ Assertions.assertEquals(Integer.valueOf(pinnedSortOrderId),
file.sortOrderId());
+ Assertions.assertEquals(FileFormat.ORC, file.format());
+ }
+
@Test
public void overwriteDynamicReplacesPartitions() {
InMemoryCatalog catalog = freshCatalog();
@@ -1144,6 +1238,57 @@ public class IcebergConnectorTransactionTest {
"a stale static overwrite must fail at begin before it can
degrade to an always-true filter");
}
+ @Test
+ public void
staticOverwriteRejectsConcurrentCurrentSpecReplacementBeforeOpening() {
+ // The pinned spec is still retained with its definition, but it is no
longer the current spec: an overwrite
+ // must fail before an SDK transaction opens instead of replacing data
through a superseded spec.
+ InMemoryCatalog catalog = freshCatalog();
+ TableIdentifier id = TableIdentifier.of("db1", "t1");
+ Table table = catalog.createTable(id, PART_SCHEMA,
+
PartitionSpec.builderFor(PART_SCHEMA).identity("region").build(),
props("format-version", "2"));
+ IcebergWriteContext boundContext = overwriteStaticCtx(table,
Collections.singletonMap("region", "us"));
+ table.updateSpec().removeField("region").addField("id").commit();
+ AtomicInteger refreshes = new AtomicInteger();
+ IcebergConnectorTransaction txn = txnFor(
+ opsReturning(countingRefreshes(catalog.loadTable(id),
refreshes)), new RecordingConnectorContext());
+
+ DorisConnectorException ex =
Assertions.assertThrows(DorisConnectorException.class,
+ () -> txn.beginWrite(SESSION, "db1", "t1", boundContext));
+ Assertions.assertTrue(ex.getMessage().contains("current partition spec
changed during overwrite planning"),
+ ex.getMessage());
+ Assertions.assertTrue(ex.getMessage().contains("retry the statement"),
ex.getMessage());
+ Assertions.assertEquals(0, refreshes.get(), "the overwrite must be
rejected before a transaction opens");
+ Assertions.assertNull(reloadCurrentSnapshot(catalog, id));
+ }
+
+ @Test
+ public void
dynamicOverwriteRejectsPartitionedToUnpartitionedSpecDriftAtCommit() {
+ // Dropping the last partition field after beginWrite must fail the
overwrite at commit, whether or not the
+ // statement wrote files; neither path may replace or clear data
through the superseded partitioned spec.
+ assertDynamicOverwriteRejectsUnpartitionedSpecDrift(true);
+ assertDynamicOverwriteRejectsUnpartitionedSpecDrift(false);
+ }
+
+ private static void
assertDynamicOverwriteRejectsUnpartitionedSpecDrift(boolean writesFiles) {
+ InMemoryCatalog catalog = freshCatalog();
+ TableIdentifier id = TableIdentifier.of("db1", "t1");
+ Table table = catalog.createTable(id, PART_SCHEMA,
+
PartitionSpec.builderFor(PART_SCHEMA).identity("region").build(),
props("format-version", "2"));
+ IcebergConnectorTransaction txn = txnFor(opsReturning(table), new
RecordingConnectorContext());
+ txn.beginWrite(SESSION, "db1", "t1",
pinnedCtx(WriteOperation.OVERWRITE, table));
+ if (writesFiles) {
+ txn.addCommitData(commitBytes(dataFileItem(
+ "s3://b/db1/t1/region=us/f1.parquet", 1L, 1024L,
Collections.singletonList("us"))));
+ }
+
+ table.updateSpec().removeField("region").commit();
+
+ DorisConnectorException ex =
Assertions.assertThrows(DorisConnectorException.class, txn::commit);
+ Assertions.assertTrue(ex.getMessage().contains("current partition spec
changed"), ex.getMessage());
+ Assertions.assertTrue(ex.getMessage().contains("retry the statement"),
ex.getMessage());
+ Assertions.assertNull(reloadCurrentSnapshot(catalog, id),
"writesFiles=" + writesFiles);
+ }
+
@Test
public void overwriteStaticNullBinaryKeyReplacesOnlyTheNullPartition() {
// A NULL static key must become key IS NULL for full and hybrid
specs; matching bytes or text
@@ -1386,6 +1531,139 @@ public class IcebergConnectorTransactionTest {
"a replacement exposed by newTransaction refresh must not
become the write baseline");
}
+ @Test
+ public void
beginWriteRejectsPinnedSchemaSkewBeforeOpeningInsertOrMergeTransaction() {
+ // A schema commit between binding and beginWrite is rejected from the
loaded metadata, before an SDK
+ // transaction refreshes the table and could become the conflict
baseline of the stale plan.
+ InMemoryCatalog catalog = freshCatalog();
+ TableIdentifier id = TableIdentifier.of("db1", "t1");
+ Table table = catalog.createTable(id, SCHEMA,
PartitionSpec.unpartitioned(), props("format-version", "2"));
+ List<IcebergWriteContext> contexts = Arrays.asList(
+ pinnedCtx(WriteOperation.INSERT, table),
pinnedCtx(WriteOperation.MERGE, table));
+ table.updateSchema().addColumn("added_after_binding",
Types.IntegerType.get()).commit();
+
+ for (IcebergWriteContext ctx : contexts) {
+ AtomicInteger refreshes = new AtomicInteger();
+ IcebergConnectorTransaction txn = txnFor(
+ opsReturning(countingRefreshes(catalog.loadTable(id),
refreshes)),
+ new RecordingConnectorContext());
+ DorisConnectorException ex =
Assertions.assertThrows(DorisConnectorException.class,
+ () -> txn.beginWrite(SESSION, "db1", "t1", ctx));
+ Assertions.assertTrue(ex.getMessage().contains("schema changed
during write planning"), ex.getMessage());
+ Assertions.assertTrue(ex.getMessage().contains("retry the
statement"), ex.getMessage());
+ Assertions.assertEquals(0, refreshes.get(),
ctx.getWriteOperation() + " opened a transaction");
+ }
+ Assertions.assertNull(reloadCurrentSnapshot(catalog, id));
+ }
+
+ @Test
+ public void beginWriteRejectsRecreatedTableForInsertOverwriteAndMerge() {
+ // A same-name DROP/CREATE after binding fails INSERT, OVERWRITE and
MERGE on the pinned table UUID before a
+ // transaction opens, so the replacement never becomes the baseline
for rows bound against the old table.
+ InMemoryCatalog catalog = freshCatalog();
+ TableIdentifier id = TableIdentifier.of("db1", "t1");
+ Table original = catalog.createTable(id, SCHEMA,
PartitionSpec.unpartitioned(), props("format-version", "2"));
+ List<IcebergWriteContext> contexts =
Arrays.asList(pinnedCtx(WriteOperation.INSERT, original),
+ pinnedCtx(WriteOperation.OVERWRITE, original),
pinnedCtx(WriteOperation.MERGE, original));
+ catalog.dropTable(id, false);
+ Table replacement = catalog.createTable(id, SCHEMA,
PartitionSpec.unpartitioned(),
+ props("format-version", "2"));
+
+ for (IcebergWriteContext ctx : contexts) {
+ AtomicInteger refreshes = new AtomicInteger();
+ IcebergConnectorTransaction txn = txnFor(
+ opsReturning(countingRefreshes(replacement, refreshes)),
new RecordingConnectorContext());
+ DorisConnectorException ex =
Assertions.assertThrows(DorisConnectorException.class,
+ () -> txn.beginWrite(SESSION, "db1", "t1", ctx));
+ Assertions.assertTrue(ex.getMessage().contains("identity
changed"), ex.getMessage());
+ Assertions.assertEquals(0, refreshes.get(),
ctx.getWriteOperation() + " opened a transaction");
+ }
+ Assertions.assertNull(reloadCurrentSnapshot(catalog, id));
+ }
+
+ @Test
+ public void
commitReplayRejectsRequiredSchemaChangeStagedBeforeTheMetadataSwap() {
+ // A required column added while the commit stages its snapshot passes
the commit preflight. The final
+ // metadata swap re-validates the replayed update and must reject it
rather than publish files without it.
+ InMemoryCatalog catalog = freshCatalog();
+ TableIdentifier id = TableIdentifier.of("db1", "t1");
+ Table table = catalog.createTable(id, SCHEMA,
PartitionSpec.unpartitioned(), props("format-version", "2"));
+ Table writeTable = stagingTable(table.name(),
+ new AtomicReference<>(((HasTableOperations)
table).operations()),
+ () -> addRequiredColumn(catalog, id,
"required_after_staging"));
+ IcebergConnectorTransaction txn = txnFor(opsReturning(writeTable), new
RecordingConnectorContext());
+
+ txn.beginWrite(SESSION, "db1", "t1", pinnedCtx(WriteOperation.INSERT,
table));
+ txn.addCommitData(commitBytes(dataFileItem("s3://b/db1/t1/f1.parquet",
1L, 1024L)));
+ DorisConnectorException ex =
Assertions.assertThrows(DorisConnectorException.class, txn::commit);
+
+ assertRejectedAfterStagedSchemaChange(catalog, id, ex);
+ }
+
+ @Test
+ public void
mergeCommitReplayRejectsRequiredSchemaChangeStagedBeforeTheMetadataSwap() {
+ // The same metadata-swap fence covers the RowDelta a MERGE stages for
its data and position-delete files.
+ InMemoryCatalog catalog = freshCatalog();
+ TableIdentifier id = TableIdentifier.of("db1", "t1");
+ Table table = catalog.createTable(id, SCHEMA,
PartitionSpec.unpartitioned(), props("format-version", "2"));
+ Table writeTable = stagingTable(table.name(),
+ new AtomicReference<>(((HasTableOperations)
table).operations()),
+ () -> addRequiredColumn(catalog, id,
"required_after_staging"));
+ IcebergConnectorTransaction txn = txnFor(opsReturning(writeTable), new
RecordingConnectorContext());
+
+ txn.beginWrite(SESSION, "db1", "t1", pinnedCtx(WriteOperation.MERGE,
table));
+
txn.addCommitData(commitBytes(dataFileItem("s3://b/db1/t1/new.parquet", 2L,
1024L)));
+ txn.addCommitData(commitBytes(
+ positionDeleteItem("s3://b/db1/t1/del.parquet", 2L,
"s3://b/db1/t1/old.parquet")));
+ DorisConnectorException ex =
Assertions.assertThrows(DorisConnectorException.class, txn::commit);
+
+ assertRejectedAfterStagedSchemaChange(catalog, id, ex);
+ }
+
+ private static void addRequiredColumn(InMemoryCatalog catalog,
TableIdentifier id, String column) {
+ catalog.loadTable(id).updateSchema().allowIncompatibleChanges()
+ .addRequiredColumn(column, Types.IntegerType.get()).commit();
+ }
+
+ private static void assertRejectedAfterStagedSchemaChange(InMemoryCatalog
catalog, TableIdentifier id,
+ DorisConnectorException ex) {
+ Assertions.assertTrue(ex.getMessage().contains("schema changed during
write planning"), ex.getMessage());
+ Assertions.assertTrue(ex.getMessage().contains("retry the statement"),
ex.getMessage());
+ Table current = catalog.loadTable(id);
+
Assertions.assertNotNull(current.schema().findField("required_after_staging"),
+ "the schema change must land while the commit stages its
snapshot");
+ Assertions.assertNull(current.currentSnapshot(), "the staged update
must not be published");
+ }
+
+ @Test
+ public void commitRejectsTableReplacementStagedBeforeTheMetadataSwap() {
+ // A same-name DROP/CREATE while the commit stages its snapshot must
not let the staged files land in the
+ // replacement: the metadata swap re-reads the current metadata and
fences the pinned table identity.
+ InMemoryCatalog catalog = freshCatalog();
+ TableIdentifier id = TableIdentifier.of("db1", "t1");
+ Table table = catalog.createTable(id, SCHEMA,
PartitionSpec.unpartitioned(), props("format-version", "2"));
+ UUID originalUuid = table.uuid();
+ AtomicReference<TableOperations> operations =
+ new AtomicReference<>(((HasTableOperations)
table).operations());
+ AtomicReference<Table> replacement = new AtomicReference<>();
+ Table writeTable = stagingTable(table.name(), operations, () -> {
+ catalog.dropTable(id, false);
+ replacement.set(catalog.createTable(id, SCHEMA,
PartitionSpec.unpartitioned(),
+ props("format-version", "2")));
+ // A catalog that does not compare UUIDs on refresh (e.g. a Hadoop
catalog) now resolves the replacement.
+ operations.set(((HasTableOperations)
replacement.get()).operations());
+ });
+ IcebergConnectorTransaction txn = txnFor(opsReturning(writeTable), new
RecordingConnectorContext());
+
+ txn.beginWrite(SESSION, "db1", "t1", pinnedCtx(WriteOperation.INSERT,
table));
+ txn.addCommitData(commitBytes(dataFileItem("s3://b/db1/t1/f1.parquet",
1L, 1024L)));
+ DorisConnectorException ex =
Assertions.assertThrows(DorisConnectorException.class, txn::commit);
+
+ Assertions.assertTrue(ex.getMessage().contains("identity changed"),
ex.getMessage());
+ Assertions.assertNotEquals(originalUuid, replacement.get().uuid());
+ Assertions.assertNull(reloadCurrentSnapshot(catalog, id), "the
replacement must not receive staged files");
+ }
+
// ─────────────────── commit-time conflict-detection validation suite
(T05) ───────────────────
@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 5cbea55cfd4..b0a6866de3a 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
@@ -42,7 +42,9 @@ import org.apache.doris.thrift.TFileRangeDesc;
import org.apache.doris.thrift.TFileScanRangeParams;
import org.apache.doris.thrift.TIcebergDeleteFileDesc;
import org.apache.doris.thrift.TIcebergFileDesc;
+import org.apache.doris.thrift.TPrimitiveType;
import org.apache.doris.thrift.TTableFormatFileDesc;
+import org.apache.doris.thrift.schema.external.TField;
import org.apache.doris.thrift.schema.external.TFieldPtr;
import com.google.common.collect.ImmutableMap;
@@ -62,6 +64,7 @@ import org.apache.iceberg.PartitionData;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
import org.apache.iceberg.Snapshot;
+import org.apache.iceberg.SnapshotSummary;
import org.apache.iceberg.StructLike;
import org.apache.iceberg.SupportsDistributedScanPlanning;
import org.apache.iceberg.Table;
@@ -79,6 +82,10 @@ import org.apache.iceberg.io.InputFile;
import org.apache.iceberg.io.OutputFile;
import org.apache.iceberg.io.StorageCredential;
import org.apache.iceberg.io.SupportsStorageCredentials;
+import org.apache.iceberg.mapping.MappedField;
+import org.apache.iceberg.mapping.MappedFields;
+import org.apache.iceberg.mapping.NameMapping;
+import org.apache.iceberg.mapping.NameMappingParser;
import org.apache.iceberg.metrics.ScanReport;
import org.apache.iceberg.types.Conversions;
import org.apache.iceberg.types.Types;
@@ -109,6 +116,7 @@ import java.util.OptionalLong;
import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.UnaryOperator;
+import java.util.stream.Collectors;
/**
* Tests for {@link IcebergScanPlanProvider}. T01 pinned the capability
constants + that {@code planScan}
@@ -190,6 +198,16 @@ public class IcebergScanPlanProviderTest {
(proxy, method, args) -> method.getName().equals("io") ?
fileIO : invoke(method, table, args));
}
+ private static Table tableWithSchemas(Table table, Schema... schemas) {
+ Map<Integer, Schema> schemasById = new HashMap<>();
+ for (Schema schema : schemas) {
+ schemasById.put(schema.schemaId(), schema);
+ }
+ return (Table) Proxy.newProxyInstance(Table.class.getClassLoader(),
new Class<?>[] {Table.class},
+ (proxy, method, args) -> method.getName().equals("schemas")
+ ? schemasById : invoke(method, table, args));
+ }
+
private static Table serverPlannedTable(Table table, AtomicBoolean
localMetadataAccessed) {
return (Table) Proxy.newProxyInstance(Table.class.getClassLoader(),
new Class<?>[] {Table.class,
SupportsDistributedScanPlanning.class},
@@ -290,6 +308,25 @@ public class IcebergScanPlanProviderTest {
.orElseThrow(() -> new AssertionError("no range ending in " +
suffix));
}
+ /** Round-trip the scan properties onto real params and return the
top-level fields of the dictionary. */
+ private static List<TFieldPtr> dictRootFields(IcebergScanPlanProvider
provider, Map<String, String> props) {
+ TFileScanRangeParams params = new TFileScanRangeParams();
+ provider.populateScanLevelParams(params, props);
+ return params.getHistorySchemaInfo().get(0).getRootField().getFields();
+ }
+
+ private static TField dictFieldById(List<TFieldPtr> fields, int fieldId) {
+ return fields.stream().map(TFieldPtr::getFieldPtr).filter(field ->
field.getId() == fieldId)
+ .findFirst().orElseThrow(() -> new AssertionError("no
dictionary field with ID " + fieldId));
+ }
+
+ /** The rolling-upgrade fence the scan properties carry for the projected
columns, or null when absent. */
+ private static String requiredBackendSemantics(IcebergScanPlanProvider
provider, IcebergTableHandle handle,
+ ConnectorColumnHandle... columns) {
+ return provider.getScanNodeProperties(null, handle,
Arrays.asList(columns), Optional.empty())
+ .get(ScanNodePropertyKeys.REQUIRED_CURRENT_BACKEND_SEMANTICS);
+ }
+
private static RecordingIcebergCatalogOps opsReturning(Table table) {
RecordingIcebergCatalogOps ops = new RecordingIcebergCatalogOps();
ops.table = table;
@@ -1885,6 +1922,130 @@ public class IcebergScanPlanProviderTest {
"a dropped field still referenced by a live equality delete
must remain resolvable by ID");
}
+ @Test
+ public void equalityCarrierKeepsLatestDefinitionOfDroppedEqualityKey()
throws Exception {
+ // BE fills an equality key that an old data file lacks from the key's
carrier entry, so a dropped key must
+ // resolve to its latest historical definition: its last name, its
optionality and its initial default.
+ Schema schema = new Schema(
+ Types.NestedField.required(1, "id", Types.IntegerType.get()),
+
Types.NestedField.optional("k").withId(2).ofType(Types.IntegerType.get())
+ .withInitialDefault(7).build());
+ Table table = createTable("renamed_dropped_eq_key", schema,
PartitionSpec.unpartitioned(),
+ Collections.singletonMap(TableProperties.FORMAT_VERSION, "3"));
+ int keyId = table.schema().findField("k").fieldId();
+ table.newAppend().appendFile(dataFile(table.spec(),
+ "s3://b/db/renamed_dropped_eq_key/f1.parquet", 1024, null,
null)).commit();
+ table.newRowDelta().addDeletes(equalityDeleteFile(
+ "s3://b/db/renamed_dropped_eq_key/eq.parquet",
FileFormat.PARQUET, keyId)).commit();
+ table.updateSchema().renameColumn("k", "k2").commit();
+ table.updateSchema().deleteColumn("k2").commit();
+
+ IcebergScanPlanProvider provider = providerOver(table);
+ Map<String, String> props = provider.getScanNodeProperties(
+ null, new IcebergTableHandle("db1", "renamed_dropped_eq_key"),
+ Collections.singletonList(new IcebergColumnHandle("id",
table.schema().findField("id").fieldId())),
+ Optional.empty());
+
+ TField key = dictFieldById(dictRootFields(provider, props), keyId);
+ Assertions.assertEquals("k2", key.getName());
+ Assertions.assertTrue(key.isIsOptional());
+ Assertions.assertEquals("7", key.getInitialDefaultValue());
+ }
+
+ @Test
+ public void
equalityCarrierRestoresDroppedNestedEqualityKeyInsideItsParent() throws
Exception {
+ // A dropped nested equality key must come back inside the struct that
still exists, after the current
+ // children and with its own initial default, so BE resolves the key's
full field-ID path.
+ Schema schema = new Schema(
+ Types.NestedField.required(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "payload", Types.StructType.of(
+ Types.NestedField.optional(3, "existing",
Types.IntegerType.get()),
+
Types.NestedField.optional("k").withId(4).ofType(Types.IntegerType.get())
+ .withInitialDefault(7).build())));
+ Table table = createTable("dropped_nested_eq_key", schema,
PartitionSpec.unpartitioned(),
+ Collections.singletonMap(TableProperties.FORMAT_VERSION, "3"));
+ int payloadId = table.schema().findField("payload").fieldId();
+ int existingId =
table.schema().findField("payload.existing").fieldId();
+ int keyId = table.schema().findField("payload.k").fieldId();
+ table.newAppend().appendFile(dataFile(table.spec(),
+ "s3://b/db/dropped_nested_eq_key/f1.parquet", 1024, null,
null)).commit();
+ table.newRowDelta().addDeletes(equalityDeleteFile(
+ "s3://b/db/dropped_nested_eq_key/eq.parquet",
FileFormat.PARQUET, keyId)).commit();
+ table.updateSchema().deleteColumn("payload.k").commit();
+
+ List<NestedField> carrier =
IcebergScanPlanProvider.schemaForPotentialEqualityDeletes(
+ table, table.newScan(), table.schema());
+ Assertions.assertEquals(2, carrier.size(), "the nested key must not
add a second payload column");
+ NestedField payload = carrier.get(1);
+ Assertions.assertEquals(payloadId, payload.fieldId());
+ Assertions.assertTrue(payload.isOptional());
+ Assertions.assertEquals(Arrays.asList(existingId, keyId),
payload.type().asStructType().fields()
+
.stream().map(NestedField::fieldId).collect(Collectors.toList()));
+
+ IcebergScanPlanProvider provider = providerOver(table);
+ Map<String, String> props = provider.getScanNodeProperties(
+ null, new IcebergTableHandle("db1", "dropped_nested_eq_key"),
+ Collections.singletonList(new IcebergColumnHandle("id",
table.schema().findField("id").fieldId())),
+ Optional.empty());
+ TField payloadField = dictFieldById(dictRootFields(provider, props),
payloadId);
+ TField key =
dictFieldById(payloadField.getNestedField().getStructField().getFields(),
keyId);
+ Assertions.assertEquals("7", key.getInitialDefaultValue());
+ }
+
+ @Test
+ public void
equalityCarrierKeepsDroppedNanosecondEqualityKeyAsTimestampNs() throws
Exception {
+ // Unzoned nanosecond timestamps are readable, so a dropped nanosecond
equality key stays in the carrier as
+ // TIMESTAMP_NS instead of the STRING placeholder that stands in for
types Doris cannot read.
+ Schema schema = new Schema(
+ Types.NestedField.required(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "equality_nanos",
Types.TimestampNanoType.withoutZone()));
+ Table table = createTable("dropped_nanos_eq_key", schema,
PartitionSpec.unpartitioned(),
+ Collections.singletonMap(TableProperties.FORMAT_VERSION, "3"));
+ int keyId = table.schema().findField("equality_nanos").fieldId();
+ table.newAppend().appendFile(dataFile(table.spec(),
+ "s3://b/db/dropped_nanos_eq_key/f1.parquet", 1024, null,
null)).commit();
+ table.newRowDelta().addDeletes(equalityDeleteFile(
+ "s3://b/db/dropped_nanos_eq_key/eq.parquet",
FileFormat.PARQUET, keyId)).commit();
+ table.updateSchema().deleteColumn("equality_nanos").commit();
+
+ IcebergScanPlanProvider provider = providerOver(table);
+ Map<String, String> props = provider.getScanNodeProperties(
+ null, new IcebergTableHandle("db1", "dropped_nanos_eq_key"),
+ Collections.singletonList(new IcebergColumnHandle("id",
table.schema().findField("id").fieldId())),
+ Optional.empty());
+
+ TField key = dictFieldById(dictRootFields(provider, props), keyId);
+ Assertions.assertEquals("equality_nanos", key.getName());
+ Assertions.assertEquals(TPrimitiveType.TIMESTAMP_NS,
key.getType().getType());
+ }
+
+ @Test
+ public void
equalityCarrierTakesOnlyReferencedFieldsFromAHistoricalSchema() throws
Exception {
+ // Only the historical fields that live equality deletes reference may
enter the carrier. A sibling
+ // dropped in the same change, here a TIME column Doris cannot read,
must stay out of it.
+ Schema schema = new Schema(
+ Types.NestedField.required(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "equality_key",
Types.IntegerType.get()),
+ Types.NestedField.optional(3, "dropped_time",
Types.TimeType.get()));
+ Table table = createTable("dropped_unreferenced_sibling", schema,
PartitionSpec.unpartitioned(),
+ Collections.singletonMap(TableProperties.FORMAT_VERSION, "2"));
+ int idId = table.schema().findField("id").fieldId();
+ int keyId = table.schema().findField("equality_key").fieldId();
+ table.newAppend().appendFile(dataFile(table.spec(),
+ "s3://b/db/dropped_unreferenced_sibling/f1.parquet", 1024,
null, null)).commit();
+ table.newRowDelta().addDeletes(equalityDeleteFile(
+ "s3://b/db/dropped_unreferenced_sibling/eq.parquet",
FileFormat.PARQUET, keyId)).commit();
+
table.updateSchema().deleteColumn("equality_key").deleteColumn("dropped_time").commit();
+
+ IcebergScanPlanProvider provider = providerOver(table);
+ Map<String, String> props = provider.getScanNodeProperties(
+ null, new IcebergTableHandle("db1",
"dropped_unreferenced_sibling"),
+ Collections.singletonList(new IcebergColumnHandle("id",
idId)), Optional.empty());
+
+ Assertions.assertEquals(Arrays.asList(idId, keyId),
dictRootFields(provider, props).stream()
+ .map(field ->
field.getFieldPtr().getId()).collect(Collectors.toList()));
+ }
+
@Test
public void equalityCarrierSizeDoesNotGrowWithUnrelatedSchemaHistory()
throws Exception {
Schema schema = new Schema(
@@ -2089,6 +2250,232 @@ public class IcebergScanPlanProviderTest {
"the upgrade fence must be conservative and bounded by schema
history, not snapshot count");
}
+ @Test
+ public void missingRequiredFieldFenceFollowsProjection() {
+ // Older files cannot contain a required field added later, and only
current backends reject such a file
+ // instead of reading NULL. The fence must therefore follow the
projection, nested paths included.
+ Schema schema = new Schema(
+ Types.NestedField.required(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "payload", Types.StructType.of(
+ Types.NestedField.optional(3, "existing",
Types.IntegerType.get()))));
+ Table table = createTable("required_projection_fence", schema,
PartitionSpec.unpartitioned());
+ table.newAppend().appendFile(dataFile(table.spec(),
+ "s3://b/db/required_projection_fence/old.parquet", 128, null,
null)).commit();
+ table.updateSchema().allowIncompatibleChanges()
+ .addRequiredColumn("required_added", Types.IntegerType.get())
+ .addRequiredColumn("payload", "required_nested",
Types.IntegerType.get())
+ .commit();
+ table.newAppend().appendFile(dataFile(table.spec(),
+ "s3://b/db/required_projection_fence/new.parquet", 128, null,
null)).commit();
+ Schema current = table.schema();
+ int payloadId = current.findField("payload").fieldId();
+ ConnectorColumnHandle payload = new IcebergColumnHandle("payload",
payloadId);
+ IcebergScanPlanProvider provider = providerOver(table);
+ IcebergTableHandle handle = new IcebergTableHandle("db1",
"required_projection_fence");
+
+ Assertions.assertNull(requiredBackendSemantics(provider, handle,
+ new IcebergColumnHandle("id",
current.findField("id").fieldId())));
+ Assertions.assertEquals("Current Iceberg scan semantics",
requiredBackendSemantics(provider, handle,
+ new IcebergColumnHandle("required_added",
current.findField("required_added").fieldId())));
+ Assertions.assertNull(requiredBackendSemantics(provider, handle,
payload.withProjectedFieldIds(
+ ImmutableSet.of(payloadId,
current.findField("payload.existing").fieldId()))));
+ Assertions.assertEquals("Current Iceberg scan semantics",
requiredBackendSemantics(provider, handle,
+ payload.withProjectedFieldIds(ImmutableSet.of(
+ payloadId,
current.findField("payload.required_nested").fieldId()))));
+ }
+
+ @Test
+ public void
requiredCollectionWrappersDoNotTriggerMissingRequiredFieldFence() {
+ // A list element or a map key/value has no value of its own without
its collection, so a required wrapper
+ // missing from an older schema never needs the fence; a required
struct field inside the element does.
+ Types.NestedField existing = Types.NestedField.optional(32,
"existing", Types.IntegerType.get());
+ Schema historicalSchema = new Schema(0, Arrays.asList(
+ Types.NestedField.optional(10, "items",
Types.ListType.ofOptional(90, Types.IntegerType.get())),
+ Types.NestedField.optional(20, "entries",
Types.MapType.ofOptional(
+ 91, 92, Types.StringType.get(),
Types.IntegerType.get())),
+ Types.NestedField.optional(30, "struct_items",
Types.ListType.ofOptional(
+ 31, Types.StructType.of(existing)))));
+ Schema scanSchema = new Schema(1, Arrays.asList(
+ Types.NestedField.optional(10, "items",
Types.ListType.ofRequired(11, Types.IntegerType.get())),
+ Types.NestedField.optional(20, "entries",
Types.MapType.ofRequired(
+ 21, 22, Types.StringType.get(),
Types.IntegerType.get())),
+ Types.NestedField.optional(30, "struct_items",
Types.ListType.ofOptional(
+ 31, Types.StructType.of(existing,
+ Types.NestedField.required(33,
"required_nested", Types.IntegerType.get()))))));
+ Table snapshotTable = createTable("required_collection_wrappers",
SCHEMA, PartitionSpec.unpartitioned());
+ snapshotTable.newAppend().appendFile(dataFile(snapshotTable.spec(),
+ "s3://b/db/required_collection_wrappers/f1.parquet", 128,
null, null)).commit();
+ Table table = tableWithSchemas(snapshotTable, historicalSchema,
scanSchema);
+ Snapshot snapshot = snapshotTable.currentSnapshot();
+ ConnectorColumnHandle structItems = new
IcebergColumnHandle("struct_items", 30);
+
+
Assertions.assertFalse(IcebergScanPlanProvider.selectedHistoryRequiresMissingRequiredFieldRejection(
+ table, scanSchema,
IcebergScanPlanProvider.projectedFieldIds(scanSchema, Arrays.asList(
+ new IcebergColumnHandle("items", 10), new
IcebergColumnHandle("entries", 20))),
+ snapshot));
+ // The access path struct_items.*.existing selects the element's child
without the element wrapper ID.
+
Assertions.assertFalse(IcebergScanPlanProvider.selectedHistoryRequiresMissingRequiredFieldRejection(
+ table, scanSchema,
IcebergScanPlanProvider.projectedFieldIds(scanSchema, Collections.singletonList(
+ structItems.withProjectedFieldIds(ImmutableSet.of(30,
32)))),
+ snapshot));
+
Assertions.assertTrue(IcebergScanPlanProvider.selectedHistoryRequiresMissingRequiredFieldRejection(
+ table, scanSchema,
IcebergScanPlanProvider.projectedFieldIds(scanSchema, Collections.singletonList(
+ structItems.withProjectedFieldIds(ImmutableSet.of(30,
33)))),
+ snapshot));
+ }
+
+ @Test
+ public void missingRequiredFieldFenceCoversSchemaOnlyTarget() {
+ // A schema-only change has no snapshot of its own, so every selected
file predates the required field it
+ // adds; projecting that field still needs the fence.
+ Table table = createTable("schema_only_required", new Schema(
+ Types.NestedField.required(1, "id", Types.LongType.get())),
PartitionSpec.unpartitioned());
+ table.newAppend().appendFile(dataFile(table.spec(),
+ "s3://b/db/schema_only_required/f1.parquet", 128, null,
null)).commit();
+ table.updateSchema().allowIncompatibleChanges()
+ .addRequiredColumn("required_value",
Types.IntegerType.get()).commit();
+ Assertions.assertNotEquals(table.schema().schemaId(),
table.currentSnapshot().schemaId(),
+ "the selected snapshot must predate the scanned schema");
+ IcebergScanPlanProvider provider = providerOver(table);
+ IcebergTableHandle handle = new IcebergTableHandle("db1",
"schema_only_required");
+
+ Assertions.assertEquals("Current Iceberg scan semantics",
requiredBackendSemantics(provider, handle,
+ new IcebergColumnHandle("required_value",
table.schema().findField("required_value").fieldId())));
+ Assertions.assertNull(requiredBackendSemantics(provider, handle,
+ new IcebergColumnHandle("id",
table.schema().findField("id").fieldId())));
+ }
+
+ @Test
+ public void missingRequiredFieldFenceCoversCherryPickedSourceAncestry() {
+ // A cherry-picked snapshot has no parent on the target branch, but
its files come from a source branch
+ // whose ancestors were written before the required field existed;
projecting that field needs the fence.
+ Table table = createTable("cherry_picked_required", new Schema(
+ Types.NestedField.required(1, "id", Types.LongType.get())),
PartitionSpec.unpartitioned());
+ table.newAppend().appendFile(dataFile(table.spec(),
+ "s3://b/db/cherry_picked_required/old.parquet", 128, null,
null)).toBranch("audit").commit();
+ table.updateSchema().allowIncompatibleChanges()
+ .addRequiredColumn("required_value",
Types.IntegerType.get()).commit();
+ table.newAppend().appendFile(dataFile(table.spec(),
+ "s3://b/db/cherry_picked_required/new.parquet", 128, null,
null)).toBranch("audit").commit();
+ long sourceSnapshotId = table.refs().get("audit").snapshotId();
+ table.manageSnapshots().cherrypick(sourceSnapshotId).commit();
+ Snapshot target = table.currentSnapshot();
+ Assertions.assertNull(target.parentId(), "the target branch must have
no snapshot before the picked one");
+ Assertions.assertEquals(String.valueOf(sourceSnapshotId),
+ target.summary().get(SnapshotSummary.SOURCE_SNAPSHOT_ID_PROP));
+ IcebergScanPlanProvider provider = providerOver(table);
+ IcebergTableHandle handle = new IcebergTableHandle("db1",
"cherry_picked_required");
+
+ Assertions.assertEquals("Current Iceberg scan semantics",
requiredBackendSemantics(provider, handle,
+ new IcebergColumnHandle("required_value",
table.schema().findField("required_value").fieldId())));
+ Assertions.assertNull(requiredBackendSemantics(provider, handle,
+ new IcebergColumnHandle("id",
table.schema().findField("id").fieldId())));
+ }
+
+ @Test
+ public void
nestedInitialDefaultRequiresCurrentBackendSemanticsOnlyWhenProjected() {
+ // Older backends cannot materialize an initial default below the top
level. The fence follows the
+ // projected subtree: a top-level scalar default and a struct path
that skips the defaulted child stay
+ // schedulable on every backend.
+ Schema schema = new Schema(
+ Types.NestedField.required(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "payload", Types.StructType.of(
+ Types.NestedField.optional(3, "existing",
Types.IntegerType.get()))));
+ Table table = createTable("nested_default_fence", schema,
PartitionSpec.unpartitioned(),
+ Collections.singletonMap(TableProperties.FORMAT_VERSION, "3"));
+ table.newAppend().appendFile(dataFile(table.spec(),
+ "s3://b/db/nested_default_fence/f1.parquet", 1024, null,
null)).commit();
+ table.updateSchema()
+ .addColumn("scalar", Types.IntegerType.get(), Literal.of(5))
+ .addColumn("payload", "added", Types.IntegerType.get(),
Literal.of(7))
+ .commit();
+ Schema current = table.schema();
+ int payloadId = current.findField("payload").fieldId();
+ ConnectorColumnHandle payload = new IcebergColumnHandle("payload",
payloadId);
+ IcebergScanPlanProvider provider = providerOver(table);
+ IcebergTableHandle handle = new IcebergTableHandle("db1",
"nested_default_fence");
+
+ Assertions.assertNull(requiredBackendSemantics(provider, handle,
+ new IcebergColumnHandle("scalar",
current.findField("scalar").fieldId())));
+ Assertions.assertEquals("Current Iceberg scan semantics",
+ requiredBackendSemantics(provider, handle, payload));
+ Assertions.assertNull(requiredBackendSemantics(provider, handle,
payload.withProjectedFieldIds(
+ ImmutableSet.of(payloadId,
current.findField("payload.existing").fieldId()))));
+ Assertions.assertEquals("Current Iceberg scan semantics",
requiredBackendSemantics(provider, handle,
+ payload.withProjectedFieldIds(ImmutableSet.of(
+ payloadId,
current.findField("payload.added").fieldId()))));
+ }
+
+ @Test
+ public void nestedInitialDefaultLeavesSmoothUpgradeRejectionToTheEngine() {
+ // The connector plans the scan and only names the semantics it needs.
The engine then refuses to schedule
+ // it with "Current Iceberg scan semantics are unavailable while
backend <id> is a smooth upgrade source".
+ Schema schema = new Schema(Types.NestedField.optional(1, "payload",
Types.StructType.of(
+ Types.NestedField.optional(2, "existing",
Types.IntegerType.get()))));
+ Table table = createTable("nested_default_engine_fence", schema,
PartitionSpec.unpartitioned(),
+ Collections.singletonMap(TableProperties.FORMAT_VERSION, "3"));
+ table.newAppend().appendFile(dataFile(table.spec(),
+ "s3://b/db/nested_default_engine_fence/f1.parquet", 1024,
null, null)).commit();
+ table.updateSchema().addColumn("payload", "added",
Types.IntegerType.get(), Literal.of(5)).commit();
+ IcebergScanPlanProvider provider = providerOver(table);
+ ConnectorSession session = new FakeScanSession("UTC",
+ Collections.singletonMap("enable_file_scanner_v2", "true"));
+ IcebergTableHandle handle = new IcebergTableHandle("db1",
"nested_default_engine_fence");
+ List<ConnectorColumnHandle> columns = Collections.singletonList(
+ new IcebergColumnHandle("payload",
table.schema().findField("payload").fieldId()));
+
+ Assertions.assertEquals(1,
+ provider.planScan(session,
ConnectorScanRequest.builder(handle, columns).build()).size());
+ Assertions.assertEquals("Current Iceberg scan semantics",
+ provider.getScanNodeProperties(session, handle, columns,
Optional.empty())
+
.get(ScanNodePropertyKeys.REQUIRED_CURRENT_BACKEND_SEMANTICS));
+ }
+
+ @Test
+ public void
nameMappingAliasCollisionRequiresCurrentBackendSemanticsOnlyWhenProjected() {
+ // When a name-mapping alias equals a sibling's current name, older
backends may bind the wrong physical
+ // column of an ID-less file. The fence applies when a colliding field
is projected and spares the rest.
+ Schema schema = new Schema(
+ Types.NestedField.optional(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "root", Types.StructType.of(
+ Types.NestedField.optional(3, "renamed_payload",
+ Types.ListType.ofOptional(4,
Types.IntegerType.get())),
+ Types.NestedField.optional(5, "payload",
Types.MapType.ofOptional(
+ 6, 7, Types.StringType.get(),
Types.IntegerType.get())),
+ Types.NestedField.optional(8, "safe",
Types.IntegerType.get()))));
+ Table table = createTable("name_alias_collision", schema,
PartitionSpec.unpartitioned());
+ Schema current = table.schema();
+ int rootId = current.findField("root").fieldId();
+ int renamedPayloadId =
current.findField("root.renamed_payload").fieldId();
+ int payloadId = current.findField("root.payload").fieldId();
+ table.newAppend().appendFile(dataFile(table.spec(),
+ "s3://b/db/name_alias_collision/f1.parquet", 1024, null,
null)).commit();
+ // Mapped names are unique per level, so the collision is the renamed
field's old name that the newer,
+ // unmapped sibling now uses as its current name.
+ table.updateProperties().set(TableProperties.DEFAULT_NAME_MAPPING,
NameMappingParser.toJson(
+ NameMapping.of(MappedField.of(rootId, "root", MappedFields.of(
+ MappedField.of(renamedPayloadId,
Arrays.asList("payload", "renamed_payload"))))))).commit();
+ ConnectorColumnHandle root = new IcebergColumnHandle("root", rootId);
+ IcebergScanPlanProvider provider = providerOver(table);
+ IcebergTableHandle handle = new IcebergTableHandle("db1",
"name_alias_collision");
+
+ Assertions.assertNull(requiredBackendSemantics(provider, handle,
+ new IcebergColumnHandle("id",
current.findField("id").fieldId())));
+ Assertions.assertEquals("Current Iceberg scan semantics",
+ requiredBackendSemantics(provider, handle, root));
+ Assertions.assertNull(requiredBackendSemantics(provider, handle,
root.withProjectedFieldIds(
+ ImmutableSet.of(rootId,
current.findField("root.safe").fieldId()))));
+ Assertions.assertEquals("Current Iceberg scan semantics",
requiredBackendSemantics(provider, handle,
+ root.withProjectedFieldIds(ImmutableSet.of(rootId,
renamedPayloadId))));
+
+ // An alias that no sibling currently uses as its name is not a
collision.
+ table.updateProperties().set(TableProperties.DEFAULT_NAME_MAPPING,
NameMappingParser.toJson(
+ NameMapping.of(MappedField.of(rootId, "root", MappedFields.of(
+ MappedField.of(renamedPayloadId,
Arrays.asList("legacy_payload", "renamed_payload")),
+ MappedField.of(payloadId, "payload")))))).commit();
+ Assertions.assertNull(requiredBackendSemantics(provider, handle,
root));
+ }
+
@Test
public void
getScanNodePropertiesEmitsSchemaEvolutionDictForPartitionedTableToo() {
// The dict is emitted alongside path_partition_keys (it is
unconditional, like legacy
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergSchemaUtilsTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergSchemaUtilsTest.java
index 0e9278ab8bb..832ed2149bc 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergSchemaUtilsTest.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergSchemaUtilsTest.java
@@ -362,6 +362,24 @@ public class IcebergSchemaUtilsTest {
Assertions.assertEquals(TPrimitiveType.BOOLEAN,
fields.get("added_bool").getType().getType());
}
+ @Test
+ public void dictCarriesInitialDefaultRatherThanWriteDefault() {
+ // Old files are read with the initial default; the write default only
fills omitted columns of new rows.
+ // A timestamptz default keeps its UTC offset so every session zone
reads the same instant.
+ Schema schema = new Schema(
+
Types.NestedField.optional("added_column").withId(1).ofType(Types.IntegerType.get())
+ .withInitialDefault(7).withWriteDefault(9).build(),
+
Types.NestedField.optional("added_timestamptz").withId(2).ofType(Types.TimestampType.withZone())
+ .withInitialDefault(1_704_067_200_123_456L).build());
+
+ Map<String, TField> fields =
topFields(IcebergSchemaUtils.buildCurrentSchema(schema,
+ Arrays.asList("added_column", "added_timestamptz"),
Collections.emptyMap()));
+
+ Assertions.assertEquals("7",
fields.get("added_column").getInitialDefaultValue());
+ Assertions.assertEquals("2024-01-01 00:00:00.123456+00:00",
+ fields.get("added_timestamptz").getInitialDefaultValue());
+ }
+
@Test
public void complexInitialDefaultsUseIcebergJsonAndMarkBinaryLeaves() {
Types.StructType structType = Types.StructType.of(
@@ -412,6 +430,38 @@ public class IcebergSchemaUtilsTest {
.getFieldPtr().isInitialDefaultValueIsBase64());
}
+ @Test
+ public void structChildrenCarryTheirOwnInitialDefaultsAndRequiredness() {
+ // BE fills a struct child that an old file lacks from the child's own
entry, and rejects the file only when
+ // that entry is required without a default. Each child therefore
carries its own initial default and
+ // Iceberg requiredness, independently of its parent and siblings.
+ Schema schema = new Schema(Types.NestedField.required(1, "payload",
Types.StructType.of(
+
Types.NestedField.required("nested_int").withId(2).ofType(Types.IntegerType.get())
+ .withInitialDefault(17).build(),
+
Types.NestedField.optional("nested_binary").withId(3).ofType(Types.BinaryType.get())
+ .withInitialDefault(ByteBuffer.wrap(new byte[] {0, 1,
2, (byte) 0xFF})).build(),
+ Types.NestedField.optional(4, "nested_uuid",
Types.UUIDType.get()))));
+
+ TField payload = topFields(IcebergSchemaUtils.buildCurrentSchema(
+ schema, Collections.singletonList("payload"),
Collections.emptyMap())).get("payload");
+
+ Assertions.assertFalse(payload.isIsOptional());
+ Assertions.assertFalse(payload.isSetInitialDefaultValue());
+ TField nestedInt = childByName(payload, "nested_int");
+ Assertions.assertEquals("17", nestedInt.getInitialDefaultValue());
+ Assertions.assertFalse(nestedInt.isSetInitialDefaultValueIsBase64());
+ Assertions.assertFalse(nestedInt.isIsOptional());
+ TField nestedBinary = childByName(payload, "nested_binary");
+ Assertions.assertEquals(3, nestedBinary.getId());
+ Assertions.assertEquals("AAEC/w==",
nestedBinary.getInitialDefaultValue());
+ Assertions.assertTrue(nestedBinary.isInitialDefaultValueIsBase64());
+ Assertions.assertTrue(nestedBinary.isIsOptional());
+ TField nestedUuid = childByName(payload, "nested_uuid");
+ Assertions.assertFalse(nestedUuid.isSetInitialDefaultValue());
+ Assertions.assertFalse(nestedUuid.isSetInitialDefaultValueIsBase64());
+ Assertions.assertTrue(nestedUuid.isIsOptional());
+ }
+
@Test
public void equalityCarrierAllowsDropAndReaddNamesAtEveryLevel() throws
Exception {
Types.StructType nestedDuplicateNames = Types.StructType.of(
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 9193fbbc79b..e93355a4e9b 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
@@ -66,6 +66,9 @@ import org.apache.iceberg.PartitionSpecParser;
import org.apache.iceberg.Schema;
import org.apache.iceberg.SchemaParser;
import org.apache.iceberg.Table;
+import org.apache.iceberg.TableMetadata;
+import org.apache.iceberg.TableMetadataParser;
+import org.apache.iceberg.TableOperations;
import org.apache.iceberg.TableProperties;
import org.apache.iceberg.catalog.Namespace;
import org.apache.iceberg.catalog.TableIdentifier;
@@ -75,11 +78,17 @@ import org.apache.iceberg.expressions.Literal;
import org.apache.iceberg.inmemory.InMemoryCatalog;
import org.apache.iceberg.io.FileIO;
import org.apache.iceberg.io.InputFile;
+import org.apache.iceberg.io.LocationProvider;
import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.types.Type;
import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.DateTimeUtil;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Proxy;
+import java.math.BigDecimal;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
@@ -91,6 +100,7 @@ import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
+import java.util.UUID;
/**
* Pins {@link IcebergWritePlanProvider#planWrite} for INSERT/OVERWRITE
against legacy
@@ -412,6 +422,102 @@ public class IcebergWritePlanProviderTest {
IcebergWriteSchemaContext.toDorisSql(nestedType,
nestedDefault, false, false));
}
+ @Test
+ public void primitiveWriteDefaultsBecomeTypedSqlLiterals() {
+ // Each primitive write-default reaches the engine as SQL of the
column's own type. The initial-default only
+ // describes rows written before the field existed, so it must never
become the value of a new row.
+ Schema schema = new Schema(
+ defaultField(1, "boolean_col", Types.BooleanType.get(), true,
false),
+ defaultField(2, "int_col", Types.IntegerType.get(), 34, 35),
+ defaultField(3, "long_col", Types.LongType.get(),
4_900_000_000L, 4_900_000_001L),
+ defaultField(4, "float_col", Types.FloatType.get(), 12.25F,
13.5F),
+ defaultField(5, "double_col", Types.DoubleType.get(), -123.5D,
456.75D),
+ defaultField(6, "decimal_col", Types.DecimalType.of(20, 4),
+ new BigDecimal("12345.6789"), new
BigDecimal("98765.4321")),
+ defaultField(7, "date_col", Types.DateType.get(),
+ DateTimeUtil.isoDateToDays("2024-12-17"),
DateTimeUtil.isoDateToDays("2025-01-18")),
+ defaultField(8, "timestamp_col",
Types.TimestampType.withoutZone(),
+
DateTimeUtil.isoTimestampToMicros("2024-12-17T23:59:59.123456"),
+
DateTimeUtil.isoTimestampToMicros("2025-01-18T01:02:03.654321")),
+ defaultField(9, "timestamptz_col",
Types.TimestampType.withZone(),
+
DateTimeUtil.isoTimestamptzToMicros("2024-12-17T23:59:59.123456+00:00"),
+
DateTimeUtil.isoTimestamptzToMicros("2025-01-18T01:02:03.654321+00:00")),
+ defaultField(10, "string_col", Types.StringType.get(),
"initial-default", "write-default"),
+ defaultField(11, "uuid_col", Types.UUIDType.get(),
+
UUID.fromString("123e4567-e89b-12d3-a456-426614174000"),
+
UUID.fromString("123e4567-e89b-12d3-a456-426614174001")),
+ defaultField(12, "fixed_col", Types.FixedType.ofLength(4),
+ ByteBuffer.wrap(new byte[] {0x01, 0x02, 0x03, 0x04}),
+ ByteBuffer.wrap(new byte[] {0x0a, 0x0b, 0x0c, 0x0d})),
+ defaultField(13, "binary_col", Types.BinaryType.get(),
+ ByteBuffer.wrap(new byte[] {0x05, 0x06}),
ByteBuffer.wrap(new byte[] {0x0e, 0x0f})),
+ Types.NestedField.optional(14, "optional_col",
Types.IntegerType.get()),
+ Types.NestedField.required(15, "required_col",
Types.IntegerType.get()));
+ Table table = freshCatalog().createTable(TableIdentifier.of("db1",
"typed_defaults"), schema,
+ PartitionSpec.unpartitioned(),
Collections.singletonMap("format-version", "3"));
+ IcebergWriteSchemaContext writeContext =
IcebergWriteSchemaContext.create(
+ table, "db1.typed_defaults", Optional.empty(), false, false);
+ Map<String, ConnectorColumn> columns = new HashMap<>();
+ for (ConnectorColumn column : writeContext.getColumns()) {
+ columns.put(column.getName(), column);
+ }
+
+ assertWriteDefault(columns.get("boolean_col"),
ConnectorType.of("BOOLEAN"), "false");
+ assertWriteDefault(columns.get("int_col"), ConnectorType.of("INT"),
"35");
+ assertWriteDefault(columns.get("long_col"),
ConnectorType.of("BIGINT"), "4900000001");
+ assertWriteDefault(columns.get("float_col"),
ConnectorType.of("FLOAT"), "13.5");
+ assertWriteDefault(columns.get("double_col"),
ConnectorType.of("DOUBLE"), "456.75");
+ assertWriteDefault(columns.get("decimal_col"),
ConnectorType.of("DECIMALV3", 20, 4), "98765.4321");
+ assertWriteDefault(columns.get("date_col"),
ConnectorType.of("DATEV2"), "'2025-01-18'");
+ assertWriteDefault(columns.get("timestamp_col"),
ConnectorType.of("DATETIMEV2", 6, 0),
+ "'2025-01-18 01:02:03.654321'");
+ assertWriteDefault(columns.get("timestamptz_col"),
ConnectorType.of("TIMESTAMPTZ", 6, 0),
+ "'2025-01-18 01:02:03.654321+00:00'");
+ assertWriteDefault(columns.get("string_col"),
ConnectorType.of("STRING"), "'write-default'");
+ assertWriteDefault(columns.get("uuid_col"), ConnectorType.of("UUID"),
+ "'123e4567-e89b-12d3-a456-426614174001'");
+ assertWriteDefault(columns.get("fixed_col"),
ConnectorType.of("VARBINARY", 4, 0), "X'0A0B0C0D'");
+ assertWriteDefault(columns.get("binary_col"),
ConnectorType.of("VARBINARY"), "X'0E0F'");
+ assertWriteDefault(columns.get("optional_col"),
ConnectorType.of("INT"), "NULL");
+ assertWriteDefault(columns.get("required_col"),
ConnectorType.of("INT"), null);
+ Assertions.assertFalse(columns.get("string_col").isNullable());
+ Assertions.assertEquals(table.schema().schemaId(),
writeContext.getSchema().schemaId());
+ Assertions.assertEquals(SchemaParser.toJson(table.schema()),
writeContext.getSchemaJson());
+
Assertions.assertTrue(writeContext.getMergeSchemaJson().contains("_row_id"));
+ }
+
+ private static Types.NestedField defaultField(int id, String name, Type
type,
+ Object initialDefault, Object writeDefault) {
+ return Types.NestedField.required(name).withId(id).ofType(type)
+
.withInitialDefault(initialDefault).withWriteDefault(writeDefault).build();
+ }
+
+ private static void assertWriteDefault(ConnectorColumn column,
ConnectorType type, String defaultSql) {
+ Assertions.assertEquals(type, column.getType(), column.getName());
+ Assertions.assertEquals(defaultSql, column.getDefaultValueSql(),
column.getName());
+ }
+
+ @Test
+ public void complexWriteDefaultsRenderMapsAndNullElementsRecursively() {
+ // A complex write-default keeps every nested NULL in place and
renders maps entry by entry.
+ Types.StructType structType = Types.StructType.of(
+ Types.NestedField.required(101, "count",
Types.IntegerType.get()),
+ Types.NestedField.optional(102, "items",
Types.ListType.ofOptional(103, Types.StringType.get())),
+ Types.NestedField.optional(104, "attributes",
+ Types.MapType.ofOptional(105, 106,
Types.StringType.get(), Types.IntegerType.get())));
+ Map<String, Integer> attributes = new LinkedHashMap<>();
+ attributes.put("one", 1);
+ attributes.put("none", null);
+ GenericRecord structDefault = GenericRecord.create(structType);
+ structDefault.setField("count", 7);
+ structDefault.setField("items", Arrays.asList("a", null));
+ structDefault.setField("attributes", attributes);
+
+ Assertions.assertEquals("named_struct('count', 7, 'items', array('a',
NULL), "
+ + "'attributes', map('one', 1, 'none', NULL))",
+ IcebergWriteSchemaContext.toDorisSql(structType,
structDefault, false, false));
+ }
+
@Test
public void zonedAndNestedBinaryWriteDefaultsIgnoreLegacyMappingFlags() {
// A zoned default keeps its instant, and a nested binary leaf stays
bytes, with the old flags off.
@@ -618,6 +724,97 @@ public class IcebergWritePlanProviderTest {
Assertions.assertDoesNotThrow(() ->
writeContext.validateCurrentSchema(table, false));
}
+ @Test
+ public void pinnedWriterMetadataRejectsSchemaFormatSpecAndSortOrderDrift()
{
+ // Pinned writer metadata stays valid only while its schema
generation, format version, partition spec and
+ // sort order are still the table's; a spec or sort order that is
gone, or reused with another definition,
+ // would label the written files with metadata they were not written
with.
+ Table table = partitionedSortedTable(freshCatalog());
+ IcebergWriteSchemaContext writeContext =
IcebergWriteSchemaContext.create(
+ table, "db1.t1", Optional.empty(), false, false);
+ Assertions.assertDoesNotThrow(() ->
writeContext.validateCurrentSchema(table, false));
+
+ assertWriterMetadataDrift(writeContext, withMetadata(table, "specs",
Collections.emptyMap()),
+ "partition spec changed during write planning");
+ PartitionSpec redefinedSpec = PartitionSpec.builderFor(table.schema())
+ .withSpecId(table.spec().specId()).identity("name").build();
+ assertWriterMetadataDrift(writeContext, withMetadata(table, "specs",
+ Collections.singletonMap(redefinedSpec.specId(),
redefinedSpec)),
+ "partition spec changed during write planning");
+ assertWriterMetadataDrift(writeContext, withMetadata(table,
"sortOrders", Collections.emptyMap()),
+ "sort order changed during write planning");
+
+ table.updateProperties().set(TableProperties.FORMAT_VERSION,
"3").commit();
+ assertWriterMetadataDrift(writeContext, table, "schema changed during
write planning");
+ assertWriterMetadataDrift(writeContext, table, "/format 3");
+ table.updateSchema().addColumn("added_after_planning",
Types.IntegerType.get()).commit();
+ assertWriterMetadataDrift(writeContext, table, "current schema " +
table.schema().schemaId() + "/");
+ }
+
+ private static void assertWriterMetadataDrift(IcebergWriteSchemaContext
writeContext, Table current,
+ String expectedMessage) {
+ DorisConnectorException ex =
Assertions.assertThrows(DorisConnectorException.class,
+ () -> writeContext.validateCurrentSchema(current, false));
+ Assertions.assertTrue(ex.getMessage().contains(expectedMessage),
ex.getMessage());
+ Assertions.assertTrue(ex.getMessage().contains("retry the statement"),
ex.getMessage());
+ }
+
+ /** A view of {@code table} whose {@code method} returns {@code value};
every other call reaches the table. */
+ private static Table withMetadata(Table table, String method, Object
value) {
+ return (Table) Proxy.newProxyInstance(Table.class.getClassLoader(),
+ new Class<?>[] {Table.class, HasTableOperations.class},
(proxy, invoked, args) -> {
+ if (method.equals(invoked.getName())) {
+ return value;
+ }
+ try {
+ return invoked.invoke(table, args);
+ } catch (InvocationTargetException e) {
+ throw e.getCause();
+ }
+ });
+ }
+
+ @Test
+ public void
v1MetadataIdentityAcceptsRetainedAncestorAndRejectsReplacement() {
+ // A V1 table without a UUID is identified by its metadata file and
timestamp: a later commit that keeps the
+ // pinned file in its metadata log is the same table, a reused file
name with another timestamp is not.
+ String pinnedFile = "file:///tmp/v1_table/metadata/v1.metadata.json";
+ MutableMetadataOperations operations =
+ new MutableMetadataOperations(v1Metadata(pinnedFile, 1234L,
null, 0L));
+ Assertions.assertNull(operations.current().uuid());
+ Table table = new BaseTable(operations, "db1.v1_table");
+ IcebergWriteSchemaContext writeContext =
IcebergWriteSchemaContext.create(
+ table, "db1.v1_table", Optional.empty(), false, false);
+ Assertions.assertDoesNotThrow(() ->
writeContext.validateCurrentSchema(table, false));
+
+ operations.metadata =
v1Metadata("file:///tmp/v1_table/metadata/v2.metadata.json", 2345L,
+ pinnedFile, 1234L);
+ Assertions.assertDoesNotThrow(() ->
writeContext.validateCurrentSchema(table, false));
+
+ operations.metadata = v1Metadata(pinnedFile, 3456L, null, 0L);
+ DorisConnectorException ex =
Assertions.assertThrows(DorisConnectorException.class,
+ () -> writeContext.validateCurrentSchema(table, false));
+ Assertions.assertTrue(ex.getMessage().contains("identity changed"),
ex.getMessage());
+ }
+
+ /** V1 metadata without a table UUID, as written before Iceberg assigned
one to every table. */
+ private static TableMetadata v1Metadata(String metadataFile, long
lastUpdatedMillis,
+ String previousFile, long previousTimestampMillis) {
+ String metadataLog = previousFile == null ? "[]" :
"[{\"timestamp-ms\":" + previousTimestampMillis
+ + ",\"metadata-file\":\"" + previousFile + "\"}]";
+ return TableMetadataParser.fromJson(metadataFile,
"{\"format-version\":1,"
+ + "\"location\":\"file:///tmp/v1_table\","
+ + "\"last-updated-ms\":" + lastUpdatedMillis + ","
+ + "\"last-column-id\":1,"
+ +
"\"schema\":{\"type\":\"struct\",\"schema-id\":0,\"fields\":["
+ +
"{\"id\":1,\"name\":\"id\",\"required\":false,\"type\":\"int\"}]},"
+ + "\"partition-spec\":[],"
+ + "\"properties\":{\"write.format.default\":\"parquet\"},"
+ + "\"current-snapshot-id\":-1,"
+ + "\"snapshots\":[],"
+ + "\"metadata-log\":" + metadataLog + "}");
+ }
+
// ───────────────────────────── INSERT: table-derived fields
─────────────────────────────
@Test
@@ -2266,6 +2463,76 @@ public class IcebergWritePlanProviderTest {
Assertions.assertFalse(sink.isSetRewritableDeleteFileSets());
}
+ @Test
+ public void planWriteMergeSinkUsesCompleteStatementPinnedWriterMetadata() {
+ // Every merge-sink field that labels written files comes from the
statement-pinned writer metadata, so a
+ // property, spec or sort-order change after the pin cannot make files
claim metadata they were not
+ // written with. The bound write-metadata identity fence for spec and
sort order is covered separately.
+ InMemoryCatalog catalog = freshCatalog();
+ Map<String, String> tableProps = new HashMap<>();
+ tableProps.put("write.format.default", "parquet");
+ tableProps.put("write.parquet.compression-codec", "zstd");
+ tableProps.put("write.data.path", "oss://bucket/wh/db1/pinned/data");
+ tableProps.put("format-version", "3");
+ Table table = catalog.createTable(TableIdentifier.of("db1", "pinned"),
SCHEMA,
+ PartitionSpec.builderFor(SCHEMA).identity("id").build(),
tableProps);
+ table.replaceSortOrder().asc("id", NullOrder.NULLS_FIRST).commit();
+ RecordingConnectorContext ctx = contextWithStorage();
+ IcebergWritePlanProvider provider = providerFor(table, ctx);
+ WriteSession session = sessionFor(table, ctx);
+ IcebergTableHandle tableHandle = new IcebergTableHandle("db1",
"pinned");
+ provider.getWriteColumns(session, tableHandle, Optional.empty());
+ IcebergWriteSchemaContext pinned =
IcebergStatementScope.activeWriteSchema(session, "db1", "pinned")
+ .orElseThrow(AssertionError::new);
+ int pinnedSpecId = pinned.getPartitionSpec().specId();
+
+ table.updateProperties()
+ .set("write.format.default", "orc")
+ .set("write.orc.compression-codec", "zlib")
+ .set("write.data.path", "oss://bucket/wh/db1/pinned/moved")
+ .set("write.metadata.metrics.default", "none")
+ .commit();
+ table.updateSpec().removeField("id").addField("name").commit();
+ table.replaceSortOrder().desc("name").commit();
+ TIcebergMergeSink sink = provider.planWrite(session,
+ new
WriteHandle(tableHandle).writeOperation(WriteOperation.MERGE))
+ .getDataSink().getIcebergMergeSink();
+
+ Assertions.assertNotEquals(pinnedSpecId, table.spec().specId(), "the
live spec must have moved on");
+ Assertions.assertEquals(pinned.getMergeSchemaJson(),
sink.getSchemaJson());
+ Assertions.assertTrue(sink.getSchemaJson().contains("_row_id"));
+ Assertions.assertEquals(pinnedSpecId, sink.getPartitionSpecId());
+ Assertions.assertEquals(Collections.singletonMap(pinnedSpecId,
pinned.getPartitionSpecJson()),
+ sink.getPartitionSpecsJson());
+ Assertions.assertEquals(pinnedSpecId,
sink.getPartitionSpecIdForDelete());
+ Assertions.assertEquals(TFileFormatType.FORMAT_PARQUET,
sink.getFileFormat());
+ Assertions.assertEquals(TFileCompressType.ZSTD,
sink.getCompressionType());
+ Assertions.assertEquals("oss://bucket/wh/db1/pinned/data",
sink.getOriginalOutputPath());
+ Assertions.assertEquals("oss://bucket/wh/db1/pinned/data",
sink.getTableLocation());
+ Assertions.assertTrue(sink.isCollectColumnStats(), "the pinned metrics
policy still collects stats");
+ Assertions.assertEquals(1, sink.getSortFieldsSize());
+ Assertions.assertEquals(table.schema().findField("id").fieldId(),
+ sink.getSortFields().get(0).getSourceColumnId());
+ Assertions.assertTrue(sink.getSortFields().get(0).isAscending());
+ }
+
+ @Test
+ public void planWriteMergeSinkPinsUnpartitionedSpecIdForDelete() {
+ // Once the last partition field is dropped, position deletes are
written under the current unpartitioned
+ // spec. Its id is not the initial spec 0, so the delete side must
carry the pinned id, not a default.
+ Table table = partitionedSortedTable(freshCatalog());
+ table.updateSpec().removeField("id").commit();
+
+ TIcebergMergeSink sink = planMergeSink(table, contextWithStorage(),
+ new WriteHandle(new IcebergTableHandle("db1",
"t1")).writeOperation(WriteOperation.MERGE));
+
+ Assertions.assertFalse(table.spec().isPartitioned());
+ Assertions.assertNotEquals(0, table.spec().specId());
+ Assertions.assertFalse(sink.isSetPartitionSpecId());
+ Assertions.assertFalse(sink.isSetPartitionSpecsJson());
+ Assertions.assertEquals(table.spec().specId(),
sink.getPartitionSpecIdForDelete());
+ }
+
@Test
public void planWriteMergeSinkSortFieldsFromIdentitySortOrder() {
Table table = partitionedSortedTable(freshCatalog());
@@ -2604,4 +2871,43 @@ public class IcebergWritePlanProviderTest {
throw new UnsupportedOperationException();
}
}
+
+ /** Table operations over replaceable metadata, so a test can advance or
replace a table without a catalog. */
+ private static final class MutableMetadataOperations implements
TableOperations {
+ private TableMetadata metadata;
+
+ MutableMetadataOperations(TableMetadata metadata) {
+ this.metadata = metadata;
+ }
+
+ @Override
+ public TableMetadata current() {
+ return metadata;
+ }
+
+ @Override
+ public TableMetadata refresh() {
+ return metadata;
+ }
+
+ @Override
+ public void commit(TableMetadata base, TableMetadata updated) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public FileIO io() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public String metadataFileLocation(String fileName) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public LocationProvider locationProvider() {
+ throw new UnsupportedOperationException();
+ }
+ }
}
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWriterHelperTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWriterHelperTest.java
index 7a7c61de093..75160363940 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWriterHelperTest.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWriterHelperTest.java
@@ -44,8 +44,10 @@ import java.time.ZoneOffset;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Optional;
/**
* Parity oracle for the connector-resident {@link IcebergWriterHelper} — the
self-contained port of legacy
@@ -210,6 +212,51 @@ public class IcebergWriterHelperTest {
Assertions.assertEquals(FileFormat.PARQUET, df.format());
}
+ @Test
+ public void convertToWriterResultUsesPinnedMetadataAfterTableEvolution() {
+ // Files are labeled with the spec, sort order, format and metrics
policy the writer was pinned to, even after
+ // the live table evolved every one of them.
+ InMemoryCatalog catalog = new InMemoryCatalog();
+ catalog.initialize("test", Collections.emptyMap());
+ catalog.createNamespace(Namespace.of("db1"));
+ Map<String, String> properties = new HashMap<>();
+ properties.put("write.format.default", "orc");
+ properties.put("write.metadata.metrics.default", "full");
+ Table table = catalog.createTable(TableIdentifier.of("db1", "t"),
schema,
+ PartitionSpec.builderFor(schema).identity("id").build(),
properties);
+ table.replaceSortOrder().desc("age").commit();
+ IcebergWriteSchemaContext context = IcebergWriteSchemaContext.create(
+ table, "db1.t", Optional.empty(), false, false);
+ int pinnedSortOrderId = table.sortOrder().orderId();
+ table.updateSpec().removeField("id").commit();
+ table.replaceSortOrder().asc("name").commit();
+ table.updateProperties().set("write.format.default", "parquet")
+ .set("write.metadata.metrics.default", "none").commit();
+
+ ByteBuffer ageBound =
Conversions.toByteBuffer(Types.IntegerType.get(), 42);
+ TIcebergColumnStats stats = new TIcebergColumnStats();
+ stats.setLowerBounds(Collections.singletonMap(3, ageBound));
+ stats.setUpperBounds(Collections.singletonMap(3, ageBound));
+ TIcebergCommitData commitData = new TIcebergCommitData();
+ commitData.setFilePath("/path/to/pinned-data.orc");
+ commitData.setPartitionValues(Collections.singletonList("7"));
+ commitData.setRowCount(1);
+ commitData.setFileSize(128);
+ commitData.setFileContent(TFileContent.DATA);
+ commitData.setColumnStats(stats);
+
+ DataFile dataFile = IcebergWriterHelper.convertToWriterResult(
+ context, Collections.singletonList(commitData),
ZoneOffset.UTC).dataFiles()[0];
+
+ Assertions.assertNotEquals(table.spec().specId(), dataFile.specId(),
"the live spec must have moved on");
+ Assertions.assertEquals(context.getPartitionSpec().specId(),
dataFile.specId());
+ Assertions.assertEquals(Integer.valueOf(7),
dataFile.partition().get(0, Integer.class));
+ Assertions.assertEquals(Integer.valueOf(pinnedSortOrderId),
dataFile.sortOrderId());
+ Assertions.assertEquals(FileFormat.ORC, dataFile.format());
+ Assertions.assertEquals(ageBound, dataFile.lowerBounds().get(3));
+ Assertions.assertEquals(ageBound, dataFile.upperBounds().get(3));
+ }
+
@Test
public void convertToWriterResultDecodesBinaryAndUuidPartitionValues() {
// BE commits binary-like identity partitions as 0x hex; UUID may also
arrive as canonical text.
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/ConnectorWriteSchemaUtilsTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/ConnectorWriteSchemaUtilsTest.java
index a998880094e..b61a51c1ead 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/ConnectorWriteSchemaUtilsTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/ConnectorWriteSchemaUtilsTest.java
@@ -18,19 +18,44 @@
package org.apache.doris.nereids.trees.plans.commands;
import org.apache.doris.catalog.Column;
+import org.apache.doris.catalog.PrimitiveType;
import org.apache.doris.catalog.ScalarType;
+import org.apache.doris.catalog.Type;
+import org.apache.doris.datasource.ExternalTable;
import org.apache.doris.nereids.analyzer.UnboundSlot;
+import org.apache.doris.nereids.exceptions.AnalysisException;
import org.apache.doris.nereids.parser.ParserTestBase;
+import org.apache.doris.nereids.rules.expression.rules.FoldConstantRuleOnFE;
import org.apache.doris.nereids.trees.expressions.Cast;
+import org.apache.doris.nereids.trees.expressions.Default;
import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.SlotReference;
+import org.apache.doris.nereids.trees.expressions.StatementScopeIdGenerator;
+import org.apache.doris.nereids.trees.expressions.literal.BigIntLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.BooleanLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.DecimalV3Literal;
+import org.apache.doris.nereids.trees.expressions.literal.DoubleLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.FloatLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.Literal;
+import org.apache.doris.nereids.trees.expressions.literal.TinyIntLiteral;
+import org.apache.doris.nereids.types.DataType;
+import org.apache.doris.nereids.types.DecimalV3Type;
+import org.apache.doris.nereids.util.TypeCoercionUtils;
import org.apache.doris.qe.ConnectContext;
import org.apache.doris.qe.SqlModeHelper;
+import com.google.common.collect.ImmutableList;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import java.math.BigDecimal;
+import java.util.List;
public class ConnectorWriteSchemaUtilsTest extends ParserTestBase {
+ private static final List<String> TARGET_NAME_PARTS =
ImmutableList.of("catalog", "database", "test_table");
+
@Test
public void connectorHexDefaultParsesIdenticallyInBothBackslashModes() {
Column column = new Column("path", ScalarType.createStringType());
@@ -68,4 +93,88 @@ public class ConnectorWriteSchemaUtilsTest extends
ParserTestBase {
"non-finite default must not be parsed as an identifier: "
+ defaultSql);
}
}
+
+ @Test
+ public void defaultReferencesResolveAgainstPinnedWriteColumns() {
+ // DEFAULT(column) in DML resolves the statement-pinned write default;
a name absent from the pinned writer
+ // schema is an error rather than a guess.
+ ExternalTable target = targetTable();
+ List<Column> pinned = ImmutableList.of(pinnedColumn("score", 1, "9"));
+
+ Assertions.assertEquals(new TinyIntLiteral((byte) 9),
ConnectorWriteSchemaUtils.resolveDefaultReferences(
+ new Default(new UnboundSlot("score")), pinned, target,
ConnectContext.get(),
+ TARGET_NAME_PARTS, null));
+ AnalysisException exception =
Assertions.assertThrows(AnalysisException.class,
+ () -> ConnectorWriteSchemaUtils.resolveDefaultReferences(
+ new Default(new UnboundSlot("missing")), pinned,
target, ConnectContext.get(),
+ TARGET_NAME_PARTS, null));
+ Assertions.assertTrue(exception.getMessage().contains("missing"),
exception.getMessage());
+ }
+
+ @Test
+ public void defaultOfBoundSlotDoesNotFallBackToReusedColumnName() {
+ // A slot bound to a dropped and re-added column must not borrow the
pinned default of the old field that
+ // reused its name: a bound slot resolves DEFAULT by field id only.
+ ExternalTable target = targetTable();
+ Column pinnedField = pinnedColumn("reused_name", 7, "9");
+ List<Column> pinned = ImmutableList.of(pinnedField);
+ Column replacement = new Column("reused_name",
ScalarType.createType(PrimitiveType.INT));
+ replacement.setUniqueId(8);
+
+ Assertions.assertEquals(new TinyIntLiteral((byte) 9),
ConnectorWriteSchemaUtils.resolveDefaultReferences(
+ new Default(boundSlot(target, pinnedField)), pinned, target,
ConnectContext.get(),
+ TARGET_NAME_PARTS, null));
+ AnalysisException exception =
Assertions.assertThrows(AnalysisException.class,
+ () -> ConnectorWriteSchemaUtils.resolveDefaultReferences(
+ new Default(boundSlot(target, replacement)), pinned,
target, ConnectContext.get(),
+ TARGET_NAME_PARTS, null));
+ Assertions.assertTrue(exception.getMessage().contains("Cannot find
column information for DEFAULT"),
+ exception.getMessage());
+ }
+
+ @Test
+ public void connectorScalarDefaultsCastLosslesslyToTheirWriteColumnTypes()
{
+ // Connector write-default SQL is untyped text; binding casts it to
the write column type, which must keep
+ // the whole value: BIGINT beyond INT range, decimal scale,
microseconds and dates.
+ Assertions.assertEquals(new BigIntLiteral(4_900_000_001L),
+ bindDefault(Type.BIGINT, "4900000001"));
+ Assertions.assertEquals(new
DecimalV3Literal(DecimalV3Type.createDecimalV3Type(20, 4),
+ new BigDecimal("98765.4321")),
bindDefault(ScalarType.createDecimalV3Type(20, 4), "98765.4321"));
+ Assertions.assertEquals(new DoubleLiteral(456.75),
bindDefault(Type.DOUBLE, "456.75"));
+ Assertions.assertEquals(new FloatLiteral(13.5F),
bindDefault(Type.FLOAT, "13.5"));
+ Assertions.assertEquals(BooleanLiteral.FALSE,
bindDefault(Type.BOOLEAN, "false"));
+ Assertions.assertEquals("2025-01-18",
+ ((Literal) bindDefault(ScalarType.createDateV2Type(),
"'2025-01-18'")).getStringValue());
+ Assertions.assertEquals("2025-01-18 01:02:03.654321", ((Literal)
bindDefault(
+ ScalarType.createDatetimeV2Type(6), "'2025-01-18
01:02:03.654321'")).getStringValue());
+ }
+
+ /** Parse, cast and fold a connector default the way binding fills an
omitted column. */
+ private static Expression bindDefault(Type columnType, String defaultSql) {
+ Column column = new Column("value", columnType);
+ column.setConnectorDefaultValueSql(defaultSql);
+ DataType targetType = DataType.fromCatalogType(columnType);
+ Expression bound =
FoldConstantRuleOnFE.evaluateWithoutContext(TypeCoercionUtils.castIfNotSameType(
+ ConnectorWriteSchemaUtils.resolveDefault(column), targetType));
+ Assertions.assertTrue(bound instanceof Literal, defaultSql + " -> " +
bound);
+ Assertions.assertEquals(targetType, bound.getDataType(), defaultSql);
+ return bound;
+ }
+
+ private static Column pinnedColumn(String name, int uniqueId, String
defaultSql) {
+ Column column = new Column(name,
ScalarType.createType(PrimitiveType.INT));
+ column.setUniqueId(uniqueId);
+ column.setConnectorDefaultValueSql(defaultSql);
+ return column;
+ }
+
+ private static ExternalTable targetTable() {
+ ExternalTable table = Mockito.mock(ExternalTable.class);
+ Mockito.when(table.getId()).thenReturn(9L);
+ return table;
+ }
+
+ private static SlotReference boundSlot(ExternalTable table, Column column)
{
+ return SlotReference.fromColumn(StatementScopeIdGenerator.newExprId(),
table, column, ImmutableList.of());
+ }
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/ExternalRowLevelMergePlanBuilderTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/ExternalRowLevelMergePlanBuilderTest.java
index e6287d502c2..ac13299d347 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/ExternalRowLevelMergePlanBuilderTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/ExternalRowLevelMergePlanBuilderTest.java
@@ -34,12 +34,15 @@ import
org.apache.doris.nereids.trees.expressions.SlotReference;
import
org.apache.doris.nereids.trees.expressions.functions.scalar.ShortCircuitIf;
import org.apache.doris.nereids.trees.expressions.literal.BooleanLiteral;
import org.apache.doris.nereids.trees.expressions.literal.IntegerLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.Literal;
import org.apache.doris.nereids.trees.expressions.literal.StringLiteral;
import org.apache.doris.nereids.trees.plans.RelationId;
import org.apache.doris.nereids.trees.plans.commands.merge.MergeMatchedClause;
+import
org.apache.doris.nereids.trees.plans.commands.merge.MergeNotMatchedClause;
import
org.apache.doris.nereids.trees.plans.logical.LogicalExternalRowLevelMergeSink;
import org.apache.doris.nereids.trees.plans.logical.LogicalOneRowRelation;
import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
import org.apache.doris.nereids.types.ConnectorComputeVariantType;
import org.apache.doris.nereids.types.IntegerType;
import org.apache.doris.nereids.types.VarBinaryType;
@@ -129,6 +132,58 @@ public class ExternalRowLevelMergePlanBuilderTest {
metadataOrder.verify(table).getWriteSchemaSnapshot();
}
+ @Test
+ public void
matchedUpdateKeepsTargetValuesWhileNotMatchedInsertUsesWriteDefaults() {
+ // A WHEN MATCHED UPDATE keeps the target row's value for every
unassigned column; only the WHEN NOT MATCHED
+ // INSERT branch fills an omitted column with its pinned write default.
+ ConnectContext ctx = new ConnectContext();
+ ctx.setStatementContext(new StatementContext());
+ ctx.setThreadLocalInfo();
+ LogicalPlan source = new LogicalOneRowRelation(new RelationId(0),
+ ImmutableList.of(new UnboundAlias(new IntegerLiteral(1),
"dummy")));
+ ExternalRowLevelMergePlanBuilder builder = new
ExternalRowLevelMergePlanBuilder(
+ ImmutableList.of("test_catalog", "test_db", "test_table"),
+ Optional.empty(), Optional.empty(), source,
BooleanLiteral.TRUE,
+ ImmutableList.of(new MergeMatchedClause(Optional.empty(),
+ ImmutableList.of(new EqualTo(new UnboundSlot("c1"),
new IntegerLiteral(1))), false)),
+ ImmutableList.of(new MergeNotMatchedClause(Optional.empty(),
ImmutableList.of("c1"),
+ ImmutableList.of(new UnboundAlias(new
IntegerLiteral(5))))));
+ Column rowId = new Column(Column.ICEBERG_ROWID_COL,
ScalarType.createStringType());
+ rowId.setIsVisible(false);
+ Column assigned = new Column("c1",
ScalarType.createType(PrimitiveType.INT));
+ assigned.setConnectorDefaultValueSql("4141");
+ Column omitted = new Column("c2",
ScalarType.createType(PrimitiveType.INT));
+ omitted.setConnectorDefaultValueSql("4242");
+ PluginDrivenExternalTable table =
Mockito.mock(PluginDrivenExternalTable.class);
+ PluginDrivenExternalTable.WriteSchemaSnapshot writeSchema =
+
Mockito.mock(PluginDrivenExternalTable.WriteSchemaSnapshot.class);
+ Mockito.when(table.getName()).thenReturn("test_table");
+
Mockito.doReturn(Mockito.mock(ExternalDatabase.class)).when(table).getDatabase();
+ Mockito.doReturn(writeSchema).when(table).getWriteSchemaSnapshot();
+ Mockito.doReturn(Optional.of(ImmutableList.of(assigned, omitted)))
+ .when(table).resolveWriteColumns(Optional.empty());
+ Mockito.doReturn(ImmutableList.of(assigned, omitted,
rowId)).when(table).getFullSchema();
+
+ LogicalPlan plan = builder.buildMergePlan(ctx, table);
+
+ Expression c2 = ((LogicalProject<?>)
plan.child(0)).getProjects().stream()
+ .filter(project -> project instanceof UnboundAlias
+ && ((UnboundAlias)
project).getAlias().equals(Optional.of("c2")))
+ .findFirst().orElseThrow(AssertionError::new).child(0);
+ // The outermost selection picks the NOT MATCHED insert branch; its
else-branch selects the MATCHED update.
+ Expression insertBranch = c2.child(1);
+ Expression updateBranch = c2.child(2).child(1);
+ Assertions.assertTrue(insertBranch.anyMatch(expression ->
isLiteral(expression, "4242")), c2.toString());
+ Assertions.assertTrue(updateBranch.anyMatch(expression -> expression
instanceof UnboundSlot
+ && "c2".equalsIgnoreCase(((UnboundSlot)
expression).getNameParts().get(
+ ((UnboundSlot) expression).getNameParts().size() -
1))), c2.toString());
+ Assertions.assertFalse(updateBranch.anyMatch(expression ->
isLiteral(expression, "4242")), c2.toString());
+ }
+
+ private static boolean isLiteral(Object expression, String value) {
+ return expression instanceof Literal && value.equals(((Literal)
expression).getStringValue());
+ }
+
@Test
public void mergeBranchesUseThePinnedWriterType() {
VarBinaryType uuidType = VarBinaryType.createVarBinaryType(16);
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/ExternalRowLevelUpdatePlanBuilderTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/ExternalRowLevelUpdatePlanBuilderTest.java
index 7ebc2e87af7..e78a923492c 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/ExternalRowLevelUpdatePlanBuilderTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/ExternalRowLevelUpdatePlanBuilderTest.java
@@ -26,8 +26,10 @@ import
org.apache.doris.datasource.plugin.PluginDrivenExternalTable;
import org.apache.doris.nereids.StatementContext;
import org.apache.doris.nereids.analyzer.UnboundAlias;
import org.apache.doris.nereids.analyzer.UnboundSlot;
+import org.apache.doris.nereids.trees.expressions.EqualTo;
import org.apache.doris.nereids.trees.expressions.NamedExpression;
import org.apache.doris.nereids.trees.expressions.literal.IntegerLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.Literal;
import org.apache.doris.nereids.trees.plans.RelationId;
import org.apache.doris.nereids.trees.plans.commands.merge.MergeOperation;
import
org.apache.doris.nereids.trees.plans.logical.LogicalExternalRowLevelMergeSink;
@@ -146,6 +148,42 @@ public class ExternalRowLevelUpdatePlanBuilderTest {
metadataOrder.verify(table).getWriteSchemaSnapshot();
}
+ @Test
+ public void
updateKeepsUnassignedTargetValuesInsteadOfInjectingWriteDefaults() {
+ // UPDATE rewrites only the assigned columns. An unassigned column
keeps the target row's value even when the
+ // pinned writer schema gives it a write default; write defaults only
fill values a write leaves out.
+ ConnectContext ctx = new ConnectContext();
+ ctx.setStatementContext(new StatementContext());
+ ctx.setThreadLocalInfo();
+ LogicalPlan basePlan = new LogicalOneRowRelation(new RelationId(0),
+ ImmutableList.of(new UnboundAlias(new IntegerLiteral(1),
"dummy")));
+ ExternalRowLevelUpdatePlanBuilder builder = new
ExternalRowLevelUpdatePlanBuilder(
+ ImmutableList.of("test_catalog", "test_db", "test_table"),
null, ImmutableList.of(), basePlan);
+ Column assigned = new Column("c1",
ScalarType.createType(PrimitiveType.INT));
+ assigned.setConnectorDefaultValueSql("4141");
+ Column unassigned = new Column("c2",
ScalarType.createType(PrimitiveType.INT));
+ unassigned.setConnectorDefaultValueSql("4242");
+ PluginDrivenExternalTable table =
Mockito.mock(PluginDrivenExternalTable.class);
+ PluginDrivenExternalTable.WriteSchemaSnapshot writeSchema =
+
Mockito.mock(PluginDrivenExternalTable.WriteSchemaSnapshot.class);
+ Mockito.when(table.getName()).thenReturn("test_table");
+
Mockito.doReturn(Mockito.mock(ExternalDatabase.class)).when(table).getDatabase();
+ Mockito.doReturn(writeSchema).when(table).getWriteSchemaSnapshot();
+ Mockito.doReturn(Optional.of(ImmutableList.of(assigned, unassigned)))
+ .when(table).resolveWriteColumns(Optional.empty());
+
+ LogicalPlan plan = builder.buildMergePlan(ctx, basePlan,
+ ImmutableList.of(new EqualTo(new UnboundSlot("c1"), new
IntegerLiteral(1))), table);
+
+ List<NamedExpression> outputs = ((LogicalProject<?>)
plan.child(0)).getProjects();
+ Assertions.assertTrue(outputs.stream().anyMatch(output -> output
instanceof UnboundSlot
+ && "c2".equalsIgnoreCase(((UnboundSlot)
output).getNameParts().get(
+ ((UnboundSlot) output).getNameParts().size() - 1))),
outputs.toString());
+ Assertions.assertFalse(outputs.stream().anyMatch(output ->
output.anyMatch(expression ->
+ expression instanceof Literal && (((Literal)
expression).getStringValue().equals("4141")
+ || ((Literal)
expression).getStringValue().equals("4242")))), outputs.toString());
+ }
+
@Test
public void testBuildUpdateSelectItemsSkipsHiddenColumns() {
ExternalRowLevelUpdatePlanBuilder command = new
ExternalRowLevelUpdatePlanBuilder(
diff --git a/gensrc/thrift/PlanNodes.thrift b/gensrc/thrift/PlanNodes.thrift
index a3e997ddeb6..1cb49705c5b 100644
--- a/gensrc/thrift/PlanNodes.thrift
+++ b/gensrc/thrift/PlanNodes.thrift
@@ -362,6 +362,9 @@ struct TIcebergFileDesc {
// Only for format_version >= 3, the sequence number which last updated
this file.
11: optional i64 last_updated_sequence_number;
12: optional string serialized_split;
+ // Reserved for branch-4.1 / branch-4.2 field `13: optional
ExternalTableSchema.TSchema
+ // equality_delete_schema` (the split-local schema of dropped
equality-delete keys). Master
+ // carries those schemas in TFileScanRangeParams.history_schema_info. Do
not reuse this id.
}
struct TPaimonDeletionFileDesc {
diff --git
a/regression-test/data/external_table_p0/iceberg/test_iceberg_write_default.out
b/regression-test/data/external_table_p0/iceberg/test_iceberg_write_default.out
new file mode 100644
index 00000000000..65a51659d9d
--- /dev/null
+++
b/regression-test/data/external_table_p0/iceberg/test_iceberg_write_default.out
@@ -0,0 +1,4 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !desc_write_default --
+\N
+
diff --git
a/regression-test/suites/external_table_p0/iceberg/test_iceberg_write_default.groovy
b/regression-test/suites/external_table_p0/iceberg/test_iceberg_write_default.groovy
index 9759a32461e..f356ea3103d 100644
---
a/regression-test/suites/external_table_p0/iceberg/test_iceberg_write_default.groovy
+++
b/regression-test/suites/external_table_p0/iceberg/test_iceberg_write_default.groovy
@@ -70,6 +70,15 @@ suite("test_iceberg_write_default", "p0,external") {
assertTrue(cRow != null, "column c must exist in DESC: " +
descRows.toString())
assertFalse(cRow.toString().contains("42"),
"DESC must not show connector write default 42, got: " +
cRow.toString())
+ // information_schema reads the cached catalog Column too, so
COLUMN_DEFAULT stays NULL.
+ qt_desc_write_default """
+ SELECT COLUMN_DEFAULT
+ FROM ${catalog_name}.information_schema.columns
+ WHERE TABLE_CATALOG = '${catalog_name}'
+ AND TABLE_SCHEMA = '${db}'
+ AND TABLE_NAME = '${tbl}'
+ AND COLUMN_NAME = 'c'
+ """
// 2) INSERT with column c omitted must apply the write default.
sql """ INSERT INTO ${tbl} (id) VALUES (1) """
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
