This is an automated email from the ASF dual-hosted git repository.
Gabriel39 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new dc4f9985337 [fix](iceberg) Promote equality-delete keys to current
schema type (#68267)
dc4f9985337 is described below
commit dc4f99853371b1cf2a55c8ae090f06d8fde74aee
Author: daidai <[email protected]>
AuthorDate: Tue Sep 22 16:37:27 2026 +0800
[fix](iceberg) Promote equality-delete keys to current schema type (#68267)
### What problem does this PR solve?
Issue Number: N/A
Related PR: N/A
Problem Summary:
Iceberg equality deletes are compared in the physical type of the delete
file.
When a column is legally promoted (for example INT to LONG), the data
file can
hold values outside the old type, but the data key was cast into that
old type.
An out-of-range value becomes NULL and then matches a NULL
equality-delete key
through NULL-safe equality, so a valid row is silently deleted. With
`enable_strict_cast` the same cast makes the whole scan fail with
`Value ... out of range for type int`.
Resolve the comparison type from the current snapshot schema instead of
the
delete file, and promote historical delete values into that domain. With
both
sides in the wider type the cast is always lossless for an Iceberg-legal
type
promotion, so a non-NULL key can no longer turn into NULL. Include the
current
schema id in the delete-file cache key because the loaded block now
depends on
the comparison type.
Covered by a new BE unit test: a data file holding LONG `NULL, 0, 1,
4294967296`
plus a NULL equality-delete key written under the old INT schema. Before
the fix
the scan returned `{0, 1}`; after the fix it returns `{0, 1,
4294967296}`.
### Release note
Fix Iceberg equality deletes being evaluated in the delete file's
historical
type, which could silently drop valid rows after a column type
promotion.
### Check List (For Author)
- Test
- [ ] Regression test
- [x] Unit Test
- [ ] Manual test (add detailed scripts or steps below)
- [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
- [ ] Previous test can cover this change.
- [ ] No code files have been changed.
- [ ] Other reason
- Behavior changed:
- [ ] No.
- [x] Yes.
- Does this need documentation?
- [x] No.
- [ ] Yes.
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label
---
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 e50c77597ff..2a3fa1b1a49 100644
--- a/be/src/format_v2/table/iceberg_reader.cpp
+++ b/be/src/format_v2/table/iceberg_reader.cpp
@@ -1768,7 +1768,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);
@@ -1814,11 +1824,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));
@@ -1862,6 +1882,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 f91c237f575..fa07e2d3135 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,
@@ -3772,6 +3804,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]
