diff --git a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdReassigner.java b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdReassigner.java index ec12b3fd2131..904c0eb587ef 100644 --- a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdReassigner.java +++ b/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( diff --git a/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionGlobalIndexScanner.java b/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionGlobalIndexScanner.java index d39174378587..6c515e898a05 100644 --- a/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionGlobalIndexScanner.java +++ b/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionGlobalIndexScanner.java @@ -249,9 +249,12 @@ public static Optional create( @Nullable PartitionPredicate partitionFilter, @Nullable Predicate filter) { @Nullable Snapshot snapshot = tryTravelOrLatest(table); + List indexEntries = + table.store() + .newIndexFileHandler() + .scan(snapshot, indexFileFilter(table, partitionFilter, filter)); List 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 createForTopN( DataField indexField = table.rowType().getField(topN.orders().get(0).field().name()); int fieldId = indexField.id(); @Nullable Snapshot snapshot = tryTravelOrLatest(table); + List indexEntries = + table.store() + .newIndexFileHandler() + .scan(snapshot, topNIndexFileFilter(partitionFilter, fieldId)); List indexFiles = - table.store().newIndexFileHandler() - .scan(snapshot, topNIndexFileFilter(partitionFilter, fieldId)).stream() + GlobalIndexSchemaCompatibility.filterCompatible(table, indexEntries).stream() .map(IndexManifestEntry::indexFile) .collect(Collectors.toList()); if (indexFiles.isEmpty()) { diff --git a/paimon-core/src/main/java/org/apache/paimon/globalindex/GlobalIndexBuilderUtils.java b/paimon-core/src/main/java/org/apache/paimon/globalindex/GlobalIndexBuilderUtils.java index aea1efee5dbd..40e4c09bc2da 100644 --- a/paimon-core/src/main/java/org/apache/paimon/globalindex/GlobalIndexBuilderUtils.java +++ b/paimon-core/src/main/java/org/apache/paimon/globalindex/GlobalIndexBuilderUtils.java @@ -72,7 +72,8 @@ public static List toIndexFileMetas( Range range, int indexFieldId, String indexType, - List entries) + List entries, + long schemaId) throws IOException { return toIndexFileMetas( fileIO, @@ -83,7 +84,8 @@ public static List toIndexFileMetas( null, indexType, entries, - null); + null, + schemaId); } /** @@ -100,7 +102,8 @@ public static List toIndexFileMetas( List fields, String indexType, List entries, - @Nullable byte[] sourceMeta) + @Nullable byte[] sourceMeta, + long schemaId) throws IOException { return toIndexFileMetas( fileIO, @@ -111,7 +114,8 @@ public static List toIndexFileMetas( extraFieldIds(fields), indexType, entries, - sourceMeta); + sourceMeta, + schemaId); } public static List unindexedRowRanges( @@ -572,7 +576,8 @@ private static List toIndexFileMetas( @Nullable int[] extraFieldIds, String indexType, List entries, - @Nullable byte[] sourceMeta) + @Nullable byte[] sourceMeta, + long schemaId) throws IOException { List results = new ArrayList<>(); for (ResultEntry entry : entries) { @@ -599,8 +604,10 @@ private static List toIndexFileMetas( fileName, fileSize, entry.rowCount(), + null, + externalPathString, globalIndexMeta, - externalPathString); + schemaId); results.add(indexFileMeta); } return results; diff --git a/paimon-core/src/main/java/org/apache/paimon/globalindex/GlobalIndexSchemaCompatibility.java b/paimon-core/src/main/java/org/apache/paimon/globalindex/GlobalIndexSchemaCompatibility.java new file mode 100644 index 000000000000..2ae02b85ed22 --- /dev/null +++ b/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 filterCompatible( + FileStoreTable table, Collection entries) { + RowType currentRowType = table.rowType(); + Map historicalRowTypes = new HashMap<>(); + historicalRowTypes.put(table.schema().id(), currentRowType); + Set missingSchemaIds = new HashSet<>(); + List 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() {} +} diff --git a/paimon-core/src/main/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexWriter.java b/paimon-core/src/main/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexWriter.java index 52cfd6479f02..9f7afca8f2d8 100644 --- a/paimon-core/src/main/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexWriter.java +++ b/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()); diff --git a/paimon-core/src/main/java/org/apache/paimon/index/IndexFileMeta.java b/paimon-core/src/main/java/org/apache/paimon/index/IndexFileMeta.java index cc77e30438f3..c18c677f28cd 100644 --- a/paimon-core/src/main/java/org/apache/paimon/index/IndexFileMeta.java +++ b/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,6 +72,8 @@ public class IndexFileMeta { private final @Nullable String externalPath; + @Nullable private final Long schemaId; + public IndexFileMeta( String indexType, String fileName, @@ -78,7 +81,7 @@ public IndexFileMeta( long rowCount, @Nullable LinkedHashMap dvRanges, @Nullable String externalPath) { - this(indexType, fileName, fileSize, rowCount, dvRanges, externalPath, null); + this(indexType, fileName, fileSize, rowCount, dvRanges, externalPath, null, null); } public IndexFileMeta( @@ -89,6 +92,26 @@ public IndexFileMeta( @Nullable LinkedHashMap 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 dvRanges, + @Nullable String externalPath, + @Nullable GlobalIndexMeta globalIndexMeta, + @Nullable Long schemaId) { this.indexType = indexType; this.fileName = fileName; this.fileSize = fileSize; @@ -96,6 +119,7 @@ public IndexFileMeta( 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 + '}'; } } diff --git a/paimon-core/src/main/java/org/apache/paimon/index/IndexFileMetaSerializer.java b/paimon-core/src/main/java/org/apache/paimon/index/IndexFileMetaSerializer.java index 6e71c5f74a5b..42d738375440 100644 --- a/paimon-core/src/main/java/org/apache/paimon/index/IndexFileMetaSerializer.java +++ b/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( diff --git a/paimon-core/src/main/java/org/apache/paimon/index/IndexFileMetaV5Deserializer.java b/paimon-core/src/main/java/org/apache/paimon/index/IndexFileMetaV5Deserializer.java new file mode 100644 index 000000000000..77ea40561493 --- /dev/null +++ b/paimon-core/src/main/java/org/apache/paimon/index/IndexFileMetaV5Deserializer.java @@ -0,0 +1,118 @@ +/* + * 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.index; + +import org.apache.paimon.data.InternalRow; +import org.apache.paimon.data.serializer.InternalRowSerializer; +import org.apache.paimon.data.serializer.InternalSerializers; +import org.apache.paimon.io.DataInputView; +import org.apache.paimon.types.ArrayType; +import org.apache.paimon.types.BigIntType; +import org.apache.paimon.types.DataField; +import org.apache.paimon.types.DataTypes; +import org.apache.paimon.types.IntType; +import org.apache.paimon.types.RowType; + +import java.io.IOException; +import java.io.Serializable; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; + +import static org.apache.paimon.index.IndexFileMetaSerializer.rowArrayDataToDvMetas; +import static org.apache.paimon.utils.SerializationUtils.newStringType; + +/** Deserializer for {@link IndexFileMeta} before the schema ID was added. */ +public class IndexFileMetaV5Deserializer implements Serializable { + + private static final long serialVersionUID = 1L; + + private static final RowType GLOBAL_INDEX_SCHEMA = + new RowType( + true, + Arrays.asList( + new DataField(0, "_ROW_RANGE_START", new BigIntType(false)), + new DataField(1, "_ROW_RANGE_END", new BigIntType(false)), + 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()))); + + public static final RowType SCHEMA = + new RowType( + false, + Arrays.asList( + new DataField(0, "_INDEX_TYPE", newStringType(false)), + new DataField(1, "_FILE_NAME", newStringType(false)), + new DataField(2, "_FILE_SIZE", new BigIntType(false)), + new DataField(3, "_ROW_COUNT", new BigIntType(false)), + new DataField( + 4, + "_DELETIONS_VECTORS_RANGES", + new ArrayType(true, DeletionVectorMeta.SCHEMA)), + new DataField(5, "_EXTERNAL_PATH", newStringType(true)), + new DataField(6, "_GLOBAL_INDEX", GLOBAL_INDEX_SCHEMA))); + + private final InternalRowSerializer rowSerializer; + + public IndexFileMetaV5Deserializer() { + this.rowSerializer = InternalSerializers.create(SCHEMA); + } + + private IndexFileMeta fromRow(InternalRow row) { + GlobalIndexMeta globalIndexMeta = null; + if (!row.isNullAt(6)) { + InternalRow globalIndexRow = row.getRow(6, GLOBAL_INDEX_SCHEMA.getFieldCount()); + long rowRangeStart = globalIndexRow.getLong(0); + long rowRangeEnd = globalIndexRow.getLong(1); + int indexFieldId = globalIndexRow.getInt(2); + int[] extraFields = + 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); + globalIndexMeta = + new GlobalIndexMeta( + rowRangeStart, + rowRangeEnd, + indexFieldId, + extraFields, + indexMeta, + sourceMeta); + } + + return new IndexFileMeta( + row.getString(0).toString(), + row.getString(1).toString(), + row.getLong(2), + row.getLong(3), + row.isNullAt(4) ? null : rowArrayDataToDvMetas(row.getArray(4)), + row.isNullAt(5) ? null : row.getString(5).toString(), + globalIndexMeta); + } + + public List deserializeList(DataInputView source) throws IOException { + int size = source.readInt(); + List records = new ArrayList<>(size); + for (int i = 0; i < size; i++) { + records.add(fromRow(rowSerializer.deserialize(source))); + } + return records; + } +} diff --git a/paimon-core/src/main/java/org/apache/paimon/manifest/IndexManifestEntry.java b/paimon-core/src/main/java/org/apache/paimon/manifest/IndexManifestEntry.java index f69716d455e6..70637beaed21 100644 --- a/paimon-core/src/main/java/org/apache/paimon/manifest/IndexManifestEntry.java +++ b/paimon-core/src/main/java/org/apache/paimon/manifest/IndexManifestEntry.java @@ -30,6 +30,8 @@ import org.apache.paimon.types.RowType; import org.apache.paimon.types.TinyIntType; +import javax.annotation.Nullable; + import java.util.Arrays; import java.util.Objects; @@ -55,6 +57,7 @@ public class IndexManifestEntry { public static final String DELETION_VECTORS_RANGES = "_DELETIONS_VECTORS_RANGES"; public static final String EXTERNAL_PATH = "_EXTERNAL_PATH"; public static final String GLOBAL_INDEX = "_GLOBAL_INDEX"; + public static final String SCHEMA_ID = "_SCHEMA_ID"; public static final RowType SCHEMA = new RowType( @@ -72,7 +75,8 @@ public class IndexManifestEntry { DELETION_VECTORS_RANGES, new ArrayType(true, DeletionVectorMeta.SCHEMA)), new DataField(8, EXTERNAL_PATH, newStringType(true)), - new DataField(9, GLOBAL_INDEX, GlobalIndexMeta.SCHEMA))); + new DataField(9, GLOBAL_INDEX, GlobalIndexMeta.SCHEMA), + new DataField(10, SCHEMA_ID, new BigIntType(true)))); public static final RowType MANIFEST_ROW_TYPE = ManifestSchemaUtils.withFormatIdentifier(SCHEMA); @@ -81,18 +85,29 @@ public class IndexManifestEntry { private final BinaryRow partition; private final int bucket; private final IndexFileMeta indexFile; + @Nullable private final Long schemaId; public IndexManifestEntry( FileKind kind, BinaryRow partition, int bucket, IndexFileMeta indexFile) { + this(kind, partition, bucket, indexFile, indexFile.schemaId()); + } + + public IndexManifestEntry( + FileKind kind, + BinaryRow partition, + int bucket, + IndexFileMeta indexFile, + @Nullable Long schemaId) { this.kind = kind; this.partition = partition; this.bucket = bucket; - this.indexFile = indexFile; + this.indexFile = indexFile.withSchemaId(schemaId); + this.schemaId = schemaId; } public IndexManifestEntry toDeleteEntry() { checkArgument(kind == FileKind.ADD); - return new IndexManifestEntry(FileKind.DELETE, partition, bucket, indexFile); + return new IndexManifestEntry(FileKind.DELETE, partition, bucket, indexFile, schemaId); } public FileKind kind() { @@ -111,6 +126,11 @@ public IndexFileMeta indexFile() { return indexFile; } + @Nullable + public Long schemaId() { + return schemaId; + } + @Override public boolean equals(Object o) { if (this == o) { @@ -123,12 +143,13 @@ public boolean equals(Object o) { return bucket == entry.bucket && kind == entry.kind && Objects.equals(partition, entry.partition) - && Objects.equals(indexFile, entry.indexFile); + && Objects.equals(indexFile, entry.indexFile) + && Objects.equals(schemaId, entry.schemaId); } @Override public int hashCode() { - return Objects.hash(kind, partition, bucket, indexFile); + return Objects.hash(kind, partition, bucket, indexFile, schemaId); } @Override @@ -142,6 +163,8 @@ public String toString() { + bucket + ", indexFile=" + indexFile + + ", schemaId=" + + schemaId + '}'; } } diff --git a/paimon-core/src/main/java/org/apache/paimon/manifest/IndexManifestEntrySerializer.java b/paimon-core/src/main/java/org/apache/paimon/manifest/IndexManifestEntrySerializer.java index c37bb77a0022..5afd8bae28ec 100644 --- a/paimon-core/src/main/java/org/apache/paimon/manifest/IndexManifestEntrySerializer.java +++ b/paimon-core/src/main/java/org/apache/paimon/manifest/IndexManifestEntrySerializer.java @@ -76,7 +76,8 @@ public InternalRow toRow(IndexManifestEntry record) { indexFile.rowCount(), dvMetasToRowArrayData(indexFile.dvRanges()), fromString(indexFile.externalPath()), - globalIndexRow); + globalIndexRow, + record.schemaId()); } @Override @@ -112,6 +113,7 @@ private IndexManifestEntry fromDataRow(InternalRow row) { sourceMeta); } + Long schemaId = row.getFieldCount() <= 10 || row.isNullAt(10) ? null : row.getLong(10); return new IndexManifestEntry( FileKind.fromByteValue(row.getByte(0)), deserializeBinaryRow(row.getBinary(1)), @@ -123,7 +125,8 @@ private IndexManifestEntry fromDataRow(InternalRow row) { row.getLong(6), row.isNullAt(7) ? null : rowArrayDataToDvMetas(row.getArray(7)), row.isNullAt(8) ? null : row.getString(8).toString(), - globalIndexMeta)); + globalIndexMeta), + schemaId); } public static Function partitionGetter() { diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/commit/ManifestEntryChanges.java b/paimon-core/src/main/java/org/apache/paimon/operation/commit/ManifestEntryChanges.java index faa65bba4750..04b34ea5acf1 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/commit/ManifestEntryChanges.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/commit/ManifestEntryChanges.java @@ -80,7 +80,8 @@ public void collect(CommitMessage message) { FileKind.DELETE, commitMessage.partition(), commitMessage.bucket(), - m))); + m, + m.schemaId()))); commitMessage .newFilesIncrement() .newIndexFiles() @@ -91,7 +92,8 @@ public void collect(CommitMessage message) { FileKind.ADD, commitMessage.partition(), commitMessage.bucket(), - m))); + m, + m.schemaId()))); commitMessage .compactIncrement() @@ -115,7 +117,8 @@ public void collect(CommitMessage message) { FileKind.DELETE, commitMessage.partition(), commitMessage.bucket(), - m))); + m, + m.schemaId()))); commitMessage .compactIncrement() .newIndexFiles() @@ -126,7 +129,8 @@ public void collect(CommitMessage message) { FileKind.ADD, commitMessage.partition(), commitMessage.bucket(), - m))); + m, + m.schemaId()))); } private ManifestEntry makeEntry(FileKind kind, CommitMessage commitMessage, DataFileMeta file) { diff --git a/paimon-core/src/main/java/org/apache/paimon/table/sink/CommitMessageSerializer.java b/paimon-core/src/main/java/org/apache/paimon/table/sink/CommitMessageSerializer.java index 8222b07c8c8c..0382e3c96368 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/sink/CommitMessageSerializer.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/sink/CommitMessageSerializer.java @@ -26,6 +26,7 @@ import org.apache.paimon.index.IndexFileMetaV2Deserializer; import org.apache.paimon.index.IndexFileMetaV3Deserializer; import org.apache.paimon.index.IndexFileMetaV4Deserializer; +import org.apache.paimon.index.IndexFileMetaV5Deserializer; import org.apache.paimon.io.CompactIncrement; import org.apache.paimon.io.DataFileMeta; import org.apache.paimon.io.DataFileMeta08Serializer; @@ -53,7 +54,7 @@ /** {@link VersionedSerializer} for {@link CommitMessage}. */ public class CommitMessageSerializer implements VersionedSerializer { - public static final int CURRENT_VERSION = 13; + public static final int CURRENT_VERSION = 14; private final DataFileMetaSerializer dataFileSerializer; private final IndexFileMetaSerializer indexEntrySerializer; @@ -68,6 +69,7 @@ public class CommitMessageSerializer implements VersionedSerializer> fileDeserializer( private IOExceptionSupplier> indexEntryDeserializer( int version, DataInputView view) { - if (version >= 12) { + if (version >= 14) { return () -> indexEntrySerializer.deserializeList(view); + } else if (version >= 12) { + if (indexEntryV5Deserializer == null) { + indexEntryV5Deserializer = new IndexFileMetaV5Deserializer(); + } + return () -> indexEntryV5Deserializer.deserializeList(view); } else if (version == 11) { if (indexEntryV4Deserializer == null) { indexEntryV4Deserializer = new IndexFileMetaV4Deserializer(); diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/BucketVectorSearchSplit.java b/paimon-core/src/main/java/org/apache/paimon/table/source/BucketVectorSearchSplit.java index 75ca1f58a137..dfbcd773f6f7 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/BucketVectorSearchSplit.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/BucketVectorSearchSplit.java @@ -45,7 +45,7 @@ public class BucketVectorSearchSplit extends VectorSearchSplit { private static final long serialVersionUID = 1L; private static final long MAGIC = 0x504B5653504C4954L; - private static final int VERSION = 1; + private static final int VERSION = 2; private DataSplit dataSplit; private List payloadFiles; diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionFullTextRead.java b/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionFullTextRead.java index 40e492968ce2..308053ffab0b 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionFullTextRead.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionFullTextRead.java @@ -101,6 +101,7 @@ private GlobalIndexResult read( Map> splitsByColumn = new HashMap<>(); List rawRowRanges = new ArrayList<>(); + @Nullable String rawIndexType = null; for (FullTextSearchSplit split : splits) { if (split instanceof IndexFullTextSearchSplit) { IndexFullTextSearchSplit indexSplit = (IndexFullTextSearchSplit) split; @@ -108,7 +109,11 @@ private GlobalIndexResult read( .computeIfAbsent(indexSplit.columnName(), k -> new ArrayList<>()) .add(indexSplit); } else if (split instanceof RawFullTextSearchSplit) { - rawRowRanges.addAll(((RawFullTextSearchSplit) split).rowRanges()); + RawFullTextSearchSplit rawSplit = (RawFullTextSearchSplit) split; + rawRowRanges.addAll(rawSplit.rowRanges()); + if (rawIndexType == null) { + rawIndexType = rawSplit.indexType(); + } } } @@ -125,6 +130,7 @@ private GlobalIndexResult read( partitionFilter, limit, textColumn, + rawIndexType, this::evalQuery) .withRawSearch(result, rawRowRanges, splitsByColumn, executor); } diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionFullTextScan.java b/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionFullTextScan.java index fbd1bd83d133..235854de3b9b 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionFullTextScan.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionFullTextScan.java @@ -21,6 +21,7 @@ import org.apache.paimon.Snapshot; import org.apache.paimon.annotation.VisibleForTesting; import org.apache.paimon.globalindex.DataEvolutionGlobalIndexCoverage; +import org.apache.paimon.globalindex.GlobalIndexSchemaCompatibility; import org.apache.paimon.globalindex.GlobalIndexerFactory; import org.apache.paimon.globalindex.GlobalIndexerFactoryUtils; import org.apache.paimon.index.GlobalIndexMeta; @@ -115,14 +116,23 @@ public Plan scan() { && supportsFullTextSearch(entry.indexFile().indexType()); }; - List allIndexFiles = - indexFileHandler.scan(snapshot, indexFileFilter).stream() + List discoveredEntries = + indexFileHandler.scan(snapshot, indexFileFilter); + List discoveredIndexFiles = + discoveredEntries.stream() .map(IndexManifestEntry::indexFile) .collect(Collectors.toList()); + List discoveredSelections = + chooseIndexRanges(discoveredIndexFiles, textColumnIds, idToColumn); + List compatibleIndexFiles = + GlobalIndexSchemaCompatibility.filterCompatible(table, discoveredEntries).stream() + .map(IndexManifestEntry::indexFile) + .collect(Collectors.toList()); + List compatibleSelections = + chooseIndexRanges(compatibleIndexFiles, textColumnIds, idToColumn); List splits = new ArrayList<>(); - for (IndexRangeSelection selection : - chooseIndexRanges(allIndexFiles, textColumnIds, idToColumn)) { + for (IndexRangeSelection selection : compatibleSelections) { splits.add( new IndexFullTextSearchSplit( selection.columnName, @@ -132,18 +142,22 @@ public Plan scan() { selection.searchRanges)); } - if (!allIndexFiles.isEmpty()) { - List rawRowRanges = - new DataEvolutionGlobalIndexCoverage( - table, - snapshot, - partitionFilter, - allIndexFiles, - table.coreOptions().fullTextIndexSearchMode()) - .unindexedRanges(textColumnIds); - if (!rawRowRanges.isEmpty()) { - splits.add(new RawFullTextSearchSplit(rawRowRanges)); - } + List rawRowRanges = + new DataEvolutionGlobalIndexCoverage( + table, + snapshot, + partitionFilter, + compatibleIndexFiles, + table.coreOptions().fullTextIndexSearchMode()) + .unindexedRanges(textColumnIds); + @Nullable + String rawIndexType = + firstIndexType( + compatibleSelections.isEmpty() + ? discoveredSelections + : compatibleSelections); + if (!rawRowRanges.isEmpty() && rawIndexType != null) { + splits.add(new RawFullTextSearchSplit(rawRowRanges, rawIndexType)); } @Nullable Snapshot planSnapshot = snapshot; @@ -161,6 +175,14 @@ public Snapshot snapshot() { }; } + @Nullable + private static String firstIndexType(List selections) { + if (selections.isEmpty() || selections.get(0).files.isEmpty()) { + return null; + } + return selections.get(0).files.get(0).indexType(); + } + /** * Returns the searched text-column ids served by {@code meta}: its primary {@code indexFieldId} * plus any {@code extraFieldIds} present in {@code textColumnIds}. This lets a multi-column diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionVectorScan.java b/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionVectorScan.java index fdb57385abf3..a2c96a9e7a7f 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionVectorScan.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionVectorScan.java @@ -22,6 +22,7 @@ import org.apache.paimon.CoreOptions.GlobalIndexSearchMode; import org.apache.paimon.Snapshot; import org.apache.paimon.globalindex.DataEvolutionGlobalIndexCoverage; +import org.apache.paimon.globalindex.GlobalIndexSchemaCompatibility; import org.apache.paimon.index.GlobalIndexMeta; import org.apache.paimon.index.IndexFileHandler; import org.apache.paimon.index.IndexFileMeta; @@ -114,8 +115,9 @@ public Plan scan() { return false; }; + List indexEntries = indexFileHandler.scan(snapshot, indexFileFilter); List allIndexFiles = - indexFileHandler.scan(snapshot, indexFileFilter).stream() + GlobalIndexSchemaCompatibility.filterCompatible(table, indexEntries).stream() .map(IndexManifestEntry::indexFile) .collect(Collectors.toList()); String vectorIndexType = vectorIndexType(allIndexFiles); diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/IndexFullTextSearchSplit.java b/paimon-core/src/main/java/org/apache/paimon/table/source/IndexFullTextSearchSplit.java index 54a55536f179..e487494838ec 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/IndexFullTextSearchSplit.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/IndexFullTextSearchSplit.java @@ -37,7 +37,7 @@ public class IndexFullTextSearchSplit extends FullTextSearchSplit { private static final long serialVersionUID = 1L; - private static final int VERSION = 1; + private static final int VERSION = 2; private static final ThreadLocal INDEX_SERIALIZER = ThreadLocal.withInitial(IndexFileMetaSerializer::new); diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/IndexVectorSearchSplit.java b/paimon-core/src/main/java/org/apache/paimon/table/source/IndexVectorSearchSplit.java index 355b60d8b75e..d5844aecef1a 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/IndexVectorSearchSplit.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/IndexVectorSearchSplit.java @@ -34,7 +34,7 @@ public class IndexVectorSearchSplit extends VectorSearchSplit { private static final long serialVersionUID = 1L; - private static final int VERSION = 1; + private static final int VERSION = 2; private static final ThreadLocal INDEX_SERIALIZER = ThreadLocal.withInitial(IndexFileMetaSerializer::new); diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeyFullTextSearchSplit.java b/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeyFullTextSearchSplit.java index 8a5bd09963db..b1286031fcee 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeyFullTextSearchSplit.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeyFullTextSearchSplit.java @@ -42,7 +42,7 @@ public class PrimaryKeyFullTextSearchSplit extends FullTextSearchSplit { private static final long serialVersionUID = 1L; - private static final int VERSION = 1; + private static final int VERSION = 2; private DataSplit dataSplit; private transient List payloadFiles; diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/RawFullTextReadImpl.java b/paimon-core/src/main/java/org/apache/paimon/table/source/RawFullTextReadImpl.java index 86c6d2a1d2fd..2a680d4c0c4d 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/RawFullTextReadImpl.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/RawFullTextReadImpl.java @@ -70,6 +70,7 @@ class RawFullTextReadImpl { @Nullable private final PartitionPredicate partitionFilter; private final int limit; private final DataField textColumn; + @Nullable private final String rawIndexType; private final IndexSearch indexSearch; RawFullTextReadImpl( @@ -78,12 +79,14 @@ class RawFullTextReadImpl { @Nullable PartitionPredicate partitionFilter, int limit, DataField textColumn, + @Nullable String rawIndexType, IndexSearch indexSearch) { this.table = table; this.planSnapshot = planSnapshot; this.partitionFilter = partitionFilter; this.limit = limit; this.textColumn = textColumn; + this.rawIndexType = rawIndexType; this.indexSearch = indexSearch; } @@ -178,11 +181,13 @@ private Map createRawFullTextIndexes( Map rawIndexes = new HashMap<>(); long rowRangeStart = rawRowRanges.get(0).from; long rowRangeEnd = rawRowRanges.get(rawRowRanges.size() - 1).to; - String fallbackIndexType = firstIndexType(splitsByColumn); String column = textColumn.name(); String indexType = indexType(column, splitsByColumn); if (indexType == null) { - indexType = checkNotNull(fallbackIndexType); + indexType = rawIndexType; + } + if (indexType == null) { + indexType = checkNotNull(firstIndexType(splitsByColumn)); } GlobalIndexer globalIndexer = GlobalIndexerFactoryUtils.load(indexType).create(textColumn, rawSearchOptions()); diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/RawFullTextSearchSplit.java b/paimon-core/src/main/java/org/apache/paimon/table/source/RawFullTextSearchSplit.java index a95ea76255e8..0e0416ef0dc2 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/RawFullTextSearchSplit.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/RawFullTextSearchSplit.java @@ -20,6 +20,8 @@ import org.apache.paimon.utils.Range; +import javax.annotation.Nullable; + import java.util.ArrayList; import java.util.Collections; import java.util.List; @@ -31,31 +33,49 @@ public class RawFullTextSearchSplit extends FullTextSearchSplit { private static final long serialVersionUID = 1L; private final List rowRanges; + @Nullable private final String indexType; public RawFullTextSearchSplit(List rowRanges) { + this(rowRanges, null); + } + + public RawFullTextSearchSplit(List rowRanges, @Nullable String indexType) { this.rowRanges = Collections.unmodifiableList(new ArrayList<>(rowRanges)); + this.indexType = indexType; } public List rowRanges() { return rowRanges; } + @Nullable + public String indexType() { + return indexType; + } + @Override public boolean equals(Object o) { if (o == null || getClass() != o.getClass()) { return false; } RawFullTextSearchSplit that = (RawFullTextSearchSplit) o; - return Objects.equals(rowRanges, that.rowRanges); + return Objects.equals(rowRanges, that.rowRanges) + && Objects.equals(indexType, that.indexType); } @Override public int hashCode() { - return Objects.hash(rowRanges); + return Objects.hash(rowRanges, indexType); } @Override public String toString() { - return "RawFullTextSearchSplit{" + "rowRanges=" + rowRanges + '}'; + return "RawFullTextSearchSplit{" + + "rowRanges=" + + rowRanges + + ", indexType='" + + indexType + + '\'' + + '}'; } } diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/RawVectorSearchSplit.java b/paimon-core/src/main/java/org/apache/paimon/table/source/RawVectorSearchSplit.java index 36221e8ae8d7..f1459d08e839 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/RawVectorSearchSplit.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/RawVectorSearchSplit.java @@ -39,7 +39,7 @@ public class RawVectorSearchSplit extends VectorSearchSplit { private static final long serialVersionUID = 1L; - private static final int VERSION = 1; + private static final int VERSION = 2; private static final ThreadLocal INDEX_SERIALIZER = ThreadLocal.withInitial(IndexFileMetaSerializer::new); diff --git a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdReassignerTest.java b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdReassignerTest.java index 54e89507c974..20e119b06716 100644 --- a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdReassignerTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdReassignerTest.java @@ -1740,6 +1740,7 @@ public void testSkipUnpartitionedTable() throws Exception { public void testReassignGlobalIndexRowRanges() throws Exception { FileStoreTable table = createTableWithInterleavedPartitions(); createBTreeIndex(table); + long buildSchemaId = table.schema().id(); assertThat(table.snapshotManager().latestSnapshot().nextRowId()).isEqualTo(5L); @@ -1758,6 +1759,12 @@ public void testReassignGlobalIndexRowRanges() throws Exception { new Range(7, 7), new Range(8, 8), new Range(9, 9)); + assertThat(table.store().newIndexFileHandler().scanEntries()) + .allSatisfy( + entry -> { + assertThat(entry.schemaId()).isEqualTo(buildSchemaId); + assertThat(entry.indexFile().schemaId()).isEqualTo(buildSchemaId); + }); Predicate predicate = new PredicateBuilder(table.rowType()).equal(table.rowType().getFieldIndex("id"), 4); diff --git a/paimon-core/src/test/java/org/apache/paimon/globalindex/GlobalIndexBuilderUtilsTest.java b/paimon-core/src/test/java/org/apache/paimon/globalindex/GlobalIndexBuilderUtilsTest.java index 063114a99611..6c9102838b06 100644 --- a/paimon-core/src/test/java/org/apache/paimon/globalindex/GlobalIndexBuilderUtilsTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/globalindex/GlobalIndexBuilderUtilsTest.java @@ -108,7 +108,8 @@ void testToIndexFileMetasMultiColumn() throws IOException { fields, "test-type", entries, - null); + null, + 11L); assertThat(metas).hasSize(1); assertThat(metas.get(0).globalIndexMeta().indexFieldId()).isEqualTo(1); @@ -136,7 +137,8 @@ void testToIndexFileMetasSingleColumn() throws IOException { fields, "test-type", entries, - null); + null, + 11L); assertThat(metas).hasSize(1); assertThat(metas.get(0).globalIndexMeta().indexFieldId()).isEqualTo(1); @@ -157,9 +159,11 @@ void testToIndexFileMetasWithSourceMeta() throws IOException { Collections.singletonList(field), "lumina", createDummyResultEntries(), - sourceMeta); + sourceMeta, + 11L); assertThat(metas.get(0).globalIndexMeta().sourceMeta()).containsExactly(sourceMeta); + assertThat(metas.get(0).schemaId()).isEqualTo(11L); } // Test: 3 columns (title + vec + id), primary column title is indexFieldId, rest in @@ -183,7 +187,8 @@ void testToIndexFileMetasThreeColumns() throws IOException { fields, "test-type", entries, - null); + null, + 11L); assertThat(metas).hasSize(1); assertThat(metas.get(0).globalIndexMeta().indexFieldId()).isEqualTo(1); diff --git a/paimon-core/src/test/java/org/apache/paimon/globalindex/GlobalIndexSchemaCompatibilityTest.java b/paimon-core/src/test/java/org/apache/paimon/globalindex/GlobalIndexSchemaCompatibilityTest.java new file mode 100644 index 000000000000..de3059651b3e --- /dev/null +++ b/paimon-core/src/test/java/org/apache/paimon/globalindex/GlobalIndexSchemaCompatibilityTest.java @@ -0,0 +1,108 @@ +/* + * 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.CoreOptions; +import org.apache.paimon.data.BinaryRow; +import org.apache.paimon.index.GlobalIndexMeta; +import org.apache.paimon.index.IndexFileMeta; +import org.apache.paimon.manifest.FileKind; +import org.apache.paimon.manifest.IndexManifestEntry; +import org.apache.paimon.schema.Schema; +import org.apache.paimon.schema.SchemaChange; +import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.TableTestBase; +import org.apache.paimon.types.DataTypes; + +import org.junit.jupiter.api.Test; + +import java.util.Arrays; +import java.util.Collections; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Tests for {@link GlobalIndexSchemaCompatibility}. */ +public class GlobalIndexSchemaCompatibilityTest extends TableTestBase { + + @Override + protected Schema schemaDefault() { + return Schema.newBuilder() + .column("indexed", DataTypes.INT()) + .option(CoreOptions.BUCKET.key(), "-1") + .option(CoreOptions.ROW_TRACKING_ENABLED.key(), "true") + .option(CoreOptions.DATA_EVOLUTION_ENABLED.key(), "true") + .build(); + } + + @Test + public void testCompatibilityUsesIndexedFieldTypes() throws Exception { + createTableDefault(); + FileStoreTable table = getTableDefault(); + long buildSchemaId = table.schema().id(); + IndexManifestEntry entry = globalIndexEntry(buildSchemaId); + + catalog.alterTable( + identifier(), + Collections.singletonList(SchemaChange.addColumn("unrelated", DataTypes.STRING())), + false); + table = getTableDefault(); + assertThat( + GlobalIndexSchemaCompatibility.filterCompatible( + table, Collections.singleton(entry))) + .containsExactly(entry); + + catalog.alterTable( + identifier(), + Collections.singletonList( + SchemaChange.updateColumnType("indexed", DataTypes.BIGINT())), + false); + table = getTableDefault(); + assertThat( + GlobalIndexSchemaCompatibility.filterCompatible( + table, Collections.singleton(entry))) + .isEmpty(); + } + + @Test + public void testMissingSchemaIdentityFailsClosed() throws Exception { + createTableDefault(); + FileStoreTable table = getTableDefault(); + + assertThat( + GlobalIndexSchemaCompatibility.filterCompatible( + table, + Arrays.asList( + globalIndexEntry(null), globalIndexEntry(Long.MAX_VALUE)))) + .isEmpty(); + } + + private static IndexManifestEntry globalIndexEntry(Long schemaId) { + IndexFileMeta indexFile = + new IndexFileMeta( + "btree", + "index-file", + 1L, + 1L, + null, + null, + new GlobalIndexMeta(0L, 0L, 0, null, null), + schemaId); + return new IndexManifestEntry(FileKind.ADD, BinaryRow.EMPTY_ROW, 0, indexFile, schemaId); + } +} diff --git a/paimon-core/src/test/java/org/apache/paimon/index/IndexFileMetaSerializerTest.java b/paimon-core/src/test/java/org/apache/paimon/index/IndexFileMetaSerializerTest.java index 33373741c667..22133f6ecee9 100644 --- a/paimon-core/src/test/java/org/apache/paimon/index/IndexFileMetaSerializerTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/index/IndexFileMetaSerializerTest.java @@ -41,14 +41,17 @@ void testGlobalIndexSourceMetaRoundTrip() { "index-file", 100, 10, + null, + null, new GlobalIndexMeta(0, 9, 7, null, new byte[] {3, 4}, new byte[] {1, 2}), - null); + 11L); - GlobalIndexMeta restored = - serializer.fromRow(serializer.toRow(indexFile)).globalIndexMeta(); + IndexFileMeta restoredIndexFile = serializer.fromRow(serializer.toRow(indexFile)); + GlobalIndexMeta restored = restoredIndexFile.globalIndexMeta(); assertThat(restored.sourceMeta()).containsExactly(1, 2); assertThat(restored.indexMeta()).containsExactly(3, 4); + assertThat(restoredIndexFile.schemaId()).isEqualTo(11L); } @Test diff --git a/paimon-core/src/test/java/org/apache/paimon/manifest/IndexManifestEntrySerializerTest.java b/paimon-core/src/test/java/org/apache/paimon/manifest/IndexManifestEntrySerializerTest.java index 945f54fdf465..9d1b3c6ea03e 100644 --- a/paimon-core/src/test/java/org/apache/paimon/manifest/IndexManifestEntrySerializerTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/manifest/IndexManifestEntrySerializerTest.java @@ -19,6 +19,7 @@ package org.apache.paimon.manifest; import org.apache.paimon.data.BinaryRow; +import org.apache.paimon.data.GenericRow; import org.apache.paimon.data.InternalRow; import org.apache.paimon.index.GlobalIndexMeta; import org.apache.paimon.index.IndexFileMeta; @@ -30,8 +31,10 @@ import java.io.IOException; import java.util.Random; +import static org.apache.paimon.data.BinaryString.fromString; import static org.apache.paimon.index.IndexFileMetaSerializerTest.randomIndexFile; import static org.apache.paimon.io.DataFileTestUtils.row; +import static org.apache.paimon.utils.SerializationUtils.serializeBinaryRow; import static org.assertj.core.api.Assertions.assertThat; /** Test for {@link IndexManifestEntrySerializer}. */ @@ -78,17 +81,43 @@ void testGlobalIndexSourceMetaRoundTrip() throws IOException { 10, new GlobalIndexMeta( 0, 9, 7, null, new byte[] {3, 4}, new byte[] {1, 2}), - null)); + null), + 11L); assertThat(serializer.toRow(entry).getInt(0)).isEqualTo(1); - GlobalIndexMeta restored = - serializer - .deserializeFromBytes(serializer.serializeToBytes(entry)) - .indexFile() - .globalIndexMeta(); + IndexManifestEntry restoredEntry = + serializer.deserializeFromBytes(serializer.serializeToBytes(entry)); + GlobalIndexMeta restored = restoredEntry.indexFile().globalIndexMeta(); assertThat(restored.indexMeta()).containsExactly(3, 4); assertThat(restored.sourceMeta()).containsExactly(1, 2); + assertThat(restoredEntry.schemaId()).isEqualTo(11L); + assertThat(restoredEntry.indexFile().schemaId()).isEqualTo(11L); + assertThat(restoredEntry.toDeleteEntry().schemaId()).isEqualTo(11L); + } + + @Test + void testReadsLegacyEntryWithoutSchemaId() { + IndexManifestEntrySerializer serializer = new IndexManifestEntrySerializer(); + InternalRow globalIndex = GenericRow.of(0L, 9L, 7, null, null, null); + InternalRow legacyRow = + GenericRow.of( + 1, + FileKind.ADD.toByteValue(), + serializeBinaryRow(BinaryRow.EMPTY_ROW), + 0, + fromString("btree"), + fromString("index-file"), + 100L, + 10L, + null, + null, + globalIndex); + + IndexManifestEntry restored = serializer.fromRow(legacyRow); + + assertThat(restored.schemaId()).isNull(); + assertThat(restored.indexFile().schemaId()).isNull(); } @Override diff --git a/paimon-core/src/test/java/org/apache/paimon/manifest/IndexManifestFileHandlerTest.java b/paimon-core/src/test/java/org/apache/paimon/manifest/IndexManifestFileHandlerTest.java index 3c0002c2adeb..7b463abfb555 100644 --- a/paimon-core/src/test/java/org/apache/paimon/manifest/IndexManifestFileHandlerTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/manifest/IndexManifestFileHandlerTest.java @@ -143,11 +143,16 @@ public void testNewIndexManifestReadableWithLegacySchema() throws Exception { IndexManifestFileHandler handler = new IndexManifestFileHandler(indexManifestFile, BucketMode.HASH_FIXED); - String manifestFile = handler.write(null, Arrays.asList(pkVectorEntry("btree", "index"))); + IndexManifestEntry entry = pkVectorEntry("btree", "index"); + entry = + new IndexManifestEntry( + entry.kind(), entry.partition(), entry.bucket(), entry.indexFile(), 11L); + String manifestFile = handler.write(null, Arrays.asList(entry)); RowType legacyGlobalIndexSchema = GlobalIndexMeta.SCHEMA.copy(GlobalIndexMeta.SCHEMA.getFields().subList(0, 5)); - List legacyEntryFields = new ArrayList<>(IndexManifestEntry.SCHEMA.getFields()); + List legacyEntryFields = + new ArrayList<>(IndexManifestEntry.SCHEMA.getFields().subList(0, 10)); legacyEntryFields.set(9, legacyEntryFields.get(9).newType(legacyGlobalIndexSchema)); RowType legacySchema = ManifestSchemaUtils.withFormatIdentifier(new RowType(false, legacyEntryFields)); diff --git a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestCommittableSerializerCompatibilityTest.java b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestCommittableSerializerCompatibilityTest.java index dbe1ebfab09e..94568a7602a2 100644 --- a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestCommittableSerializerCompatibilityTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestCommittableSerializerCompatibilityTest.java @@ -50,7 +50,7 @@ public class ManifestCommittableSerializerCompatibilityTest { "generateManifestCommittableGoldenFiles"; @Test - public void testCompatibilityToV5CommitV13() throws IOException { + public void testCompatibilityToV5CommitV14() throws IOException { DataFileMeta dataFile = DataFileMeta.create( "column-sequence-file", @@ -78,7 +78,14 @@ public void testCompatibilityToV5CommitV13() throws IOException { .withColumnMaxSequenceNumbers(new long[] {3L, 5L}); IndexFileMeta indexFile = new IndexFileMeta( - "index-type", "index-file", 100L, 10L, (GlobalIndexMeta) null, null); + "index-type", + "index-file", + 100L, + 10L, + null, + null, + (GlobalIndexMeta) null, + 11L); ManifestCommittable committable = createManifestCommittable( Collections.singletonList(dataFile), indexFile, indexFile); @@ -88,7 +95,7 @@ public void testCompatibilityToV5CommitV13() throws IOException { byte[] serialized; if (Boolean.parseBoolean( System.getProperties().getProperty(GENERATE_GOLDEN_FILES_PROPERTY))) { - CompatibilityUtils.writeCompatibilityFile("manifest-committable-v13-v5", current); + CompatibilityUtils.writeCompatibilityFile("manifest-committable-v14-v5", current); serialized = current; } else { serialized = @@ -96,7 +103,7 @@ public void testCompatibilityToV5CommitV13() throws IOException { ManifestCommittableSerializerCompatibilityTest.class .getClassLoader() .getResourceAsStream( - "compatibility/manifest-committable-v13-v5"), + "compatibility/manifest-committable-v14-v5"), true); } @@ -104,6 +111,54 @@ public void testCompatibilityToV5CommitV13() throws IOException { assertThat(serializer.deserialize(5, serialized)).isEqualTo(committable); } + @Test + public void testCompatibilityToV5CommitV13() throws IOException { + byte[] serialized = + IOUtils.readFully( + ManifestCommittableSerializerCompatibilityTest.class + .getClassLoader() + .getResourceAsStream("compatibility/manifest-committable-v13-v5"), + true); + + ManifestCommittable restored = + new ManifestCommittableSerializer().deserialize(5, serialized); + IndexFileMeta restoredIndexFile = + ((CommitMessageImpl) restored.fileCommittables().get(0)) + .newFilesIncrement() + .newIndexFiles() + .get(0); + assertThat(restoredIndexFile.schemaId()).isNull(); + } + + @Test + public void testCompatibilityToV5CommitV13WithGlobalIndex() throws IOException { + byte[] serialized = + IOUtils.readFully( + ManifestCommittableSerializerCompatibilityTest.class + .getClassLoader() + .getResourceAsStream( + "compatibility/manifest-committable-v13-global-index-v5"), + true); + + ManifestCommittable restored = + new ManifestCommittableSerializer().deserialize(5, serialized); + IndexFileMeta restoredIndexFile = + ((CommitMessageImpl) restored.fileCommittables().get(0)) + .newFilesIncrement() + .newIndexFiles() + .get(0); + assertThat(restoredIndexFile.schemaId()).isNull(); + assertThat(restoredIndexFile.globalIndexMeta()) + .isEqualTo( + new GlobalIndexMeta( + 0L, + 9L, + 7, + new int[] {8, 9}, + new byte[] {0x12, 0x34}, + new byte[] {0x56, 0x78})); + } + @Test public void testCompatibilityToV5CommitV11() throws IOException { String fileName = "manifest-committable-v11-v5"; diff --git a/paimon-core/src/test/java/org/apache/paimon/operation/commit/ManifestEntryChangesTest.java b/paimon-core/src/test/java/org/apache/paimon/operation/commit/ManifestEntryChangesTest.java new file mode 100644 index 000000000000..74d8939594aa --- /dev/null +++ b/paimon-core/src/test/java/org/apache/paimon/operation/commit/ManifestEntryChangesTest.java @@ -0,0 +1,68 @@ +/* + * 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.operation.commit; + +import org.apache.paimon.data.BinaryRow; +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.sink.CommitMessageImpl; + +import org.junit.jupiter.api.Test; + +import java.util.Collections; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Tests for {@link ManifestEntryChanges}. */ +public class ManifestEntryChangesTest { + + @Test + public void testCollectPreservesIndexSchemaId() { + IndexFileMeta indexFile = + new IndexFileMeta( + "btree", + "index-file", + 1L, + 1L, + null, + null, + new GlobalIndexMeta(0L, 0L, 0, null, null), + 11L); + CommitMessageImpl message = + new CommitMessageImpl( + BinaryRow.EMPTY_ROW, + 0, + null, + DataIncrement.indexIncrement(Collections.singletonList(indexFile)), + CompactIncrement.emptyIncrement()); + + ManifestEntryChanges changes = new ManifestEntryChanges(1); + changes.collect(message); + + assertThat(changes.appendIndexFiles) + .singleElement() + .satisfies( + entry -> { + assertThat(entry.schemaId()).isEqualTo(11L); + assertThat(entry.indexFile().schemaId()).isEqualTo(11L); + }); + } +} diff --git a/paimon-core/src/test/java/org/apache/paimon/table/BitmapGlobalIndexTableTest.java b/paimon-core/src/test/java/org/apache/paimon/table/BitmapGlobalIndexTableTest.java index fb8f980dccd8..0edb25ec6c9b 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/BitmapGlobalIndexTableTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/BitmapGlobalIndexTableTest.java @@ -260,7 +260,8 @@ private CommitMessage buildIndex( rowRange, indexField.id(), INDEX_TYPE, - resultEntries); + resultEntries, + table.schema().id()); DataIncrement dataIncrement = DataIncrement.indexIncrement(indexFileMetas); return new CommitMessageImpl( partition(split), 0, null, dataIncrement, CompactIncrement.emptyIncrement()); diff --git a/paimon-core/src/test/java/org/apache/paimon/table/MultiValueGlobalIndexTableTest.java b/paimon-core/src/test/java/org/apache/paimon/table/MultiValueGlobalIndexTableTest.java index 90d440a98c2f..9817b2ec70ce 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/MultiValueGlobalIndexTableTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/MultiValueGlobalIndexTableTest.java @@ -21,6 +21,7 @@ import org.apache.paimon.CoreOptions; import org.apache.paimon.data.GenericArray; import org.apache.paimon.data.GenericRow; +import org.apache.paimon.globalindex.DataEvolutionGlobalIndexScanner; import org.apache.paimon.globalindex.IndexedSplit; import org.apache.paimon.globalindex.ScanResult; import org.apache.paimon.globalindex.sorted.SortedGlobalIndexScanner; @@ -29,7 +30,9 @@ import org.apache.paimon.io.DataIncrement; import org.apache.paimon.predicate.Predicate; import org.apache.paimon.predicate.PredicateBuilder; +import org.apache.paimon.schema.NestedSchemaUtils; import org.apache.paimon.schema.Schema; +import org.apache.paimon.schema.SchemaChange; import org.apache.paimon.table.sink.BatchTableCommit; import org.apache.paimon.table.sink.CommitMessage; import org.apache.paimon.table.sink.CommitMessageImpl; @@ -37,6 +40,7 @@ import org.apache.paimon.table.source.ReadBuilder; import org.apache.paimon.table.source.TableScan; import org.apache.paimon.types.DataTypes; +import org.apache.paimon.utils.Range; import org.junit.jupiter.api.Test; @@ -106,6 +110,48 @@ public void testCoreScanUsesMultiValueIndexAndPreservesCoverage() throws Excepti assertThat(readIds(fullSearchTable, containsRed)).containsExactlyInAnyOrder(1, 5, 6); } + @Test + public void testIndexCompatibilityAcrossSchemaEvolution() throws Exception { + createTableDefault(); + FileStoreTable table = getTableDefault(); + write(table, GenericRow.of(1, array(-1))); + long firstBuildSchemaId = table.schema().id(); + buildIndex(table); + + catalog.alterTable(identifier(), SchemaChange.addColumn("note", DataTypes.STRING()), false); + table = (FileStoreTable) catalog.getTable(identifier()); + FileStoreTable fullSearchTable = fullSearchTable(table); + Predicate sameTypePredicate = + new PredicateBuilder(fullSearchTable.rowType()).arrayContains(1, -1); + assertThat(readIds(fullSearchTable, sameTypePredicate)).containsExactly(1); + + List schemaChanges = new ArrayList<>(); + NestedSchemaUtils.generateNestedColumnUpdates( + Collections.singletonList("tags"), + table.rowType().getTypeAt(1), + DataTypes.ARRAY(DataTypes.BIGINT()), + schemaChanges); + table.schemaManager().commitChanges(schemaChanges); + table = table.copyWithLatestSchema(); + write(table, GenericRow.of(2, array(-1L), null)); + buildIndex(table); + + fullSearchTable = fullSearchTable(table.copyWithLatestSchema()); + Predicate evolvedTypePredicate = + new PredicateBuilder(fullSearchTable.rowType()).arrayContains(1, -1L); + try (DataEvolutionGlobalIndexScanner scanner = + DataEvolutionGlobalIndexScanner.create(fullSearchTable, null, evolvedTypePredicate) + .get()) { + assertThat(scanner.scan(evolvedTypePredicate).get().results().toRangeList()) + .containsExactly(new Range(1, 1)); + assertThat(scanner.unindexedRows(evolvedTypePredicate).results().toRangeList()) + .containsExactly(new Range(0, 0)); + } + assertThat(readIdsWithoutSplitAssertion(fullSearchTable, evolvedTypePredicate)) + .containsExactly(1, 2); + assertThat(firstBuildSchemaId).isNotEqualTo(fullSearchTable.schema().id()); + } + private void buildIndex(FileStoreTable table) throws Exception { SortedGlobalIndexScanner scanner = new SortedGlobalIndexScanner(table, "multivalue").withIndexField("tags"); @@ -168,6 +214,24 @@ private List readIds( return ids; } + private List readIdsWithoutSplitAssertion(FileStoreTable table, Predicate predicate) + throws Exception { + ReadBuilder readBuilder = table.newReadBuilder().withFilter(predicate); + TableScan.Plan plan = readBuilder.newScan().plan(); + List ids = new ArrayList<>(); + readBuilder + .newRead() + .executeFilter() + .createReader(plan) + .forEachRemaining(row -> ids.add(row.getInt(0))); + return ids; + } + + private FileStoreTable fullSearchTable(FileStoreTable table) { + return table.copy( + Collections.singletonMap(CoreOptions.GLOBAL_INDEX_SEARCH_MODE.key(), "full")); + } + private GenericArray array(Object... elements) { return new GenericArray(elements); } diff --git a/paimon-core/src/test/java/org/apache/paimon/table/source/BucketVectorSearchSplitTest.java b/paimon-core/src/test/java/org/apache/paimon/table/source/BucketVectorSearchSplitTest.java index c2f5c7826181..17336afc6340 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/source/BucketVectorSearchSplitTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/source/BucketVectorSearchSplitTest.java @@ -137,10 +137,13 @@ void testRejectsForeignBytes() throws Exception { .hasMessageContaining("wrong magic number"); byte[] unsupportedVersion = serialize(split()); - ByteBuffer.wrap(unsupportedVersion).putInt(Long.BYTES, 2); + ByteBuffer versionBuffer = ByteBuffer.wrap(unsupportedVersion); + int futureVersion = versionBuffer.getInt(Long.BYTES) + 1; + versionBuffer.putInt(Long.BYTES, futureVersion); assertThatThrownBy(() -> deserialize(unsupportedVersion)) .isInstanceOf(IOException.class) - .hasMessageContaining("Unsupported BucketVectorSearchSplit version: 2"); + .hasMessageContaining( + "Unsupported BucketVectorSearchSplit version: " + futureVersion); } private static byte[] serialize(BucketVectorSearchSplit split) throws IOException { diff --git a/paimon-core/src/test/java/org/apache/paimon/table/source/FullTextSearchBuilderTest.java b/paimon-core/src/test/java/org/apache/paimon/table/source/FullTextSearchBuilderTest.java index 0c4f2b512257..e2d7aff832ec 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/source/FullTextSearchBuilderTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/source/FullTextSearchBuilderTest.java @@ -44,6 +44,7 @@ import org.apache.paimon.predicate.PredicateBuilder; import org.apache.paimon.reader.RecordReader; import org.apache.paimon.schema.Schema; +import org.apache.paimon.schema.SchemaChange; import org.apache.paimon.table.FileStoreTable; import org.apache.paimon.table.TableTestBase; import org.apache.paimon.table.sink.BatchTableCommit; @@ -58,6 +59,8 @@ import org.junit.jupiter.api.Test; +import javax.annotation.Nullable; + import java.io.ByteArrayInputStream; import java.io.ByteArrayOutputStream; import java.io.ObjectInputStream; @@ -66,6 +69,7 @@ import java.util.Arrays; import java.util.Collections; import java.util.List; +import java.util.stream.Collectors; import static org.apache.paimon.table.source.DeletionVectorTestUtils.commitDeletionVectors; import static org.assertj.core.api.Assertions.assertThat; @@ -277,6 +281,52 @@ public void testFullTextSearchNonFastModesScanUnindexedData() throws Exception { } } + @Test + public void testFullTextSearchNonFastModesScanDataWithLegacyIndex() throws Exception { + createTableDefault(); + FileStoreTable table = getTableDefault(); + + String[] documents = {"legacy needle", "other document"}; + writeDocuments(table, documents); + buildAndCommitIndexWithFields( + table, + documents, + Collections.singletonList(table.rowType().getField(TEXT_FIELD_NAME)), + null); + + assertNonFastModesUseRawFallback(table, "needle", 0); + } + + @Test + public void testFullTextSearchNonFastModesScanDataWithIncompatibleIndex() throws Exception { + Identifier identifier = identifier("full_text_incompatible_index"); + Schema schema = + Schema.newBuilder() + .column("id", DataTypes.INT()) + .column(TEXT_FIELD_NAME, DataTypes.VARCHAR(32)) + .option(CoreOptions.BUCKET.key(), "-1") + .option(CoreOptions.ROW_TRACKING_ENABLED.key(), "true") + .option(CoreOptions.DATA_EVOLUTION_ENABLED.key(), "true") + .build(); + catalog.createTable(identifier, schema, false); + FileStoreTable table = getTable(identifier); + + String[] documents = {"incompatible needle", "other document"}; + writeDocuments(table, documents); + buildAndCommitIndex(table, documents); + long buildSchemaId = table.schema().id(); + + catalog.alterTable( + identifier, + Collections.singletonList( + SchemaChange.updateColumnType(TEXT_FIELD_NAME, DataTypes.STRING())), + false); + table = getTable(identifier); + assertThat(table.schema().id()).isNotEqualTo(buildSchemaId); + + assertNonFastModesUseRawFallback(table, "needle", 0); + } + @Test public void testFullTextSearchRawSearchRespectsPartitionFilter() throws Exception { Identifier identifier = identifier("PartitionedTextTable"); @@ -881,7 +931,9 @@ public void testFullTextSearchSplitSerialization() throws Exception { } RawFullTextSearchSplit rawOriginal = - new RawFullTextSearchSplit(Collections.singletonList(new Range(2, 3))); + new RawFullTextSearchSplit( + Collections.singletonList(new Range(2, 3)), + TestFullTextGlobalIndexerFactory.IDENTIFIER); bos = new ByteArrayOutputStream(); try (ObjectOutputStream out = new ObjectOutputStream(bos)) { out.writeObject(rawOriginal); @@ -894,6 +946,7 @@ public void testFullTextSearchSplitSerialization() throws Exception { } assertThat(rawDeserialized.rowRanges()).isEqualTo(rawOriginal.rowRanges()); + assertThat(rawDeserialized.indexType()).isEqualTo(rawOriginal.indexType()); } // ====================== Helper methods ====================== @@ -962,6 +1015,15 @@ private void buildAndCommitIndex(FileStoreTable table, String[] documents) throw private void buildAndCommitIndexWithFields( FileStoreTable table, String[] documents, List indexFields) throws Exception { + buildAndCommitIndexWithFields(table, documents, indexFields, table.schema().id()); + } + + private void buildAndCommitIndexWithFields( + FileStoreTable table, + String[] documents, + List indexFields, + @Nullable Long schemaId) + throws Exception { Options options = table.coreOptions().toConfiguration(); DataField textField = table.rowType().getField(TEXT_FIELD_NAME); @@ -987,7 +1049,14 @@ private void buildAndCommitIndexWithFields( indexFields, TestFullTextGlobalIndexerFactory.IDENTIFIER, entries, - null); + null, + schemaId == null ? table.schema().id() : schemaId); + if (schemaId == null) { + indexFiles = + indexFiles.stream() + .map(indexFile -> indexFile.withSchemaId(null)) + .collect(Collectors.toList()); + } DataIncrement dataIncrement = DataIncrement.indexIncrement(indexFiles); CommitMessage message = @@ -1002,6 +1071,30 @@ private void buildAndCommitIndexWithFields( } } + private void assertNonFastModesUseRawFallback( + FileStoreTable table, String query, int expectedId) throws Exception { + for (String searchMode : Arrays.asList("full", "detail")) { + FileStoreTable nonFastModeTable = + (FileStoreTable) + table.copy( + Collections.singletonMap( + CoreOptions.FULL_TEXT_INDEX_SEARCH_MODE.key(), + searchMode)); + FullTextSearchBuilder searchBuilder = + nonFastModeTable + .newFullTextSearchBuilder() + .withQuery(TEXT_FIELD_NAME, matchQuery(query)) + .withLimit(10); + + List splits = searchBuilder.newFullTextScan().scan().splits(); + assertThat(splits).singleElement().isInstanceOf(RawFullTextSearchSplit.class); + RawFullTextSearchSplit rawSplit = (RawFullTextSearchSplit) splits.get(0); + assertThat(rawSplit.indexType()).isEqualTo(TestFullTextGlobalIndexerFactory.IDENTIFIER); + assertThat(readIds(nonFastModeTable, searchBuilder.executeLocal())) + .containsExactly(expectedId); + } + } + private void buildAndCommitSourceBackedIndex(FileStoreTable table, String[] documents) throws Exception { Options options = table.coreOptions().toConfiguration(); @@ -1026,7 +1119,8 @@ private void buildAndCommitSourceBackedIndex(FileStoreTable table, String[] docu Collections.singletonList(textField), TestFullTextGlobalIndexerFactory.IDENTIFIER, writer.finish(), - null); + null, + table.schema().id()); byte[] sourceMeta = new PrimaryKeyIndexSourceMeta( 1, new PrimaryKeyIndexSourceFile("data-file", documents.length)) @@ -1040,6 +1134,8 @@ private void buildAndCommitSourceBackedIndex(FileStoreTable table, String[] docu indexFile.fileName(), indexFile.fileSize(), indexFile.rowCount(), + indexFile.dvRanges(), + indexFile.externalPath(), new GlobalIndexMeta( meta.rowRangeStart(), meta.rowRangeEnd(), @@ -1047,7 +1143,7 @@ private void buildAndCommitSourceBackedIndex(FileStoreTable table, String[] docu meta.extraFieldIds(), meta.indexMeta(), sourceMeta), - indexFile.externalPath())); + indexFile.schemaId())); } CommitMessage message = @@ -1098,7 +1194,8 @@ private void buildAndCommitIndexRange( indexFields, TestFullTextGlobalIndexerFactory.IDENTIFIER, entries, - null); + null, + table.schema().id()); DataIncrement dataIncrement = DataIncrement.indexIncrement(indexFiles); CommitMessage message = @@ -1176,7 +1273,8 @@ private void buildAndCommitIndexForColumn( rowRange, textField.id(), TestFullTextGlobalIndexerFactory.IDENTIFIER, - entries); + entries, + table.schema().id()); DataIncrement dataIncrement = DataIncrement.indexIncrement(indexFiles); CommitMessage message = @@ -1210,7 +1308,8 @@ private void buildAndCommitBTreeIndex(FileStoreTable table, String[] documents) rowRange, textField.id(), BTreeGlobalIndexerFactory.IDENTIFIER, - entries); + entries, + table.schema().id()); DataIncrement dataIncrement = DataIncrement.indexIncrement(indexFiles); CommitMessage message = @@ -1262,7 +1361,8 @@ private void buildAndCommitMultipleIndexFiles(FileStoreTable table, String[] doc rowRange1, textField.id(), TestFullTextGlobalIndexerFactory.IDENTIFIER, - entries1); + entries1, + table.schema().id()); // Build second index file covering rows [mid, end) GlobalIndexSingleColumnWriter writer2 = @@ -1285,7 +1385,8 @@ private void buildAndCommitMultipleIndexFiles(FileStoreTable table, String[] doc rowRange2, textField.id(), TestFullTextGlobalIndexerFactory.IDENTIFIER, - entries2); + entries2, + table.schema().id()); // Combine all index files and commit together List allIndexFiles = new ArrayList<>(); diff --git a/paimon-core/src/test/java/org/apache/paimon/table/source/VectorSearchBuilderTest.java b/paimon-core/src/test/java/org/apache/paimon/table/source/VectorSearchBuilderTest.java index efec98c1f370..3edc7f021f73 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/source/VectorSearchBuilderTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/source/VectorSearchBuilderTest.java @@ -473,6 +473,34 @@ public void testFullModeRawOnlyUsesConfiguredMetric() throws Exception { assertThat(result.results()).containsExactly(0L); } + @Test + public void testFullModeFallsBackForLegacyIndex() throws Exception { + catalog.createTable( + identifier("full_search_legacy_vector_index_table"), + vectorSchemaBuilder(VECTOR_FIELD_NAME) + .option(CoreOptions.VECTOR_INDEX_SEARCH_MODE.key(), "full") + .build(), + false); + FileStoreTable table = getTable(identifier("full_search_legacy_vector_index_table")); + + float[][] vectors = {{0.0f, 0.0f}, {1.0f, 0.0f}}; + writeVectors(table, vectors); + buildAndCommitIndex(table, VECTOR_FIELD_NAME, vectors, false); + + VectorSearchBuilder builder = + table.newVectorSearchBuilder() + .withVector(new float[] {0.0f, 0.0f}) + .withLimit(2) + .withVectorColumn(VECTOR_FIELD_NAME); + VectorScan.Plan plan = builder.newVectorScan().scan(); + + assertThat(indexVectorSearchSplits(plan.splits())).isEmpty(); + assertThat(rawVectorSearchSplits(plan.splits())).hasSize(1); + assertThat(rawVectorSearchSplits(plan.splits()).get(0).rowRanges()) + .containsExactly(new Range(0, 1)); + assertThat(builder.newVectorRead().read(plan).results()).containsExactly(0L, 1L); + } + @Test public void testVectorSearchFullModeScansUnindexedData() throws Exception { catalog.createTable( @@ -1797,6 +1825,12 @@ private static ScoredGlobalIndexResult scoredResult(float score, long... rowIds) private void buildAndCommitIndex(FileStoreTable table, String fieldName, float[][] vectors) throws Exception { + buildAndCommitIndex(table, fieldName, vectors, true); + } + + private void buildAndCommitIndex( + FileStoreTable table, String fieldName, float[][] vectors, boolean includeSchemaId) + throws Exception { Options options = table.coreOptions().toConfiguration(); DataField vectorField = table.rowType().getField(fieldName); @@ -1821,7 +1855,15 @@ private void buildAndCommitIndex(FileStoreTable table, String fieldName, float[] rowRange, vectorField.id(), TestVectorGlobalIndexerFactory.IDENTIFIER, - entries); + entries, + table.schema().id()); + if (!includeSchemaId) { + List legacyIndexFiles = new ArrayList<>(indexFiles.size()); + for (IndexFileMeta indexFile : indexFiles) { + legacyIndexFiles.add(indexFile.withSchemaId(null)); + } + indexFiles = legacyIndexFiles; + } DataIncrement dataIncrement = DataIncrement.indexIncrement(indexFiles); CommitMessage message = @@ -1863,7 +1905,8 @@ private void buildAndCommitMultipleIndexFiles(FileStoreTable table, float[][] ve rowRange1, vectorField.id(), TestVectorGlobalIndexerFactory.IDENTIFIER, - entries1); + entries1, + table.schema().id()); // Build second index file covering rows [mid, end) GlobalIndexSingleColumnWriter writer2 = @@ -1886,7 +1929,8 @@ private void buildAndCommitMultipleIndexFiles(FileStoreTable table, float[][] ve rowRange2, vectorField.id(), TestVectorGlobalIndexerFactory.IDENTIFIER, - entries2); + entries2, + table.schema().id()); // Combine all index files and commit together List allIndexFiles = new ArrayList<>(); @@ -2060,7 +2104,8 @@ private void buildAndCommitVectorIndexWithFields( indexFields, TestVectorGlobalIndexerFactory.IDENTIFIER, entries, - null); + null, + table.schema().id()); DataIncrement dataIncrement = DataIncrement.indexIncrement(indexFiles); CommitMessage message = @@ -2098,7 +2143,8 @@ private void buildAndCommitBTreeIndex(FileStoreTable table, int[] ids, Range row rowRange, idField.id(), BTreeGlobalIndexerFactory.IDENTIFIER, - entries); + entries, + table.schema().id()); DataIncrement dataIncrement = DataIncrement.indexIncrement(indexFiles); CommitMessage message = @@ -2139,7 +2185,8 @@ private void buildAndCommitPartitionedIndex( rowRange, vectorField.id(), TestVectorGlobalIndexerFactory.IDENTIFIER, - entries); + entries, + table.schema().id()); DataIncrement dataIncrement = DataIncrement.indexIncrement(indexFiles); CommitMessage message = diff --git a/paimon-core/src/test/resources/compatibility/manifest-committable-v13-global-index-v5 b/paimon-core/src/test/resources/compatibility/manifest-committable-v13-global-index-v5 new file mode 100644 index 000000000000..b2201de23e68 Binary files /dev/null and b/paimon-core/src/test/resources/compatibility/manifest-committable-v13-global-index-v5 differ diff --git a/paimon-core/src/test/resources/compatibility/manifest-committable-v14-v5 b/paimon-core/src/test/resources/compatibility/manifest-committable-v14-v5 new file mode 100644 index 000000000000..65fd9f3aca6d Binary files /dev/null and b/paimon-core/src/test/resources/compatibility/manifest-committable-v14-v5 differ diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/globalindex/GenericIndexTopoBuilder.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/globalindex/GenericIndexTopoBuilder.java index 4740a3ea805c..df23bc883ca9 100644 --- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/globalindex/GenericIndexTopoBuilder.java +++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/globalindex/GenericIndexTopoBuilder.java @@ -554,7 +554,8 @@ private static CommitMessage flushIndex( indexFields, indexType, resultEntries, - sourceMeta); + sourceMeta, + table.schema().id()); return new CommitMessageImpl( partition, 0, null, indexIncrement(indexFileMetas), emptyIncrement()); } diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/VectorSearchProcedureITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/VectorSearchProcedureITCase.java index 31324f2ec7f3..2c48bdc34a10 100644 --- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/VectorSearchProcedureITCase.java +++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/VectorSearchProcedureITCase.java @@ -1072,7 +1072,8 @@ private void buildAndCommitVectorIndex(FileStoreTable table, float[][] vectors, rowRange, vectorField.id(), TestVectorGlobalIndexerFactory.IDENTIFIER, - entries); + entries, + table.schema().id()); DataIncrement dataIncrement = DataIncrement.indexIncrement(indexFiles); CommitMessage message = @@ -1115,7 +1116,8 @@ private void buildAndCommitMultiFieldVectorIndex( Arrays.asList(vectorField, idField), TestVectorGlobalIndexerFactory.IDENTIFIER, entries, - null); + null, + table.schema().id()); DataIncrement dataIncrement = DataIncrement.indexIncrement(indexFiles); CommitMessage message = diff --git a/paimon-full-text/src/test/java/org/apache/paimon/fulltext/index/JavaPyNativeFullTextE2ETest.java b/paimon-full-text/src/test/java/org/apache/paimon/fulltext/index/JavaPyNativeFullTextE2ETest.java index 7056909d5bf3..d4971a570ca5 100644 --- a/paimon-full-text/src/test/java/org/apache/paimon/fulltext/index/JavaPyNativeFullTextE2ETest.java +++ b/paimon-full-text/src/test/java/org/apache/paimon/fulltext/index/JavaPyNativeFullTextE2ETest.java @@ -225,7 +225,8 @@ private void writeTableWithNativeFullTextIndex( rowRange, contentField.id(), NativeFullTextGlobalIndexerFactory.IDENTIFIER, - entries); + entries, + table.schema().id()); // Commit the index DataIncrement dataIncrement = DataIncrement.indexIncrement(indexFiles); diff --git a/paimon-lumina/src/test/java/org/apache/paimon/lumina/index/JavaPyLuminaE2ETest.java b/paimon-lumina/src/test/java/org/apache/paimon/lumina/index/JavaPyLuminaE2ETest.java index d402a40ac942..4c1be3f98ce3 100644 --- a/paimon-lumina/src/test/java/org/apache/paimon/lumina/index/JavaPyLuminaE2ETest.java +++ b/paimon-lumina/src/test/java/org/apache/paimon/lumina/index/JavaPyLuminaE2ETest.java @@ -194,7 +194,8 @@ public void testLuminaVectorIndexWrite(String fileFormat) throws Exception { rowRange, embeddingField.id(), LuminaVectorGlobalIndexerFactory.IDENTIFIER, - entries); + entries, + table.schema().id()); DataIncrement dataIncrement = DataIncrement.indexIncrement(indexFiles); CommitMessage message = @@ -298,7 +299,8 @@ public void testLuminaVectorWithBTreeIndexWrite() throws Exception { rowRange, embeddingField.id(), LuminaVectorGlobalIndexerFactory.IDENTIFIER, - vectorEntries); + vectorEntries, + table.schema().id()); // Build BTree global index on "id". DataField idField = table.rowType().getField("id"); @@ -318,7 +320,8 @@ public void testLuminaVectorWithBTreeIndexWrite() throws Exception { rowRange, idField.id(), BTreeGlobalIndexerFactory.IDENTIFIER, - idEntries); + idEntries, + table.schema().id()); // Commit both index sets in a single index-only commit. java.util.List allIndexFiles = new java.util.ArrayList<>(); diff --git a/paimon-python/pypaimon/globalindex/create_global_index.py b/paimon-python/pypaimon/globalindex/create_global_index.py index bd9d34a05a7a..e470a10dcd87 100644 --- a/paimon-python/pypaimon/globalindex/create_global_index.py +++ b/paimon-python/pypaimon/globalindex/create_global_index.py @@ -502,6 +502,7 @@ def _to_index_manifest_entries( partition=partition, bucket=0, index_file=index_file, + schema_id=table.table_schema.id, ) ) return entries diff --git a/paimon-python/pypaimon/globalindex/data_evolution_global_index_scanner.py b/paimon-python/pypaimon/globalindex/data_evolution_global_index_scanner.py index dde7f24f570e..6bf82aa1a6ff 100644 --- a/paimon-python/pypaimon/globalindex/data_evolution_global_index_scanner.py +++ b/paimon-python/pypaimon/globalindex/data_evolution_global_index_scanner.py @@ -27,6 +27,9 @@ from pypaimon.globalindex.global_index_meta import GlobalIndexIOMeta from pypaimon.globalindex.global_index_reader import GlobalIndexReader, _map_future from pypaimon.globalindex.global_index_result import GlobalIndexResult +from pypaimon.globalindex.global_index_schema_compatibility import ( + filter_compatible_global_indexes, +) from pypaimon.common.options.core_options import CoreOptions from pypaimon.common.options.options import Options from pypaimon.common.predicate import Predicate @@ -207,6 +210,7 @@ def index_file_filter(entry): snapshot = _resolve_snapshot(table, None) index_file_handler = IndexFileHandler(table=table) entries = index_file_handler.scan(snapshot, index_file_filter) + entries = filter_compatible_global_indexes(table, entries) scanned_index_files = [entry.index_file for entry in entries] if len(scanned_index_files) == 0: diff --git a/paimon-python/pypaimon/globalindex/drop_global_index.py b/paimon-python/pypaimon/globalindex/drop_global_index.py index 6be45499e8b3..72aa9641c95a 100644 --- a/paimon-python/pypaimon/globalindex/drop_global_index.py +++ b/paimon-python/pypaimon/globalindex/drop_global_index.py @@ -120,6 +120,7 @@ def should_delete(entry: IndexManifestEntry) -> bool: partition=entry.partition, bucket=entry.bucket, index_file=entry.index_file, + schema_id=entry.schema_id, ) ) diff --git a/paimon-python/pypaimon/globalindex/global_index_schema_compatibility.py b/paimon-python/pypaimon/globalindex/global_index_schema_compatibility.py new file mode 100644 index 000000000000..76a6b379701c --- /dev/null +++ b/paimon-python/pypaimon/globalindex/global_index_schema_compatibility.py @@ -0,0 +1,86 @@ +# 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. + +"""Validates global indexes against the current table schema.""" + +from typing import Collection, Dict, List, Set + +from pypaimon.schema.data_types import DataTypeParser + + +def filter_compatible_global_indexes(table, entries: Collection) -> List: + """Keep entries whose indexed fields have the same logical types.""" + current_schema = table.table_schema + current_fields = _fields_by_id(current_schema.fields) + historical_fields = {current_schema.id: current_fields} + missing_schema_ids: Set[int] = set() + compatibility_cache = {} + compatible = [] + + for entry in entries: + global_index = entry.index_file.global_index_meta + schema_id = entry.schema_id + if global_index is None or schema_id is None: + continue + + fields = historical_fields.get(schema_id) + if fields is None and schema_id not in missing_schema_ids: + historical_schema = table.schema_manager.get_schema(schema_id) + if historical_schema is None: + missing_schema_ids.add(schema_id) + else: + fields = _fields_by_id(historical_schema.fields) + historical_fields[schema_id] = fields + + if fields is None: + continue + field_ids = tuple( + [global_index.index_field_id] + + list(global_index.extra_field_ids or []) + ) + cache_key = (schema_id, field_ids) + if cache_key not in compatibility_cache: + compatibility_cache[cache_key] = _compatible_indexed_fields( + field_ids, fields, current_fields) + if compatibility_cache[cache_key]: + compatible.append(entry) + + return compatible + + +def _compatible_indexed_fields(field_ids, historical_fields, current_fields): + for field_id in field_ids: + historical_field = historical_fields.get(field_id) + current_field = current_fields.get(field_id) + if historical_field is None or current_field is None: + return False + if not _equals_ignore_nullable( + historical_field.type, current_field.type): + return False + return True + + +def _fields_by_id(fields) -> Dict: + return {field.id: field for field in fields} + + +def _equals_ignore_nullable(left, right) -> bool: + left_copy = DataTypeParser.parse_data_type(left.to_dict()) + right_copy = DataTypeParser.parse_data_type(right.to_dict()) + left_copy.nullable = True + right_copy.nullable = True + return left_copy == right_copy diff --git a/paimon-python/pypaimon/index/dynamic_bucket.py b/paimon-python/pypaimon/index/dynamic_bucket.py index c5bbd08b7dfa..fe70878470db 100644 --- a/paimon-python/pypaimon/index/dynamic_bucket.py +++ b/paimon-python/pypaimon/index/dynamic_bucket.py @@ -438,6 +438,7 @@ def prepare_commit(self) -> Dict[Tuple[Tuple, int], DynamicBucketIndexChanges]: partition=old_entry.partition, bucket=old_entry.bucket, index_file=old_entry.index_file, + schema_id=old_entry.schema_id, ) ] if old_entry is not None else [], ) diff --git a/paimon-python/pypaimon/manifest/index_manifest_entry.py b/paimon-python/pypaimon/manifest/index_manifest_entry.py index 7a5e7d1a4f53..38c9dd4950d8 100644 --- a/paimon-python/pypaimon/manifest/index_manifest_entry.py +++ b/paimon-python/pypaimon/manifest/index_manifest_entry.py @@ -16,6 +16,7 @@ # under the License. from dataclasses import dataclass +from typing import Optional from pypaimon.index.index_file_meta import IndexFileMeta from pypaimon.table.row.generic_row import GenericRow @@ -29,6 +30,7 @@ class IndexManifestEntry: partition: GenericRow bucket: int index_file: IndexFileMeta + schema_id: Optional[int] = None def __eq__(self, other): if not isinstance(other, IndexManifestEntry): @@ -36,11 +38,12 @@ def __eq__(self, other): return (self.kind == other.kind and self.partition == other.partition and self.bucket == other.bucket and - self.index_file == other.index_file) + self.index_file == other.index_file and + self.schema_id == other.schema_id) def __hash__(self): return hash((self.kind, tuple(self.partition.values), - self.bucket, self.index_file)) + self.bucket, self.index_file, self.schema_id)) INDEX_MANIFEST_ENTRY = { @@ -57,6 +60,7 @@ def __hash__(self): {"name": "_ROW_COUNT", "type": "long"}, {"name": "_DELETIONS_VECTORS_RANGES", "type": {"type": "array", "elementType": "DeletionVectorMeta"}}, {"name": "_EXTERNAL_PATH", "type": ["null", "string"]}, - {"name": "_GLOBAL_INDEX", "type": "GlobalIndexMeta"} + {"name": "_GLOBAL_INDEX", "type": "GlobalIndexMeta"}, + {"name": "_SCHEMA_ID", "type": ["null", "long"], "default": None} ] } diff --git a/paimon-python/pypaimon/manifest/index_manifest_file.py b/paimon-python/pypaimon/manifest/index_manifest_file.py index 3c1c9448bac7..2b7cfce8e060 100644 --- a/paimon-python/pypaimon/manifest/index_manifest_file.py +++ b/paimon-python/pypaimon/manifest/index_manifest_file.py @@ -74,6 +74,7 @@ {"name": "_EXTERNAL_PATH", "type": ["null", "string"], "default": None}, {"name": "_GLOBAL_INDEX", "type": ["null", _GLOBAL_INDEX_META_SCHEMA], "default": None}, + {"name": "_SCHEMA_ID", "type": ["null", "long"], "default": None}, ], } @@ -157,7 +158,8 @@ def _read_avro(self, file_bytes: bytes, index_manifest_path: str) -> List[IndexM self.partition_keys_fields ), bucket=record['_BUCKET'], - index_file=index_file_meta + index_file=index_file_meta, + schema_id=record.get('_SCHEMA_ID'), ) entries.append(entry) @@ -211,7 +213,8 @@ def _read_json( kind=record.get('kind', 0), partition=partition, bucket=record.get('bucket', 0), - index_file=index_file_meta + index_file=index_file_meta, + schema_id=record.get('schema_id'), ) entries.append(entry) @@ -311,6 +314,7 @@ def _to_avro_record(self, entry: IndexManifestEntry) -> dict: "_DELETIONS_VECTORS_RANGES": dv_ranges, "_EXTERNAL_PATH": index_file.external_path, "_GLOBAL_INDEX": global_index, + "_SCHEMA_ID": entry.schema_id, } diff --git a/paimon-python/pypaimon/table/source/full_text_scan.py b/paimon-python/pypaimon/table/source/full_text_scan.py index 1387b02536bc..4c609a7c9f21 100644 --- a/paimon-python/pypaimon/table/source/full_text_scan.py +++ b/paimon-python/pypaimon/table/source/full_text_scan.py @@ -25,6 +25,9 @@ from pypaimon.globalindex.full_text.native_full_text_global_index_reader import ( FULL_TEXT_IDENTIFIER, ) +from pypaimon.globalindex.global_index_schema_compatibility import ( + filter_compatible_global_indexes, +) from pypaimon.table.source.full_text_search_split import ( FullTextSearchSplit, IndexFullTextSearchSplit, @@ -98,6 +101,7 @@ def index_file_filter(entry): ) entries = index_file_handler.scan(snapshot, index_file_filter) + entries = filter_compatible_global_indexes(self._table, entries) all_index_files = [entry.index_file for entry in entries] # Group full-text index files by column and (rowRangeStart, rowRangeEnd). @@ -116,18 +120,17 @@ def index_file_filter(entry): IndexFullTextSearchSplit( column_name, range_key.from_, range_key.to, files)) - if all_index_files: - raw_row_ranges = DataEvolutionGlobalIndexCoverage( - self._table, - snapshot, - partition_filter, - all_index_files, - ).unindexed_ranges( - list(text_column_ids), - search_mode=self._table.options.full_text_index_search_mode(), - ) - if raw_row_ranges: - splits.append(RawFullTextSearchSplit(raw_row_ranges)) + raw_row_ranges = DataEvolutionGlobalIndexCoverage( + self._table, + snapshot, + partition_filter, + all_index_files, + ).unindexed_ranges( + list(text_column_ids), + search_mode=self._table.options.full_text_index_search_mode(), + ) + if raw_row_ranges: + splits.append(RawFullTextSearchSplit(raw_row_ranges)) return FullTextScanPlan(splits) diff --git a/paimon-python/pypaimon/table/source/vector_search_scan.py b/paimon-python/pypaimon/table/source/vector_search_scan.py index 8c94390b511f..4b2c73bc5c92 100644 --- a/paimon-python/pypaimon/table/source/vector_search_scan.py +++ b/paimon-python/pypaimon/table/source/vector_search_scan.py @@ -25,6 +25,9 @@ from pypaimon.globalindex.data_evolution_global_index_scanner import ( is_supported_scalar_index, ) +from pypaimon.globalindex.global_index_schema_compatibility import ( + filter_compatible_global_indexes, +) from pypaimon.table.source.vector_search_split import ( IndexVectorSearchSplit, RawVectorSearchSplit, @@ -132,6 +135,7 @@ def index_file_filter(entry): return False entries = index_file_handler.scan(snapshot, index_file_filter) + entries = filter_compatible_global_indexes(self._table, entries) all_index_files = [entry.index_file for entry in entries] # Group vector index files by (rowRangeStart, rowRangeEnd). diff --git a/paimon-python/pypaimon/tests/global_index_build_test.py b/paimon-python/pypaimon/tests/global_index_build_test.py index 186270fcb79a..6f58a2e5dff9 100644 --- a/paimon-python/pypaimon/tests/global_index_build_test.py +++ b/paimon-python/pypaimon/tests/global_index_build_test.py @@ -223,6 +223,8 @@ def test_create_btree_global_index_from_python(self): entries = IndexFileHandler(table).scan(snapshot) self.assertEqual(2, len(entries)) self.assertEqual({'btree'}, {e.index_file.index_type for e in entries}) + self.assertEqual( + {table.table_schema.id}, {e.schema_id for e in entries}) self.assertEqual({0}, {e.index_file.global_index_meta.row_range_start for e in entries}) self.assertEqual({3}, {e.index_file.global_index_meta.row_range_end for e in entries}) diff --git a/paimon-python/pypaimon/tests/global_index_schema_compatibility_test.py b/paimon-python/pypaimon/tests/global_index_schema_compatibility_test.py new file mode 100644 index 000000000000..3f1753a40009 --- /dev/null +++ b/paimon-python/pypaimon/tests/global_index_schema_compatibility_test.py @@ -0,0 +1,104 @@ +# 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. + +import types +import unittest + +from pypaimon.globalindex.global_index_meta import GlobalIndexMeta +from pypaimon.globalindex.global_index_schema_compatibility import ( + filter_compatible_global_indexes, +) +from pypaimon.index.index_file_meta import IndexFileMeta +from pypaimon.manifest.index_manifest_entry import IndexManifestEntry +from pypaimon.schema.data_types import ArrayType, AtomicType, DataField +from pypaimon.schema.table_schema import TableSchema + + +def _field(field_id, name, data_type): + return DataField(field_id, name, data_type) + + +def _schema(schema_id, fields): + return TableSchema(id=schema_id, fields=fields) + + +def _entry(file_name, schema_id, field_id, extra_field_ids=None): + index_file = IndexFileMeta( + index_type='bitmap', + file_name=file_name, + file_size=1, + row_count=1, + global_index_meta=GlobalIndexMeta( + row_range_start=0, + row_range_end=0, + index_field_id=field_id, + extra_field_ids=extra_field_ids, + ), + ) + return IndexManifestEntry(0, None, 0, index_file, schema_id) + + +class GlobalIndexSchemaCompatibilityTest(unittest.TestCase): + + def test_filters_by_indexed_field_types(self): + historical = _schema(1, [ + _field(1, 'numbers', ArrayType(True, AtomicType('INT'))), + _field(2, 'name', AtomicType('STRING', nullable=False)), + _field(3, 'age', AtomicType('INT')), + ]) + current = _schema(2, [ + _field(1, 'numbers', ArrayType(True, AtomicType('BIGINT'))), + _field(2, 'renamed', AtomicType('STRING')), + _field(3, 'age', AtomicType('BIGINT')), + ]) + schema_lookups = [] + + def get_schema(schema_id): + schema_lookups.append(schema_id) + return historical if schema_id == historical.id else None + + table = types.SimpleNamespace( + table_schema=current, + schema_manager=types.SimpleNamespace(get_schema=get_schema), + ) + compatible = _entry('compatible', 1, 2) + current_schema = _entry('current', 2, 2) + changed_primary = _entry('changed-primary', 1, 1) + changed_extra = _entry('changed-extra', 1, 2, [3]) + legacy = _entry('legacy', None, 2) + missing_schema_a = _entry('missing-a', 99, 2) + missing_schema_b = _entry('missing-b', 99, 2) + + result = filter_compatible_global_indexes(table, [ + compatible, + current_schema, + changed_primary, + changed_extra, + legacy, + missing_schema_a, + missing_schema_b, + ]) + + self.assertEqual( + ['compatible', 'current'], + [entry.index_file.file_name for entry in result], + ) + self.assertEqual([1, 99], schema_lookups) + + +if __name__ == '__main__': + unittest.main() diff --git a/paimon-python/pypaimon/tests/index_manifest_write_test.py b/paimon-python/pypaimon/tests/index_manifest_write_test.py index e6e428f6b593..c2e72fd2c214 100644 --- a/paimon-python/pypaimon/tests/index_manifest_write_test.py +++ b/paimon-python/pypaimon/tests/index_manifest_write_test.py @@ -20,14 +20,20 @@ import tempfile import unittest import uuid +from copy import deepcopy +from io import BytesIO +import fastavro import pyarrow as pa from pypaimon import CatalogFactory, Schema from pypaimon.globalindex.global_index_meta import GlobalIndexMeta from pypaimon.index.index_file_meta import IndexFileMeta from pypaimon.manifest.index_manifest_entry import IndexManifestEntry -from pypaimon.manifest.index_manifest_file import IndexManifestFile +from pypaimon.manifest.index_manifest_file import ( + INDEX_MANIFEST_ENTRY_SCHEMA, + IndexManifestFile, +) from pypaimon.table.row.generic_row import GenericRow @@ -55,7 +61,9 @@ def _table(self): self.catalog.create_table(name, s, False) return self.catalog.get_table(name) - def _entry(self, file_name, field_id, meta=b'm', source_meta=None): + def _entry( + self, file_name, field_id, meta=b'm', source_meta=None, + schema_id=7): partition = GenericRow([], []) index_file = IndexFileMeta( index_type='BTREE', @@ -71,7 +79,13 @@ def _entry(self, file_name, field_id, meta=b'm', source_meta=None): source_meta=source_meta, ), ) - return IndexManifestEntry(kind=0, partition=partition, bucket=0, index_file=index_file) + return IndexManifestEntry( + kind=0, + partition=partition, + bucket=0, + index_file=index_file, + schema_id=schema_id, + ) def test_write_read_roundtrip(self): imf = IndexManifestFile(self._table()) @@ -90,6 +104,28 @@ def test_write_read_roundtrip(self): self.assertEqual(10, gim.row_range_end) self.assertEqual([2], gim.extra_field_ids) self.assertEqual(b'm', bytes(gim.index_meta)) + self.assertEqual(7, a.schema_id) + + def test_read_legacy_manifest_without_schema_id(self): + table = self._table() + imf = IndexManifestFile(table) + entry = self._entry('legacy-index', 1) + record = imf._to_avro_record(entry) + record.pop('_SCHEMA_ID') + legacy_schema = deepcopy(INDEX_MANIFEST_ENTRY_SCHEMA) + legacy_schema['fields'] = [ + field for field in legacy_schema['fields'] + if field['name'] != '_SCHEMA_ID' + ] + + buffer = BytesIO() + fastavro.writer(buffer, legacy_schema, [record]) + file_name = 'index-manifest-legacy-' + uuid.uuid4().hex + path = table.table_path.rstrip('/') + '/manifest/' + file_name + with table.file_io.new_output_stream(path) as output_stream: + output_stream.write(buffer.getvalue()) + + self.assertIsNone(imf.read(file_name)[0].schema_id) def test_write_read_primary_key_source_meta(self): source_meta = b'primary-key-source-meta' @@ -106,8 +142,9 @@ def test_combine_drops_named_files(self): deletes = [self._entry('idx-a', 1)] new_name = imf.combine_deletes(previous, deletes) self.assertNotEqual(previous, new_name) - survivors = {e.index_file.file_name for e in imf.read(new_name)} - self.assertEqual({'idx-b'}, survivors) + survivors = imf.read(new_name) + self.assertEqual({'idx-b'}, {e.index_file.file_name for e in survivors}) + self.assertEqual(7, survivors[0].schema_id) def test_combine_unknown_delete_is_noop_on_content(self): imf = IndexManifestFile(self._table()) diff --git a/paimon-python/pypaimon/tests/vector_search_filter_test.py b/paimon-python/pypaimon/tests/vector_search_filter_test.py index e3bbccc911b5..72c4f37f746b 100644 --- a/paimon-python/pypaimon/tests/vector_search_filter_test.py +++ b/paimon-python/pypaimon/tests/vector_search_filter_test.py @@ -85,7 +85,9 @@ def boost_query(positive, negative, negative_boost): class _StubSchema: - def __init__(self): + def __init__(self, fields=None, schema_id=0): + self.id = schema_id + self.fields = list(fields or []) self.options = {} @@ -97,7 +99,14 @@ def __init__(self, fields, entries, partition_fields=None): self.partition_keys_fields = partition_fields or [] self.partition_keys: List[str] = [ f.name for f in self.partition_keys_fields] - self.table_schema = _StubSchema() + self.table_schema = _StubSchema(fields) + self.schema_manager = types.SimpleNamespace( + get_schema=lambda schema_id: ( + self.table_schema + if schema_id == self.table_schema.id + else None + ) + ) self.options = CoreOptions(Options.from_none()) self.file_io = object() self._entries = entries @@ -142,7 +151,7 @@ def _field(fid, name, dtype="INT"): def _entry(partition_row, field_id, index_type, file_name, - row_range_start, row_range_end, external_path=None): + row_range_start, row_range_end, external_path=None, schema_id=0): meta = GlobalIndexMeta( row_range_start=row_range_start, row_range_end=row_range_end, @@ -157,8 +166,13 @@ def _entry(partition_row, field_id, index_type, file_name, global_index_meta=meta, external_path=external_path, ) - return IndexManifestEntry(kind=0, partition=partition_row, bucket=0, - index_file=index_file) + return IndexManifestEntry( + kind=0, + partition=partition_row, + bucket=0, + index_file=index_file, + schema_id=schema_id, + ) def _bitmap(*row_ids): @@ -932,6 +946,44 @@ def test_full_text_mode_overrides_global_mode_for_uncovered_ranges(self): self.assertFalse(any(isinstance(split, RawFullTextSearchSplit) for split in splits)) + def test_full_text_scan_falls_back_for_incompatible_index_schema(self): + from pypaimon.table.source.full_text_scan import DataEvolutionFullTextScan + from pypaimon.table.source.full_text_search_split import ( + IndexFullTextSearchSplit, + RawFullTextSearchSplit, + ) + + text_field = _field(1, "content", "STRING") + entry = _entry( + None, + field_id=1, + index_type="full-text", + file_name="legacy-ft.index", + row_range_start=0, + row_range_end=9, + schema_id=1, + ) + table = _StubTable(fields=[text_field], entries=[entry]) + table.schema_manager = types.SimpleNamespace( + get_schema=lambda schema_id: _StubSchema( + [_field(1, "content", "VARCHAR(10)")], schema_id=1) + if schema_id == 1 + else None + ) + table.options = CoreOptions(Options({ + "full-text-index.search-mode": "full", + })) + _patch_snapshot( + self, [entry], types.SimpleNamespace(id=1, next_row_id=10)) + + splits = DataEvolutionFullTextScan(table, [text_field]).scan().splits() + + self.assertFalse(any(isinstance(s, IndexFullTextSearchSplit) + for s in splits)) + raw = [s for s in splits if isinstance(s, RawFullTextSearchSplit)] + self.assertEqual(1, len(raw)) + self.assertEqual([Range(0, 9)], raw[0].row_ranges) + class VectorSearchFilterTest(unittest.TestCase): """Non-partitioned wiring: scan + read + external_path plumbing.""" @@ -1036,6 +1088,54 @@ def test_unsupported_scalar_coverage_still_plans_raw_split(self): self.assertEqual([Range(0, 9)], raw[0].row_ranges) self.assertEqual([], raw[0].scalar_index_files) + def test_vector_scan_falls_back_for_incompatible_index_schema(self): + from pypaimon.table.source.vector_search_split import ( + IndexVectorSearchSplit, + RawVectorSearchSplit, + ) + + current_embedding = _field(1, "embedding", "DOUBLE") + entry = _entry( + None, + field_id=1, + index_type="lumina-vector-ann", + file_name="legacy-vector.index", + row_range_start=0, + row_range_end=9, + schema_id=1, + ) + table = _StubTable( + fields=[self.id_field, current_embedding], entries=[entry]) + table.schema_manager = types.SimpleNamespace( + get_schema=lambda schema_id: _StubSchema( + [self.id_field, _field(1, "embedding", "FLOAT")], + schema_id=1, + ) if schema_id == 1 else None + ) + table.options = CoreOptions(Options({ + "vector-index.search-mode": "full", + })) + self._scan_patch.stop() + self._travel_patch.stop() + _patch_snapshot( + self, [entry], types.SimpleNamespace(id=1, next_row_id=10)) + + splits = ( + VectorSearchBuilderImpl(table) + .with_vector_column("embedding") + .with_query_vector([1.0, 0.0]) + .with_limit(3) + .new_vector_search_scan() + .scan() + .splits() + ) + + self.assertFalse(any(isinstance(s, IndexVectorSearchSplit) + for s in splits)) + raw = [s for s in splits if isinstance(s, RawVectorSearchSplit)] + self.assertEqual(1, len(raw)) + self.assertEqual([Range(0, 9)], raw[0].row_ranges) + def test_read_threads_prefilter_bitmap_as_include_row_ids(self): """preFilter bitmap from scanner.scan(filter) must reach each split's VectorSearch, offset-rebased to local coords by OffsetGlobalIndexReader. @@ -2322,6 +2422,54 @@ def get_latest_snapshot(self_inner): self.assertEqual([Range(5, 9)], result.results().to_range_list()) + def test_scanner_ignores_incompatible_index_schema(self): + from pypaimon.globalindex.data_evolution_global_index_scanner import ( + DataEvolutionGlobalIndexScanner, + ) + from pypaimon.schema.data_types import ArrayType + + current_field = DataField( + id=0, + name="numbers", + type=ArrayType(True, AtomicType("BIGINT")), + ) + entry = _entry( + None, + field_id=0, + index_type="bitmap", + file_name="legacy-array.index", + row_range_start=0, + row_range_end=9, + schema_id=1, + ) + table = _StubTable(fields=[current_field], entries=[entry]) + historical_field = DataField( + id=0, + name="numbers", + type=ArrayType(True, AtomicType("INT")), + ) + table.schema_manager = types.SimpleNamespace( + get_schema=lambda schema_id: _StubSchema( + [historical_field], schema_id=1) + if schema_id == 1 + else None + ) + snapshot = types.SimpleNamespace(id=1, next_row_id=10) + _patch_snapshot(self, [entry], snapshot) + + scanner = DataEvolutionGlobalIndexScanner.create( + table, + predicate=Predicate( + method="equal", + index=0, + field="numbers", + literals=[[1]], + ), + snapshot=snapshot, + ) + + self.assertIsNone(scanner) + def test_scanner_create_selects_extra_field_indexes(self): from pypaimon.globalindex.data_evolution_global_index_scanner import ( DataEvolutionGlobalIndexScanner, @@ -2338,7 +2486,7 @@ def test_scanner_create_selects_extra_field_indexes(self): fields=[name_field, id_field, emb_field], entries=[ IndexManifestEntry(kind=0, partition=None, - bucket=0, index_file=multi) + bucket=0, index_file=multi, schema_id=0) ], ) _patch_snapshot(self, table._entries) diff --git a/paimon-python/pypaimon/write/file_store_commit.py b/paimon-python/pypaimon/write/file_store_commit.py index 5f8f054aa812..28cbc7b0b536 100644 --- a/paimon-python/pypaimon/write/file_store_commit.py +++ b/paimon-python/pypaimon/write/file_store_commit.py @@ -419,6 +419,7 @@ def _overwrite_hash_index_deletes(self, partition_filter, deletes): partition=entry.partition, bucket=entry.bucket, index_file=entry.index_file, + schema_id=entry.schema_id, ) return list(by_file_name.values()) diff --git a/paimon-python/pypaimon/write/global_index_update_checker.py b/paimon-python/pypaimon/write/global_index_update_checker.py index b7f270c17cc5..7b036a549f37 100644 --- a/paimon-python/pypaimon/write/global_index_update_checker.py +++ b/paimon-python/pypaimon/write/global_index_update_checker.py @@ -38,7 +38,11 @@ def build_index_delete_msgs(entries) -> list: key = tuple(e.partition.values) by_partition.setdefault(key, []).append( IndexManifestEntry( - kind=1, partition=e.partition, bucket=e.bucket, index_file=e.index_file + kind=1, + partition=e.partition, + bucket=e.bucket, + index_file=e.index_file, + schema_id=e.schema_id, ) ) return [ diff --git a/paimon-python/pypaimon/write/table_delete.py b/paimon-python/pypaimon/write/table_delete.py index 5c475c31b1f6..b7038304f5b3 100644 --- a/paimon-python/pypaimon/write/table_delete.py +++ b/paimon-python/pypaimon/write/table_delete.py @@ -220,6 +220,7 @@ def _build_commit_message( partition=entry.partition, bucket=entry.bucket, index_file=entry.index_file, + schema_id=entry.schema_id, ) for entry in old_entries ] diff --git a/paimon-python/pypaimon/write/table_update.py b/paimon-python/pypaimon/write/table_update.py index a9404b32fecb..284556560e08 100644 --- a/paimon-python/pypaimon/write/table_update.py +++ b/paimon-python/pypaimon/write/table_update.py @@ -539,6 +539,7 @@ def _delete_by_partition_filter( partition=entry.partition, bucket=entry.bucket, index_file=entry.index_file, + schema_id=entry.schema_id, )) return [message for message in messages.values() if not message.is_empty()] diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/copy/CopyFilesUtil.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/copy/CopyFilesUtil.java index 88dd1fb40311..f8fb185b19b5 100644 --- a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/copy/CopyFilesUtil.java +++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/copy/CopyFilesUtil.java @@ -28,6 +28,8 @@ import org.apache.commons.io.IOUtils; +import javax.annotation.Nullable; + import java.io.IOException; import java.util.Optional; @@ -79,7 +81,8 @@ public static DataFileMeta toNewDataFileMeta( null); } - public static IndexFileMeta toNewIndexFileMeta(IndexFileMeta oldFileMeta, String newFileName) { + public static IndexFileMeta toNewIndexFileMeta( + IndexFileMeta oldFileMeta, String newFileName, @Nullable Long newSchemaId) { String newExternalPath = externalPathDir(oldFileMeta.externalPath()) .map(dir -> dir + "/" + newFileName) @@ -90,7 +93,9 @@ public static IndexFileMeta toNewIndexFileMeta(IndexFileMeta oldFileMeta, String oldFileMeta.fileSize(), oldFileMeta.rowCount(), oldFileMeta.dvRanges(), - newExternalPath); + newExternalPath, + oldFileMeta.globalIndexMeta(), + newSchemaId); } public static Optional externalPathDir(String externalPath) { diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/copy/ListIndexFilesOperator.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/copy/ListIndexFilesOperator.java index 4e1d6c456261..4f34f5cb5836 100644 --- a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/copy/ListIndexFilesOperator.java +++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/copy/ListIndexFilesOperator.java @@ -22,9 +22,11 @@ import org.apache.paimon.catalog.Catalog; import org.apache.paimon.catalog.Identifier; import org.apache.paimon.fs.Path; +import org.apache.paimon.globalindex.GlobalIndexSchemaCompatibility; import org.apache.paimon.index.IndexFileHandler; import org.apache.paimon.index.IndexFileMeta; import org.apache.paimon.index.IndexFileMetaSerializer; +import org.apache.paimon.index.IndexPathFactory; import org.apache.paimon.manifest.IndexManifestEntry; import org.apache.paimon.partition.PartitionPredicate; import org.apache.paimon.table.FileStoreTable; @@ -37,7 +39,9 @@ import java.io.IOException; import java.util.ArrayList; +import java.util.HashSet; import java.util.List; +import java.util.Set; /** List index files. */ public class ListIndexFilesOperator extends CopyFilesOperator { @@ -66,15 +70,27 @@ public List execute( FileStoreTable targetTable = (FileStoreTable) targetCatalog.getTable(targetIdentifier); List indexFiles = new ArrayList<>(); IndexFileHandler sourceIndexHandler = sourceTable.store().newIndexFileHandler(); + FileStorePathFactory sourceFileStorePathFactory = sourceTable.store().pathFactory(); FileStorePathFactory targetFileStorePathFactory = targetTable.store().pathFactory(); List indexManifestEntries = sourceIndexHandler.readManifestWithIOException(snapshot.indexManifest()); + Set compatibleGlobalIndexes = + new HashSet<>( + GlobalIndexSchemaCompatibility.filterCompatible( + sourceTable, indexManifestEntries)); for (IndexManifestEntry indexManifestEntry : indexManifestEntries) { + boolean globalIndex = indexManifestEntry.indexFile().globalIndexMeta() != null; + if (globalIndex && !compatibleGlobalIndexes.contains(indexManifestEntry)) { + continue; + } if (partitionPredicate == null || partitionPredicate.test(indexManifestEntry.partition())) { CopyFileInfo indexFile = pickIndexFiles( - indexManifestEntry, sourceIndexHandler, targetFileStorePathFactory); + indexManifestEntry, + sourceFileStorePathFactory, + targetFileStorePathFactory, + globalIndex ? targetTable.schema().id() : null); indexFiles.add(indexFile); } } @@ -83,18 +99,20 @@ public List execute( private CopyFileInfo pickIndexFiles( IndexManifestEntry indexManifestEntry, - IndexFileHandler sourceIndexFileHandler, - FileStorePathFactory targetFileStorePathFactory) + FileStorePathFactory sourceFileStorePathFactory, + FileStorePathFactory targetFileStorePathFactory, + @Nullable Long targetSchemaId) throws IOException { - Path indexFilePath = sourceIndexFileHandler.filePath(indexManifestEntry); - Path targetIndexFilePath = - targetFileStorePathFactory - .indexFileFactory( - indexManifestEntry.partition(), indexManifestEntry.bucket()) - .newPath(); IndexFileMeta fileMeta = indexManifestEntry.indexFile(); + IndexPathFactory sourceIndexPathFactory = + indexPathFactory(sourceFileStorePathFactory, indexManifestEntry); + IndexPathFactory targetIndexPathFactory = + indexPathFactory(targetFileStorePathFactory, indexManifestEntry); + Path indexFilePath = sourceIndexPathFactory.toPath(fileMeta); + Path targetIndexFilePath = targetIndexPathFactory.newPath(); IndexFileMeta targetFileMeta = - CopyFilesUtil.toNewIndexFileMeta(fileMeta, targetIndexFilePath.getName()); + CopyFilesUtil.toNewIndexFileMeta( + fileMeta, targetIndexFilePath.getName(), targetSchemaId); return new CopyFileInfo( indexFilePath.toString(), targetIndexFilePath.toString(), @@ -102,4 +120,11 @@ private CopyFileInfo pickIndexFiles( indexManifestEntry.bucket(), indexFileSerializer.serializeToBytes(targetFileMeta)); } + + private static IndexPathFactory indexPathFactory( + FileStorePathFactory pathFactory, IndexManifestEntry entry) { + return entry.indexFile().globalIndexMeta() == null + ? pathFactory.indexFileFactory(entry.partition(), entry.bucket()) + : pathFactory.globalIndexFileFactory(); + } } diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/DefaultGlobalIndexBuilder.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/DefaultGlobalIndexBuilder.java index 5734cf84e396..3c96966b6717 100644 --- a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/DefaultGlobalIndexBuilder.java +++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/DefaultGlobalIndexBuilder.java @@ -159,7 +159,8 @@ public CommitMessage build(CloseableIterator data) throws IOExcepti indexedFields(), indexType, resultEntries, - sourceMeta); + sourceMeta, + table.schema().id()); DataIncrement dataIncrement = DataIncrement.indexIncrement(indexFileMetas); return new CommitMessageImpl( partition, 0, null, dataIncrement, CompactIncrement.emptyIncrement()); diff --git a/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/copy/CopyFilesUtilTest.java b/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/copy/CopyFilesUtilTest.java index 2f973ffce3ad..cdda439d46bf 100644 --- a/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/copy/CopyFilesUtilTest.java +++ b/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/copy/CopyFilesUtilTest.java @@ -18,6 +18,8 @@ package org.apache.paimon.spark.copy; +import org.apache.paimon.index.GlobalIndexMeta; +import org.apache.paimon.index.IndexFileMeta; import org.apache.paimon.io.DataFileMeta; import org.apache.paimon.stats.SimpleStats; @@ -58,4 +60,19 @@ void testClearColumnSequencesWhenChangingSchemaId() { assertThat(copied.writeCols()).containsExactly("a", "b"); assertThat(copied.columnMaxSequenceNumbers()).isNull(); } + + @Test + void testPreserveGlobalIndexMetaAndRebindSchemaId() { + GlobalIndexMeta globalIndexMeta = + new GlobalIndexMeta(0L, 9L, 1, null, new byte[] {1, 2}, new byte[] {3, 4}); + IndexFileMeta source = + new IndexFileMeta( + "btree", "source.idx", 100L, 10L, null, null, globalIndexMeta, 5L); + + IndexFileMeta copied = CopyFilesUtil.toNewIndexFileMeta(source, "copied.idx", 8L); + + assertThat(copied.fileName()).isEqualTo("copied.idx"); + assertThat(copied.globalIndexMeta()).isEqualTo(globalIndexMeta); + assertThat(copied.schemaId()).isEqualTo(8L); + } } diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/CopyFilesProcedureTest.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/CopyFilesProcedureTest.scala index 6c35af37e8a2..dc8fd97e302f 100644 --- a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/CopyFilesProcedureTest.scala +++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/CopyFilesProcedureTest.scala @@ -24,6 +24,8 @@ import org.apache.spark.sql.Row import java.util.concurrent.ThreadLocalRandom +import scala.collection.JavaConverters._ + class CopyFilesProcedureTest extends PaimonSparkTestBase { test("Paimon copy files procedure: append table") { @@ -164,6 +166,78 @@ class CopyFilesProcedureTest extends PaimonSparkTestBase { } } + test("Paimon copy files procedure: global index schema compatibility") { + val random = ThreadLocalRandom.current().nextInt(100000) + val source = s"source_tbl$random" + val compatibleTarget = s"compatible_target_tbl$random" + val incompatibleTarget = s"incompatible_target_tbl$random" + withTable(source, compatibleTarget, incompatibleTarget) { + sql(s""" + |CREATE TABLE $source (id INT, idx INT, payload STRING) + |TBLPROPERTIES ( + | 'bucket' = '-1', + | 'global-index.enabled' = 'true', + | 'row-tracking.enabled' = 'true', + | 'data-evolution.enabled' = 'true', + | 'global-index.column-update-action' = 'IGNORE', + | 'btree-index.records-per-range' = '2') + |""".stripMargin) + sql(s"ALTER TABLE $source RENAME COLUMN payload TO payload_at_build") + sql(s"INSERT INTO $source VALUES (1, 10, 'a'), (2, 20, 'b')") + spark + .sql(s"CALL sys.create_global_index(table => 'test.$source', " + + "index_column => 'idx', index_type => 'btree')") + .collect() + + val sourceEntries = + loadTable(source).store().newIndexFileHandler().scan("btree").asScala + assert(sourceEntries.nonEmpty) + val buildSchemaId = sourceEntries.head.schemaId().longValue() + + sql(s"ALTER TABLE $source RENAME COLUMN payload_at_build TO payload_after_build") + assert(loadTable(source).schema().id() != buildSchemaId) + + checkAnswer( + sql(s"CALL sys.copy(source_table => '$source', target_table => '$compatibleTarget')"), + Row(true) :: Nil + ) + + val compatibleTargetTable = loadTable(compatibleTarget) + val targetEntries = + compatibleTargetTable.store().newIndexFileHandler().scan("btree").asScala + assert(targetEntries.size == sourceEntries.size) + assert(compatibleTargetTable.schema().id() != buildSchemaId) + targetEntries.foreach { + entry => + assert(entry.indexFile().globalIndexMeta() != null) + assert(entry.schemaId().longValue() == compatibleTargetTable.schema().id()) + assert(entry.indexFile().schemaId().longValue() == compatibleTargetTable.schema().id()) + assert( + compatibleTargetTable + .fileIO() + .exists( + compatibleTargetTable + .store() + .pathFactory() + .globalIndexFileFactory() + .toPath(entry.indexFile()))) + } + checkAnswer(sql(s"SELECT id FROM $compatibleTarget WHERE idx = 20"), Row(2) :: Nil) + + sql(s"ALTER TABLE $source ALTER COLUMN idx TYPE BIGINT") + checkAnswer( + sql(s"CALL sys.copy(source_table => '$source', target_table => '$incompatibleTarget')"), + Row(true) :: Nil + ) + assert( + loadTable(incompatibleTarget) + .store() + .newIndexFileHandler() + .scan("btree") + .isEmpty) + } + } + test("Paimon copy files procedure: copy to existed table") { val random = ThreadLocalRandom.current().nextInt(100000); withTable(s"tbl$random") { diff --git a/paimon-vector/src/test/java/org/apache/paimon/JavaPyE2ETest.java b/paimon-vector/src/test/java/org/apache/paimon/JavaPyE2ETest.java index 4c8cbfd59cc1..f0c6b39f32f3 100644 --- a/paimon-vector/src/test/java/org/apache/paimon/JavaPyE2ETest.java +++ b/paimon-vector/src/test/java/org/apache/paimon/JavaPyE2ETest.java @@ -177,7 +177,8 @@ public void testVindexVectorIndexWrite() throws Exception { rowRange, embeddingField.id(), IvfFlatVectorGlobalIndexerFactory.IDENTIFIER, - entries); + entries, + table.schema().id()); DataIncrement dataIncrement = DataIncrement.indexIncrement(indexFiles); CommitMessage message = @@ -287,7 +288,8 @@ public void testVindexVectorRawFallbackWrite() throws Exception { rowRange, embeddingField.id(), IvfFlatVectorGlobalIndexerFactory.IDENTIFIER, - entries); + entries, + table.schema().id()); DataIncrement dataIncrement = DataIncrement.indexIncrement(indexFiles); CommitMessage message =