This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/doris.git
commit f051ca12cfe1e54db82d09b75af2d418db7e587a
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Wed Sep 23 08:42:54 2026 +0800
branch-4.1: [fix](iceberg) Promote equality-delete keys to current schema
type #68267 (#68383)
Cherry-picked from #68267
Co-authored-by: daidai <[email protected]>
---
be/src/format_v2/table/iceberg_reader.cpp | 29 ++++++-
be/test/format_v2/table/iceberg_reader_test.cpp | 102 ++++++++++++++++++++++++
2 files changed, 129 insertions(+), 2 deletions(-)
diff --git a/be/src/format_v2/table/iceberg_reader.cpp
b/be/src/format_v2/table/iceberg_reader.cpp
index 1ba1686e037..3c1c098af93 100644
--- a/be/src/format_v2/table/iceberg_reader.cpp
+++ b/be/src/format_v2/table/iceberg_reader.cpp
@@ -1675,7 +1675,17 @@ Status
IcebergTableReader::_resolve_equality_delete_fields(
return Status::NotSupported(
"Iceberg equality delete does not support complex column
{}", field->name);
}
- const auto key_type = path.size() > 1 ? make_nullable(field->type) :
field->type;
+ // Equality comparison must run in the promoted (current snapshot
schema) type domain.
+ // Narrowing a wider data key into a historical delete-file type is
lossy: an INT overflow
+ // becomes NULL, and NULL-safe equality then matches a NULL delete
key. Delete values are
+ // promoted into this same domain when the delete file is read.
+ const DataTypePtr delete_file_type =
+ path.size() > 1 ? make_nullable(field->type) : field->type;
+ DataTypePtr key_type = delete_file_type;
+ if (auto table_field = _find_table_column_by_field_id(field_id,
delete_file_type, true);
+ table_field.has_value() && table_field->type != nullptr) {
+ key_type = table_field->type;
+ }
delete_paths->push_back(std::move(path));
result->field_ids.push_back(field_id);
result->field_names.push_back(field->name);
@@ -1721,11 +1731,21 @@ Status
IcebergTableReader::_load_equality_delete_file(const TIcebergDeleteFileDe
std::vector<VExprContextSPtr> key_exprs;
key_exprs.reserve(delete_paths.size());
RowDescriptor row_desc;
- for (const auto& path : delete_paths) {
+ for (size_t index = 0; index < delete_paths.size(); ++index) {
+ const auto& path = delete_paths[index];
const auto root_column_id =
format::LocalColumnId(path.front()->file_local_id());
VExprSPtr key_expr;
RETURN_IF_ERROR(build_equality_delete_key_expr(
path, request->local_positions.at(root_column_id).value(),
&key_expr));
+ const auto& key_type = result->key_types[index];
+ if (!key_expr->data_type()->equals(*key_type)) {
+ // Historical delete values are promoted into the comparison
domain. For an
+ // Iceberg-legal type promotion this cast is always widening, so
it cannot turn a
+ // non-NULL value into NULL.
+ auto cast_expr = Cast::create_shared(key_type);
+ cast_expr->add_child(key_expr);
+ key_expr = std::move(cast_expr);
+ }
auto context = VExprContext::create_shared(std::move(key_expr));
RETURN_IF_ERROR(context->prepare(_runtime_state, row_desc));
RETURN_IF_ERROR(context->open(_runtime_state));
@@ -1768,6 +1788,11 @@ Status
IcebergTableReader::_read_equality_delete_file(const TIcebergDeleteFileDe
}
std::ostringstream cache_key;
cache_key << _delete_file_cache_key("iceberg_v2_equality_delete_",
delete_file.path);
+ if (scan_params.__isset.current_schema_id) {
+ // The promoted comparison type depends on the current snapshot
schema, so a cached filter
+ // must not be reused across schemas.
+ cache_key << ":schema=" << scan_params.current_schema_id;
+ }
cache_key << ':' << delete_file.field_ids.size();
for (const auto field_id : delete_file.field_ids) {
cache_key << ':' << field_id;
diff --git a/be/test/format_v2/table/iceberg_reader_test.cpp
b/be/test/format_v2/table/iceberg_reader_test.cpp
index c9b033dcb00..234531335af 100644
--- a/be/test/format_v2/table/iceberg_reader_test.cpp
+++ b/be/test/format_v2/table/iceberg_reader_test.cpp
@@ -932,6 +932,38 @@ void
write_iceberg_equality_delete_bigint_parquet_file(const std::string& file_p
builder.build()));
}
+void write_nullable_int64_parquet_file(const std::string& file_path, int32_t
field_id,
+ const std::string& field_name,
+ const
std::vector<std::optional<int64_t>>& values) {
+ const auto metadata =
+ arrow::key_value_metadata({"PARQUET:field_id"},
{std::to_string(field_id)});
+ auto schema = arrow::schema({
+ arrow::field(field_name, arrow::int64(),
true)->WithMetadata(metadata),
+ });
+ arrow::Int64Builder value_builder;
+ for (const auto& value : values) {
+ if (value.has_value()) {
+ ASSERT_TRUE(value_builder.Append(*value).ok());
+ } else {
+ ASSERT_TRUE(value_builder.AppendNull().ok());
+ }
+ }
+ auto value_result = value_builder.Finish();
+ ASSERT_TRUE(value_result.ok()) << value_result.status();
+ auto table = arrow::Table::Make(schema, {*value_result});
+
+ auto file_result = arrow::io::FileOutputStream::Open(file_path);
+ ASSERT_TRUE(file_result.ok()) << file_result.status();
+ std::shared_ptr<arrow::io::FileOutputStream> out = *file_result;
+
+ ::parquet::WriterProperties::Builder builder;
+ builder.version(::parquet::ParquetVersion::PARQUET_2_6);
+ builder.data_page_version(::parquet::ParquetDataPageVersion::V2);
+ builder.compression(::parquet::Compression::UNCOMPRESSED);
+ PARQUET_THROW_NOT_OK(::parquet::arrow::WriteTable(*table,
arrow::default_memory_pool(), out, 1,
+ builder.build()));
+}
+
void write_int_pair_parquet_file(const std::string& file_path, const
std::vector<int32_t>& ids,
const std::vector<int32_t>& scores,
const std::vector<std::string>& values,
@@ -3679,6 +3711,76 @@ TEST(IcebergV2ReaderTest,
IcebergEqualityDeleteCastsDataColumnToDeleteKeyType) {
std::filesystem::remove_all(test_dir);
}
+TEST(IcebergV2ReaderTest,
IcebergEqualityDeletePromotesHistoricalDeleteKeyToCurrentType) {
+ const auto test_dir =
+ std::filesystem::temp_directory_path() /
"doris_iceberg_equality_delete_promotion_test";
+ std::filesystem::remove_all(test_dir);
+ std::filesystem::create_directories(test_dir);
+
+ const auto file_path = (test_dir / "split.parquet").string();
+ const auto delete_file_path = (test_dir /
"equality-delete.parquet").string();
+ // The data file was written after INT -> LONG promotion and holds a value
outside the old INT
+ // domain. The delete file was written under the old INT schema and
deletes the NULL key.
+ write_nullable_int64_parquet_file(
+ file_path, 0, "x", {std::nullopt, int64_t {0}, int64_t {1},
int64_t {4294967296}});
+ write_iceberg_null_equality_delete_parquet_file(delete_file_path, 0, "x");
+
+ std::vector<ColumnDefinition> projected_columns;
+ projected_columns.push_back(
+ make_table_column(0, "x",
make_nullable(std::make_shared<DataTypeInt64>())));
+
+ RuntimeProfile profile("test_profile");
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ auto scan_params = make_local_parquet_scan_params();
+
scan_params.__set_iceberg_scan_semantics_version(ICEBERG_SCAN_SEMANTICS_VERSION_2);
+ scan_params.__set_current_schema_id(100);
+ scan_params.__set_history_schema_info({external_schema(
+ 100, {external_schema_field("x", 0, {}, std::nullopt,
+
external_primitive_type(TPrimitiveType::BIGINT), false,
+ true)})});
+ io::FileReaderStats file_reader_stats;
+ io::FileCacheStatistics file_cache_stats;
+ auto io_ctx = make_io_context(&file_reader_stats, &file_cache_stats);
+ ShardedKVCache cache(1);
+ doris::format::iceberg::IcebergTableReader reader;
+ init_iceberg_reader(&reader, projected_columns, &scan_params, io_ctx,
&state, &profile);
+
+ auto split_options = build_split_options(file_path);
+ split_options.cache = &cache;
+
split_options.current_range.__set_table_format_params(make_iceberg_table_format_desc(
+ file_path, {make_iceberg_equality_delete_file(delete_file_path,
{0})}));
+ ASSERT_TRUE(reader.prepare_split(split_options).ok());
+
+ std::vector<std::optional<int64_t>> values;
+ bool eos = false;
+ while (!eos) {
+ Block block = build_table_block(projected_columns);
+ ASSERT_TRUE(reader.get_block(&block, &eos).ok());
+ if (block.rows() == 0) {
+ continue;
+ }
+ const auto full_column =
block.get_by_position(0).column->convert_to_full_column_if_const();
+ const auto& nullable_column = assert_cast<const
ColumnNullable&>(*full_column);
+ const auto& data =
+ assert_cast<const
ColumnInt64&>(nullable_column.get_nested_column()).get_data();
+ for (size_t row = 0; row < nullable_column.size(); ++row) {
+ if (nullable_column.get_null_map_data()[row] != 0) {
+ values.push_back(std::nullopt);
+ } else {
+ values.push_back(data[row]);
+ }
+ }
+ }
+
+ // The NULL row is deleted, while 4294967296 must not be narrowed into the
old INT domain and
+ // deleted together with it.
+ EXPECT_EQ(values, (std::vector<std::optional<int64_t>> {int64_t {0},
int64_t {1},
+ int64_t
{4294967296}}));
+
+ ASSERT_TRUE(reader.close().ok());
+ std::filesystem::remove_all(test_dir);
+}
+
TEST(IcebergV2ReaderTest,
IcebergEqualityDeleteMatchesNullForMissingDataColumn) {
const auto test_dir = std::filesystem::temp_directory_path() /
"doris_iceberg_equality_delete_missing_column_test";
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
