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";