From 4f72136cd3d3a3fb34c112f9983da2b3fa83f837 Mon Sep 17 00:00:00 2001 From: zhoulii Date: Sun, 23 Aug 2026 22:42:51 +0800 Subject: [PATCH 1/2] [core] Guard global indexes during schema changes --- .../apache/paimon/schema/SchemaManager.java | 142 ++++++++++++++- .../schema/SchemaManagerGlobalIndexTest.java | 169 ++++++++++++++++++ 2 files changed, 310 insertions(+), 1 deletion(-) create mode 100644 paimon-core/src/test/java/org/apache/paimon/schema/SchemaManagerGlobalIndexTest.java diff --git a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java index 73e947d9fa0e..b598602191aa 100644 --- a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java +++ b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java @@ -19,6 +19,7 @@ package org.apache.paimon.schema; import org.apache.paimon.CoreOptions; +import org.apache.paimon.Snapshot; import org.apache.paimon.annotation.VisibleForTesting; import org.apache.paimon.casting.CastExecutors; import org.apache.paimon.catalog.Catalog; @@ -26,6 +27,10 @@ import org.apache.paimon.fs.FileIO; import org.apache.paimon.fs.Path; import org.apache.paimon.iceberg.IcebergOptions; +import org.apache.paimon.index.GlobalIndexMeta; +import org.apache.paimon.index.IndexFileMeta; +import org.apache.paimon.manifest.IndexManifestEntry; +import org.apache.paimon.options.Options; import org.apache.paimon.schema.ColumnDirectiveUtils.ConvertedColumn; import org.apache.paimon.schema.SchemaChange.AddColumn; import org.apache.paimon.schema.SchemaChange.DropColumn; @@ -38,6 +43,8 @@ import org.apache.paimon.schema.SchemaChange.UpdateColumnPosition; import org.apache.paimon.schema.SchemaChange.UpdateColumnType; import org.apache.paimon.schema.SchemaChange.UpdateComment; +import org.apache.paimon.table.CatalogEnvironment; +import org.apache.paimon.table.FileStoreTable; import org.apache.paimon.table.FileStoreTableFactory; import org.apache.paimon.table.SchemaModification; import org.apache.paimon.types.ArrayType; @@ -73,6 +80,8 @@ import java.util.Collections; import java.util.HashMap; import java.util.HashSet; +import java.util.LinkedHashMap; +import java.util.LinkedHashSet; import java.util.List; import java.util.Map; import java.util.Objects; @@ -275,8 +284,16 @@ public TableSchema commitChanges(List changes) tableRoot.toString(), true, branch))); LazyField lazyIdentifier = new LazyField<>(() -> identifierFromPath(tableRoot.toString(), true, branch)); + Set affectedGlobalIndexFieldIds = new LinkedHashSet<>(); TableSchema newTableSchema = - generateTableSchema(oldTableSchema, changes, hasSnapshots, lazyIdentifier); + generateTableSchema( + oldTableSchema, + changes, + hasSnapshots, + lazyIdentifier, + affectedGlobalIndexFieldIds); + assertNoGlobalIndexReferences( + snapshotManager, oldTableSchema, affectedGlobalIndexFieldIds); try { boolean success = commit(newTableSchema); if (success) { @@ -294,6 +311,17 @@ public static TableSchema generateTableSchema( LazyField hasSnapshots, LazyField lazyIdentifier) throws Catalog.ColumnAlreadyExistException, Catalog.ColumnNotExistException { + return generateTableSchema( + oldTableSchema, changes, hasSnapshots, lazyIdentifier, new HashSet<>()); + } + + private static TableSchema generateTableSchema( + TableSchema oldTableSchema, + List changes, + LazyField hasSnapshots, + LazyField lazyIdentifier, + Set affectedGlobalIndexFieldIds) + throws Catalog.ColumnAlreadyExistException, Catalog.ColumnNotExistException { Map oldOptions = new HashMap<>(oldTableSchema.options()); Map newOptions = new HashMap<>(oldTableSchema.options()); boolean disableNullToNotNull = @@ -443,6 +471,8 @@ protected void updateLastColumn( }.updateIntermediateColumn(newFields, 0); } else if (change instanceof RenameColumn) { RenameColumn rename = (RenameColumn) change; + collectAffectedGlobalIndexFieldIds( + newFields, rename.fieldNames(), affectedGlobalIndexFieldIds); assertNotUpdatingPartitionKeys(oldTableSchema, rename.fieldNames(), "rename"); assertNotUpdatingPrimaryKeyIndexColumn( oldTableSchema, rename.fieldNames(), "rename"); @@ -475,6 +505,8 @@ protected void updateLastColumn( }.updateIntermediateColumn(newFields, 0); } else if (change instanceof DropColumn) { DropColumn drop = (DropColumn) change; + collectAffectedGlobalIndexFieldIds( + newFields, drop.fieldNames(), affectedGlobalIndexFieldIds); dropColumnValidation(oldTableSchema, drop); if (drop.fieldNames().length == 1) { String dropName = drop.fieldNames()[0]; @@ -500,6 +532,8 @@ protected void updateLastColumn( }.updateIntermediateColumn(newFields, 0); } else if (change instanceof UpdateColumnType) { UpdateColumnType update = (UpdateColumnType) change; + collectAffectedGlobalIndexFieldIds( + newFields, update.fieldNames(), affectedGlobalIndexFieldIds); assertNotUpdatingPartitionKeys(oldTableSchema, update.fieldNames(), "update"); assertNotUpdatingPrimaryKeys(oldTableSchema, update.fieldNames(), "update"); assertNotUpdatingPrimaryKeyIndexColumn( @@ -1035,6 +1069,112 @@ private static void assertNotUpdatingPrimaryKeyIndexColumn( } } + private void assertNoGlobalIndexReferences( + SnapshotManager snapshotManager, TableSchema schema, Set affectedFieldIds) { + if (affectedFieldIds.isEmpty()) { + return; + } + + Snapshot snapshot = snapshotManager.latestSnapshot(); + if (snapshot == null || snapshot.indexManifest() == null) { + return; + } + + Options dynamicOptions = new Options(); + dynamicOptions.set(CoreOptions.BRANCH, branch); + FileStoreTable table = + FileStoreTableFactory.createWithoutFallbackBranch( + fileIO, tableRoot, schema, dynamicOptions, CatalogEnvironment.empty()); + List references = + table.store() + .newIndexFileHandler() + .scan( + snapshot, + entry -> { + GlobalIndexMeta meta = entry.indexFile().globalIndexMeta(); + return meta != null + && meta.getIndexedFieldIds().stream() + .anyMatch(affectedFieldIds::contains); + }); + if (references.isEmpty()) { + return; + } + + Map referenceSummary = new LinkedHashMap<>(); + for (IndexManifestEntry reference : references) { + IndexFileMeta indexFile = reference.indexFile(); + String key = + String.format( + "type=%s, indexed-field-ids=%s", + indexFile.indexType(), + indexFile.globalIndexMeta().getIndexedFieldIds()); + referenceSummary.put(key, referenceSummary.getOrDefault(key, 0) + 1); + } + List summaries = new ArrayList<>(); + for (Map.Entry entry : referenceSummary.entrySet()) { + summaries.add(entry.getKey() + ", files=" + entry.getValue()); + } + + throw new UnsupportedOperationException( + String.format( + "Cannot drop, rename, or update the type of columns with field ids %s " + + "because they are referenced by live Global Index files: %s. " + + "Drop the complete Global Index before altering the indexed columns.", + affectedFieldIds, String.join("; ", summaries))); + } + + private static void collectAffectedGlobalIndexFieldIds( + List fields, String[] fieldNames, Set affectedFieldIds) { + collectAffectedGlobalIndexFieldIds(fields, fieldNames, 0, affectedFieldIds); + } + + private static void collectAffectedGlobalIndexFieldIds( + List fields, String[] fieldNames, int depth, Set affectedFieldIds) { + DataField field = null; + for (DataField candidate : fields) { + if (candidate.name().equals(fieldNames[depth])) { + field = candidate; + break; + } + } + if (field == null) { + return; + } + + affectedFieldIds.add(field.id()); + if (depth == fieldNames.length - 1) { + field.type().collectFieldIds(affectedFieldIds); + return; + } + + DataType nestedType = field.type(); + int nestedDepth = depth; + while (true) { + switch (nestedType.getTypeRoot()) { + case ROW: + nestedDepth++; + if (nestedDepth < fieldNames.length) { + collectAffectedGlobalIndexFieldIds( + ((RowType) nestedType).getFields(), + fieldNames, + nestedDepth, + affectedFieldIds); + } + return; + case ARRAY: + nestedDepth++; + nestedType = ((ArrayType) nestedType).getElementType(); + break; + case MAP: + nestedDepth++; + nestedType = ((MapType) nestedType).getValueType(); + break; + default: + return; + } + } + } + private static void assertNotChangingBlobColumnType( List fields, String[] fieldNames, DataType newType) { if (fieldNames.length > 1) { diff --git a/paimon-core/src/test/java/org/apache/paimon/schema/SchemaManagerGlobalIndexTest.java b/paimon-core/src/test/java/org/apache/paimon/schema/SchemaManagerGlobalIndexTest.java new file mode 100644 index 000000000000..81fedf8a0041 --- /dev/null +++ b/paimon-core/src/test/java/org/apache/paimon/schema/SchemaManagerGlobalIndexTest.java @@ -0,0 +1,169 @@ +/* + * 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.schema; + +import org.apache.paimon.CoreOptions; +import org.apache.paimon.data.BinaryRow; +import org.apache.paimon.data.GenericRow; +import org.apache.paimon.fs.FileIO; +import org.apache.paimon.fs.Path; +import org.apache.paimon.fs.local.LocalFileIO; +import org.apache.paimon.index.GlobalIndexMeta; +import org.apache.paimon.index.IndexFileMeta; +import org.apache.paimon.io.CompactIncrement; +import org.apache.paimon.io.DataIncrement; +import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.FileStoreTableFactory; +import org.apache.paimon.table.sink.CommitMessageImpl; +import org.apache.paimon.table.sink.TableCommitImpl; +import org.apache.paimon.table.sink.TableWriteImpl; +import org.apache.paimon.types.DataField; +import org.apache.paimon.types.DataTypes; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; + +import static org.apache.paimon.table.BucketMode.UNAWARE_BUCKET; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatCode; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Tests schema changes which affect columns referenced by live Global Index files. */ +public class SchemaManagerGlobalIndexTest { + + private static final int INDEX_FIELD_ID = 1; + private static final int EXTRA_FIELD_ID = 2; + + @TempDir java.nio.file.Path tempDir; + + private SchemaManager schemaManager; + private FileStoreTable table; + private IndexFileMeta globalIndex; + + @BeforeEach + public void beforeEach() throws Exception { + FileIO fileIO = LocalFileIO.create(); + Path tablePath = new Path(tempDir.toString()); + schemaManager = new SchemaManager(fileIO, tablePath); + + Map options = new HashMap<>(); + options.put(CoreOptions.BUCKET.key(), "-1"); + options.put(CoreOptions.ROW_TRACKING_ENABLED.key(), "true"); + schemaManager.createTable( + new Schema( + Arrays.asList( + new DataField(0, "id", DataTypes.INT()), + new DataField(INDEX_FIELD_ID, "indexed_col", DataTypes.INT()), + new DataField(EXTRA_FIELD_ID, "extra_col", DataTypes.INT()), + new DataField(3, "other_col", DataTypes.INT())), + Collections.emptyList(), + Collections.emptyList(), + options, + null)); + + table = FileStoreTableFactory.create(fileIO, tablePath); + writeOneRow(); + globalIndex = + new IndexFileMeta( + "test-global-index", + "global-index-file", + 1, + 1, + new GlobalIndexMeta(0, 0, INDEX_FIELD_ID, new int[] {EXTRA_FIELD_ID}, null), + null); + commitIndex(DataIncrement.indexIncrement(Collections.singletonList(globalIndex))); + } + + @Test + public void testRejectReferencedGlobalIndexColumnChanges() { + assertReferencedChangeRejected(SchemaChange.dropColumn("indexed_col")); + assertReferencedChangeRejected( + SchemaChange.renameColumn("indexed_col", "renamed_indexed_col")); + assertReferencedChangeRejected( + SchemaChange.updateColumnType("indexed_col", DataTypes.BIGINT())); + + assertReferencedChangeRejected(SchemaChange.dropColumn("extra_col")); + assertReferencedChangeRejected(SchemaChange.renameColumn("extra_col", "renamed_extra_col")); + assertReferencedChangeRejected( + SchemaChange.updateColumnType("extra_col", DataTypes.BIGINT())); + } + + @Test + public void testAllowUnrelatedSchemaChanges() { + assertThatCode( + () -> + schemaManager.commitChanges( + SchemaChange.renameColumn("other_col", "renamed_other"), + SchemaChange.updateColumnComment( + new String[] {"indexed_col"}, "comment"))) + .doesNotThrowAnyException(); + + assertThat(schemaManager.latest().get().fieldNames()) + .containsExactly("id", "indexed_col", "extra_col", "renamed_other"); + } + + @Test + public void testAllowSchemaChangeAfterDroppingGlobalIndex() throws Exception { + commitIndex(DataIncrement.deleteIndexIncrement(Collections.singletonList(globalIndex))); + + assertThatCode( + () -> + schemaManager.commitChanges( + SchemaChange.updateColumnType( + "indexed_col", DataTypes.BIGINT()), + SchemaChange.renameColumn("extra_col", "renamed_extra"))) + .doesNotThrowAnyException(); + } + + private void assertReferencedChangeRejected(SchemaChange change) { + assertThatThrownBy(() -> schemaManager.commitChanges(change)) + .isInstanceOf(UnsupportedOperationException.class) + .hasMessageContaining("referenced by live Global Index files") + .hasMessageContaining("indexed-field-ids=[1, 2]") + .hasMessageContaining("Drop the complete Global Index"); + } + + private void writeOneRow() throws Exception { + String commitUser = "write-row"; + try (TableWriteImpl write = table.newWrite(commitUser); + TableCommitImpl commit = table.newCommit(commitUser)) { + write.write(GenericRow.of(0, 1, 2, 3)); + commit.commit(write.prepareCommit(false, 1)); + } + } + + private void commitIndex(DataIncrement increment) throws Exception { + try (TableCommitImpl commit = table.newCommit("global-index")) { + commit.commit( + Collections.singletonList( + new CommitMessageImpl( + BinaryRow.EMPTY_ROW, + UNAWARE_BUCKET, + null, + increment, + CompactIncrement.emptyIncrement()))); + } + } +} From 232a855dc9fb8537b070c43b6c5a920195be708c Mon Sep 17 00:00:00 2001 From: zhoulii Date: Wed, 26 Aug 2026 21:13:05 +0800 Subject: [PATCH 2/2] [core] Align global index guard with top-level columns --- .../apache/paimon/schema/SchemaManager.java | 49 +++---------------- .../schema/SchemaManagerGlobalIndexTest.java | 6 ++- 2 files changed, 10 insertions(+), 45 deletions(-) diff --git a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java index b598602191aa..3820e0450996 100644 --- a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java +++ b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java @@ -1125,52 +1125,15 @@ private void assertNoGlobalIndexReferences( private static void collectAffectedGlobalIndexFieldIds( List fields, String[] fieldNames, Set affectedFieldIds) { - collectAffectedGlobalIndexFieldIds(fields, fieldNames, 0, affectedFieldIds); - } - - private static void collectAffectedGlobalIndexFieldIds( - List fields, String[] fieldNames, int depth, Set affectedFieldIds) { - DataField field = null; - for (DataField candidate : fields) { - if (candidate.name().equals(fieldNames[depth])) { - field = candidate; - break; - } - } - if (field == null) { + if (fieldNames.length == 0) { return; } - affectedFieldIds.add(field.id()); - if (depth == fieldNames.length - 1) { - field.type().collectFieldIds(affectedFieldIds); - return; - } - - DataType nestedType = field.type(); - int nestedDepth = depth; - while (true) { - switch (nestedType.getTypeRoot()) { - case ROW: - nestedDepth++; - if (nestedDepth < fieldNames.length) { - collectAffectedGlobalIndexFieldIds( - ((RowType) nestedType).getFields(), - fieldNames, - nestedDepth, - affectedFieldIds); - } - return; - case ARRAY: - nestedDepth++; - nestedType = ((ArrayType) nestedType).getElementType(); - break; - case MAP: - nestedDepth++; - nestedType = ((MapType) nestedType).getValueType(); - break; - default: - return; + // Global Index metadata resolves field IDs against the table's top-level row type. + for (DataField field : fields) { + if (field.name().equals(fieldNames[0])) { + affectedFieldIds.add(field.id()); + return; } } } diff --git a/paimon-core/src/test/java/org/apache/paimon/schema/SchemaManagerGlobalIndexTest.java b/paimon-core/src/test/java/org/apache/paimon/schema/SchemaManagerGlobalIndexTest.java index 81fedf8a0041..c18ddf0b8689 100644 --- a/paimon-core/src/test/java/org/apache/paimon/schema/SchemaManagerGlobalIndexTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/schema/SchemaManagerGlobalIndexTest.java @@ -131,10 +131,12 @@ public void testAllowSchemaChangeAfterDroppingGlobalIndex() throws Exception { assertThatCode( () -> schemaManager.commitChanges( - SchemaChange.updateColumnType( - "indexed_col", DataTypes.BIGINT()), + SchemaChange.dropColumn("indexed_col"), SchemaChange.renameColumn("extra_col", "renamed_extra"))) .doesNotThrowAnyException(); + + assertThat(schemaManager.latest().get().fieldNames()) + .containsExactly("id", "renamed_extra", "other_col"); } private void assertReferencedChangeRejected(SchemaChange change) {