[core] Validate global-index schema compatibility before reader and coverage

paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdReassigner.java

@@ -786,10 +786,15 @@ private RewrittenIndexManifest rewriteIndexManifest(Assignment assignment) {
                             indexFile.rowCount(),
                             indexFile.dvRanges(),
                             indexFile.externalPath(),
-                            newGlobalIndex);
+                            newGlobalIndex,
+                            entry.schemaId());
             rewritten.add(
                     new IndexManifestEntry(
-                            entry.kind(), entry.partition(), entry.bucket(), newIndexFile));
+                            entry.kind(),
+                            entry.partition(),
+                            entry.bucket(),
+                            newIndexFile,
+                            entry.schemaId()));
         }
 
         return new RewrittenIndexManifest(

paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionGlobalIndexScanner.java

@@ -249,9 +249,12 @@ public static Optional<DataEvolutionGlobalIndexScanner> create(
             @Nullable PartitionPredicate partitionFilter,
             @Nullable Predicate filter) {
         @Nullable Snapshot snapshot = tryTravelOrLatest(table);
+        List<IndexManifestEntry> indexEntries =
+                table.store()
+                        .newIndexFileHandler()
+                        .scan(snapshot, indexFileFilter(table, partitionFilter, filter));
         List<IndexFileMeta> indexFiles =
-                table.store().newIndexFileHandler()
-                        .scan(snapshot, indexFileFilter(table, partitionFilter, filter)).stream()
+                GlobalIndexSchemaCompatibility.filterCompatible(table, indexEntries).stream()
                         .map(IndexManifestEntry::indexFile)
                         .collect(Collectors.toList());
         if (indexFiles.isEmpty()) {
@@ -284,9 +287,12 @@ public static Optional<DataEvolutionGlobalIndexScanner> createForTopN(
         DataField indexField = table.rowType().getField(topN.orders().get(0).field().name());
         int fieldId = indexField.id();
         @Nullable Snapshot snapshot = tryTravelOrLatest(table);
+        List<IndexManifestEntry> indexEntries =
+                table.store()
+                        .newIndexFileHandler()
+                        .scan(snapshot, topNIndexFileFilter(partitionFilter, fieldId));
         List<IndexFileMeta> indexFiles =
-                table.store().newIndexFileHandler()
-                        .scan(snapshot, topNIndexFileFilter(partitionFilter, fieldId)).stream()
+                GlobalIndexSchemaCompatibility.filterCompatible(table, indexEntries).stream()
                         .map(IndexManifestEntry::indexFile)
                         .collect(Collectors.toList());
         if (indexFiles.isEmpty()) {

paimon-core/src/main/java/org/apache/paimon/globalindex/GlobalIndexBuilderUtils.java

@@ -72,7 +72,8 @@ public static List<IndexFileMeta> toIndexFileMetas(
             Range range,
             int indexFieldId,
             String indexType,
-            List<ResultEntry> entries)
+            List<ResultEntry> entries,
+            long schemaId)
             throws IOException {
         return toIndexFileMetas(
                 fileIO,
@@ -83,7 +84,8 @@ public static List<IndexFileMeta> toIndexFileMetas(
                 null,
                 indexType,
                 entries,
-                null);
+                null,
+                schemaId);
     }
 
     /**
@@ -100,7 +102,8 @@ public static List<IndexFileMeta> toIndexFileMetas(
             List<DataField> fields,
             String indexType,
             List<ResultEntry> entries,
-            @Nullable byte[] sourceMeta)
+            @Nullable byte[] sourceMeta,
+            long schemaId)
             throws IOException {
         return toIndexFileMetas(
                 fileIO,
@@ -111,7 +114,8 @@ public static List<IndexFileMeta> toIndexFileMetas(
                 extraFieldIds(fields),
                 indexType,
                 entries,
-                sourceMeta);
+                sourceMeta,
+                schemaId);
     }
 
     public static List<Range> unindexedRowRanges(
@@ -572,7 +576,8 @@ private static List<IndexFileMeta> toIndexFileMetas(
             @Nullable int[] extraFieldIds,
             String indexType,
             List<ResultEntry> entries,
-            @Nullable byte[] sourceMeta)
+            @Nullable byte[] sourceMeta,
+            long schemaId)
             throws IOException {
         List<IndexFileMeta> results = new ArrayList<>();
         for (ResultEntry entry : entries) {
@@ -599,8 +604,10 @@ private static List<IndexFileMeta> toIndexFileMetas(
                             fileName,
                             fileSize,
                             entry.rowCount(),
+                            null,
+                            externalPathString,
                             globalIndexMeta,
-                            externalPathString);
+                            schemaId);
             results.add(indexFileMeta);
         }
         return results;

paimon-core/src/main/java/org/apache/paimon/globalindex/GlobalIndexSchemaCompatibility.java

@@ -0,0 +1,88 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.globalindex;
+
+import org.apache.paimon.index.GlobalIndexMeta;
+import org.apache.paimon.manifest.IndexManifestEntry;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.types.RowType;
+
+import java.io.FileNotFoundException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+/** Validates global index manifest entries against the current table schema. */
+public final class GlobalIndexSchemaCompatibility {
+
+    public static List<IndexManifestEntry> filterCompatible(
+            FileStoreTable table, Collection<IndexManifestEntry> entries) {
+        RowType currentRowType = table.rowType();
+        Map<Long, RowType> historicalRowTypes = new HashMap<>();
+        historicalRowTypes.put(table.schema().id(), currentRowType);
+        Set<Long> missingSchemaIds = new HashSet<>();
+        List<IndexManifestEntry> compatible = new ArrayList<>();
+        for (IndexManifestEntry entry : entries) {
+            GlobalIndexMeta globalIndex = entry.indexFile().globalIndexMeta();
+            Long schemaId = entry.schemaId();
+            if (globalIndex == null || schemaId == null) {
+                continue;
+            }
+
+            RowType historicalRowType = historicalRowTypes.get(schemaId);
+            if (historicalRowType == null && !missingSchemaIds.contains(schemaId)) {
+                try {
+                    historicalRowType =
+                            table.schemaManager().tryGetSchema(schemaId).logicalRowType();
+                    historicalRowTypes.put(schemaId, historicalRowType);
+                } catch (FileNotFoundException e) {
+                    missingSchemaIds.add(schemaId);
+                }
+            }
+            if (historicalRowType != null
+                    && compatibleIndexedFields(globalIndex, historicalRowType, currentRowType)) {
+                compatible.add(entry);
+            }
+        }
+        return compatible;
+    }
+
+    private static boolean compatibleIndexedFields(
+            GlobalIndexMeta globalIndex, RowType historicalRowType, RowType currentRowType) {
+        for (int fieldId : globalIndex.getIndexedFieldIds()) {
+            if (!historicalRowType.containsField(fieldId)
+                    || !currentRowType.containsField(fieldId)) {
+                return false;
+            }
+            if (!historicalRowType
+                    .getField(fieldId)
+                    .type()
+                    .equalsIgnoreNullable(currentRowType.getField(fieldId).type())) {
+                return false;
+            }
+        }
+        return true;
+    }
+
+    private GlobalIndexSchemaCompatibility() {}
+}

paimon-core/src/main/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexWriter.java

@@ -158,7 +158,8 @@ public CommitMessage flushIndex(
                         Collections.singletonList(indexField),
                         indexType,
                         resultEntries,
-                        sourceMeta);
+                        sourceMeta,
+                        table.schema().id());
         DataIncrement dataIncrement = DataIncrement.indexIncrement(indexFileMetas);
         return new CommitMessageImpl(
                 partition, 0, null, dataIncrement, CompactIncrement.emptyIncrement());

paimon-core/src/main/java/org/apache/paimon/index/IndexFileMeta.java

@@ -54,7 +54,8 @@ public class IndexFileMeta {
                                     "_DELETIONS_VECTORS_RANGES",
                                     new ArrayType(true, DeletionVectorMeta.SCHEMA)),
                             new DataField(5, "_EXTERNAL_PATH", newStringType(true)),
-                            new DataField(6, "_GLOBAL_INDEX", GlobalIndexMeta.SCHEMA)));
+                            new DataField(6, "_GLOBAL_INDEX", GlobalIndexMeta.SCHEMA),
+                            new DataField(7, "_SCHEMA_ID", new BigIntType(true))));
 
     private final String indexType;
     private final String fileName;
