[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

@@ -228,7 +228,9 @@ public static Optional<DataEvolutionGlobalIndexScanner> create(
             @Nullable Snapshot pinnedSnapshot,
             @Nullable PartitionPredicate partitionFilter,
             Collection<IndexFileMeta> indexFiles) {
-        List<IndexFileMeta> globalIndexFiles = globalIndexFiles(indexFiles);
+        List<IndexFileMeta> globalIndexFiles =
+                GlobalIndexSchemaCompatibility.filterCompatibleFiles(
+                        table, globalIndexFiles(indexFiles));
         if (globalIndexFiles.isEmpty()) {
             return Optional.empty();
         }
@@ -249,9 +251,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 +289,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,144 @@
+/*
+ * 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.index.IndexFileMeta;
+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) {
+        return partitionByCompatibility(table, entries).compatible();
+    }
+
+    public static CompatibilityResult partitionByCompatibility(
+            FileStoreTable table, Collection<IndexManifestEntry> entries) {
+        CompatibilityChecker checker = new CompatibilityChecker(table);
+        List<IndexManifestEntry> compatible = new ArrayList<>();
+        List<IndexManifestEntry> incompatible = new ArrayList<>();
+        for (IndexManifestEntry entry : entries) {
+            if (checker.isCompatible(entry.indexFile(), entry.schemaId())) {
+                compatible.add(entry);
+            } else {
+                incompatible.add(entry);
+            }
+        }
+        return new CompatibilityResult(compatible, incompatible);
+    }
+
+    public static List<IndexFileMeta> filterCompatibleFiles(
+            FileStoreTable table, Collection<IndexFileMeta> indexFiles) {
+        CompatibilityChecker checker = new CompatibilityChecker(table);
+        List<IndexFileMeta> compatible = new ArrayList<>();
+        for (IndexFileMeta indexFile : indexFiles) {
+            if (checker.isCompatible(indexFile, indexFile.schemaId())) {
+                compatible.add(indexFile);
+            }
+        }
+        return compatible;
+    }
+
+    /** Global index manifest entries grouped by compatibility with the current table schema. */
+    public static final class CompatibilityResult {
+
+        private final List<IndexManifestEntry> compatible;
+        private final List<IndexManifestEntry> incompatible;
+
+        private CompatibilityResult(
+                List<IndexManifestEntry> compatible, List<IndexManifestEntry> incompatible) {
+            this.compatible = compatible;
+            this.incompatible = incompatible;
+        }
+
+        public List<IndexManifestEntry> compatible() {
+            return compatible;
+        }
+
+        public List<IndexManifestEntry> incompatible() {
+            return incompatible;
+        }
+    }
+
+    private static class CompatibilityChecker {
+
+        private final FileStoreTable table;
+        private final RowType currentRowType;
+        private final Map<Long, RowType> historicalRowTypes = new HashMap<>();
+        private final Set<Long> missingSchemaIds = new HashSet<>();
+
+        private CompatibilityChecker(FileStoreTable table) {
+            this.table = table;
+            this.currentRowType = table.rowType();
+            historicalRowTypes.put(table.schema().id(), currentRowType);
+        }
+
+        private boolean isCompatible(IndexFileMeta indexFile, Long schemaId) {
+            GlobalIndexMeta globalIndex = indexFile.globalIndexMeta();
+            if (globalIndex == null || schemaId == null) {
+                return false;
+            }
+
+            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);
+                }
+            }
+            return historicalRowType != null
+                    && compatibleIndexedFields(globalIndex, historicalRowType, currentRowType);
+        }
+    }
+
+    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/generic/GenericGlobalIndexScanner.java

@@ -21,6 +21,7 @@
 import org.apache.paimon.CoreOptions;
 import org.apache.paimon.Snapshot;
 import org.apache.paimon.globalindex.DataEvolutionGlobalIndexRefreshPlanner;
+import org.apache.paimon.globalindex.GlobalIndexSchemaCompatibility;
 import org.apache.paimon.globalindex.ScanResult;
 import org.apache.paimon.manifest.IndexManifestEntry;
 import org.apache.paimon.manifest.ManifestEntry;
