[fix](be) Align cast input nullability for Iceberg struct evolution

be/src/format_v2/table_reader.h

@@ -994,8 +994,12 @@ class TableReader {
         }
 
         DataTypePtr input_type = file_type;
+        // Cast wrappers unwrap nullable inputs according to the declared input type, so keep the
+        // root nullability of the declared type aligned with the actual column shape.
         if ((*column)->is_nullable() && !input_type->is_nullable()) {
             input_type = make_nullable(input_type);
+        } else if (!(*column)->is_nullable() && input_type->is_nullable()) {
+            input_type = remove_nullable(input_type);
         }
         Block cast_block;
         cast_block.insert({*column, input_type, column_name});

be/test/format_v2/table_reader_test.cpp

@@ -896,6 +896,11 @@ class TableReaderCharVarcharTestHelper final : public TableReader {
     using TableReader::_truncate_char_or_varchar_column;
 };
 
+class TableReaderCastTestHelper final : public TableReader {
+public:
+    using TableReader::_cast_column_to_type;
+};
+
 TEST(TableReaderTest, TruncateCharOrVarcharPredicateOnlyAppliesToParquetStringWidthMismatch) {
     ColumnMapping mapping;
     mapping.table_type = std::make_shared<DataTypeString>(3, TYPE_VARCHAR);
@@ -1516,6 +1521,41 @@ TEST(TableReaderTest, ComplexRematerializeCastsScalarChildToTableType) {
     EXPECT_EQ(city_values.get_data_at(1).to_string(), "London");
 }
 
+TEST(TableReaderTest, ComplexRematerializeCastsNonNullableScalarChildWithNullableFileType) {
+    const auto int_type = std::make_shared<DataTypeInt32>();
+    const auto bigint_type = std::make_shared<DataTypeInt64>();
+    const auto nullable_int_type = make_nullable(int_type);
+    const auto nullable_bigint_type = make_nullable(bigint_type);
+    RuntimeState state {TQueryOptions(), TQueryGlobals()};
+    TableReaderCastTestHelper reader;
+    ASSERT_TRUE(reader.init({
+                                    .projected_columns = {},
+                                    .conjuncts = {},
+                                    .format = FileFormat::PARQUET,
+                                    .scan_params = nullptr,
+                                    .io_ctx = nullptr,
+                                    .runtime_state = &state,
+                                    .scanner_profile = nullptr,
+                            })
+                        .ok());
+
+    auto column = ColumnInt32::create();
+    column->insert_value(10);
+    column->insert_value(20);
+    ColumnPtr result_column = std::move(column);
+    const auto status = reader._cast_column_to_type(&result_column, nullable_int_type,
+                                                    nullable_bigint_type, "struct_column.a");
+    ASSERT_TRUE(status.ok()) << status.to_string();
+
+    const auto& result_nullable = assert_cast<const ColumnNullable&>(*result_column);
+    const auto& child_values = assert_cast<const ColumnInt64&>(result_nullable.get_nested_column());
+    ASSERT_EQ(result_nullable.size(), 2);
+    EXPECT_FALSE(result_nullable.is_null_at(0));
+    EXPECT_FALSE(result_nullable.is_null_at(1));
+    EXPECT_EQ(child_values.get_element(0), 10);
+    EXPECT_EQ(child_values.get_element(1), 20);
+}
+
 TEST(TableReaderTest, ReopenSplitAfterClose) {
     const auto test_dir = std::filesystem::temp_directory_path() / "doris_table_reader_test";
     std::filesystem::remove_all(test_dir);

regression-test/suites/external_table_p0/iceberg/test_iceberg_struct_schema_evolution.groovy

@@ -95,6 +95,28 @@ suite("test_iceberg_struct_schema_evolution", "p0,external") {
     // Test 8: DISTINCT query on struct fields
     qt_struct_distinct """SELECT DISTINCT element_at(a_struct, 'renamed'), element_at(a_struct, 'added'), element_at(a_struct, 'keep') FROM ${table_name} ORDER BY 1, 2, 3"""
 
+    // Reproduce Spark Iceberg struct child type evolution: old files keep col.a as INT while
+    // current Iceberg schema exposes it as BIGINT. Reading col.a must cast the materialized struct
+    // child without assuming the declared nullable file type matches the actual column nullability.
+    def type_evolution_table_name = "test_struct_child_type_evolution"
+    spark_iceberg_multi """
+        DROP TABLE IF EXISTS demo.test_db.${type_evolution_table_name};
+        CREATE TABLE demo.test_db.${type_evolution_table_name} (
+            id INT,
+            col STRUCT<a: INT, b: INT, c: INT>
+        ) USING iceberg
+        TBLPROPERTIES ('write.format.default' = 'parquet');
+        INSERT INTO demo.test_db.${type_evolution_table_name}
+            SELECT 1, named_struct('a', 10, 'b', 20, 'c', 30);
+        ALTER TABLE demo.test_db.${type_evolution_table_name} ALTER COLUMN col.a TYPE BIGINT;
+    """
+    sql """REFRESH CATALOG ${catalog_name}"""
+    sql """
+        SELECT /*+ SET_VAR(enable_prune_nested_column=true) */ col.a, col.b, col.c
+        FROM ${type_evolution_table_name}
+        ORDER BY id
+    """
+
     // ============================================================
     // Test with ORC format (for completeness)
     // ============================================================