[fix](iceberg) Preserve schema for partition evolution

fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java

@@ -507,11 +507,16 @@ public void createScanRangeLocations() throws UserException {
         super.createScanRangeLocations();
         enableCurrentIcebergScanSemantics();
         // Extract name mapping from Iceberg table properties
-        Optional<Map<Integer, List<String>>> nameMapping = extractNameMapping();
+        initializeIcebergSchemaInfo(extractNameMapping());
+    }
 
+    @VisibleForTesting
+    void initializeIcebergSchemaInfo(Optional<Map<Integer, List<String>>> nameMapping) throws UserException {
         // Equality-delete keys are hidden scan dependencies and need not appear in the query
         // projection. Both scanners need the complete current schema to resolve field ids,
         // historical names, types, and initial defaults when an old data file lacks such a key.
+        // An identity partition column can also be a physical field in files written by an older
+        // partition spec, so preserving the complete schema is required for partition evolution.
         ExternalUtil.initSchemaInfoForAllColumn(params, -1L, source.getTargetTable().getColumns(),
                 nameMapping.orElse(Collections.emptyMap()), nameMapping.isPresent(),
                 getBase64EncodedInitialDefaultsForScan());

fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/source/IcebergScanNodeTest.java

@@ -138,6 +138,12 @@ int enableAndGetIcebergScanSemanticsVersion() {
             enableCurrentIcebergScanSemantics();
             return params.getIcebergScanSemanticsVersion();
         }
+
+        TFileScanRangeParams initializeAndGetIcebergSchemaInfo() throws UserException {
+            params = new TFileScanRangeParams();
+            initializeIcebergSchemaInfo(Optional.empty());
+            return params;
+        }
     }
 
     @Test
@@ -148,6 +154,33 @@ public void testEmitsCurrentIcebergScanSemanticsCapability() {
                 node.enableAndGetIcebergScanSemanticsVersion());
     }
 
+    @Test
+    public void testPartitionEvolutionKeepsNonFileSlotInReaderSchema() throws Exception {
+        Column evolvedIdentityColumn = new Column("int_col", Type.BIGINT, true);
+        evolvedIdentityColumn.setUniqueId(1);
+        Column projectedColumn = new Column("payload", Type.STRING, true);
+        projectedColumn.setUniqueId(2);
+
+        IcebergExternalTable targetTable = Mockito.mock(IcebergExternalTable.class);
+        Mockito.when(targetTable.getColumns()).thenReturn(
+                List.of(evolvedIdentityColumn, projectedColumn));
+        IcebergSource source = Mockito.mock(IcebergSource.class);
+        Mockito.when(source.getTargetTable()).thenReturn(targetTable);
+
+        TestIcebergScanNode node = Mockito.spy(new TestIcebergScanNode(new SessionVariable()));
+        node.addSlot(1, projectedColumn);
+        setIcebergSource(node, source);
+        Mockito.doReturn(Collections.emptyMap()).when(node).getBase64EncodedInitialDefaultsForScan();
+
+        TFileScanRangeParams scanParams = node.initializeAndGetIcebergSchemaInfo();
+
+        Assert.assertEquals(2, scanParams.getHistorySchemaInfo().get(0).getRootField().getFieldsSize());
+        Assert.assertEquals("int_col", scanParams.getHistorySchemaInfo().get(0).getRootField()
+                .getFields().get(0).getFieldPtr().getName());
+        Assert.assertEquals("payload", scanParams.getHistorySchemaInfo().get(0).getRootField()
+                .getFields().get(1).getFieldPtr().getName());
+    }
+
     @Test
     public void testExtractNameMappingDistinguishesAbsentAndEmpty() throws Exception {
         TestIcebergScanNode node = new TestIcebergScanNode(new SessionVariable());