[fix](fe) Resolve dropped Iceberg equality-delete keys after a schema ID is reused

fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java

@@ -2108,7 +2108,7 @@ public Map<String, String> getScanNodeProperties(
             // every branch so the default and current field type match BE's read.
             if (hasApplicableEqualityDeletes) {
                 List<NestedField> equalityFields = schemaForPotentialEqualityDeletes(
-                        table, scanSchema, applicableEqualityDeleteFieldIds);
+                        table, exactScan.snapshot(), scanSchema, applicableEqualityDeleteFieldIds);
                 dict = IcebergSchemaUtils.encodeEqualitySchemaEvolutionProp(
                         table, equalityFields, appendRowLineage,
                         enableVarbinary, enableTimestampTz);
@@ -2280,38 +2280,52 @@ private Set<Integer> cachedApplicableEqualityDeleteFieldIds(Table table, TableSc
      * delete manifests does not enumerate data files or byte-split tasks, and retained state is bounded by the
      * number of equality keys rather than the table's entire schema history.
      *
-     * <p>The metadata schema list is searched in reverse chronology up to the selected schema (schema IDs are
-     * identifiers, not a sequence), so the latest definition of each stable field ID wins. Current fields remain
-     * first, allowing a dropped/re-added name to resolve the projected field by name while a live historical
-     * equality key resolves by its stable field ID.</p>
+     * <p>Schema IDs are identifiers, not a sequence: Iceberg reuses an ID when evolution returns to an earlier
+     * schema, so a schema's position in the metadata list does not tell when it was last active, and the list may
+     * also hold schemas committed after a time-travel or branch target. The scan snapshot's parent chain is
+     * therefore searched first, so the definition active on that lineage wins; the whole metadata list, newest
+     * first, is the fallback for schema-only changes and expired ancestors. A dropped key nevertheless takes the
+     * widest type it has in any retained schema: a delete committed after a schema-only promotion holds values of
+     * the promoted type even when no snapshot records it. Current fields remain first, allowing a dropped/re-added
+     * name to resolve the projected field by name while a live historical equality key resolves by its stable
+     * field ID.</p>
      */
     @VisibleForTesting
     static List<NestedField> schemaForPotentialEqualityDeletes(
             Table table, TableScan scan, Schema scanSchema) {
         return schemaForPotentialEqualityDeletes(
-                table, scanSchema, applicableEqualityDeleteFieldIds(table, scan));
+                table, scan.snapshot(), scanSchema, applicableEqualityDeleteFieldIds(table, scan));
     }
 
-    private static List<NestedField> schemaForPotentialEqualityDeletes(
-            Table table, Schema scanSchema, Set<Integer> equalityFieldIds) {
-        List<Schema> metadataSchemas = metadataSchemaHistory(table);
-        int selectedSchemaIndex = -1;
-        for (int index = 0; index < metadataSchemas.size(); index++) {
-            if (metadataSchemas.get(index).schemaId() == scanSchema.schemaId()) {
-                selectedSchemaIndex = index;
-            }
-        }
-        int lastRelevantIndex = selectedSchemaIndex >= 0
-                ? selectedSchemaIndex : metadataSchemas.size() - 1;
+    @VisibleForTesting
+    static List<NestedField> schemaForPotentialEqualityDeletes(
+            Table table, Snapshot scanSnapshot, Schema scanSchema, Set<Integer> equalityFieldIds) {
         Set<Integer> missing = new HashSet<>(equalityFieldIds);
         missing.removeAll(TypeUtil.indexById(scanSchema.asStruct()).keySet());
         if (missing.isEmpty()) {
             return scanSchema.columns();
         }
 
         List<NestedField> fields = new ArrayList<>(scanSchema.columns());
-        for (int index = lastRelevantIndex; index >= 0 && !missing.isEmpty(); index--) {
-            addHistoricalEqualityFields(fields, missing, metadataSchemas.get(index));
+        List<Schema> metadataSchemas = metadataSchemaHistory(table);
+        Map<Integer, Type> widestTypes = widestPromotedTypes(missing, metadataSchemas);
+        // Snapshots on one lineage mostly share a schema, so each schema is indexed at most once.
+        Set<Integer> lineageSchemaIds = new HashSet<>();
+        for (Snapshot snapshot = scanSnapshot; snapshot != null && !missing.isEmpty();
+                snapshot = snapshot.parentId() == null ? null : table.snapshot(snapshot.parentId())) {
+            Integer schemaId = snapshot.schemaId();
+            if (schemaId == null || !lineageSchemaIds.add(schemaId)) {
+                continue;
+            }
+            Schema lineageSchema = table.schemas().get(schemaId);
+            if (lineageSchema == null) {
+                throw new IllegalStateException(
+                        "Iceberg snapshot schema " + schemaId + " is absent from table metadata");
+            }
+            addHistoricalEqualityFields(fields, missing, lineageSchema, widestTypes);
+        }
+        for (int index = metadataSchemas.size() - 1; index >= 0 && !missing.isEmpty(); index--) {
+            addHistoricalEqualityFields(fields, missing, metadataSchemas.get(index), widestTypes);
         }
         if (!missing.isEmpty()) {
             throw new IllegalStateException(
@@ -2327,8 +2341,40 @@ private static List<Schema> metadataSchemaHistory(Table table) {
         return new ArrayList<>(table.schemas().values());
     }
 
-    private static void addHistoricalEqualityFields(
-            List<NestedField> fields, Set<Integer> missingFieldIds, Schema historicalSchema) {
+    /**
+     * Field IDs are never reused and a field's type only changes by promotion, which is monotonic, so the widest
+     * type a field has in any retained schema covers every file written for it.
+     */
+    private static Map<Integer, Type> widestPromotedTypes(Set<Integer> fieldIds, List<Schema> schemas) {
+        Map<Integer, Type> widestTypes = new HashMap<>();
+        for (Schema schema : schemas) {
+            for (Integer fieldId : fieldIds) {
+                Type type = schema.findType(fieldId);
+                if (type == null) {
+                    continue;
+                }
+                Type widest = widestTypes.get(fieldId);
+                if (widest == null || TypeUtil.isPromotionAllowed(widest, type.asPrimitiveType())) {
+                    widestTypes.put(fieldId, type);
+                }
+            }
+        }
+        return widestTypes;
+    }
+
+    private static List<NestedField> withWidestTypes(List<NestedField> fields, Map<Integer, Type> widestTypes) {
+        List<NestedField> widened = new ArrayList<>(fields.size());
+        for (NestedField field : fields) {
+            Type type = field.type().isStructType()
+                    ? Types.StructType.of(withWidestTypes(field.type().asStructType().fields(), widestTypes))
+                    : widestTypes.getOrDefault(field.fieldId(), field.type());
+            widened.add(type.equals(field.type()) ? field : Types.NestedField.from(field).ofType(type).build());
+        }
+        return widened;
+    }
+
+    private static void addHistoricalEqualityFields(List<NestedField> fields, Set<Integer> missingFieldIds,
+            Schema historicalSchema, Map<Integer, Type> widestTypes) {
         Map<Integer, NestedField> historicalFields = TypeUtil.indexById(historicalSchema.asStruct());
         Set<Integer> selectedFieldIds = new HashSet<>();
         for (Integer fieldId : missingFieldIds) {
@@ -2345,7 +2391,7 @@ private static void addHistoricalEqualityFields(
             return;
         }
         Schema selectedSchema = TypeUtil.select(historicalSchema, selectedFieldIds);
-        mergeHistoricalEqualityFields(fields, selectedSchema.columns());
+        mergeHistoricalEqualityFields(fields, withWidestTypes(selectedSchema.columns(), widestTypes));
         missingFieldIds.removeAll(selectedFieldIds);
     }