[fix](be) Compare Iceberg V1 equality-delete keys in the promoted type (#69691)

### What problem does this PR solve?

Issue Number: None

Related PR: #66538 (branch-4.1), #68267

Problem Summary:

**In short.** This brings the V1-reader part of branch-4.1 #66538
(Iceberg V3 default values) to master: Iceberg equality deletes keep
working after the type of their key column is promoted (INT to BIGINT,
FLOAT to DOUBLE, a wider DECIMAL precision). It does not copy
branch-4.1's approach, which compares a key in the delete file's older
type; it compares in the scan schema's type, as #68267 already does in
the V2 reader.

**Background.** Iceberg allows widening a column's type, and files keep
the type they were written with: an equality delete written while `id`
was `int` stores its key as INT32, and so does a data file written then.
The V1 reader (`enable_file_scanner_v2 = false`) loads each delete file
into a delete block and compares it with the data rows, which the
Parquet and ORC readers already return in the table's current type.

**The problem, and what it cost.** The V1 reader compared every key in
the type of the delete file it came from. After a promotion, any of
these fails:

| Case (V1 reader, `id` or `payload.existing` promoted from INT to
BIGINT) | master |
|---|---|
| `SELECT id`, delete written before the promotion | Query fails: `Bad
cast from type ColumnVector<BIGINT> to ColumnVector<INT>` |
| `SELECT id`, one delete written before the promotion and one after |
Query fails: the INT and BIGINT keys are merged into one delete block
(debug builds abort on the block's type check) |
| `SELECT v` (key not projected), same two deletes | Same |
| `SELECT id`, delete on `payload.existing` written after the promotion,
data file written before it | The INT values of the old file are copied
into a BIGINT column: debug builds abort with `Bad cast`, release builds
read the wrong bytes and keep or delete the wrong rows |

branch-4.1 / branch-4.2 handle these by casting the data key down to the
delete file's type. That narrowing is what #68267 fixed in the V2
reader: a data file written after the promotion can hold 4294967296,
which becomes NULL as INT and is then deleted by the NULL key of a
delete file written before the promotion (equality deletes use null-safe
equality); with `enable_strict_cast` the whole scan fails instead.

**How this PR fixes it.**
- The type a key is compared in is the field's type in the scan schema
(`equality_delete_key_type`, the same rule as V2's
`restore_current_primitive_type`). A plan without a scan schema keeps
the delete file's type, as before, and STRING / VARBINARY keys keep
theirs too, since they match by bytes.
- Delete keys, and nested keys extracted from the physical root of an
older data file, are widened to that type with
`converter::ColumnTypeConverter`, the converter the V1 Parquet and ORC
readers already apply to the data column of a promoted field, so both
sides hold the same value (`widen_equality_delete_key`).
`NestedEqualityDeleteColumn` records the leaf's source type for that.
- Delete blocks and hidden key carriers are built in the comparison
type, so deletes written before and after a promotion share one block.

**Results.**

| Case (V1 reader) | master | This PR |
|---|---|---|
| `SELECT id`, INT delete `id = 2`, data 1, 2, 3 | `Bad cast` | 1, 3 |
| `SELECT id`, INT delete `id = 2` + BIGINT delete `id = 3` | aborts
(debug) / fails | 1 |
| `SELECT v`, same two deletes | aborts (debug) / fails | 10 |
| `SELECT id`, BIGINT delete `payload.existing = 20` on an old INT file
| `Bad cast` (debug) / wrong rows (release) | 1, 3 |
| `SELECT id`, data `NULL, 0, 1, 4294967296`, INT delete holding NULL |
`Bad cast` (branch-4.x: 0, 1) | 0, 1, 4294967296 |

be/src/format/table/equality_delete.cpp

@@ -17,23 +17,82 @@
 
 #include "format/table/equality_delete.h"
 
+#include <gen_cpp/ExternalTableSchema_types.h>
+
+#include "core/assert_cast.h"
 #include "core/column/column_const.h"
 #include "core/column/column_nullable.h"
+#include "core/data_type/data_type_factory.hpp"
 #include "exprs/create_predicate_function.h"
+#include "format/column_type_convert.h"
 #include "util/hash_util.hpp"
 
 namespace doris {
 
-namespace {
+DataTypePtr equality_delete_key_type(const schema::external::TField* field,
+                                     const DataTypePtr& delete_key_type) {
+    DORIS_CHECK(delete_key_type->is_nullable());
+    // A plan from an FE that sends no scan schema, or no type for the field, leaves only the delete
+    // file's own type, which is how keys were compared before.
+    if (field == nullptr || !field->__isset.type || is_equality_delete_byte_type(delete_key_type)) {
+        return delete_key_type;
+    }
+    const auto primitive_type = thrift_to_type(field->type.type);
+    DORIS_CHECK(!is_complex_type(primitive_type));
+    return DataTypeFactory::instance().create_data_type(
+            primitive_type, true, field->type.__isset.precision ? field->type.precision : 0,
+            field->type.__isset.scale ? field->type.scale : 0,
+            field->type.__isset.len ? field->type.len : -1);
+}
 
-// Iceberg BINARY / FIXED keys are read as VARBINARY and string keys as STRING, and the delete file and
-// the data file may carry the same key in different ones of these classes (a delete block holding
-// STRING bytes for a VARBINARY data column, say). Keys of these types therefore match by their bytes.
 bool is_equality_delete_byte_type(const DataTypePtr& type) {
     const auto primitive_type = remove_nullable(type)->get_primitive_type();
     return is_string_type(primitive_type) || is_varbinary(primitive_type);
 }
 
+Status widen_equality_delete_key(ColumnPtr column, const DataTypePtr& column_type,
+                                 const DataTypePtr& key_type, ColumnPtr* result) {
+    DORIS_CHECK(result != nullptr);
+    DORIS_CHECK(column_type->is_nullable());
+    DORIS_CHECK(key_type->is_nullable());
+    if (column_type->equals(*key_type)) {
+        *result = std::move(column);
+        return Status::OK();
+    }
+    // Byte keys keep the type they were read in and match by their bytes, and the converters below
+    // would turn VARBINARY into its text form.
+    if (is_equality_delete_byte_type(column_type) || is_equality_delete_byte_type(key_type))
+            [[unlikely]] {
+        return Status::InternalError("Unexpected type change of equality delete key from {} to {}",
+                                     column_type->get_name(), key_type->get_name());
+    }
+    // The Parquet and ORC readers widen the data column of a promoted field with the same
+    // converters, so a widened delete key holds exactly the value of the data key it must match.
+    auto converter =
+            converter::ColumnTypeConverter::get_converter(column_type, key_type, converter::COMMON);
+    if (!converter->support()) {
+        return Status::InternalError(
+                "Not support type change of equality delete key from {} to {}: {}",
+                column_type->get_name(), key_type->get_name(), converter->get_error_msg());
+    }
+    if (converter->is_consistent()) {
+        // The same column layout, for example DATETIMEV2 of another scale.
+        *result = std::move(column);
+        return Status::OK();
+    }
+    MutableColumnPtr widened = key_type->create_column();
+    // The converters fill only the nested values; as in OrcReader::_orc_column_to_doris_column, the
+    // null map is copied before converting.
+    const auto& source_null_map = assert_cast<const ColumnNullable&>(*column).get_null_map_data();
+    assert_cast<ColumnNullable&>(*widened).get_null_map_data().insert(source_null_map.begin(),
+                                                                      source_null_map.end());
+    RETURN_IF_ERROR(converter->convert(column, widened));
+    *result = std::move(widened);
+    return Status::OK();
+}
+
+namespace {
+
 // The rows of a byte-key column. A const column repeats its single value, which can be NULL: when the FE does
 // not send Iceberg scan semantics v2, a key column that the data file lacks is filled from
 // Const(Nullable(String) NULL), and its rows must match a NULL delete key rather than an empty string.

be/src/format/table/equality_delete.h

@@ -19,8 +19,36 @@
 #include "exprs/hybrid_set.h"
 #include "runtime/runtime_profile.h"
 
+namespace doris::schema::external {
+class TField;
+} // namespace doris::schema::external
+
 namespace doris {
 
+// The type an equality-delete key is compared in: the field's type in the scan schema (`field`, which
+// is null when the plan carries no schema), as in the V2 reader. Data files are read in that type, so
+// an equality delete written before a type promotion is widened to it rather than the data narrowed
+// to the delete file's type. Example: `id` promoted from INT to BIGINT, a data file written after the
+// promotion holds 4294967296, and a delete file written before it holds the NULL key. Narrowed to
+// INT, 4294967296 became NULL and was deleted by that NULL key; widened, the delete key is compared
+// as BIGINT and deletes only rows whose id is NULL. STRING / VARBINARY keys keep `delete_key_type`:
+// they match by their bytes whatever the two sides were read as.
+DataTypePtr equality_delete_key_type(const schema::external::TField* field,
+                                     const DataTypePtr& delete_key_type);
+
+// Iceberg BINARY / FIXED keys are read as VARBINARY and string keys as STRING, and the delete file and
+// the data file may carry the same key in different ones of these classes (a delete block holding
+// STRING bytes for a VARBINARY data column, say). Keys of these types therefore match by their bytes.
+bool is_equality_delete_byte_type(const DataTypePtr& type);
+
+// Widens an equality-delete key read in `column_type` to `key_type`, the type the key is compared in.
+// A file written before an Iceberg type promotion holds the key in the old type: the INT key of a
+// column since promoted to BIGINT becomes BIGINT here. Iceberg's promotions (INT to BIGINT, FLOAT to
+// DOUBLE, a wider DECIMAL precision) only widen, so every value is kept. Returns `column` itself when
+// the types are equal.
+Status widen_equality_delete_key(ColumnPtr column, const DataTypePtr& column_type,
+                                 const DataTypePtr& key_type, ColumnPtr* result);
+
 /**
  * Support Iceberg equality delete.
  * If there's only one delete column in delete file, use `SimpleEqualityDelete`,

be/src/format/table/iceberg_reader.cpp

@@ -728,8 +728,10 @@ Status IcebergParquetReader::on_before_init_reader(ReaderInitContext* ctx) {
         RETURN_IF_ERROR(find_projected_equality_key_path(*root_path.front(), key_path,
                                                          slot->second->type(), child_indexes,
                                                          &leaf_type, &found));
-        // The projected leaf has the table's current type. A key written before a type promotion
-        // (INT promoted to BIGINT) keeps its own type, so it still reads the physical root.
+        // The projected leaf has the table's current type, which is the type a key is compared in
+        // (equality_delete_key_type) even when the key was written before a type promotion. Only a
+        // STRING / VARBINARY key keeps its delete file's type: a STRING key under a VARBINARY
+        // projected leaf still reads the physical root, where the leaf is read as STRING too.
         if (found && remove_nullable(leaf_type)->equals(*remove_nullable(key_type)) &&
             projected_root_reads_physical_path(*ctx->table_info_node, slot->second->col_name(),
                                                slot->second->type(), *child_indexes, file_path)) {
@@ -834,6 +836,7 @@ Status IcebergParquetReader::on_before_init_reader(ReaderInitContext* ctx) {
                         .field_id = field_id,
                         .block_name = table_col_name,
                         .leaf_type = _expand_columns[i].type,
+                        .source_leaf_type = _expand_columns[i].type,
                         .child_indexes = std::move(projected_child_indexes),
                         .missing_value = nullptr,
                         .source_block_name = std::move(projected_root_name),
@@ -846,6 +849,14 @@ Status IcebergParquetReader::on_before_init_reader(ReaderInitContext* ctx) {
                         field_id, _expand_columns[i].type, file_path.fields.size(),
                         &missing_value));
             }
+            // The physical root holds the leaf in the data file's own type, which is narrower than
+            // the compared type when the file was written before a type promotion: payload.existing
+            // promoted from INT to BIGINT is INT in older files, and copying those INT values into a
+            // BIGINT column read the wrong bytes. A leaf the file lacks is the missing default,
+            // which is built in the compared type.
+            const DataTypePtr source_leaf_type =
+                    complete_file_path ? make_nullable(file_path.fields.back()->data_type)
+                                       : _expand_columns[i].type;
             const auto root_carrier = physical_root_carriers.find(file_column->name);
             if (root_carrier != physical_root_carriers.end()) {
                 // Keys payload.a and payload.b: the carrier of payload.a reads payload, and
@@ -854,6 +865,7 @@ Status IcebergParquetReader::on_before_init_reader(ReaderInitContext* ctx) {
                         .field_id = field_id,
                         .block_name = table_col_name,
                         .leaf_type = _expand_columns[i].type,
+                        .source_leaf_type = source_leaf_type,
                         .child_indexes = file_path.child_indexes,
                         .missing_value = std::move(missing_value),
                         .source_block_name = root_carrier->second,
@@ -865,6 +877,7 @@ Status IcebergParquetReader::on_before_init_reader(ReaderInitContext* ctx) {
                     .field_id = field_id,
                     .block_name = table_col_name,
                     .leaf_type = _expand_columns[i].type,
+                    .source_leaf_type = source_leaf_type,
                     .child_indexes = file_path.child_indexes,
                     .missing_value = std::move(missing_value),
             });
@@ -1234,16 +1247,21 @@ Status IcebergOrcReader::on_before_init_reader(ReaderInitContext* ctx) {
                     .field_id = field_id,
                     .block_name = table_col_name,
                     .leaf_type = _expand_columns[i].type,
+                    .source_leaf_type = _expand_columns[i].type,
                     .child_indexes = file_path.child_indexes,
                     .missing_value = std::move(missing_value),
             });
             _expand_columns[i].type = make_nullable(convert_to_doris_type(file_column));
             _expand_columns[i].column = _expand_columns[i].type->create_column();
         } else if (!file_path.child_indexes.empty()) {
+            // The root is read in the data file's own types; a leaf written before a type
+            // promotion (INT payload.existing of a column now BIGINT) is widened after extraction.
             _nested_equality_delete_columns.push_back({
                     .field_id = field_id,
                     .block_name = table_col_name,
                     .leaf_type = _expand_columns[i].type,
+                    .source_leaf_type =
+                            make_nullable(convert_to_doris_type(file_path.fields.back())),
                     .child_indexes = file_path.child_indexes,
                     .missing_value = nullptr,
             });

be/src/format/table/iceberg_reader_mixin.h

@@ -362,7 +362,12 @@ protected:
     struct NestedEqualityDeleteColumn {
         int32_t field_id = -1;
         std::string block_name;
+        // The type the key is compared in (equality_delete_key_type).
         DataTypePtr leaf_type;
+        // The type of the leaf in the column it is extracted from, widened to leaf_type afterwards.
+        // A file written before a type promotion holds the old type: payload.existing promoted from
+        // INT to BIGINT is still INT in the delete files and data files written before.
+        DataTypePtr source_leaf_type;
         std::vector<size_t> child_indexes;
         ColumnPtr missing_value;
         // Block column holding the key's root struct. Empty when the key's own column reads the
@@ -677,12 +682,15 @@ Status IcebergReaderMixin<BaseReader>::_build_parquet_equality_delete_read_specs
             return Status::NotSupported(
                     "Iceberg equality delete does not support complex column {}", leaf->name);
         }
+        const auto delete_key_type = make_nullable(leaf->data_type);
         read_specs->push_back({
                 .nested_field =
                         {
                                 .field_id = field_id,
                                 .block_name = leaf->name,
-                                .leaf_type = make_nullable(leaf->data_type),
+                                .leaf_type = equality_delete_key_type(_find_schema_field(field_id),
+                                                                      delete_key_type),
+                                .source_leaf_type = delete_key_type,
                                 .child_indexes = std::move(child_indexes),
                                 .missing_value = nullptr,
                         },
@@ -724,12 +732,15 @@ Status IcebergReaderMixin<BaseReader>::_build_orc_equality_delete_read_specs(
             return Status::NotSupported(
                     "Iceberg equality delete does not support complex column {}", names.back());
         }
+        const auto delete_key_type = make_nullable(reader->convert_to_doris_type(leaf));
         read_specs->push_back({
                 .nested_field =
                         {
                                 .field_id = field_id,
                                 .block_name = names.back(),
-                                .leaf_type = make_nullable(reader->convert_to_doris_type(leaf)),
+                                .leaf_type = equality_delete_key_type(_find_schema_field(field_id),
+                                                                      delete_key_type),
+                                .source_leaf_type = delete_key_type,
                                 .child_indexes = std::move(child_indexes),
                                 .missing_value = nullptr,
                         },
@@ -1109,8 +1120,9 @@ Status IcebergReaderMixin<BaseReader>::_extract_nested_equality_delete_column(
         }
     }
 
-    auto result = ColumnNullable::create(remove_nullable(nested_field.leaf_type)->create_column(),
-                                         ColumnUInt8::create());
+    DORIS_CHECK(nested_field.source_leaf_type != nullptr);
+    auto result = ColumnNullable::create(
+            remove_nullable(nested_field.source_leaf_type)->create_column(), ColumnUInt8::create());
     auto& result_data = result->get_nested_column();
     auto& result_null_map = result->get_null_map_data();
     result_data.reserve(root_column->size());
@@ -1131,8 +1143,9 @@ Status IcebergReaderMixin<BaseReader>::_extract_nested_equality_delete_column(
             result_null_map.push_back(0);
         }
     }
-    *leaf_column = std::move(result);
-    return Status::OK();
+    return widen_equality_delete_key(std::move(result),
+                                     make_nullable(nested_field.source_leaf_type),
+                                     make_nullable(nested_field.leaf_type), leaf_column);
 }
 
 template <typename BaseReader>
@@ -1232,6 +1245,7 @@ Status IcebergReaderMixin<BaseReader>::_create_missing_equality_delete_value(
             .field_id = field_id,
             .block_name = "",
             .leaf_type = delete_key_type,
+            .source_leaf_type = delete_key_type,
             .child_indexes = std::vector<size_t>(table_path.size() - missing_index - 1, 0),
             .missing_value = nullptr,
     };

be/test/format/table/iceberg/iceberg_reader_test.cpp

@@ -84,9 +84,10 @@
 namespace doris {
 
 void write_iceberg_int_orc_file(const std::string& file_path, const std::string& field_name,
-                                int32_t field_id, const std::vector<int64_t>& values) {
+                                int32_t field_id, const std::vector<int64_t>& values,
+                                const std::string& orc_type = "int") {
     auto type = std::unique_ptr<::orc::Type>(
-            ::orc::Type::buildTypeFromString("struct<" + field_name + ":int>"));
+            ::orc::Type::buildTypeFromString("struct<" + field_name + ":" + orc_type + ">"));
     type->getSubtype(0)->setAttribute("iceberg.id", std::to_string(field_id));
 
     MemoryOutputStream memory_stream(1024 * 1024);
@@ -456,10 +457,11 @@ void write_iceberg_nested_struct_orc_file(const std::string& file_path,
                                           const std::vector<int64_t>& values,
                                           const std::vector<bool>& parent_nulls,
                                           const std::string& child_name = "existing",
-                                          int32_t child_id = 2) {
+                                          int32_t child_id = 2,
+                                          const std::string& child_type = "int") {
     DORIS_CHECK(values.size() == parent_nulls.size());
-    auto type = std::unique_ptr<::orc::Type>(
-            ::orc::Type::buildTypeFromString("struct<payload:struct<" + child_name + ":int>>"));
+    auto type = std::unique_ptr<::orc::Type>(::orc::Type::buildTypeFromString(
+            "struct<payload:struct<" + child_name + ":" + child_type + ">>"));
     type->getSubtype(0)->setAttribute("iceberg.id", "1");
     type->getSubtype(0)->getSubtype(0)->setAttribute("iceberg.id", std::to_string(child_id));
 
@@ -592,9 +594,82 @@ void write_iceberg_int_equality_delete_parquet_file(const std::string& file_path
                                                       1, properties.build()));
 }
 
-Status create_single_int_tuple_descriptor(ObjectPool* object_pool, const std::string& column_name,
-                                          int32_t field_id, DescriptorTbl** descriptor_table,
-                                          const TupleDescriptor** tuple_descriptor) {
+// One optional INT32 or INT64 column `field_name` (field `field_id`); std::nullopt writes NULL.
+void write_iceberg_single_integer_parquet_file(const std::string& file_path,
+                                               const std::string& field_name, int32_t field_id,
+                                               bool int64_column,
+                                               const std::vector<std::optional<int64_t>>& values) {
+    std::shared_ptr<arrow::Array> array;
+    if (int64_column) {
+        arrow::Int64Builder builder;
+        for (const auto& value : values) {
+            DORIS_CHECK((value.has_value() ? builder.Append(*value) : builder.AppendNull()).ok());
+        }
+        DORIS_CHECK(builder.Finish(&array).ok());
+    } else {
+        arrow::Int32Builder builder;
+        for (const auto& value : values) {
+            DORIS_CHECK((value.has_value() ? builder.Append(static_cast<int32_t>(*value))
+                                           : builder.AppendNull())
+                                .ok());
+        }
+        DORIS_CHECK(builder.Finish(&array).ok());
+    }
+    auto field = arrow::field(field_name, int64_column ? arrow::int64() : arrow::int32(), true)
+                         ->WithMetadata(arrow::key_value_metadata({"PARQUET:field_id"},
+                                                                  {std::to_string(field_id)}));
+    auto table = arrow::Table::Make(arrow::schema({field}), {array});
+    auto output = arrow::io::FileOutputStream::Open(file_path);
+    DORIS_CHECK(output.ok());
+    ::parquet::WriterProperties::Builder properties;
+    properties.version(::parquet::ParquetVersion::PARQUET_2_6);
+    properties.data_page_version(::parquet::ParquetDataPageVersion::V2);
+    properties.compression(::parquet::Compression::UNCOMPRESSED);
+    PARQUET_THROW_NOT_OK(::parquet::arrow::WriteTable(*table, arrow::default_memory_pool(), *output,
+                                                      static_cast<int64_t>(values.size()),
+                                                      properties.build()));
+}
+
+// `payload STRUCT<existing BIGINT>` (field 1, child 2): an equality delete on payload.existing
+// written after the field was promoted from INT.
+void write_iceberg_bigint_nested_struct_parquet_file(const std::string& file_path,
+                                                     const std::vector<int64_t>& values) {
+    auto existing_field =
+            arrow::field("existing", arrow::int64(), false)
+                    ->WithMetadata(arrow::key_value_metadata({"PARQUET:field_id"}, {"2"}));
+    auto struct_type = arrow::struct_({existing_field});
+    std::vector<std::shared_ptr<arrow::ArrayBuilder>> field_builders;
+    field_builders.emplace_back(std::make_shared<arrow::Int64Builder>());
+    arrow::StructBuilder builder(struct_type, arrow::default_memory_pool(),
+                                 std::move(field_builders));
+    auto* existing_builder = assert_cast<arrow::Int64Builder*>(builder.field_builder(0));
+    for (const auto value : values) {
+        DORIS_CHECK(builder.Append().ok());
+        DORIS_CHECK(existing_builder->Append(value).ok());
+    }
+    std::shared_ptr<arrow::Array> payload;
+    DORIS_CHECK(builder.Finish(&payload).ok());
+    auto payload_field =
+            arrow::field("payload", struct_type, true)
+                    ->WithMetadata(arrow::key_value_metadata({"PARQUET:field_id"}, {"1"}));
+    auto table = arrow::Table::Make(arrow::schema({payload_field}), {payload});
+
+    auto output = arrow::io::FileOutputStream::Open(file_path);
+    DORIS_CHECK(output.ok());
+    ::parquet::WriterProperties::Builder properties;
+    properties.version(::parquet::ParquetVersion::PARQUET_2_6);
+    properties.data_page_version(::parquet::ParquetDataPageVersion::V2);
+    properties.compression(::parquet::Compression::UNCOMPRESSED);
+    PARQUET_THROW_NOT_OK(::parquet::arrow::WriteTable(*table, arrow::default_memory_pool(), *output,
+                                                      static_cast<int64_t>(values.size()),
+                                                      properties.build()));
+}
+
+Status create_single_scalar_tuple_descriptor(ObjectPool* object_pool,
+                                             const std::string& column_name, int32_t field_id,
+                                             TPrimitiveType::type slot_type,
+                                             DescriptorTbl** descriptor_table,
+                                             const TupleDescriptor** tuple_descriptor) {
     TDescriptorTable thrift_table;
     TTableDescriptor table_descriptor;
     table_descriptor.__set_id(0);
@@ -606,7 +681,7 @@ Status create_single_int_tuple_descriptor(ObjectPool* object_pool, const std::st
     TTypeNode type_node;
     type_node.__set_type(TTypeNodeType::SCALAR);
     TScalarType scalar_type;
-    scalar_type.__set_type(TPrimitiveType::INT);
+    scalar_type.__set_type(slot_type);
     type_node.__set_scalar_type(scalar_type);
     TTypeDesc type;
     type.__set_types({type_node});
@@ -637,6 +712,14 @@ Status create_single_int_tuple_descriptor(ObjectPool* object_pool, const std::st
     return Status::OK();
 }
 
+Status create_single_int_tuple_descriptor(ObjectPool* object_pool, const std::string& column_name,
+                                          int32_t field_id, DescriptorTbl** descriptor_table,
+                                          const TupleDescriptor** tuple_descriptor) {
+    return create_single_scalar_tuple_descriptor(object_pool, column_name, field_id,
+                                                 TPrimitiveType::INT, descriptor_table,
+                                                 tuple_descriptor);
+}
+
 Status create_two_int_tuple_descriptor(
         ObjectPool* object_pool, const std::array<std::pair<std::string, int32_t>, 2>& columns,
         DescriptorTbl** descriptor_table, const TupleDescriptor** tuple_descriptor) {
@@ -1045,6 +1128,15 @@ schema::external::TFieldPtr make_external_int_field(const std::string& name, int
     return field_ptr;
 }
 
+// A field promoted from INT to BIGINT: its type in the scan schema is BIGINT.
+schema::external::TFieldPtr make_external_bigint_field(const std::string& name, int32_t field_id) {
+    auto field_ptr = make_external_int_field(name, field_id, std::nullopt);
+    TColumnType type;
+    type.__set_type(TPrimitiveType::BIGINT);
+    field_ptr.field_ptr->__set_type(type);
+    return field_ptr;
+}
+
 schema::external::TFieldPtr make_external_struct_field(
         const std::string& name, int32_t field_id,
         std::vector<schema::external::TFieldPtr> children, std::vector<std::string> aliases = {},
@@ -3506,6 +3598,291 @@ TEST_F(IcebergReaderTest,
                                                           &timezone_obj, cache.get(), true);
 }
 
+// Reads `data_file` through the V1 Iceberg reader with the equality deletes `delete_files` (path and
+// key field ids) applied, under the scan schema `current_root`. `block` holds the projected slots of
+// `tuple_descriptor` in order; all rows fit in one batch.
+Status read_v1_with_equality_deletes(
+        TFileFormatType::type file_format, const std::string& data_file,
+        const std::vector<std::pair<std::string, std::vector<int32_t>>>& delete_files,
+        schema::external::TStructField current_root, const TupleDescriptor* tuple_descriptor,
+        Block* block, const cctz::time_zone* timezone, FileMetaCache* meta_cache) {
+    schema::external::TSchema current_schema;
+    current_schema.__set_schema_id(100);
+    current_schema.__set_root_field(std::move(current_root));
+    TFileScanRangeParams scan_params;
+    scan_params.__set_iceberg_scan_semantics_version(ICEBERG_SCAN_SEMANTICS_VERSION_2);
+    scan_params.__set_file_type(TFileType::FILE_LOCAL);
+    scan_params.__set_format_type(file_format);
+    scan_params.__set_current_schema_id(100);
+    scan_params.__set_history_schema_info({current_schema});
+
+    std::vector<TIcebergDeleteFileDesc> delete_descriptors;
+    for (const auto& [delete_file, field_ids] : delete_files) {
+        TIcebergDeleteFileDesc delete_descriptor;
+        delete_descriptor.__set_content(2);
+        delete_descriptor.__set_path(delete_file);
+        delete_descriptor.__set_field_ids(field_ids);
+        delete_descriptor.__set_file_format(file_format);
+        delete_descriptors.push_back(std::move(delete_descriptor));
+    }
+    TIcebergFileDesc iceberg_descriptor;
+    iceberg_descriptor.__set_format_version(2);
+    iceberg_descriptor.__set_original_file_path(data_file);
+    iceberg_descriptor.__set_delete_files(std::move(delete_descriptors));
+    TTableFormatFileDesc table_format_descriptor;
+    table_format_descriptor.__set_iceberg_params(std::move(iceberg_descriptor));
+
+    TFileRangeDesc scan_range;
+    scan_range.__set_fs_name("");
+    scan_range.__set_path(data_file);
+    scan_range.__set_start_offset(0);
+    scan_range.__set_size(static_cast<int64_t>(std::filesystem::file_size(data_file)));
+    scan_range.__set_file_size(static_cast<int64_t>(std::filesystem::file_size(data_file)));
+    scan_range.__set_table_format_params(std::move(table_format_descriptor));
+
+    RuntimeProfile profile("test_profile");
+    RuntimeState runtime_state {TQueryOptions(), TQueryGlobals()};
+    runtime_state.set_timezone("UTC");
+    io::IOContext io_ctx;
+    ShardedKVCache kv_cache(8);
+    std::vector<ColumnDescriptor> column_descriptors(block->columns());
+    std::unordered_map<std::string, uint32_t> block_positions;
+    for (uint32_t position = 0; position < block->columns(); ++position) {
+        column_descriptors[position].name = block->get_by_position(position).name;
+        block_positions.emplace(column_descriptors[position].name, position);
+    }
+    size_t read_rows = 0;
+    bool eof = false;
+    if (file_format == TFileFormatType::FORMAT_PARQUET) {
+        IcebergParquetReader reader(&kv_cache, &profile, scan_params, scan_range, 1024, timezone,
+                                    &io_ctx, &runtime_state, meta_cache);
+        io::FileReaderSPtr file_reader;
+        RETURN_IF_ERROR(io::global_local_filesystem()->open_file(data_file, &file_reader));
+        reader.set_file_reader(file_reader);
+        ParquetInitContext context;
+        context.column_descs = &column_descriptors;
+        context.col_name_to_block_idx = &block_positions;
+        context.tuple_descriptor = tuple_descriptor;
+        context.params = &scan_params;
+        context.range = &scan_range;
+        RETURN_IF_ERROR(reader.init_reader(&context));
+        return reader.get_next_block(block, &read_rows, &eof);
+    }
+    IcebergOrcReader reader(&kv_cache, &profile, &runtime_state, scan_params, scan_range, 1024,
+                            "UTC", &io_ctx, meta_cache);
+    OrcInitContext context;
+    context.column_descs = &column_descriptors;
+    context.col_name_to_block_idx = &block_positions;
+    context.tuple_descriptor = tuple_descriptor;
+    context.params = &scan_params;
+    context.range = &scan_range;
+    RETURN_IF_ERROR(reader.init_reader(&context));
+    return reader.get_next_block(block, &read_rows, &eof);
+}
+
+// Iceberg allows promoting int to long, and an equality delete keeps the key type it was written
+// with. Data `id, v, w` holds (1,10,100), (2,20,200) and (3,30,300) with `id` still int32 in the
+// file, the table type of `id` is now long, a delete written before the promotion holds id = 2 as
+// INT and one written after it id = 3 as BIGINT. Both are compared as BIGINT, the type the data is
+// read in. `project_key` selects `SELECT id` (the key is the projected column) or `SELECT v` (the
+// key is read into a hidden carrier).
+// NOLINTNEXTLINE(readability-function-cognitive-complexity,readability-function-size)
+void check_equality_deletes_match_promoted_key(TFileFormatType::type file_format, bool project_key,
+                                               bool delete_after_promotion,
+                                               const cctz::time_zone* timezone,
+                                               FileMetaCache* meta_cache) {
+    const bool is_parquet = file_format == TFileFormatType::FORMAT_PARQUET;
+    const std::string format_name = is_parquet ? "parquet" : "orc";
+    const std::string case_name = format_name + (project_key ? "_id" : "_v") +
+                                  (delete_after_promotion ? "_both_deletes" : "_int_delete");
+    SCOPED_TRACE(case_name);
+    const auto test_dir = std::filesystem::temp_directory_path() /
+                          ("doris_v1_equality_delete_promoted_key_" + case_name);
+    std::filesystem::remove_all(test_dir);
+    std::filesystem::create_directories(test_dir);
+    const auto data_file = (test_dir / ("data." + format_name)).string();
+    const auto int_delete_file = (test_dir / ("int-delete." + format_name)).string();
+    const auto bigint_delete_file = (test_dir / ("bigint-delete." + format_name)).string();
+    if (is_parquet) {
+        write_iceberg_three_int_parquet_file(data_file, "id", 0, {1, 2, 3}, "v", 1, {10, 20, 30},
+                                             "w", 2, {100, 200, 300});
+        write_iceberg_single_integer_parquet_file(int_delete_file, "id", 0, false, {2});
+        write_iceberg_single_integer_parquet_file(bigint_delete_file, "id", 0, true, {3});
+    } else {
+        write_iceberg_three_int_orc_file(data_file, "id", 0, {1, 2, 3}, "v", 1, {10, 20, 30}, "w",
+                                         2, {100, 200, 300});
+        write_iceberg_int_orc_file(int_delete_file, "id", 0, {2});
+        write_iceberg_int_orc_file(bigint_delete_file, "id", 0, {3}, "bigint");
+    }
+    std::vector<std::pair<std::string, std::vector<int32_t>>> delete_files = {
+            {int_delete_file, {0}}};
+    if (delete_after_promotion) {
+        delete_files.emplace_back(bigint_delete_file, std::vector<int32_t> {0});
+    }
+
+    schema::external::TStructField current_root;
+    current_root.__set_fields({make_external_bigint_field("id", 0),
+                               make_external_int_field("v", 1, std::nullopt),
+                               make_external_int_field("w", 2, std::nullopt)});
+    ObjectPool object_pool;
+    DescriptorTbl* descriptor_table = nullptr;
+    const TupleDescriptor* tuple_descriptor = nullptr;
+    ASSERT_TRUE(create_single_scalar_tuple_descriptor(
+                        &object_pool, project_key ? "id" : "v", project_key ? 0 : 1,
+                        project_key ? TPrimitiveType::BIGINT : TPrimitiveType::INT,
+                        &descriptor_table, &tuple_descriptor)
+                        .ok());
+    const DataTypePtr type = project_key ? make_nullable(std::make_shared<DataTypeInt64>())
+                                         : make_nullable(std::make_shared<DataTypeInt32>());
+    Block block;
+    block.insert({type->create_column(), type, project_key ? "id" : "v"});
+    const Status status = read_v1_with_equality_deletes(file_format, data_file, delete_files,
+                                                        std::move(current_root), tuple_descriptor,
+                                                        &block, timezone, meta_cache);
+
+    ASSERT_TRUE(status.ok()) << status;
+    std::vector<std::string> expected = project_key ? std::vector<std::string> {"1", "3"}
+                                                    : std::vector<std::string> {"10", "30"};
+    if (delete_after_promotion) {
+        expected.pop_back();
+    }
+    ASSERT_EQ(block.rows(), expected.size());
+    for (size_t row = 0; row < expected.size(); ++row) {
+        EXPECT_EQ(type->to_string(*block.get_by_position(0).column, row), expected[row]);
+    }
+    std::filesystem::remove_all(test_dir);
+}
+
+TEST_F(IcebergReaderTest, v1_parquet_equality_delete_matches_key_after_int_to_long_promotion) {
+    check_equality_deletes_match_promoted_key(TFileFormatType::FORMAT_PARQUET, true, false,
+                                              &timezone_obj, cache.get());
+}
+
+TEST_F(IcebergReaderTest, v1_orc_equality_delete_matches_key_after_int_to_long_promotion) {
+    check_equality_deletes_match_promoted_key(TFileFormatType::FORMAT_ORC, true, false,
+                                              &timezone_obj, cache.get());
+}
+
+TEST_F(IcebergReaderTest, v1_parquet_equality_deletes_before_and_after_int_to_long_promotion) {
+    check_equality_deletes_match_promoted_key(TFileFormatType::FORMAT_PARQUET, true, true,
+                                              &timezone_obj, cache.get());
+}
+
+TEST_F(IcebergReaderTest, v1_orc_equality_deletes_before_and_after_int_to_long_promotion) {
+    check_equality_deletes_match_promoted_key(TFileFormatType::FORMAT_ORC, true, true,
+                                              &timezone_obj, cache.get());
+}
+
+TEST_F(IcebergReaderTest,
+       v1_parquet_hidden_equality_deletes_before_and_after_int_to_long_promotion) {
+    check_equality_deletes_match_promoted_key(TFileFormatType::FORMAT_PARQUET, false, true,
+                                              &timezone_obj, cache.get());
+}
+
+TEST_F(IcebergReaderTest, v1_orc_hidden_equality_deletes_before_and_after_int_to_long_promotion) {
+    check_equality_deletes_match_promoted_key(TFileFormatType::FORMAT_ORC, false, true,
+                                              &timezone_obj, cache.get());
+}
+
+// A nested key promoted from int to long: data `id, payload STRUCT<existing>` holds (1,{10}),
+// (2,{20}) and (3,{30}) with payload.existing still int32 in the file, and a delete written after
+// the promotion holds payload.existing = 20 as BIGINT. The query projects only `id`, so the key is
+// extracted from the physical payload root and must be widened before it is compared; copying the
+// INT values into a BIGINT column read the wrong bytes.
+void check_nested_equality_delete_after_promotion_reads_older_file(
+        TFileFormatType::type file_format, const cctz::time_zone* timezone,
+        FileMetaCache* meta_cache) {
+    const bool is_parquet = file_format == TFileFormatType::FORMAT_PARQUET;
+    const std::string format_name = is_parquet ? "parquet" : "orc";
+    SCOPED_TRACE(format_name);
+    const auto test_dir = std::filesystem::temp_directory_path() /
+                          ("doris_v1_nested_equality_delete_promoted_key_" + format_name);
+    std::filesystem::remove_all(test_dir);
+    std::filesystem::create_directories(test_dir);
+    const auto data_file = (test_dir / ("data." + format_name)).string();
+    const auto delete_file = (test_dir / ("equality-delete." + format_name)).string();
+    if (is_parquet) {
+        write_iceberg_id_and_nested_struct_parquet_file(data_file, {1, 2, 3}, {10, 20, 30},
+                                                        {false, false, false});
+        write_iceberg_bigint_nested_struct_parquet_file(delete_file, {20});
+    } else {
+        write_iceberg_id_and_nested_struct_orc_file(data_file, {1, 2, 3}, {10, 20, 30},
+                                                    {false, false, false});
+        write_iceberg_nested_struct_orc_file(delete_file, {20}, {false}, "existing", 2, "bigint");
+    }
+
+    schema::external::TStructField current_root;
+    current_root.__set_fields({make_external_int_field("id", 0, std::nullopt),
+                               make_external_struct_field(
+                                       "payload", 1, {make_external_bigint_field("existing", 2)})});
+    ObjectPool object_pool;
+    DescriptorTbl* descriptor_table = nullptr;
+    const TupleDescriptor* tuple_descriptor = nullptr;
+    ASSERT_TRUE(create_single_int_tuple_descriptor(&object_pool, "id", 0, &descriptor_table,
+                                                   &tuple_descriptor)
+                        .ok());
+    const auto int_type = make_nullable(std::make_shared<DataTypeInt32>());
+    Block block;
+    block.insert({int_type->create_column(), int_type, "id"});
+    const Status status = read_v1_with_equality_deletes(
+            file_format, data_file, {{delete_file, {2}}}, std::move(current_root), tuple_descriptor,
+            &block, timezone, meta_cache);
+
+    ASSERT_TRUE(status.ok()) << status;
+    ASSERT_EQ(block.rows(), 2);
+    EXPECT_EQ(int_type->to_string(*block.get_by_position(0).column, 0), "1");
+    EXPECT_EQ(int_type->to_string(*block.get_by_position(0).column, 1), "3");
+    std::filesystem::remove_all(test_dir);
+}
+
+TEST_F(IcebergReaderTest, v1_parquet_nested_equality_delete_after_promotion_reads_older_file) {
+    check_nested_equality_delete_after_promotion_reads_older_file(TFileFormatType::FORMAT_PARQUET,
+                                                                  &timezone_obj, cache.get());
+}
+
+TEST_F(IcebergReaderTest, v1_orc_nested_equality_delete_after_promotion_reads_older_file) {
+    check_nested_equality_delete_after_promotion_reads_older_file(TFileFormatType::FORMAT_ORC,
+                                                                  &timezone_obj, cache.get());
+}
+
+// The key is compared in the promoted type, not in the delete file's older one. A data file written
+// after `id` became long holds NULL, 0, 1 and 4294967296, and a delete written before the promotion
+// holds the NULL key as INT. Narrowed to INT, 4294967296 would become NULL and be deleted by that
+// key; compared as BIGINT, only the NULL row is deleted.
+TEST_F(IcebergReaderTest, v1_parquet_equality_delete_keeps_out_of_range_promoted_value) {
+    const auto test_dir = std::filesystem::temp_directory_path() /
+                          "doris_v1_equality_delete_out_of_range_promoted_value";
+    std::filesystem::remove_all(test_dir);
+    std::filesystem::create_directories(test_dir);
+    const auto data_file = (test_dir / "data.parquet").string();
+    const auto delete_file = (test_dir / "equality-delete.parquet").string();
+    write_iceberg_single_integer_parquet_file(data_file, "id", 0, true,
+                                              {std::nullopt, 0, 1, 4294967296});
+    write_iceberg_single_integer_parquet_file(delete_file, "id", 0, false, {std::nullopt});
+
+    schema::external::TStructField current_root;
+    current_root.__set_fields({make_external_bigint_field("id", 0)});
+    ObjectPool object_pool;
+    DescriptorTbl* descriptor_table = nullptr;
+    const TupleDescriptor* tuple_descriptor = nullptr;
+    ASSERT_TRUE(create_single_scalar_tuple_descriptor(&object_pool, "id", 0, TPrimitiveType::BIGINT,
+                                                      &descriptor_table, &tuple_descriptor)
+                        .ok());
+    const auto bigint_type = make_nullable(std::make_shared<DataTypeInt64>());
+    Block block;
+    block.insert({bigint_type->create_column(), bigint_type, "id"});
+    const Status status = read_v1_with_equality_deletes(
+            TFileFormatType::FORMAT_PARQUET, data_file, {{delete_file, {0}}},
+            std::move(current_root), tuple_descriptor, &block, &timezone_obj, cache.get());
+
+    ASSERT_TRUE(status.ok()) << status;
+    ASSERT_EQ(block.rows(), 3);
+    EXPECT_EQ(bigint_type->to_string(*block.get_by_position(0).column, 0), "0");
+    EXPECT_EQ(bigint_type->to_string(*block.get_by_position(0).column, 1), "1");
+    EXPECT_EQ(bigint_type->to_string(*block.get_by_position(0).column, 2), "4294967296");
+    std::filesystem::remove_all(test_dir);
+}
+
 // Data `id INT, payload STRUCT<existing INT, k INT>` holds (1,{10,100}), (2,{20,200}) and
 // (3,{30,300}). An equality delete on `delete_field_ids` (2 = payload.existing, 3 = payload.k)
 // with the row {20,200} removes id 2. `projected_children` lists the payload children the query