@@ -71,14 +72,16 @@ public class IndexFileMeta {
 
     private final @Nullable String externalPath;
 
+    @Nullable private final Long schemaId;
+
     public IndexFileMeta(
             String indexType,
             String fileName,
             long fileSize,
             long rowCount,
             @Nullable LinkedHashMap<String, DeletionVectorMeta> dvRanges,
             @Nullable String externalPath) {
-        this(indexType, fileName, fileSize, rowCount, dvRanges, externalPath, null);
+        this(indexType, fileName, fileSize, rowCount, dvRanges, externalPath, null, null);
     }
 
     public IndexFileMeta(
@@ -89,13 +92,34 @@ public IndexFileMeta(
             @Nullable LinkedHashMap<String, DeletionVectorMeta> dvRanges,
             @Nullable String externalPath,
             @Nullable GlobalIndexMeta globalIndexMeta) {
+        this(
+                indexType,
+                fileName,
+                fileSize,
+                rowCount,
+                dvRanges,
+                externalPath,
+                globalIndexMeta,
+                null);
+    }
+
+    public IndexFileMeta(
+            String indexType,
+            String fileName,
+            long fileSize,
+            long rowCount,
+            @Nullable LinkedHashMap<String, DeletionVectorMeta> dvRanges,
+            @Nullable String externalPath,
+            @Nullable GlobalIndexMeta globalIndexMeta,
+            @Nullable Long schemaId) {
         this.indexType = indexType;
         this.fileName = fileName;
         this.fileSize = fileSize;
         this.rowCount = rowCount;
         this.dvRanges = dvRanges;
         this.externalPath = externalPath;
         this.globalIndexMeta = globalIndexMeta;
+        this.schemaId = schemaId;
     }
 
     public IndexFileMeta(
@@ -105,7 +129,7 @@ public IndexFileMeta(
             long rowCount,
             @Nullable GlobalIndexMeta globalIndexMeta,
             @Nullable String externalPath) {
-        this(indexType, fileName, fileSize, rowCount, null, externalPath, globalIndexMeta);
+        this(indexType, fileName, fileSize, rowCount, null, externalPath, globalIndexMeta, null);
     }
 
     public String indexType() {
@@ -138,6 +162,26 @@ public String externalPath() {
         return externalPath;
     }
 
+    @Nullable
+    public Long schemaId() {
+        return schemaId;
+    }
+
+    public IndexFileMeta withSchemaId(@Nullable Long schemaId) {
+        if (Objects.equals(this.schemaId, schemaId)) {
+            return this;
+        }
+        return new IndexFileMeta(
+                indexType,
+                fileName,
+                fileSize,
+                rowCount,
+                dvRanges,
+                externalPath,
+                globalIndexMeta,
+                schemaId);
+    }
+
     @Override
     public boolean equals(Object o) {
         if (this == o) {
@@ -153,13 +197,21 @@ public boolean equals(Object o) {
                 && rowCount == that.rowCount
                 && Objects.equals(dvRanges, that.dvRanges)
                 && Objects.equals(externalPath, that.externalPath)
-                && Objects.equals(globalIndexMeta, that.globalIndexMeta);
+                && Objects.equals(globalIndexMeta, that.globalIndexMeta)
+                && Objects.equals(schemaId, that.schemaId);
     }
 
     @Override
     public int hashCode() {
         return Objects.hash(
-                indexType, fileName, fileSize, rowCount, dvRanges, externalPath, globalIndexMeta);
+                indexType,
+                fileName,
+                fileSize,
+                rowCount,
+                dvRanges,
+                externalPath,
+                globalIndexMeta,
+                schemaId);
     }
 
     @Override
@@ -178,6 +230,9 @@ public String toString() {
                 + dvRanges
                 + ", externalPath='"
                 + externalPath
+                + '\''
+                + ", schemaId="
+                + schemaId
                 + '}';
     }
 }

paimon-core/src/main/java/org/apache/paimon/index/IndexFileMetaSerializer.java

@@ -58,7 +58,8 @@ public InternalRow toRow(IndexFileMeta record) {
                 record.rowCount(),
                 dvMetasToRowArrayData(record.dvRanges()),
                 fromString(record.externalPath()),
-                globalIndexRow);
+                globalIndexRow,
+                record.schemaId());
     }
 
     @Override
@@ -89,7 +90,8 @@ public IndexFileMeta fromRow(InternalRow row) {
                 row.getLong(3),
                 row.isNullAt(4) ? null : rowArrayDataToDvMetas(row.getArray(4)),
                 row.isNullAt(5) ? null : row.getString(5).toString(),
-                globalIndexMeta);
+                globalIndexMeta,
+                row.isNullAt(7) ? null : row.getLong(7));
     }
 
     public static InternalArray dvMetasToRowArrayData(