[core] Fix data evolution scan pruning files after a column type change

paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionFileStoreScan.java

@@ -303,6 +303,7 @@ static EvolutionStats evolutionStats(
         int[] fieldOffsets = new int[fieldsCount];
         Arrays.fill(rowOffsets, -1);
         Arrays.fill(fieldOffsets, -1);
+        Set<Integer> typeMismatchedFieldIds = new HashSet<>();
 
         InternalRow[] min = new InternalRow[metas.size()];
         InternalRow[] max = new InternalRow[metas.size()];
@@ -346,6 +347,7 @@ static EvolutionStats evolutionStats(
                             if (fieldId == fieldIdsWithStats[k]) {
                                 DataType fileType = dataFileSchemaWithStats.fields().get(k).type();
                                 if (!fileType.equalsIgnoreFieldId(targetType)) {
+                                    typeMismatchedFieldIds.add(targetFieldId);
                                     continue loop1;
                                 }
                                 rowOffsets[j] = i;
@@ -362,7 +364,9 @@ static EvolutionStats evolutionStats(
 
         long groupRowCount = metas.get(0).file().rowCount();
         for (int j = 0; j < fieldsCount; j++) {
-            if (rowOffsets[j] == -1 && excludedFileFieldIds.contains(allFields[j])) {
+            if (rowOffsets[j] == -1
+                    && (excludedFileFieldIds.contains(allFields[j])
+                            || typeMismatchedFieldIds.contains(allFields[j]))) {
                 rowOffsets[j] = -2;
             }
         }

paimon-core/src/test/java/org/apache/paimon/operation/DataEvolutionFileStoreScanTest.java

@@ -312,6 +312,49 @@ public void testEvolutionStatsSkipsStatsAfterColumnTypeChange() {
         assertThat(maxRow.getString(0).toString()).isEqualTo("yam");
     }
 
+    @Test
+    public void testTypeChangedColumnMustNotPrunePreAlterFiles() {
+        Schema baseSchema = createSchema("f0", "f1");
+        schemas.put(0L, TableSchema.create(0L, baseSchema));
+
+        Schema evolvedSchema =
+                Schema.newBuilder()
+                        .column("f0", DataTypes.BIGINT())
+                        .column("f1", DataTypes.STRING())
+                        .build();
+        TableSchema evolvedTableSchema = TableSchema.create(1L, evolvedSchema);
+        schemas.put(1L, evolvedTableSchema);
+
+        ManifestEntry preAlterFile =
+                createManifestEntry(
+                        0L,
+                        createSimpleStats(
+                                GenericRow.of(10, BinaryString.fromString("a")),
+                                GenericRow.of(99, BinaryString.fromString("z")),
+                                createBinaryArray(new int[] {0, 0}),
+                                new int[] {0, 1}));
+
+        EvolutionStats result =
+                DataEvolutionFileStoreScan.evolutionStats(
+                        evolvedTableSchema,
+                        scanTableSchema,
+                        Collections.singletonList(preAlterFile));
+
+        Predicate onChangedColumn =
+                new PredicateBuilder(evolvedTableSchema.logicalRowType()).equal(0, 50L);
+
+        boolean keepFile =
+                onChangedColumn.test(
+                        result.rowCount(),
+                        result.minValues(),
+                        result.maxValues(),
+                        result.nullCounts());
+
+        assertThat(keepFile)
+                .as("pre-ALTER file must not be pruned by a predicate on a type-changed column")
+                .isTrue();
+    }
+
     @Test
     public void testEvolutionStatsKeepDedicatedVectorFieldAsUnknown() {
         Schema schema = createSchema("f0", "f1", "f2");