From 77adabfbaeb57aa5e35d075bcd39a6ef02ad370c Mon Sep 17 00:00:00 2001 From: daidai Date: Tue, 22 Sep 2026 16:37:27 +0800 Subject: [PATCH] [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 ++++- .../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 1ba1686e0375c5..3c1c098af93610 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 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 c9b033dcb000cb..234531335aff25 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>& 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 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& ids, const std::vector& scores, const std::vector& 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 projected_columns; + projected_columns.push_back( + make_table_column(0, "x", make_nullable(std::make_shared()))); + + 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> 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(*full_column); + const auto& data = + assert_cast(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> {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";