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

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

@@ -667,7 +667,8 @@ private RewrittenIndexManifest rewriteIndexManifest(Assignment assignment) {
                             globalIndex.indexFieldId(),
                             globalIndex.extraFieldIds(),
                             globalIndex.indexMeta(),
-                            globalIndex.sourceMeta());
+                            globalIndex.sourceMeta(),
+                            globalIndex.buildSchemaId());
             IndexFileMeta newIndexFile =
                     new IndexFileMeta(
                             indexFile.indexType(),

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

@@ -228,7 +228,8 @@ public static Optional<DataEvolutionGlobalIndexScanner> create(
             @Nullable Snapshot pinnedSnapshot,
             @Nullable PartitionPredicate partitionFilter,
             Collection<IndexFileMeta> indexFiles) {
-        List<IndexFileMeta> globalIndexFiles = globalIndexFiles(indexFiles);
+        List<IndexFileMeta> globalIndexFiles =
+                GlobalIndexSchemaCompatibility.filterCompatible(table, indexFiles);
         if (globalIndexFiles.isEmpty()) {
             return Optional.empty();
         }
@@ -254,6 +255,7 @@ public static Optional<DataEvolutionGlobalIndexScanner> create(
                         .scan(snapshot, indexFileFilter(table, partitionFilter, filter)).stream()
                         .map(IndexManifestEntry::indexFile)
                         .collect(Collectors.toList());
+        indexFiles = GlobalIndexSchemaCompatibility.filterCompatible(table, indexFiles);
         if (indexFiles.isEmpty()) {
             return Optional.empty();
         }
@@ -289,6 +291,7 @@ public static Optional<DataEvolutionGlobalIndexScanner> createForTopN(
                         .scan(snapshot, topNIndexFileFilter(partitionFilter, fieldId)).stream()
                         .map(IndexManifestEntry::indexFile)
                         .collect(Collectors.toList());
+        indexFiles = GlobalIndexSchemaCompatibility.filterCompatible(table, indexFiles);
         if (indexFiles.isEmpty()) {
             return Optional.empty();
         }
@@ -362,12 +365,6 @@ private static Filter<IndexManifestEntry> indexFileFilter(
         return indexFileFilter;
     }
 
-    private static List<IndexFileMeta> globalIndexFiles(Collection<IndexFileMeta> indexFiles) {
-        return indexFiles.stream()
-                .filter(indexFile -> indexFile.globalIndexMeta() != null)
-                .collect(Collectors.toList());
-    }
-
     public Optional<GlobalIndexResult> scan(Predicate predicate) {
         return globalIndexEvaluator.evaluate(predicate);
     }

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 buildSchemaId)
             throws IOException {
         return toIndexFileMetas(
                 fileIO,
@@ -83,7 +84,8 @@ public static List<IndexFileMeta> toIndexFileMetas(
                 null,
                 indexType,
                 entries,
-                null);
+                null,
+                buildSchemaId);
     }
 
     /**
@@ -100,7 +102,8 @@ public static List<IndexFileMeta> toIndexFileMetas(
             List<DataField> fields,
             String indexType,
             List<ResultEntry> entries,
-            @Nullable byte[] sourceMeta)
+            @Nullable byte[] sourceMeta,
+            long buildSchemaId)
             throws IOException {
         return toIndexFileMetas(
                 fileIO,
@@ -111,7 +114,8 @@ public static List<IndexFileMeta> toIndexFileMetas(
                 extraFieldIds(fields),
                 indexType,
                 entries,
-                sourceMeta);
+                sourceMeta,
+                buildSchemaId);
     }
 
     public static List<Range> unindexedRowRanges(
@@ -569,7 +573,8 @@ private static List<IndexFileMeta> toIndexFileMetas(
             @Nullable int[] extraFieldIds,
             String indexType,
             List<ResultEntry> entries,
-            @Nullable byte[] sourceMeta)
+            @Nullable byte[] sourceMeta,
+            long buildSchemaId)
             throws IOException {
         List<IndexFileMeta> results = new ArrayList<>();
         for (ResultEntry entry : entries) {
@@ -582,7 +587,8 @@ private static List<IndexFileMeta> toIndexFileMetas(
                             indexFieldId,
                             extraFieldIds,
                             entry.meta(),
-                            sourceMeta);
+                            sourceMeta,
+                            buildSchemaId);
 
             Path externalPathDir = options.globalIndexExternalPath();
             String externalPathString = null;

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

@@ -0,0 +1,87 @@
+/*
+ * 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.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 indexes against the current table schema. */
+public final class GlobalIndexSchemaCompatibility {
+
+    public static List<IndexFileMeta> filterCompatible(
+            FileStoreTable table, Collection<IndexFileMeta> indexFiles) {
+        RowType currentRowType = table.rowType();
+        Map<Long, RowType> buildRowTypes = new HashMap<>();
+        buildRowTypes.put(table.schema().id(), currentRowType);
+        Set<Long> missingSchemaIds = new HashSet<>();
+        List<IndexFileMeta> compatible = new ArrayList<>();
+        for (IndexFileMeta indexFile : indexFiles) {
+            GlobalIndexMeta globalIndex = indexFile.globalIndexMeta();
+            if (globalIndex == null || globalIndex.buildSchemaId() == null) {
+                continue;
+            }
+
+            long buildSchemaId = globalIndex.buildSchemaId();
+            RowType buildRowType = buildRowTypes.get(buildSchemaId);
+            if (buildRowType == null && !missingSchemaIds.contains(buildSchemaId)) {
+                try {
+                    buildRowType =
+                            table.schemaManager().tryGetSchema(buildSchemaId).logicalRowType();
+                    buildRowTypes.put(buildSchemaId, buildRowType);
+                } catch (FileNotFoundException e) {
+                    missingSchemaIds.add(buildSchemaId);
+                }
+            }
+            if (buildRowType != null
+                    && compatibleIndexedFields(globalIndex, buildRowType, currentRowType)) {
+                compatible.add(indexFile);
+            }
+        }
+        return compatible;
+    }
+
+    private static boolean compatibleIndexedFields(
+            GlobalIndexMeta globalIndex, RowType buildRowType, RowType currentRowType) {
+        for (int fieldId : globalIndex.getIndexedFieldIds()) {
+            if (!buildRowType.containsField(fieldId) || !currentRowType.containsField(fieldId)) {
+                return false;
+            }
+            if (!buildRowType
+                    .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/GlobalIndexMeta.java

@@ -41,6 +41,7 @@ public class GlobalIndexMeta {
     public static final String EXTRA_FIELD_IDS = "_EXTRA_FIELD_IDS";
     public static final String INDEX_META = "_INDEX_META";
     public static final String SOURCE_META = "_SOURCE_META";
+    public static final String BUILD_SCHEMA_ID = "_BUILD_SCHEMA_ID";
 
     public static final RowType SCHEMA =
             new RowType(
@@ -51,14 +52,16 @@ public class GlobalIndexMeta {
                             new DataField(2, INDEX_FIELD_ID, new IntType(false)),
                             new DataField(3, EXTRA_FIELD_IDS, DataTypes.ARRAY(new IntType(false))),
                             new DataField(4, INDEX_META, DataTypes.BYTES()),
-                            new DataField(5, SOURCE_META, DataTypes.BYTES())));
+                            new DataField(5, SOURCE_META, DataTypes.BYTES()),
+                            new DataField(6, BUILD_SCHEMA_ID, new BigIntType())));
 
     private final long rowRangeStart;
     private final long rowRangeEnd;
     private final int indexFieldId;
     @Nullable private final int[] extraFieldIds;
     @Nullable private final byte[] indexMeta;
     @Nullable private final byte[] sourceMeta;
+    @Nullable private final Long buildSchemaId;
 
     public GlobalIndexMeta(
             long rowRangeStart,
@@ -76,12 +79,24 @@ public GlobalIndexMeta(
             @Nullable int[] extraFieldIds,
             @Nullable byte[] indexMeta,
             @Nullable byte[] sourceMeta) {
+        this(rowRangeStart, rowRangeEnd, indexFieldId, extraFieldIds, indexMeta, sourceMeta, null);
+    }
+
+    public GlobalIndexMeta(
+            long rowRangeStart,
+            long rowRangeEnd,
+            int indexFieldId,
+            @Nullable int[] extraFieldIds,
+            @Nullable byte[] indexMeta,
+            @Nullable byte[] sourceMeta,
+            @Nullable Long buildSchemaId) {
         this.rowRangeStart = rowRangeStart;
         this.rowRangeEnd = rowRangeEnd;
         this.indexFieldId = indexFieldId;
         this.extraFieldIds = extraFieldIds;
         this.indexMeta = indexMeta;
         this.sourceMeta = sourceMeta;
+        this.buildSchemaId = buildSchemaId;
     }
 
     public long rowRangeStart() {
@@ -117,6 +132,12 @@ public byte[] sourceMeta() {
         return sourceMeta;
     }
 
+    /** Schema used to build this global index. */
+    @Nullable
+    public Long buildSchemaId() {
+        return buildSchemaId;
+    }
+
     /** All indexed field ids in order: the primary {@link #indexFieldId} followed by the rest. */
     public List<Integer> getIndexedFieldIds() {
         List<Integer> ids = new ArrayList<>();
@@ -175,12 +196,13 @@ public boolean equals(Object o) {
                 && indexFieldId == that.indexFieldId
                 && Arrays.equals(extraFieldIds, that.extraFieldIds)
                 && Arrays.equals(indexMeta, that.indexMeta)
-                && Arrays.equals(sourceMeta, that.sourceMeta);
+                && Arrays.equals(sourceMeta, that.sourceMeta)
+                && Objects.equals(buildSchemaId, that.buildSchemaId);
     }
 
     @Override
     public int hashCode() {
-        int result = Objects.hash(rowRangeStart, rowRangeEnd, indexFieldId);
+        int result = Objects.hash(rowRangeStart, rowRangeEnd, indexFieldId, buildSchemaId);
         result = 31 * result + Arrays.hashCode(extraFieldIds);
         result = 31 * result + Arrays.hashCode(indexMeta);
         result = 31 * result + Arrays.hashCode(sourceMeta);

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

@@ -50,7 +50,8 @@ public InternalRow toRow(IndexFileMeta record) {
                                         ? null
                                         : new GenericArray(globalIndexMeta.extraFieldIds()),
                                 globalIndexMeta.indexMeta(),
-                                globalIndexMeta.sourceMeta());
+                                globalIndexMeta.sourceMeta(),
+                                globalIndexMeta.buildSchemaId());
         return GenericRow.of(
                 fromString(record.indexType()),
                 fromString(record.fileName()),
@@ -65,22 +66,24 @@ public InternalRow toRow(IndexFileMeta record) {
     public IndexFileMeta fromRow(InternalRow row) {
         GlobalIndexMeta globalIndexMeta = null;
         if (!row.isNullAt(6)) {
-            InternalRow globalIndexRow = row.getRow(6, 6);
+            InternalRow globalIndexRow = row.getRow(6, GlobalIndexMeta.SCHEMA.getFieldCount());
             Long rowRangeStart = globalIndexRow.getLong(0);
             Long rowRangeEnd = globalIndexRow.getLong(1);
             Integer indexFieldId = globalIndexRow.getInt(2);
             int[] extralFields =
                     globalIndexRow.isNullAt(3) ? null : globalIndexRow.getArray(3).toIntArray();
             byte[] indexMeta = globalIndexRow.isNullAt(4) ? null : globalIndexRow.getBinary(4);
             byte[] sourceMeta = globalIndexRow.isNullAt(5) ? null : globalIndexRow.getBinary(5);
+            Long buildSchemaId = globalIndexRow.isNullAt(6) ? null : globalIndexRow.getLong(6);
             globalIndexMeta =
                     new GlobalIndexMeta(
                             rowRangeStart,
                             rowRangeEnd,
                             indexFieldId,
                             extralFields,
                             indexMeta,
-                            sourceMeta);
+                            sourceMeta,
+                            buildSchemaId);
         }
         return new IndexFileMeta(
                 row.getString(0).toString(),