@@ -139,19 +140,27 @@ public Optional<ScanResult<ManifestEntry>> incrementalScan() {
         List<IndexManifestEntry> currentIndexes =
                 currentIndexEntries(
                         table, scanSnapshot, indexType, indexFields, partitionPredicate);
+        GlobalIndexSchemaCompatibility.CompatibilityResult compatibility =
+                GlobalIndexSchemaCompatibility.partitionByCompatibility(table, currentIndexes);
+        List<IndexManifestEntry> compatibleIndexes = compatibility.compatible();
         List<Range> rangesToBuild =
-                new ArrayList<>(unindexedRowRanges(scanSnapshot, currentIndexes));
-        List<IndexManifestEntry> deletedIndexEntries = Collections.emptyList();
+                new ArrayList<>(unindexedRowRanges(scanSnapshot, compatibleIndexes));
+        List<IndexManifestEntry> deletedIndexEntries =
+                new ArrayList<>(compatibility.incompatible());
+        for (IndexManifestEntry entry : compatibility.incompatible()) {
+            rangesToBuild.add(entry.indexFile().globalIndexMeta().rowRange());
+        }
         Options mergedOptions = new Options(table.options(), options.toMap());
         if (mergedOptions.get(CoreOptions.GLOBAL_INDEX_COLUMN_UPDATE_ACTION)
                 == CoreOptions.GlobalIndexColumnUpdateAction.IGNORE) {
-            deletedIndexEntries =
+            List<IndexManifestEntry> indexesToRefresh =
                     DataEvolutionGlobalIndexRefreshPlanner.findIndexesToRefresh(
                             table.schemaManager(),
                             scanResult.entries(),
-                            currentIndexes,
+                            compatibleIndexes,
                             indexFields);
-            for (IndexManifestEntry entry : deletedIndexEntries) {
+            deletedIndexEntries.addAll(indexesToRefresh);
+            for (IndexManifestEntry entry : indexesToRefresh) {
                 rangesToBuild.add(entry.indexFile().globalIndexMeta().rowRange());
             }
         }

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

@@ -21,6 +21,7 @@
 import org.apache.paimon.CoreOptions;
 import org.apache.paimon.Snapshot;
 import org.apache.paimon.globalindex.DataEvolutionGlobalIndexRefreshPlanner;
+import org.apache.paimon.globalindex.GlobalIndexSchemaCompatibility;
 import org.apache.paimon.globalindex.ScanResult;
 import org.apache.paimon.manifest.IndexManifestEntry;
 import org.apache.paimon.options.Options;
@@ -140,18 +141,27 @@ public Optional<ScanResult<DataSplit>> incrementalScan() {
                         indexType,
                         Collections.singletonList(indexField),
                         partitionPredicate);
-        List<Range> rangesToBuild = new ArrayList<>(unindexedRowRanges(snapshot, currentIndexes));
-        List<IndexManifestEntry> deletedIndexEntries = Collections.emptyList();
+        GlobalIndexSchemaCompatibility.CompatibilityResult compatibility =
+                GlobalIndexSchemaCompatibility.partitionByCompatibility(table, currentIndexes);
+        List<IndexManifestEntry> compatibleIndexes = compatibility.compatible();
+        List<Range> rangesToBuild =
+                new ArrayList<>(unindexedRowRanges(snapshot, compatibleIndexes));
+        List<IndexManifestEntry> deletedIndexEntries =
+                new ArrayList<>(compatibility.incompatible());
+        for (IndexManifestEntry entry : compatibility.incompatible()) {
+            rangesToBuild.add(entry.indexFile().globalIndexMeta().rowRange());
+        }
         if (detectDataFileChange()) {
             // Scans data manifests through reusable binary views without materializing entries.
-            deletedIndexEntries =
+            List<IndexManifestEntry> indexesToRefresh =
                     DataEvolutionGlobalIndexRefreshPlanner.findIndexesToRefresh(
                             table,
                             snapshot,
                             partitionPredicate,
-                            currentIndexes,
+                            compatibleIndexes,
                             Collections.singletonList(indexField));
-            for (IndexManifestEntry entry : deletedIndexEntries) {
+            deletedIndexEntries.addAll(indexesToRefresh);
+            for (IndexManifestEntry entry : indexesToRefresh) {
                 rangesToBuild.add(entry.indexFile().globalIndexMeta().rowRange());
             }
         }

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());