From 0a2930f786e8a5332dd71c057c4f90d021a4ecd9 Mon Sep 17 00:00:00 2001 From: QuakeWang Date: Wed, 26 Aug 2026 14:20:23 +0800 Subject: [PATCH 1/3] [core] Validate global-index schema compatibility before reader and coverage Global indexes are serialized with the indexed field types from their build schema, while readers use the current table schema. Reusing an incompatible index can miss matches, and counting it in coverage can skip the required data scan. Persist the build schema ID in global-index metadata and compare indexed field types before reader grouping and coverage. Fail closed for legacy metadata and preserve the field across serializers and row-id reassignment. Signed-off-by: QuakeWang --- .../DataEvolutionRowIdReassigner.java | 3 +- .../DataEvolutionGlobalIndexScanner.java | 11 +- .../globalindex/GlobalIndexBuilderUtils.java | 18 ++- .../GlobalIndexSchemaCompatibility.java | 87 ++++++++++++ .../sorted/SortedGlobalIndexWriter.java | 3 +- .../apache/paimon/index/GlobalIndexMeta.java | 28 +++- .../paimon/index/IndexFileMetaSerializer.java | 9 +- .../index/IndexFileMetaV5Deserializer.java | 118 +++++++++++++++ .../IndexManifestEntrySerializer.java | 7 +- .../table/sink/CommitMessageSerializer.java | 11 +- .../source/DataEvolutionFullTextRead.java | 8 +- .../source/DataEvolutionFullTextScan.java | 48 +++++-- .../table/source/DataEvolutionVectorScan.java | 2 + .../table/source/RawFullTextReadImpl.java | 9 +- .../table/source/RawFullTextSearchSplit.java | 26 +++- .../DataEvolutionRowIdReassignerTest.java | 17 ++- .../GlobalIndexBuilderUtilsTest.java | 13 +- .../index/IndexFileMetaSerializerTest.java | 16 ++- .../IndexManifestEntrySerializerTest.java | 8 +- ...ommittableSerializerCompatibilityTest.java | 96 ++++++++++--- .../table/BitmapGlobalIndexTableTest.java | 3 +- .../table/BtreeGlobalIndexTableTest.java | 8 +- .../table/MultiValueGlobalIndexTableTest.java | 95 +++++++++++++ .../sink/CommitMessageSerializerTest.java | 12 ++ .../source/FullTextSearchBuilderTest.java | 134 ++++++++++++++++-- .../table/source/VectorSearchBuilderTest.java | 18 ++- .../manifest-committable-v13-global-index-v5 | Bin 0 -> 3362 bytes .../compatibility/manifest-committable-v14-v5 | Bin 0 -> 3338 bytes .../globalindex/GenericIndexTopoBuilder.java | 3 +- .../VectorSearchProcedureITCase.java | 6 +- .../index/JavaPyNativeFullTextE2ETest.java | 3 +- .../lumina/index/JavaPyLuminaE2ETest.java | 9 +- .../LuminaVectorGlobalIndexScanTest.java | 18 ++- .../DefaultGlobalIndexBuilder.java | 3 +- .../java/org/apache/paimon/JavaPyE2ETest.java | 6 +- 35 files changed, 744 insertions(+), 112 deletions(-) create mode 100644 paimon-core/src/main/java/org/apache/paimon/globalindex/GlobalIndexSchemaCompatibility.java create mode 100644 paimon-core/src/main/java/org/apache/paimon/index/IndexFileMetaV5Deserializer.java create mode 100644 paimon-core/src/test/resources/compatibility/manifest-committable-v13-global-index-v5 create mode 100644 paimon-core/src/test/resources/compatibility/manifest-committable-v14-v5 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 361b9a812a86..0a57e7c231ca 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 @@ -667,7 +667,8 @@ private RewrittenIndexManifest rewriteIndexManifest(Assignment assignment) { globalIndex.indexFieldId(), globalIndex.extraFieldIds(), globalIndex.indexMeta(), - globalIndex.sourceMeta()); + globalIndex.sourceMeta(), + globalIndex.buildSchemaId()); IndexFileMeta newIndexFile = new IndexFileMeta( indexFile.indexType(), 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..40f5d753427e 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 @@ -228,7 +228,8 @@ public static Optional create( @Nullable Snapshot pinnedSnapshot, @Nullable PartitionPredicate partitionFilter, Collection indexFiles) { - List globalIndexFiles = globalIndexFiles(indexFiles); + List globalIndexFiles = + GlobalIndexSchemaCompatibility.filterCompatible(table, indexFiles); if (globalIndexFiles.isEmpty()) { return Optional.empty(); } @@ -254,6 +255,7 @@ public static Optional create( .scan(snapshot, indexFileFilter(table, partitionFilter, filter)).stream() .map(IndexManifestEntry::indexFile) .collect(Collectors.toList()); + indexFiles = GlobalIndexSchemaCompatibility.filterCompatible(table, indexFiles); if (indexFiles.isEmpty()) { return Optional.empty(); } @@ -289,6 +291,7 @@ public static Optional createForTopN( .scan(snapshot, topNIndexFileFilter(partitionFilter, fieldId)).stream() .map(IndexManifestEntry::indexFile) .collect(Collectors.toList()); + indexFiles = GlobalIndexSchemaCompatibility.filterCompatible(table, indexFiles); if (indexFiles.isEmpty()) { return Optional.empty(); } @@ -362,12 +365,6 @@ private static Filter indexFileFilter( return indexFileFilter; } - private static List globalIndexFiles(Collection indexFiles) { - return indexFiles.stream() - .filter(indexFile -> indexFile.globalIndexMeta() != null) - .collect(Collectors.toList()); - } - public Optional scan(Predicate predicate) { return globalIndexEvaluator.evaluate(predicate); } 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 657ba0098c10..926546634d8d 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 buildSchemaId) throws IOException { return toIndexFileMetas( fileIO, @@ -83,7 +84,8 @@ public static List toIndexFileMetas( null, indexType, entries, - null); + null, + buildSchemaId); } /** @@ -100,7 +102,8 @@ public static List toIndexFileMetas( List fields, String indexType, List entries, - @Nullable byte[] sourceMeta) + @Nullable byte[] sourceMeta, + long buildSchemaId) throws IOException { return toIndexFileMetas( fileIO, @@ -111,7 +114,8 @@ public static List toIndexFileMetas( extraFieldIds(fields), indexType, entries, - sourceMeta); + sourceMeta, + buildSchemaId); } public static List unindexedRowRanges( @@ -569,7 +573,8 @@ private static List toIndexFileMetas( @Nullable int[] extraFieldIds, String indexType, List entries, - @Nullable byte[] sourceMeta) + @Nullable byte[] sourceMeta, + long buildSchemaId) throws IOException { List results = new ArrayList<>(); for (ResultEntry entry : entries) { @@ -582,7 +587,8 @@ private static List toIndexFileMetas( indexFieldId, extraFieldIds, entry.meta(), - sourceMeta); + sourceMeta, + buildSchemaId); Path externalPathDir = options.globalIndexExternalPath(); String externalPathString = null; 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..ad57ed8ad73d --- /dev/null +++ b/paimon-core/src/main/java/org/apache/paimon/globalindex/GlobalIndexSchemaCompatibility.java @@ -0,0 +1,87 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.paimon.globalindex; + +import org.apache.paimon.index.GlobalIndexMeta; +import org.apache.paimon.index.IndexFileMeta; +import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.types.RowType; + +import java.io.FileNotFoundException; +import java.util.ArrayList; +import java.util.Collection; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +/** Validates global indexes against the current table schema. */ +public final class GlobalIndexSchemaCompatibility { + + public static List filterCompatible( + FileStoreTable table, Collection indexFiles) { + RowType currentRowType = table.rowType(); + Map buildRowTypes = new HashMap<>(); + buildRowTypes.put(table.schema().id(), currentRowType); + Set missingSchemaIds = new HashSet<>(); + List compatible = new ArrayList<>(); + for (IndexFileMeta indexFile : indexFiles) { + GlobalIndexMeta globalIndex = indexFile.globalIndexMeta(); + if (globalIndex == null || globalIndex.buildSchemaId() == null) { + continue; + } + + long buildSchemaId = globalIndex.buildSchemaId(); + RowType buildRowType = buildRowTypes.get(buildSchemaId); + if (buildRowType == null && !missingSchemaIds.contains(buildSchemaId)) { + try { + buildRowType = + table.schemaManager().tryGetSchema(buildSchemaId).logicalRowType(); + buildRowTypes.put(buildSchemaId, buildRowType); + } catch (FileNotFoundException e) { + missingSchemaIds.add(buildSchemaId); + } + } + if (buildRowType != null + && compatibleIndexedFields(globalIndex, buildRowType, currentRowType)) { + compatible.add(indexFile); + } + } + return compatible; + } + + private static boolean compatibleIndexedFields( + GlobalIndexMeta globalIndex, RowType buildRowType, RowType currentRowType) { + for (int fieldId : globalIndex.getIndexedFieldIds()) { + if (!buildRowType.containsField(fieldId) || !currentRowType.containsField(fieldId)) { + return false; + } + if (!buildRowType + .getField(fieldId) + .type() + .equalsIgnoreNullable(currentRowType.getField(fieldId).type())) { + return false; + } + } + return true; + } + + private GlobalIndexSchemaCompatibility() {} +} 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/GlobalIndexMeta.java b/paimon-core/src/main/java/org/apache/paimon/index/GlobalIndexMeta.java index 026db0786792..18901a890e1e 100644 --- a/paimon-core/src/main/java/org/apache/paimon/index/GlobalIndexMeta.java +++ b/paimon-core/src/main/java/org/apache/paimon/index/GlobalIndexMeta.java @@ -41,6 +41,7 @@ public class GlobalIndexMeta { public static final String EXTRA_FIELD_IDS = "_EXTRA_FIELD_IDS"; public static final String INDEX_META = "_INDEX_META"; public static final String SOURCE_META = "_SOURCE_META"; + public static final String BUILD_SCHEMA_ID = "_BUILD_SCHEMA_ID"; public static final RowType SCHEMA = new RowType( @@ -51,7 +52,8 @@ public class GlobalIndexMeta { new DataField(2, INDEX_FIELD_ID, new IntType(false)), new DataField(3, EXTRA_FIELD_IDS, DataTypes.ARRAY(new IntType(false))), new DataField(4, INDEX_META, DataTypes.BYTES()), - new DataField(5, SOURCE_META, DataTypes.BYTES()))); + new DataField(5, SOURCE_META, DataTypes.BYTES()), + new DataField(6, BUILD_SCHEMA_ID, new BigIntType()))); private final long rowRangeStart; private final long rowRangeEnd; @@ -59,6 +61,7 @@ public class GlobalIndexMeta { @Nullable private final int[] extraFieldIds; @Nullable private final byte[] indexMeta; @Nullable private final byte[] sourceMeta; + @Nullable private final Long buildSchemaId; public GlobalIndexMeta( long rowRangeStart, @@ -76,12 +79,24 @@ public GlobalIndexMeta( @Nullable int[] extraFieldIds, @Nullable byte[] indexMeta, @Nullable byte[] sourceMeta) { + this(rowRangeStart, rowRangeEnd, indexFieldId, extraFieldIds, indexMeta, sourceMeta, null); + } + + public GlobalIndexMeta( + long rowRangeStart, + long rowRangeEnd, + int indexFieldId, + @Nullable int[] extraFieldIds, + @Nullable byte[] indexMeta, + @Nullable byte[] sourceMeta, + @Nullable Long buildSchemaId) { this.rowRangeStart = rowRangeStart; this.rowRangeEnd = rowRangeEnd; this.indexFieldId = indexFieldId; this.extraFieldIds = extraFieldIds; this.indexMeta = indexMeta; this.sourceMeta = sourceMeta; + this.buildSchemaId = buildSchemaId; } public long rowRangeStart() { @@ -117,6 +132,12 @@ public byte[] sourceMeta() { return sourceMeta; } + /** Schema used to build this global index. */ + @Nullable + public Long buildSchemaId() { + return buildSchemaId; + } + /** All indexed field ids in order: the primary {@link #indexFieldId} followed by the rest. */ public List getIndexedFieldIds() { List ids = new ArrayList<>(); @@ -175,12 +196,13 @@ public boolean equals(Object o) { && indexFieldId == that.indexFieldId && Arrays.equals(extraFieldIds, that.extraFieldIds) && Arrays.equals(indexMeta, that.indexMeta) - && Arrays.equals(sourceMeta, that.sourceMeta); + && Arrays.equals(sourceMeta, that.sourceMeta) + && Objects.equals(buildSchemaId, that.buildSchemaId); } @Override public int hashCode() { - int result = Objects.hash(rowRangeStart, rowRangeEnd, indexFieldId); + int result = Objects.hash(rowRangeStart, rowRangeEnd, indexFieldId, buildSchemaId); result = 31 * result + Arrays.hashCode(extraFieldIds); result = 31 * result + Arrays.hashCode(indexMeta); result = 31 * result + Arrays.hashCode(sourceMeta); 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..c45f16c0ef9b 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 @@ -50,7 +50,8 @@ public InternalRow toRow(IndexFileMeta record) { ? null : new GenericArray(globalIndexMeta.extraFieldIds()), globalIndexMeta.indexMeta(), - globalIndexMeta.sourceMeta()); + globalIndexMeta.sourceMeta(), + globalIndexMeta.buildSchemaId()); return GenericRow.of( fromString(record.indexType()), fromString(record.fileName()), @@ -65,7 +66,7 @@ public InternalRow toRow(IndexFileMeta record) { public IndexFileMeta fromRow(InternalRow row) { GlobalIndexMeta globalIndexMeta = null; if (!row.isNullAt(6)) { - InternalRow globalIndexRow = row.getRow(6, 6); + InternalRow globalIndexRow = row.getRow(6, GlobalIndexMeta.SCHEMA.getFieldCount()); Long rowRangeStart = globalIndexRow.getLong(0); Long rowRangeEnd = globalIndexRow.getLong(1); Integer indexFieldId = globalIndexRow.getInt(2); @@ -73,6 +74,7 @@ public IndexFileMeta fromRow(InternalRow row) { globalIndexRow.isNullAt(3) ? null : globalIndexRow.getArray(3).toIntArray(); byte[] indexMeta = globalIndexRow.isNullAt(4) ? null : globalIndexRow.getBinary(4); byte[] sourceMeta = globalIndexRow.isNullAt(5) ? null : globalIndexRow.getBinary(5); + Long buildSchemaId = globalIndexRow.isNullAt(6) ? null : globalIndexRow.getLong(6); globalIndexMeta = new GlobalIndexMeta( rowRangeStart, @@ -80,7 +82,8 @@ public IndexFileMeta fromRow(InternalRow row) { indexFieldId, extralFields, indexMeta, - sourceMeta); + sourceMeta, + buildSchemaId); } return new IndexFileMeta( row.getString(0).toString(), 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..814a45d0e2e2 --- /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} in commit message versions 12 and 13. */ +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/IndexManifestEntrySerializer.java b/paimon-core/src/main/java/org/apache/paimon/manifest/IndexManifestEntrySerializer.java index c37bb77a0022..a1465938d895 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 @@ -64,7 +64,8 @@ public InternalRow toRow(IndexManifestEntry record) { ? null : new GenericArray(globalIndexMeta.extraFieldIds()), globalIndexMeta.indexMeta(), - globalIndexMeta.sourceMeta()); + globalIndexMeta.sourceMeta(), + globalIndexMeta.buildSchemaId()); return GenericRow.of( FORMAT_IDENTIFIER, record.kind().toByteValue(), @@ -102,6 +103,7 @@ private IndexManifestEntry fromDataRow(InternalRow row) { globalIndexRow.isNullAt(3) ? null : globalIndexRow.getArray(3).toIntArray(); byte[] indexMeta = globalIndexRow.isNullAt(4) ? null : globalIndexRow.getBinary(4); byte[] sourceMeta = globalIndexRow.isNullAt(5) ? null : globalIndexRow.getBinary(5); + Long buildSchemaId = globalIndexRow.isNullAt(6) ? null : globalIndexRow.getLong(6); globalIndexMeta = new GlobalIndexMeta( rowRangeStart, @@ -109,7 +111,8 @@ private IndexManifestEntry fromDataRow(InternalRow row) { indexFieldId, extralFields, indexMeta, - sourceMeta); + sourceMeta, + buildSchemaId); } return new IndexManifestEntry( 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/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..9c2edb354b0a 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,19 @@ public Plan scan() { && supportsFullTextSearch(entry.indexFile().indexType()); }; - List allIndexFiles = + List discoveredIndexFiles = indexFileHandler.scan(snapshot, indexFileFilter).stream() .map(IndexManifestEntry::indexFile) .collect(Collectors.toList()); + List discoveredSelections = + chooseIndexRanges(discoveredIndexFiles, textColumnIds, idToColumn); + List compatibleIndexFiles = + GlobalIndexSchemaCompatibility.filterCompatible(table, discoveredIndexFiles); + 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 +138,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 +171,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..fbac57183f26 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; @@ -118,6 +119,7 @@ public Plan scan() { indexFileHandler.scan(snapshot, indexFileFilter).stream() .map(IndexManifestEntry::indexFile) .collect(Collectors.toList()); + allIndexFiles = GlobalIndexSchemaCompatibility.filterCompatible(table, allIndexFiles); String vectorIndexType = vectorIndexType(allIndexFiles); if (vectorIndexType == null) { vectorIndexType = configuredVectorIndexType(); 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/test/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdReassignerTest.java b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdReassignerTest.java index 08b35b7bdfc0..c0aa694abad6 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 @@ -1538,6 +1538,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); @@ -1556,6 +1557,11 @@ 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.indexFile().globalIndexMeta().buildSchemaId()) + .isEqualTo(buildSchemaId)); Predicate predicate = new PredicateBuilder(table.rowType()).equal(table.rowType().getFieldIndex("id"), 4); @@ -2787,7 +2793,8 @@ private void setGlobalIndexSourceMeta(FileStoreTable table, long scanSnapshotId) globalIndex.indexFieldId(), globalIndex.extraFieldIds(), globalIndex.indexMeta(), - sourceMeta)))); + sourceMeta, + globalIndex.buildSchemaId())))); } replaceLatestSnapshotIndexManifest( table, latest, indexManifestFile.writeWithoutRolling(rewritten)); @@ -2818,7 +2825,9 @@ private void replaceGlobalIndexRangesWithPartitionSpanningRanges(FileStoreTable staleRowRange.to, globalIndex.indexFieldId(), globalIndex.extraFieldIds(), - globalIndex.indexMeta()); + globalIndex.indexMeta(), + globalIndex.sourceMeta(), + globalIndex.buildSchemaId()); IndexFileMeta indexFile = entry.indexFile(); rewritten.add( new IndexManifestEntry( @@ -2875,7 +2884,9 @@ private void appendGlobalIndexRange(FileStoreTable table, String partition, Rang rowRange.to, globalIndex.indexFieldId(), globalIndex.extraFieldIds(), - globalIndex.indexMeta())))); + globalIndex.indexMeta(), + globalIndex.sourceMeta(), + globalIndex.buildSchemaId())))); replaceLatestSnapshotIndexManifest( table, latest, indexManifestFile.writeWithoutRolling(entries)); } 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..730a8e2e91b6 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).globalIndexMeta().buildSchemaId()).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/index/IndexFileMetaSerializerTest.java b/paimon-core/src/test/java/org/apache/paimon/index/IndexFileMetaSerializerTest.java index 33373741c667..3b67f429b6bc 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 @@ -33,7 +33,7 @@ public class IndexFileMetaSerializerTest extends ObjectSerializerTestBase { @Test - void testGlobalIndexSourceMetaRoundTrip() { + void testGlobalIndexMetadataRoundTrip() { IndexFileMetaSerializer serializer = new IndexFileMetaSerializer(); IndexFileMeta indexFile = new IndexFileMeta( @@ -41,7 +41,8 @@ void testGlobalIndexSourceMetaRoundTrip() { "index-file", 100, 10, - new GlobalIndexMeta(0, 9, 7, null, new byte[] {3, 4}, new byte[] {1, 2}), + new GlobalIndexMeta( + 0, 9, 7, null, new byte[] {3, 4}, new byte[] {1, 2}, 11L), null); GlobalIndexMeta restored = @@ -49,6 +50,7 @@ void testGlobalIndexSourceMetaRoundTrip() { assertThat(restored.sourceMeta()).containsExactly(1, 2); assertThat(restored.indexMeta()).containsExactly(3, 4); + assertThat(restored.buildSchemaId()).isEqualTo(11L); } @Test @@ -65,8 +67,16 @@ void testEqualityIncludesGlobalIndexMeta() { globalIndexFile( new GlobalIndexMeta( 0, 9, 7, new int[] {8}, new byte[] {3}, new byte[] {2})); + IndexFileMeta differentBuildSchema = + globalIndexFile( + new GlobalIndexMeta( + 0, 9, 7, new int[] {8}, new byte[] {3}, new byte[] {1}, 1L)); - assertThat(first).isEqualTo(equal).hasSameHashCodeAs(equal).isNotEqualTo(different); + assertThat(first) + .isEqualTo(equal) + .hasSameHashCodeAs(equal) + .isNotEqualTo(different) + .isNotEqualTo(differentBuildSchema); } private static IndexFileMeta globalIndexFile(GlobalIndexMeta globalIndexMeta) { 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..5aeff7ec03dd 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 @@ -55,16 +55,17 @@ void testReadsGlobalIndexWithoutSourceMeta() { InternalRow serialized = serializer.toRow(entry); assertThat(serialized.getInt(0)).isEqualTo(1); assertThat(serialized.getRow(10, GlobalIndexMeta.SCHEMA.getFieldCount()).getFieldCount()) - .isEqualTo(6); + .isEqualTo(7); GlobalIndexMeta restored = serializer.fromRow(serialized).indexFile().globalIndexMeta(); assertThat(restored.indexMeta()).containsExactly(1); assertThat(restored.sourceMeta()).isNull(); + assertThat(restored.buildSchemaId()).isNull(); } @Test - void testGlobalIndexSourceMetaRoundTrip() throws IOException { + void testGlobalIndexMetadataRoundTrip() throws IOException { IndexManifestEntrySerializer serializer = new IndexManifestEntrySerializer(); IndexManifestEntry entry = new IndexManifestEntry( @@ -77,7 +78,7 @@ void testGlobalIndexSourceMetaRoundTrip() throws IOException { 100, 10, new GlobalIndexMeta( - 0, 9, 7, null, new byte[] {3, 4}, new byte[] {1, 2}), + 0, 9, 7, null, new byte[] {3, 4}, new byte[] {1, 2}, 11L), null)); assertThat(serializer.toRow(entry).getInt(0)).isEqualTo(1); @@ -89,6 +90,7 @@ void testGlobalIndexSourceMetaRoundTrip() throws IOException { assertThat(restored.indexMeta()).containsExactly(3, 4); assertThat(restored.sourceMeta()).containsExactly(1, 2); + assertThat(restored.buildSchemaId()).isEqualTo(11L); } @Override 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..2e908648379b 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 @@ -49,8 +49,80 @@ public class ManifestCommittableSerializerCompatibilityTest { private static final String GENERATE_GOLDEN_FILES_PROPERTY = "generateManifestCommittableGoldenFiles"; + @Test + public void testCompatibilityToV5CommitV14() throws IOException { + ManifestCommittable committable = + createCurrentCommitCommittable(new GlobalIndexMeta(0, 9, 7, null, null, null, 11L)); + + ManifestCommittableSerializer serializer = new ManifestCommittableSerializer(); + byte[] current = serializer.serialize(committable); + byte[] serialized; + if (Boolean.parseBoolean( + System.getProperties().getProperty(GENERATE_GOLDEN_FILES_PROPERTY))) { + CompatibilityUtils.writeCompatibilityFile("manifest-committable-v14-v5", current); + serialized = current; + } else { + serialized = + IOUtils.readFully( + ManifestCommittableSerializerCompatibilityTest.class + .getClassLoader() + .getResourceAsStream( + "compatibility/manifest-committable-v14-v5"), + true); + } + + assertThat(current).isEqualTo(serialized); + 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); + + assertThat(new ManifestCommittableSerializer().deserialize(5, serialized)) + .isEqualTo(createCurrentCommitCommittable(null)); + } + + @Test + public void testCompatibilityToV5CommitV13WithGlobalIndex() throws IOException { + byte[] serialized = + IOUtils.readFully( + ManifestCommittableSerializerCompatibilityTest.class + .getClassLoader() + .getResourceAsStream( + "compatibility/manifest-committable-v13-global-index-v5"), + true); + GlobalIndexMeta expectedGlobalIndex = + new GlobalIndexMeta( + 0L, + 9L, + 7, + new int[] {8, 9}, + new byte[] {0x12, 0x34}, + new byte[] {0x56, 0x78}, + null); + + ManifestCommittable deserialized = + new ManifestCommittableSerializer().deserialize(5, serialized); + assertThat(deserialized).isEqualTo(createCurrentCommitCommittable(expectedGlobalIndex)); + GlobalIndexMeta actualGlobalIndex = + ((CommitMessageImpl) deserialized.fileCommittables().get(0)) + .newFilesIncrement() + .newIndexFiles() + .get(0) + .globalIndexMeta(); + assertThat(actualGlobalIndex).isEqualTo(expectedGlobalIndex); + assertThat(actualGlobalIndex.sourceMeta()).containsExactly(0x56, 0x78); + assertThat(actualGlobalIndex.buildSchemaId()).isNull(); + } + + private static ManifestCommittable createCurrentCommitCommittable( + GlobalIndexMeta globalIndexMeta) { DataFileMeta dataFile = DataFileMeta.create( "column-sequence-file", @@ -77,31 +149,11 @@ public void testCompatibilityToV5CommitV13() throws IOException { null) .withColumnMaxSequenceNumbers(new long[] {3L, 5L}); IndexFileMeta indexFile = - new IndexFileMeta( - "index-type", "index-file", 100L, 10L, (GlobalIndexMeta) null, null); + new IndexFileMeta("index-type", "index-file", 100L, 10L, globalIndexMeta, null); ManifestCommittable committable = createManifestCommittable( Collections.singletonList(dataFile), indexFile, indexFile); - - ManifestCommittableSerializer serializer = new ManifestCommittableSerializer(); - byte[] current = serializer.serialize(committable); - byte[] serialized; - if (Boolean.parseBoolean( - System.getProperties().getProperty(GENERATE_GOLDEN_FILES_PROPERTY))) { - CompatibilityUtils.writeCompatibilityFile("manifest-committable-v13-v5", current); - serialized = current; - } else { - serialized = - IOUtils.readFully( - ManifestCommittableSerializerCompatibilityTest.class - .getClassLoader() - .getResourceAsStream( - "compatibility/manifest-committable-v13-v5"), - true); - } - - assertThat(current).isEqualTo(serialized); - assertThat(serializer.deserialize(5, serialized)).isEqualTo(committable); + return committable; } @Test 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/BtreeGlobalIndexTableTest.java b/paimon-core/src/test/java/org/apache/paimon/table/BtreeGlobalIndexTableTest.java index 0349b3290a53..2feded08603b 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/BtreeGlobalIndexTableTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/BtreeGlobalIndexTableTest.java @@ -592,7 +592,8 @@ public void testDataEvolutionSourceBackedIndexParticipatesInGlobalRowIdScan() th "source-backed-index", 0, 10, - new GlobalIndexMeta(0, 9, 1, null, null, new byte[] {1}), + new GlobalIndexMeta( + 0, 9, 1, null, null, new byte[] {1}, table.schema().id()), null); assertThat( @@ -623,7 +624,7 @@ public void testOrdinaryAndSourceBackedBTreeIndexCoverageCanCoexist() throws Exc "ordinary-index", 0, 5, - new GlobalIndexMeta(0, 4, 1, null, null), + new GlobalIndexMeta(0, 4, 1, null, null, null, table.schema().id()), null)); mixedIndexes.add( new IndexFileMeta( @@ -631,7 +632,8 @@ public void testOrdinaryAndSourceBackedBTreeIndexCoverageCanCoexist() throws Exc "source-backed-index", 0, 5, - new GlobalIndexMeta(5, 9, 1, null, null, new byte[] {1}), + new GlobalIndexMeta( + 5, 9, 1, null, null, new byte[] {1}, table.schema().id()), null)); DataEvolutionGlobalIndexCoverage coverage = 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..fbf39c8498b4 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,15 +21,21 @@ 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.GlobalIndexSchemaCompatibility; import org.apache.paimon.globalindex.IndexedSplit; import org.apache.paimon.globalindex.ScanResult; import org.apache.paimon.globalindex.sorted.SortedGlobalIndexScanner; import org.apache.paimon.globalindex.sorted.SortedGlobalIndexTestUtils; +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.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 +43,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 +113,89 @@ 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); + IndexFileMeta incompatibleMultiColumnIndex = + new IndexFileMeta( + "multivalue", + "incompatible-index", + 0, + 1, + new GlobalIndexMeta(0, 0, 0, new int[] {1}, null, null, firstBuildSchemaId), + null); + assertThat( + GlobalIndexSchemaCompatibility.filterCompatible( + fullSearchTable, + Collections.singletonList(incompatibleMultiColumnIndex))) + .isEmpty(); + 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(readIds(fullSearchTable, evolvedTypePredicate)).containsExactly(1, 2); + + assertThat(firstBuildSchemaId).isNotEqualTo(fullSearchTable.schema().id()); + } + + @Test + public void testIndexWithoutResolvableBuildSchemaIsIgnored() throws Exception { + createTableDefault(); + FileStoreTable table = getTableDefault(); + write(table, GenericRow.of(1, array(RED))); + IndexFileMeta legacyIndex = + new IndexFileMeta( + "multivalue", + "legacy-index", + 0, + 1, + new GlobalIndexMeta(0, 0, 1, null, null), + null); + IndexFileMeta missingSchemaIndex = + new IndexFileMeta( + "multivalue", + "missing-schema-index", + 0, + 1, + new GlobalIndexMeta(0, 0, 1, null, null, null, Long.MAX_VALUE), + null); + + assertThat( + DataEvolutionGlobalIndexScanner.create( + table, Arrays.asList(legacyIndex, missingSchemaIndex))) + .isEmpty(); + } + private void buildIndex(FileStoreTable table) throws Exception { SortedGlobalIndexScanner scanner = new SortedGlobalIndexScanner(table, "multivalue").withIndexField("tags"); @@ -148,6 +238,11 @@ private List readIdsWithFallback(FileStoreTable table, Predicate predic return readIds(table, predicate, false); } + private FileStoreTable fullSearchTable(FileStoreTable table) { + return table.copy( + Collections.singletonMap(CoreOptions.GLOBAL_INDEX_SEARCH_MODE.key(), "full")); + } + private List readIds( FileStoreTable table, Predicate predicate, boolean expectIndexedSplits) throws Exception { diff --git a/paimon-core/src/test/java/org/apache/paimon/table/sink/CommitMessageSerializerTest.java b/paimon-core/src/test/java/org/apache/paimon/table/sink/CommitMessageSerializerTest.java index bc36deedca9a..887efc6686ca 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/sink/CommitMessageSerializerTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/sink/CommitMessageSerializerTest.java @@ -18,6 +18,8 @@ package org.apache.paimon.table.sink; +import org.apache.paimon.index.GlobalIndexMeta; +import org.apache.paimon.index.IndexFileMeta; import org.apache.paimon.io.CompactIncrement; import org.apache.paimon.io.DataIncrement; @@ -49,6 +51,16 @@ public void test() throws IOException { .get(0) .withColumnMaxSequenceNumbers(new long[] {3L, 42L})); dataIncrement.newIndexFiles().addAll(Arrays.asList(randomIndexFile(), randomIndexFile())); + dataIncrement + .newIndexFiles() + .add( + new IndexFileMeta( + "btree", + "global-index-file", + 100, + 10, + new GlobalIndexMeta(0, 9, 7, null, null, null, 11L), + null)); dataIncrement .deletedIndexFiles() .addAll(Arrays.asList(randomIndexFile(), randomIndexFile())); 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..eaaa5d7e9d82 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; @@ -277,6 +280,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 +930,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 +945,7 @@ public void testFullTextSearchSplitSerialization() throws Exception { } assertThat(rawDeserialized.rowRanges()).isEqualTo(rawOriginal.rowRanges()); + assertThat(rawDeserialized.indexType()).isEqualTo(rawOriginal.indexType()); } // ====================== Helper methods ====================== @@ -962,6 +1014,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 buildSchemaId) + throws Exception { Options options = table.coreOptions().toConfiguration(); DataField textField = table.rowType().getField(TEXT_FIELD_NAME); @@ -987,7 +1048,31 @@ private void buildAndCommitIndexWithFields( indexFields, TestFullTextGlobalIndexerFactory.IDENTIFIER, entries, - null); + null, + buildSchemaId == null ? table.schema().id() : buildSchemaId); + if (buildSchemaId == null) { + List legacyIndexFiles = new ArrayList<>(); + for (IndexFileMeta indexFile : indexFiles) { + GlobalIndexMeta globalIndex = indexFile.globalIndexMeta(); + legacyIndexFiles.add( + new IndexFileMeta( + indexFile.indexType(), + indexFile.fileName(), + indexFile.fileSize(), + indexFile.rowCount(), + indexFile.dvRanges(), + indexFile.externalPath(), + new GlobalIndexMeta( + globalIndex.rowRangeStart(), + globalIndex.rowRangeEnd(), + globalIndex.indexFieldId(), + globalIndex.extraFieldIds(), + globalIndex.indexMeta(), + globalIndex.sourceMeta(), + null))); + } + indexFiles = legacyIndexFiles; + } DataIncrement dataIncrement = DataIncrement.indexIncrement(indexFiles); CommitMessage message = @@ -1002,6 +1087,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 +1135,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)) @@ -1046,7 +1156,8 @@ private void buildAndCommitSourceBackedIndex(FileStoreTable table, String[] docu meta.indexFieldId(), meta.extraFieldIds(), meta.indexMeta(), - sourceMeta), + sourceMeta, + meta.buildSchemaId()), indexFile.externalPath())); } @@ -1098,7 +1209,8 @@ private void buildAndCommitIndexRange( indexFields, TestFullTextGlobalIndexerFactory.IDENTIFIER, entries, - null); + null, + table.schema().id()); DataIncrement dataIncrement = DataIncrement.indexIncrement(indexFiles); CommitMessage message = @@ -1176,7 +1288,8 @@ private void buildAndCommitIndexForColumn( rowRange, textField.id(), TestFullTextGlobalIndexerFactory.IDENTIFIER, - entries); + entries, + table.schema().id()); DataIncrement dataIncrement = DataIncrement.indexIncrement(indexFiles); CommitMessage message = @@ -1210,7 +1323,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 +1376,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 +1400,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..619bf3fac8f3 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 @@ -1821,7 +1821,8 @@ private void buildAndCommitIndex(FileStoreTable table, String fieldName, float[] rowRange, vectorField.id(), TestVectorGlobalIndexerFactory.IDENTIFIER, - entries); + entries, + table.schema().id()); DataIncrement dataIncrement = DataIncrement.indexIncrement(indexFiles); CommitMessage message = @@ -1863,7 +1864,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 +1888,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 +2063,8 @@ private void buildAndCommitVectorIndexWithFields( indexFields, TestVectorGlobalIndexerFactory.IDENTIFIER, entries, - null); + null, + table.schema().id()); DataIncrement dataIncrement = DataIncrement.indexIncrement(indexFiles); CommitMessage message = @@ -2098,7 +2102,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 +2144,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 0000000000000000000000000000000000000000..b2201de23e689676d3facba6e78cca1a6313522a GIT binary patch literal 3362 zcmeHIyH3L}6unJBl$V4MNQ?}uk)UE=%*e>Z+J@FhQ5v^Eijol|1S5aI+L2FS>@TqK z8H}8JZPQ4wfV@rS_N`^-o)XDeHq_ibfDurT9#L=$q-}5;u}z=>aGqJl2Iegw zD?6k3I#2Rss7;2}c@gKpjW?pG$%!4{m5n8yh}H~AdHe)}d$ z)PnIOiyq!z++vpm{}Pg{i|@gEhQmAo_lARUro$U`HCDQ(!pmf!WJ9G?mgr8Z%Euq^ zxTwY@%@?%~FdqVZ-U#^X@Kqqzu`-X{Iso&Ap{rjrp-m88`h|t3I zBTYn1(44HJk>xXRAK>Bi)p)a+^|cEmy03~bFGp70j9*c^*%q-@t-oTK1^xusK9<1! vd12Unm#lMk5A&!Z~Fzx_l z-5clEm8|4QnF@7y5!b-2ccQG#ggBybJ%VE%#BL4 zj^t46a3G*wsKN?O-mEMS}iBNr!Udo}+Kk! allIndexFiles = new java.util.ArrayList<>(); diff --git a/paimon-lumina/src/test/java/org/apache/paimon/lumina/index/LuminaVectorGlobalIndexScanTest.java b/paimon-lumina/src/test/java/org/apache/paimon/lumina/index/LuminaVectorGlobalIndexScanTest.java index 85d4ebe6be98..451f42bf09c9 100644 --- a/paimon-lumina/src/test/java/org/apache/paimon/lumina/index/LuminaVectorGlobalIndexScanTest.java +++ b/paimon-lumina/src/test/java/org/apache/paimon/lumina/index/LuminaVectorGlobalIndexScanTest.java @@ -223,7 +223,14 @@ public PositionOutputStream newOutputStream(String fileName) for (ResultEntry entry : entries) { long fileSize = fileIO.getFileSize(new Path(indexDir, entry.fileName())); GlobalIndexMeta globalMeta = - new GlobalIndexMeta(0, vectors.length - 1, fieldId, null, entry.meta()); + new GlobalIndexMeta( + 0, + vectors.length - 1, + fieldId, + null, + entry.meta(), + null, + ipTable.schema().id()); metas.add( new IndexFileMeta( @@ -314,7 +321,14 @@ public PositionOutputStream newOutputStream(String fileName) for (ResultEntry entry : entries) { long fileSize = fileIO.getFileSize(new Path(indexDir, entry.fileName())); GlobalIndexMeta globalMeta = - new GlobalIndexMeta(0, vectors.length - 1, fieldId, null, entry.meta()); + new GlobalIndexMeta( + 0, + vectors.length - 1, + fieldId, + null, + entry.meta(), + null, + table.schema().id()); metas.add( new IndexFileMeta( 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-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 = From 0633e6785598313b4064c54579c9bf4de70babbd Mon Sep 17 00:00:00 2001 From: QuakeWang Date: Mon, 31 Aug 2026 09:19:31 +0800 Subject: [PATCH 2/3] [lumina] Set schema ID in global index scan test --- .../lumina/index/LuminaVectorGlobalIndexScanTest.java | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/paimon-lumina/src/test/java/org/apache/paimon/lumina/index/LuminaVectorGlobalIndexScanTest.java b/paimon-lumina/src/test/java/org/apache/paimon/lumina/index/LuminaVectorGlobalIndexScanTest.java index 85d4ebe6be98..b426638daf60 100644 --- a/paimon-lumina/src/test/java/org/apache/paimon/lumina/index/LuminaVectorGlobalIndexScanTest.java +++ b/paimon-lumina/src/test/java/org/apache/paimon/lumina/index/LuminaVectorGlobalIndexScanTest.java @@ -231,8 +231,10 @@ public PositionOutputStream newOutputStream(String fileName) entry.fileName(), fileSize, entry.rowCount(), + null, + null, globalMeta, - (String) null)); + ipTable.schema().id())); } DataIncrement dataIncrement = DataIncrement.indexIncrement(metas); @@ -322,8 +324,10 @@ public PositionOutputStream newOutputStream(String fileName) entry.fileName(), fileSize, entry.rowCount(), + null, + null, globalMeta, - (String) null)); + table.schema().id())); } return metas; } From fbc7c507d18efa8a7a53fe77a510dd3bfcc72f60 Mon Sep 17 00:00:00 2001 From: QuakeWang Date: Fri, 4 Sep 2026 16:28:34 +0800 Subject: [PATCH 3/3] [core][python][spark] Address global-index compatibility review --- .../DataEvolutionGlobalIndexScanner.java | 4 +- .../GlobalIndexSchemaCompatibility.java | 80 ++++++++-- .../generic/GenericGlobalIndexScanner.java | 19 ++- .../sorted/SortedGlobalIndexScanner.java | 20 ++- .../GenericGlobalIndexScannerTest.java | 111 +++++++++++++- .../sorted/SortedGlobalIndexScannerTest.java | 103 +++++++++++++ .../table/BtreeGlobalIndexTableTest.java | 13 +- .../table/MultiValueGlobalIndexTableTest.java | 17 ++- .../pypaimon/globalindex/build_plan.py | 70 ++++++++- .../globalindex/create_global_index.py | 37 ++++- .../data_evolution_global_index_scanner.py | 4 +- .../global_index_schema_compatibility.py | 73 ++++++--- .../pypaimon/index/index_file_meta.py | 2 + .../pypaimon/manifest/index_manifest_entry.py | 5 +- .../pypaimon/tests/global_index_build_test.py | 143 ++++++++++++++++++ .../global_index_schema_compatibility_test.py | 25 +++ .../tests/vector_search_filter_test.py | 5 + .../paimon/spark/copy/CopyFilesUtil.java | 49 +++++- .../spark/copy/ListIndexFilesOperator.java | 72 +++++++-- .../DefaultGlobalIndexTopoBuilder.java | 18 ++- .../spark/procedure/CopyFilesProcedure.java | 6 +- .../procedure/CopyFilesProcedureTest.scala | 127 +++++++++++++++- 22 files changed, 913 insertions(+), 90 deletions(-) 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 6c515e898a05..5821f9f619f5 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 @@ -228,7 +228,9 @@ public static Optional create( @Nullable Snapshot pinnedSnapshot, @Nullable PartitionPredicate partitionFilter, Collection indexFiles) { - List globalIndexFiles = globalIndexFiles(indexFiles); + List globalIndexFiles = + GlobalIndexSchemaCompatibility.filterCompatibleFiles( + table, globalIndexFiles(indexFiles)); if (globalIndexFiles.isEmpty()) { return Optional.empty(); } 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 index 2ae02b85ed22..6b8f55b3138f 100644 --- a/paimon-core/src/main/java/org/apache/paimon/globalindex/GlobalIndexSchemaCompatibility.java +++ b/paimon-core/src/main/java/org/apache/paimon/globalindex/GlobalIndexSchemaCompatibility.java @@ -19,6 +19,7 @@ package org.apache.paimon.globalindex; import org.apache.paimon.index.GlobalIndexMeta; +import org.apache.paimon.index.IndexFileMeta; import org.apache.paimon.manifest.IndexManifestEntry; import org.apache.paimon.table.FileStoreTable; import org.apache.paimon.types.RowType; @@ -37,16 +38,74 @@ 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<>(); + return partitionByCompatibility(table, entries).compatible(); + } + + public static CompatibilityResult partitionByCompatibility( + FileStoreTable table, Collection entries) { + CompatibilityChecker checker = new CompatibilityChecker(table); List compatible = new ArrayList<>(); + List incompatible = new ArrayList<>(); for (IndexManifestEntry entry : entries) { - GlobalIndexMeta globalIndex = entry.indexFile().globalIndexMeta(); - Long schemaId = entry.schemaId(); + if (checker.isCompatible(entry.indexFile(), entry.schemaId())) { + compatible.add(entry); + } else { + incompatible.add(entry); + } + } + return new CompatibilityResult(compatible, incompatible); + } + + public static List filterCompatibleFiles( + FileStoreTable table, Collection indexFiles) { + CompatibilityChecker checker = new CompatibilityChecker(table); + List compatible = new ArrayList<>(); + for (IndexFileMeta indexFile : indexFiles) { + if (checker.isCompatible(indexFile, indexFile.schemaId())) { + compatible.add(indexFile); + } + } + return compatible; + } + + /** Global index manifest entries grouped by compatibility with the current table schema. */ + public static final class CompatibilityResult { + + private final List compatible; + private final List incompatible; + + private CompatibilityResult( + List compatible, List incompatible) { + this.compatible = compatible; + this.incompatible = incompatible; + } + + public List compatible() { + return compatible; + } + + public List incompatible() { + return incompatible; + } + } + + private static class CompatibilityChecker { + + private final FileStoreTable table; + private final RowType currentRowType; + private final Map historicalRowTypes = new HashMap<>(); + private final Set missingSchemaIds = new HashSet<>(); + + private CompatibilityChecker(FileStoreTable table) { + this.table = table; + this.currentRowType = table.rowType(); + historicalRowTypes.put(table.schema().id(), currentRowType); + } + + private boolean isCompatible(IndexFileMeta indexFile, Long schemaId) { + GlobalIndexMeta globalIndex = indexFile.globalIndexMeta(); if (globalIndex == null || schemaId == null) { - continue; + return false; } RowType historicalRowType = historicalRowTypes.get(schemaId); @@ -59,12 +118,9 @@ public static List filterCompatible( missingSchemaIds.add(schemaId); } } - if (historicalRowType != null - && compatibleIndexedFields(globalIndex, historicalRowType, currentRowType)) { - compatible.add(entry); - } + return historicalRowType != null + && compatibleIndexedFields(globalIndex, historicalRowType, currentRowType); } - return compatible; } private static boolean compatibleIndexedFields( diff --git a/paimon-core/src/main/java/org/apache/paimon/globalindex/generic/GenericGlobalIndexScanner.java b/paimon-core/src/main/java/org/apache/paimon/globalindex/generic/GenericGlobalIndexScanner.java index 619467daad7d..5da04938ef22 100644 --- a/paimon-core/src/main/java/org/apache/paimon/globalindex/generic/GenericGlobalIndexScanner.java +++ b/paimon-core/src/main/java/org/apache/paimon/globalindex/generic/GenericGlobalIndexScanner.java @@ -21,6 +21,7 @@ import org.apache.paimon.CoreOptions; import org.apache.paimon.Snapshot; import org.apache.paimon.globalindex.DataEvolutionGlobalIndexRefreshPlanner; +import org.apache.paimon.globalindex.GlobalIndexSchemaCompatibility; import org.apache.paimon.globalindex.ScanResult; import org.apache.paimon.manifest.IndexManifestEntry; import org.apache.paimon.manifest.ManifestEntry; @@ -139,19 +140,27 @@ public Optional> incrementalScan() { List currentIndexes = currentIndexEntries( table, scanSnapshot, indexType, indexFields, partitionPredicate); + GlobalIndexSchemaCompatibility.CompatibilityResult compatibility = + GlobalIndexSchemaCompatibility.partitionByCompatibility(table, currentIndexes); + List compatibleIndexes = compatibility.compatible(); List rangesToBuild = - new ArrayList<>(unindexedRowRanges(scanSnapshot, currentIndexes)); - List deletedIndexEntries = Collections.emptyList(); + new ArrayList<>(unindexedRowRanges(scanSnapshot, compatibleIndexes)); + List deletedIndexEntries = + new ArrayList<>(compatibility.incompatible()); + for (IndexManifestEntry entry : compatibility.incompatible()) { + rangesToBuild.add(entry.indexFile().globalIndexMeta().rowRange()); + } Options mergedOptions = new Options(table.options(), options.toMap()); if (mergedOptions.get(CoreOptions.GLOBAL_INDEX_COLUMN_UPDATE_ACTION) == CoreOptions.GlobalIndexColumnUpdateAction.IGNORE) { - deletedIndexEntries = + List indexesToRefresh = DataEvolutionGlobalIndexRefreshPlanner.findIndexesToRefresh( table.schemaManager(), scanResult.entries(), - currentIndexes, + compatibleIndexes, indexFields); - for (IndexManifestEntry entry : deletedIndexEntries) { + deletedIndexEntries.addAll(indexesToRefresh); + for (IndexManifestEntry entry : indexesToRefresh) { rangesToBuild.add(entry.indexFile().globalIndexMeta().rowRange()); } } diff --git a/paimon-core/src/main/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexScanner.java b/paimon-core/src/main/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexScanner.java index 7f44a334515b..9df3bdd764eb 100644 --- a/paimon-core/src/main/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexScanner.java +++ b/paimon-core/src/main/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexScanner.java @@ -21,6 +21,7 @@ import org.apache.paimon.CoreOptions; import org.apache.paimon.Snapshot; import org.apache.paimon.globalindex.DataEvolutionGlobalIndexRefreshPlanner; +import org.apache.paimon.globalindex.GlobalIndexSchemaCompatibility; import org.apache.paimon.globalindex.ScanResult; import org.apache.paimon.manifest.IndexManifestEntry; import org.apache.paimon.options.Options; @@ -140,18 +141,27 @@ public Optional> incrementalScan() { indexType, Collections.singletonList(indexField), partitionPredicate); - List rangesToBuild = new ArrayList<>(unindexedRowRanges(snapshot, currentIndexes)); - List deletedIndexEntries = Collections.emptyList(); + GlobalIndexSchemaCompatibility.CompatibilityResult compatibility = + GlobalIndexSchemaCompatibility.partitionByCompatibility(table, currentIndexes); + List compatibleIndexes = compatibility.compatible(); + List rangesToBuild = + new ArrayList<>(unindexedRowRanges(snapshot, compatibleIndexes)); + List deletedIndexEntries = + new ArrayList<>(compatibility.incompatible()); + for (IndexManifestEntry entry : compatibility.incompatible()) { + rangesToBuild.add(entry.indexFile().globalIndexMeta().rowRange()); + } if (detectDataFileChange()) { // Scans data manifests through reusable binary views without materializing entries. - deletedIndexEntries = + List indexesToRefresh = DataEvolutionGlobalIndexRefreshPlanner.findIndexesToRefresh( table, snapshot, partitionPredicate, - currentIndexes, + compatibleIndexes, Collections.singletonList(indexField)); - for (IndexManifestEntry entry : deletedIndexEntries) { + deletedIndexEntries.addAll(indexesToRefresh); + for (IndexManifestEntry entry : indexesToRefresh) { rangesToBuild.add(entry.indexFile().globalIndexMeta().rowRange()); } } diff --git a/paimon-core/src/test/java/org/apache/paimon/globalindex/generic/GenericGlobalIndexScannerTest.java b/paimon-core/src/test/java/org/apache/paimon/globalindex/generic/GenericGlobalIndexScannerTest.java index 9b3b00c4d22d..8fa30883b033 100644 --- a/paimon-core/src/test/java/org/apache/paimon/globalindex/generic/GenericGlobalIndexScannerTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/globalindex/generic/GenericGlobalIndexScannerTest.java @@ -19,23 +19,31 @@ package org.apache.paimon.globalindex.generic; import org.apache.paimon.CoreOptions; -import org.apache.paimon.data.BinaryString; +import org.apache.paimon.data.BinaryRow; import org.apache.paimon.data.GenericRow; import org.apache.paimon.globalindex.ScanResult; +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.manifest.IndexManifestEntry; import org.apache.paimon.manifest.ManifestEntry; import org.apache.paimon.options.Options; 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; import org.apache.paimon.table.sink.BatchTableWrite; import org.apache.paimon.table.sink.BatchWriteBuilder; +import org.apache.paimon.table.sink.CommitMessageImpl; import org.apache.paimon.types.DataTypes; import org.apache.paimon.utils.Range; import org.junit.jupiter.api.Test; import java.util.Collections; +import java.util.List; import static org.assertj.core.api.Assertions.assertThat; @@ -46,7 +54,7 @@ public class GenericGlobalIndexScannerTest extends TableTestBase { public Schema schemaDefault() { return Schema.newBuilder() .column("id", DataTypes.INT()) - .column("v", DataTypes.STRING()) + .column("v", DataTypes.INT()) .option(CoreOptions.BUCKET.key(), "-1") .option(CoreOptions.ROW_TRACKING_ENABLED.key(), "true") .option(CoreOptions.DATA_EVOLUTION_ENABLED.key(), "true") @@ -87,6 +95,69 @@ public void testIncrementalScanWithoutExistingIndex() throws Exception { assertThat(scanResult.deletedIndexEntries()).isEmpty(); } + @Test + public void testIncrementalScanReplacesLegacyIndex() throws Exception { + FileStoreTable table = writeRows(); + IndexFileMeta legacyIndex = globalIndex("legacy-index", null); + commitIndexes(table, Collections.singletonList(legacyIndex), Collections.emptyList()); + + ScanResult scanResult = + new GenericGlobalIndexScanner(table) + .withIndex("test-index", Collections.singletonList("v"), new Options()) + .incrementalScan() + .orElseThrow( + () -> + new IllegalStateException( + "Expected legacy index replacement plan.")); + + assertThat(scanResult.rowRangeIndex().ranges()).containsExactly(new Range(0, 9)); + assertThat(scanResult.deletedIndexEntries()) + .extracting(entry -> entry.indexFile().fileName()) + .containsExactly("legacy-index"); + + IndexFileMeta replacement = globalIndex("replacement-index", table.schema().id()); + commitIndexes( + table, + Collections.singletonList(replacement), + scanResult.deletedIndexEntries().stream() + .map(IndexManifestEntry::indexFile) + .collect(java.util.stream.Collectors.toList())); + + List currentIndexes = + table.store() + .newIndexFileHandler() + .scan(table.snapshotManager().latestSnapshot(), "test-index"); + assertThat(currentIndexes).hasSize(1); + assertThat(currentIndexes.get(0).indexFile().fileName()).isEqualTo("replacement-index"); + assertThat(currentIndexes.get(0).schemaId()).isEqualTo(table.schema().id()); + } + + @Test + public void testIncrementalScanReplacesIndexAfterIndexedTypeChange() throws Exception { + FileStoreTable table = writeRows(); + long buildSchemaId = table.schema().id(); + IndexFileMeta oldIndex = globalIndex("old-index", buildSchemaId); + commitIndexes(table, Collections.singletonList(oldIndex), Collections.emptyList()); + + table.schemaManager().commitChanges(SchemaChange.updateColumnType("v", DataTypes.BIGINT())); + table = table.copyWithLatestSchema(); + + ScanResult scanResult = + new GenericGlobalIndexScanner(table) + .withIndex("test-index", Collections.singletonList("v"), new Options()) + .incrementalScan() + .orElseThrow( + () -> + new IllegalStateException( + "Expected incompatible index replacement plan.")); + + assertThat(table.schema().id()).isNotEqualTo(buildSchemaId); + assertThat(scanResult.rowRangeIndex().ranges()).containsExactly(new Range(0, 9)); + assertThat(scanResult.deletedIndexEntries()) + .extracting(entry -> entry.indexFile().fileName()) + .containsExactly("old-index"); + } + @Test public void testScanEmptyTable() throws Exception { createTableDefault(); @@ -100,7 +171,7 @@ private FileStoreTable writeRows() throws Exception { BatchWriteBuilder builder = table.newBatchWriteBuilder(); try (BatchTableWrite write = builder.newWrite()) { for (int i = 0; i < 10; i++) { - write.write(GenericRow.of(i, BinaryString.fromString("v-" + i))); + write.write(GenericRow.of(i, i)); } try (BatchTableCommit commit = builder.newCommit()) { commit.commit(write.prepareCommit()); @@ -108,4 +179,38 @@ private FileStoreTable writeRows() throws Exception { } return table; } + + private IndexFileMeta globalIndex(String fileName, Long schemaId) { + return new IndexFileMeta( + "test-index", + fileName, + 1L, + 10L, + null, + null, + new GlobalIndexMeta(0, 9, 1, null, null), + schemaId); + } + + private void commitIndexes( + FileStoreTable table, List additions, List deletions) + throws Exception { + DataIncrement increment = + new DataIncrement( + Collections.emptyList(), + Collections.emptyList(), + Collections.emptyList(), + additions, + deletions); + try (BatchTableCommit commit = table.newBatchWriteBuilder().newCommit()) { + commit.commit( + Collections.singletonList( + new CommitMessageImpl( + BinaryRow.EMPTY_ROW, + 0, + null, + increment, + CompactIncrement.emptyIncrement()))); + } + } } diff --git a/paimon-core/src/test/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexScannerTest.java b/paimon-core/src/test/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexScannerTest.java index 524f54a85c46..d563c0a6ee5b 100644 --- a/paimon-core/src/test/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexScannerTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexScannerTest.java @@ -35,7 +35,9 @@ import org.apache.paimon.index.GlobalIndexMeta; import org.apache.paimon.index.IndexFileHandler; import org.apache.paimon.index.IndexFileMeta; +import org.apache.paimon.io.CompactIncrement; import org.apache.paimon.io.DataFileMeta; +import org.apache.paimon.io.DataIncrement; import org.apache.paimon.manifest.IndexManifestEntry; import org.apache.paimon.manifest.ManifestEntry; import org.apache.paimon.memory.MemorySlice; @@ -247,6 +249,51 @@ public void testIncrementalScanAllIndexed() throws Exception { "incrementalScan should return empty when all data is already indexed"); } + @Test + public void testIncrementalScanReplacesLegacyIndex() throws Exception { + FileStoreTable table = writeSinglePartitionRows(10); + ManifestEntry dataEntry = table.store().newScan().plan().files().get(0); + IndexFileMeta legacyIndex = globalIndex("legacy-index", null); + commitIndexes( + table, + dataEntry.partition(), + dataEntry.bucket(), + Collections.singletonList(legacyIndex), + Collections.emptyList()); + + ScanResult scanResult = + new SortedGlobalIndexScanner(table, "btree") + .withIndexField("f0") + .incrementalScan() + .orElseThrow( + () -> + new IllegalStateException( + "Expected legacy index replacement plan.")); + + assertThat(scanResult.rowRangeIndex().ranges()).containsExactly(new Range(0, 9)); + assertThat(scanResult.deletedIndexEntries()) + .extracting(entry -> entry.indexFile().fileName()) + .containsExactly("legacy-index"); + + IndexFileMeta replacement = globalIndex("replacement-index", table.schema().id()); + commitIndexes( + table, + dataEntry.partition(), + dataEntry.bucket(), + Collections.singletonList(replacement), + scanResult.deletedIndexEntries().stream() + .map(IndexManifestEntry::indexFile) + .collect(Collectors.toList())); + + List currentIndexes = + table.store() + .newIndexFileHandler() + .scan(table.snapshotManager().latestSnapshot(), "btree"); + assertThat(currentIndexes).hasSize(1); + assertThat(currentIndexes.get(0).indexFile().fileName()).isEqualTo("replacement-index"); + assertThat(currentIndexes.get(0).schemaId()).isEqualTo(table.schema().id()); + } + @Test public void testIncrementalScanWithNewData() throws Exception { write(); @@ -352,6 +399,62 @@ private SortedGlobalIndexScanner dataEvolutionScanner(FileStoreTable table) { return new SortedGlobalIndexScanner(table, "btree", options); } + private FileStoreTable writeSinglePartitionRows(int rowCount) throws Exception { + createTableDefault(); + FileStoreTable table = getTableDefault(); + BatchWriteBuilder builder = table.newBatchWriteBuilder(); + try (BatchTableWrite write = builder.newWrite(); + BatchTableCommit commit = builder.newCommit()) { + for (int i = 0; i < rowCount; i++) { + write.write( + GenericRow.of( + BinaryString.fromString("p0"), + i, + BinaryString.fromString("f1_" + i))); + } + commit.commit(write.prepareCommit()); + } + return table; + } + + private IndexFileMeta globalIndex(String fileName, Long schemaId) { + return new IndexFileMeta( + "btree", + fileName, + 1L, + 10L, + null, + null, + new GlobalIndexMeta(0, 9, 1, null, null), + schemaId); + } + + private void commitIndexes( + FileStoreTable table, + BinaryRow partition, + int bucket, + List additions, + List deletions) + throws Exception { + DataIncrement increment = + new DataIncrement( + Collections.emptyList(), + Collections.emptyList(), + Collections.emptyList(), + additions, + deletions); + try (BatchTableCommit commit = table.newBatchWriteBuilder().newCommit()) { + commit.commit( + Collections.singletonList( + new CommitMessageImpl( + partition, + bucket, + null, + increment, + CompactIncrement.emptyIncrement()))); + } + } + private DataFileMeta updateColumnAndCompact(String column, int updateRound) throws Exception { Map writeOptions = new HashMap<>(); writeOptions.put( diff --git a/paimon-core/src/test/java/org/apache/paimon/table/BtreeGlobalIndexTableTest.java b/paimon-core/src/test/java/org/apache/paimon/table/BtreeGlobalIndexTableTest.java index 0349b3290a53..7b93044c9466 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/BtreeGlobalIndexTableTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/BtreeGlobalIndexTableTest.java @@ -588,12 +588,13 @@ public void testDataEvolutionSourceBackedIndexParticipatesInGlobalRowIdScan() th Snapshot snapshot = table.snapshotManager().latestSnapshot(); IndexFileMeta sourceBacked = new IndexFileMeta( - "btree", - "source-backed-index", - 0, - 10, - new GlobalIndexMeta(0, 9, 1, null, null, new byte[] {1}), - null); + "btree", + "source-backed-index", + 0, + 10, + new GlobalIndexMeta(0, 9, 1, null, null, new byte[] {1}), + null) + .withSchemaId(table.schema().id()); assertThat( DataEvolutionGlobalIndexScanner.create( 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 9817b2ec70ce..7016891e8c7b 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 @@ -26,8 +26,10 @@ import org.apache.paimon.globalindex.ScanResult; import org.apache.paimon.globalindex.sorted.SortedGlobalIndexScanner; import org.apache.paimon.globalindex.sorted.SortedGlobalIndexTestUtils; +import org.apache.paimon.index.IndexFileMeta; import org.apache.paimon.io.CompactIncrement; import org.apache.paimon.io.DataIncrement; +import org.apache.paimon.manifest.IndexManifestEntry; import org.apache.paimon.predicate.Predicate; import org.apache.paimon.predicate.PredicateBuilder; import org.apache.paimon.schema.NestedSchemaUtils; @@ -117,6 +119,10 @@ public void testIndexCompatibilityAcrossSchemaEvolution() throws Exception { write(table, GenericRow.of(1, array(-1))); long firstBuildSchemaId = table.schema().id(); buildIndex(table); + List firstIndexFiles = + table.store().newIndexFileHandler().scanEntries().stream() + .map(IndexManifestEntry::indexFile) + .collect(java.util.stream.Collectors.toList()); catalog.alterTable(identifier(), SchemaChange.addColumn("note", DataTypes.STRING()), false); table = (FileStoreTable) catalog.getTable(identifier()); @@ -134,6 +140,7 @@ public void testIndexCompatibilityAcrossSchemaEvolution() throws Exception { table.schemaManager().commitChanges(schemaChanges); table = table.copyWithLatestSchema(); write(table, GenericRow.of(2, array(-1L), null)); + assertThat(DataEvolutionGlobalIndexScanner.create(table, firstIndexFiles)).isEmpty(); buildIndex(table); fullSearchTable = fullSearchTable(table.copyWithLatestSchema()); @@ -143,13 +150,17 @@ public void testIndexCompatibilityAcrossSchemaEvolution() throws Exception { 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)); + .containsExactly(new Range(0, 1)); + assertThat(scanner.unindexedRows(evolvedTypePredicate).results().isEmpty()).isTrue(); } assertThat(readIdsWithoutSplitAssertion(fullSearchTable, evolvedTypePredicate)) .containsExactly(1, 2); assertThat(firstBuildSchemaId).isNotEqualTo(fullSearchTable.schema().id()); + long currentSchemaId = fullSearchTable.schema().id(); + assertThat(fullSearchTable.store().newIndexFileHandler().scanEntries()) + .isNotEmpty() + .allSatisfy(entry -> assertThat(entry.schemaId()).isEqualTo(currentSchemaId)) + .noneMatch(entry -> firstIndexFiles.contains(entry.indexFile())); } private void buildIndex(FileStoreTable table) throws Exception { diff --git a/paimon-python/pypaimon/globalindex/build_plan.py b/paimon-python/pypaimon/globalindex/build_plan.py index 74893d16b248..a8d08b94fe6c 100644 --- a/paimon-python/pypaimon/globalindex/build_plan.py +++ b/paimon-python/pypaimon/globalindex/build_plan.py @@ -17,7 +17,7 @@ """Reusable global index build planning helpers.""" -from typing import List, Optional, Sequence +from typing import List, Optional, Sequence, Tuple from pypaimon.globalindex.indexed_split import IndexedSplit from pypaimon.read.split import DataSplit @@ -65,9 +65,32 @@ def indexed_row_ranges( index_field_id: int, index_type: str, ) -> List[Range]: + from pypaimon.globalindex.global_index_schema_compatibility import ( + filter_compatible_global_indexes, + ) + + entries = current_index_entries( + table, snapshot, partition_filter, index_field_id, index_type) + ranges = [ + Range( + entry.index_file.global_index_meta.row_range_start, + entry.index_file.global_index_meta.row_range_end, + ) + for entry in filter_compatible_global_indexes(table, entries) + ] + return Range.sort_and_merge_overlap(ranges, True) + + +def current_index_entries( + table, + snapshot, + partition_filter, + index_field_id: int, + index_type: str, +) -> List: from pypaimon.index.index_file_handler import IndexFileHandler - ranges = [] + entries = [] for entry in IndexFileHandler(table).scan(snapshot): if getattr(entry, "kind", 0) != 0: continue @@ -84,8 +107,47 @@ def indexed_row_ranges( or meta.extra_field_ids ): continue - ranges.append(Range(meta.row_range_start, meta.row_range_end)) - return Range.sort_and_merge_overlap(ranges, True) + entries.append(entry) + return entries + + +def index_rebuild_plan( + table, + snapshot, + partition_filter, + index_field_id: int, + index_type: str, +) -> Tuple[List[Range], List]: + from pypaimon.globalindex.global_index_schema_compatibility import ( + partition_global_indexes_by_compatibility, + ) + + current_indexes = current_index_entries( + table, snapshot, partition_filter, index_field_id, index_type) + compatible, incompatible = partition_global_indexes_by_compatibility( + table, current_indexes) + + next_row_id = getattr(snapshot, "next_row_id", None) + ranges_to_build = [] + if snapshot is not None and next_row_id is not None and next_row_id > 0: + indexed_ranges = [ + Range( + entry.index_file.global_index_meta.row_range_start, + entry.index_file.global_index_meta.row_range_end, + ) + for entry in compatible + ] + ranges_to_build.extend( + Range(0, next_row_id - 1).exclude( + Range.sort_and_merge_overlap(indexed_ranges, True))) + ranges_to_build.extend( + Range( + entry.index_file.global_index_meta.row_range_start, + entry.index_file.global_index_meta.row_range_end, + ) + for entry in incompatible + ) + return Range.sort_and_merge_overlap(ranges_to_build, True), incompatible def split_by_contiguous_row_range(splits): diff --git a/paimon-python/pypaimon/globalindex/create_global_index.py b/paimon-python/pypaimon/globalindex/create_global_index.py index e470a10dcd87..c8abb30ce573 100644 --- a/paimon-python/pypaimon/globalindex/create_global_index.py +++ b/paimon-python/pypaimon/globalindex/create_global_index.py @@ -35,9 +35,9 @@ ) from pypaimon.globalindex.build_plan import ( filter_non_indexable_splits as _filter_non_indexable_splits, + index_rebuild_plan as _index_rebuild_plan, split_by_contiguous_unindexed_row_range as _split_by_contiguous_unindexed_row_range, split_by_global_index_shard as _split_by_global_index_shard, - unindexed_row_ranges as _unindexed_row_ranges, ) from pypaimon.globalindex.global_index_meta import GlobalIndexMeta from pypaimon.globalindex.key_serializer import create_serializer @@ -167,7 +167,7 @@ def build(self) -> List[CommitMessage]: index_field = self._table.field_dict[self._index_columns[0]] snapshot = self._snapshot_for_plan(plan) - unindexed_ranges = _unindexed_row_ranges( + unindexed_ranges, incompatible_indexes = _index_rebuild_plan( self._table, snapshot, partition_filter, @@ -195,10 +195,13 @@ def build(self) -> List[CommitMessage]: index_path = index_path_factory.global_index_root_path() if self._index_type in _SORTED_INDEX_IDENTIFIERS: - return self._build_sorted_index( + messages = self._build_sorted_index( splits, unindexed_ranges, index_field, table_read, index_path) - return self._build_generic_index( - splits, unindexed_ranges, index_field, table_read, index_path) + else: + messages = self._build_generic_index( + splits, unindexed_ranges, index_field, table_read, index_path) + messages.extend(_index_delete_messages(incompatible_indexes)) + return messages def _snapshot_for_plan(self, plan): snapshot_id = getattr(plan, "snapshot_id", None) @@ -506,3 +509,27 @@ def _to_index_manifest_entries( ) ) return entries + + +def _index_delete_messages(entries) -> List[CommitMessage]: + by_partition = {} + for entry in entries: + partition = tuple(entry.partition.values) + by_partition.setdefault(partition, []).append( + IndexManifestEntry( + kind=1, + partition=entry.partition, + bucket=entry.bucket, + index_file=entry.index_file, + schema_id=entry.schema_id, + ) + ) + return [ + CommitMessage( + partition=partition, + bucket=0, + new_files=[], + index_deletes=deletes, + ) + for partition, deletes in by_partition.items() + ] 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 6bf82aa1a6ff..4efd7e18dbd4 100644 --- a/paimon-python/pypaimon/globalindex/data_evolution_global_index_scanner.py +++ b/paimon-python/pypaimon/globalindex/data_evolution_global_index_scanner.py @@ -28,6 +28,7 @@ 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_index_files, filter_compatible_global_indexes, ) from pypaimon.common.options.core_options import CoreOptions @@ -164,7 +165,8 @@ def create(table, index_files=None, partition_filter=None, predicate=None, from pypaimon.index.index_file_handler import IndexFileHandler if index_files is not None: - index_files = _supported_scalar_index_files(index_files) + index_files = filter_compatible_global_index_files( + table, _supported_scalar_index_files(index_files)) if len(index_files) == 0: return None core_options = _core_options(table) diff --git a/paimon-python/pypaimon/globalindex/global_index_schema_compatibility.py b/paimon-python/pypaimon/globalindex/global_index_schema_compatibility.py index 76a6b379701c..ddc4d25ce389 100644 --- a/paimon-python/pypaimon/globalindex/global_index_schema_compatibility.py +++ b/paimon-python/pypaimon/globalindex/global_index_schema_compatibility.py @@ -17,49 +17,78 @@ """Validates global indexes against the current table schema.""" -from typing import Collection, Dict, List, Set +from typing import Collection, Dict, List, Set, Tuple 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 = {} + return partition_global_indexes_by_compatibility(table, entries)[0] + + +def partition_global_indexes_by_compatibility( + table, entries: Collection +) -> Tuple[List, List]: + """Group manifest entries by compatibility with the current schema.""" + checker = _CompatibilityChecker(table) compatible = [] + incompatible = [] for entry in entries: - global_index = entry.index_file.global_index_meta - schema_id = entry.schema_id + target = (compatible if checker.is_compatible( + entry.index_file, entry.schema_id) else incompatible) + target.append(entry) + return compatible, incompatible + + +def filter_compatible_global_index_files(table, index_files: Collection) -> List: + """Keep index files compatible with the current schema.""" + checker = _CompatibilityChecker(table) + return [ + index_file for index_file in index_files + if checker.is_compatible( + index_file, getattr(index_file, "schema_id", None)) + ] + + +class _CompatibilityChecker: + + def __init__(self, table): + self._table = table + current_schema = table.table_schema + self._current_fields = _fields_by_id(current_schema.fields) + self._historical_fields = { + current_schema.id: self._current_fields, + } + self._missing_schema_ids: Set[int] = set() + self._compatibility_cache = {} + + def is_compatible(self, index_file, schema_id) -> bool: + global_index = index_file.global_index_meta if global_index is None or schema_id is None: - continue + return False - 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) + fields = self._historical_fields.get(schema_id) + if fields is None and schema_id not in self._missing_schema_ids: + historical_schema = self._table.schema_manager.get_schema(schema_id) if historical_schema is None: - missing_schema_ids.add(schema_id) + self._missing_schema_ids.add(schema_id) else: fields = _fields_by_id(historical_schema.fields) - historical_fields[schema_id] = fields + self._historical_fields[schema_id] = fields if fields is None: - continue + return False 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 + if cache_key not in self._compatibility_cache: + self._compatibility_cache[cache_key] = _compatible_indexed_fields( + field_ids, fields, self._current_fields) + return self._compatibility_cache[cache_key] def _compatible_indexed_fields(field_ids, historical_fields, current_fields): diff --git a/paimon-python/pypaimon/index/index_file_meta.py b/paimon-python/pypaimon/index/index_file_meta.py index c0825f6b7dfb..4993787ba185 100644 --- a/paimon-python/pypaimon/index/index_file_meta.py +++ b/paimon-python/pypaimon/index/index_file_meta.py @@ -41,6 +41,8 @@ class IndexFileMeta: external_path: Optional[str] = None # For global index global_index_meta: Optional[GlobalIndexMeta] = None + # Transported from the top-level index manifest entry; not persisted here. + schema_id: Optional[int] = None def __eq__(self, other): if not isinstance(other, IndexFileMeta): diff --git a/paimon-python/pypaimon/manifest/index_manifest_entry.py b/paimon-python/pypaimon/manifest/index_manifest_entry.py index 38c9dd4950d8..35ffa97c8f6f 100644 --- a/paimon-python/pypaimon/manifest/index_manifest_entry.py +++ b/paimon-python/pypaimon/manifest/index_manifest_entry.py @@ -15,7 +15,7 @@ # specific language governing permissions and limitations # under the License. -from dataclasses import dataclass +from dataclasses import dataclass, replace from typing import Optional from pypaimon.index.index_file_meta import IndexFileMeta @@ -32,6 +32,9 @@ class IndexManifestEntry: index_file: IndexFileMeta schema_id: Optional[int] = None + def __post_init__(self): + self.index_file = replace(self.index_file, schema_id=self.schema_id) + def __eq__(self, other): if not isinstance(other, IndexManifestEntry): return False diff --git a/paimon-python/pypaimon/tests/global_index_build_test.py b/paimon-python/pypaimon/tests/global_index_build_test.py index 6f58a2e5dff9..fcbb9b5b4533 100644 --- a/paimon-python/pypaimon/tests/global_index_build_test.py +++ b/paimon-python/pypaimon/tests/global_index_build_test.py @@ -16,6 +16,7 @@ # under the License. import unittest +from dataclasses import replace from datetime import date, datetime from decimal import Decimal import os @@ -47,6 +48,8 @@ ) from pypaimon.globalindex.data_evolution_global_index_scanner import DataEvolutionGlobalIndexScanner from pypaimon.index.index_file_handler import IndexFileHandler +from pypaimon.manifest.index_manifest_entry import IndexManifestEntry +from pypaimon.schema.schema_change import SchemaChange from pypaimon.schema.data_types import ArrayType, AtomicType, RowType from pypaimon.tests.data_evolution_test_helpers import ( BatchModeMixin, @@ -54,6 +57,7 @@ ) from pypaimon.table.row.generic_row import GenericRow from pypaimon.utils.range import Range +from pypaimon.write.commit_message import CommitMessage class _FakeFile: @@ -521,6 +525,145 @@ def test_create_global_index_skips_existing_ranges(self): table.snapshot_manager().get_latest_snapshot())), ) + def test_create_global_index_replaces_legacy_schema_id(self): + table = self._create_table() + self._write_arrow(table, pa.table( + { + 'id': [3, 1, 2, 2], + 'name': ['c', 'a', 'b1', 'b2'], + 'age': [30, 10, 20, 21], + 'city': ['z', 'x', 'y', 'y2'], + }, + schema=self.pa_schema, + )) + options = {'sorted-index.records-per-range': '2'} + self.assertEqual(2, table.create_global_index('id', options=options)) + + snapshot = table.snapshot_manager().get_latest_snapshot() + current_entries = IndexFileHandler(table).scan(snapshot) + current_schema_id = table.table_schema.id + rewrite = CommitMessage( + partition=(), + bucket=0, + new_files=[], + index_deletes=[ + IndexManifestEntry( + kind=1, + partition=entry.partition, + bucket=entry.bucket, + index_file=replace(entry.index_file), + schema_id=current_schema_id, + ) + for entry in current_entries + ], + index_adds=[ + IndexManifestEntry( + kind=0, + partition=entry.partition, + bucket=entry.bucket, + index_file=replace(entry.index_file), + schema_id=None, + ) + for entry in current_entries + ], + ) + commit = table.new_batch_write_builder().new_commit() + commit.commit([rewrite]) + commit.close() + + legacy_entries = IndexFileHandler(table).scan( + table.snapshot_manager().get_latest_snapshot()) + self.assertTrue(legacy_entries) + self.assertEqual({None}, {entry.schema_id for entry in legacy_entries}) + + messages = GlobalIndexBuilder(table, 'id', options=options).build() + self.assertEqual( + {entry.index_file.file_name for entry in legacy_entries}, + { + entry.index_file.file_name + for message in messages + for entry in message.index_deletes + }, + ) + self.assertEqual( + [Range(0, 3)], + Range.sort_and_merge_overlap( + [ + Range( + entry.index_file.global_index_meta.row_range_start, + entry.index_file.global_index_meta.row_range_end, + ) + for message in messages + for entry in message.index_adds + ], + True, + ), + ) + + commit = table.new_batch_write_builder().new_commit() + commit.commit(messages) + commit.close() + replacement_entries = IndexFileHandler(table).scan( + table.snapshot_manager().get_latest_snapshot()) + self.assertTrue(replacement_entries) + self.assertEqual( + {current_schema_id}, + {entry.schema_id for entry in replacement_entries}, + ) + self.assertTrue( + {entry.index_file.file_name for entry in legacy_entries}.isdisjoint( + {entry.index_file.file_name for entry in replacement_entries}) + ) + + def test_create_global_index_replaces_changed_indexed_type(self): + table = self._create_table() + self._write_arrow(table, pa.table( + { + 'id': [1, 2], + 'name': ['a', 'b'], + 'age': [10, 20], + 'city': ['x', 'y'], + }, + schema=self.pa_schema, + )) + self.assertEqual(1, table.create_global_index('id')) + old_entries = IndexFileHandler(table).scan( + table.snapshot_manager().get_latest_snapshot()) + old_file_names = { + entry.index_file.file_name for entry in old_entries + } + + table_name = table.identifier.get_full_name() + self.catalog.alter_table( + table_name, + [SchemaChange.update_column_type('id', AtomicType('BIGINT'))], + False, + ) + table = self.catalog.get_table(table_name) + messages = GlobalIndexBuilder(table, 'id').build() + self.assertEqual( + old_file_names, + { + entry.index_file.file_name + for message in messages + for entry in message.index_deletes + }, + ) + self.assertTrue(any(message.index_adds for message in messages)) + + commit = table.new_batch_write_builder().new_commit() + commit.commit(messages) + commit.close() + replacement_entries = IndexFileHandler(table).scan( + table.snapshot_manager().get_latest_snapshot()) + self.assertTrue(replacement_entries) + self.assertEqual( + {table.table_schema.id}, + {entry.schema_id for entry in replacement_entries}, + ) + self.assertTrue(old_file_names.isdisjoint( + {entry.index_file.file_name for entry in replacement_entries})) + def test_create_btree_global_index_for_java_scalar_types(self): schema = pa.schema([ ('flag', pa.bool_()), diff --git a/paimon-python/pypaimon/tests/global_index_schema_compatibility_test.py b/paimon-python/pypaimon/tests/global_index_schema_compatibility_test.py index 3f1753a40009..1894018299ae 100644 --- a/paimon-python/pypaimon/tests/global_index_schema_compatibility_test.py +++ b/paimon-python/pypaimon/tests/global_index_schema_compatibility_test.py @@ -20,7 +20,9 @@ from pypaimon.globalindex.global_index_meta import GlobalIndexMeta from pypaimon.globalindex.global_index_schema_compatibility import ( + filter_compatible_global_index_files, filter_compatible_global_indexes, + partition_global_indexes_by_compatibility, ) from pypaimon.index.index_file_meta import IndexFileMeta from pypaimon.manifest.index_manifest_entry import IndexManifestEntry @@ -99,6 +101,29 @@ def get_schema(schema_id): ) self.assertEqual([1, 99], schema_lookups) + compatible_entries, incompatible_entries = ( + partition_global_indexes_by_compatibility(table, [ + compatible, current_schema, changed_primary, legacy, + ]) + ) + self.assertEqual( + ['compatible', 'current'], + [entry.index_file.file_name for entry in compatible_entries], + ) + self.assertEqual( + ['changed-primary', 'legacy'], + [entry.index_file.file_name for entry in incompatible_entries], + ) + self.assertEqual( + ['compatible', 'current'], + [index_file.file_name for index_file in + filter_compatible_global_index_files( + table, + [compatible.index_file, current_schema.index_file, + changed_primary.index_file, legacy.index_file], + )], + ) + if __name__ == '__main__': unittest.main() diff --git a/paimon-python/pypaimon/tests/vector_search_filter_test.py b/paimon-python/pypaimon/tests/vector_search_filter_test.py index 72c4f37f746b..b76c873783b8 100644 --- a/paimon-python/pypaimon/tests/vector_search_filter_test.py +++ b/paimon-python/pypaimon/tests/vector_search_filter_test.py @@ -2469,6 +2469,11 @@ def test_scanner_ignores_incompatible_index_schema(self): ) self.assertIsNone(scanner) + self.assertIsNone(DataEvolutionGlobalIndexScanner.create( + table, + index_files=[entry.index_file], + snapshot=snapshot, + )) def test_scanner_create_selects_extra_field_indexes(self): from pypaimon.globalindex.data_evolution_global_index_scanner import ( 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 5bcba53c1b6e..adc3b2258136 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 @@ -22,7 +22,10 @@ import org.apache.paimon.fs.Path; import org.apache.paimon.fs.PositionOutputStream; import org.apache.paimon.fs.SeekableInputStream; +import org.apache.paimon.index.GlobalIndexMeta; import org.apache.paimon.index.IndexFileMeta; +import org.apache.paimon.index.pk.PrimaryKeyIndexSourceFile; +import org.apache.paimon.index.pk.PrimaryKeyIndexSourceMeta; import org.apache.paimon.io.DataFileMeta; import org.apache.paimon.io.PojoDataFileMeta; import org.apache.paimon.schema.TableSchema; @@ -32,9 +35,13 @@ import javax.annotation.Nullable; import java.io.IOException; +import java.util.ArrayList; import java.util.List; +import java.util.Map; import java.util.Optional; +import static org.apache.paimon.utils.Preconditions.checkArgument; + /** Utils for copy files. */ public class CopyFilesUtil { @@ -100,6 +107,46 @@ public static DataFileMeta toNewDataFileMeta( public static IndexFileMeta toNewIndexFileMeta( IndexFileMeta oldFileMeta, String newFileName, @Nullable Long newSchemaId) { + return toNewIndexFileMeta( + oldFileMeta, newFileName, newSchemaId, oldFileMeta.globalIndexMeta()); + } + + public static IndexFileMeta toNewPrimaryKeyIndexFileMeta( + IndexFileMeta oldFileMeta, + String newFileName, + Map dataFileNameMapping) { + PrimaryKeyIndexSourceMeta oldSourceMeta = + PrimaryKeyIndexSourceMeta.fromIndexFile(oldFileMeta); + ArrayList newSourceFiles = new ArrayList<>(); + for (PrimaryKeyIndexSourceFile sourceFile : oldSourceMeta.sourceFiles()) { + String newSourceFileName = dataFileNameMapping.get(sourceFile.fileName()); + checkArgument( + newSourceFileName != null, + "Cannot find copied data file for primary-key index source %s.", + sourceFile.fileName()); + newSourceFiles.add( + new PrimaryKeyIndexSourceFile(newSourceFileName, sourceFile.rowCount())); + } + + GlobalIndexMeta oldGlobalIndexMeta = oldFileMeta.globalIndexMeta(); + GlobalIndexMeta newGlobalIndexMeta = + new GlobalIndexMeta( + oldGlobalIndexMeta.rowRangeStart(), + oldGlobalIndexMeta.rowRangeEnd(), + oldGlobalIndexMeta.indexFieldId(), + oldGlobalIndexMeta.extraFieldIds(), + oldGlobalIndexMeta.indexMeta(), + new PrimaryKeyIndexSourceMeta(oldSourceMeta.dataLevel(), newSourceFiles) + .serialize()); + return toNewIndexFileMeta( + oldFileMeta, newFileName, oldFileMeta.schemaId(), newGlobalIndexMeta); + } + + private static IndexFileMeta toNewIndexFileMeta( + IndexFileMeta oldFileMeta, + String newFileName, + @Nullable Long newSchemaId, + @Nullable GlobalIndexMeta newGlobalIndexMeta) { String newExternalPath = externalPathDir(oldFileMeta.externalPath()) .map(dir -> dir + "/" + newFileName) @@ -111,7 +158,7 @@ public static IndexFileMeta toNewIndexFileMeta( oldFileMeta.rowCount(), oldFileMeta.dvRanges(), newExternalPath, - oldFileMeta.globalIndexMeta(), + newGlobalIndexMeta, newSchemaId); } 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 4f34f5cb5836..f43a4a094047 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 @@ -39,8 +39,10 @@ import java.io.IOException; import java.util.ArrayList; +import java.util.HashMap; import java.util.HashSet; import java.util.List; +import java.util.Map; import java.util.Set; /** List index files. */ @@ -58,7 +60,8 @@ public List execute( Identifier sourceIdentifier, Identifier targetIdentifier, Snapshot snapshot, - @Nullable PartitionPredicate partitionPredicate) + @Nullable PartitionPredicate partitionPredicate, + List dataFiles) throws Exception { if (snapshot == null) { return null; @@ -74,13 +77,20 @@ public List execute( FileStorePathFactory targetFileStorePathFactory = targetTable.store().pathFactory(); List indexManifestEntries = sourceIndexHandler.readManifestWithIOException(snapshot.indexManifest()); + List dataEvolutionIndexes = new ArrayList<>(); + for (IndexManifestEntry entry : indexManifestEntries) { + if (isDataEvolutionIndex(sourceTable, entry)) { + dataEvolutionIndexes.add(entry); + } + } Set compatibleGlobalIndexes = new HashSet<>( GlobalIndexSchemaCompatibility.filterCompatible( - sourceTable, indexManifestEntries)); + sourceTable, dataEvolutionIndexes)); + Map dataFileNameMapping = dataFileNameMapping(dataFiles); for (IndexManifestEntry indexManifestEntry : indexManifestEntries) { - boolean globalIndex = indexManifestEntry.indexFile().globalIndexMeta() != null; - if (globalIndex && !compatibleGlobalIndexes.contains(indexManifestEntry)) { + boolean dataEvolutionIndex = isDataEvolutionIndex(sourceTable, indexManifestEntry); + if (dataEvolutionIndex && !compatibleGlobalIndexes.contains(indexManifestEntry)) { continue; } if (partitionPredicate == null @@ -90,7 +100,10 @@ public List execute( indexManifestEntry, sourceFileStorePathFactory, targetFileStorePathFactory, - globalIndex ? targetTable.schema().id() : null); + dataEvolutionIndex, + isPrimaryKeyPayload(sourceTable, indexManifestEntry), + dataEvolutionIndex ? targetTable.schema().id() : null, + dataFileNameMapping); indexFiles.add(indexFile); } } @@ -101,18 +114,26 @@ private CopyFileInfo pickIndexFiles( IndexManifestEntry indexManifestEntry, FileStorePathFactory sourceFileStorePathFactory, FileStorePathFactory targetFileStorePathFactory, - @Nullable Long targetSchemaId) + boolean dataEvolutionIndex, + boolean primaryKeyPayload, + @Nullable Long targetSchemaId, + Map dataFileNameMapping) throws IOException { IndexFileMeta fileMeta = indexManifestEntry.indexFile(); IndexPathFactory sourceIndexPathFactory = - indexPathFactory(sourceFileStorePathFactory, indexManifestEntry); + indexPathFactory( + sourceFileStorePathFactory, indexManifestEntry, dataEvolutionIndex); IndexPathFactory targetIndexPathFactory = - indexPathFactory(targetFileStorePathFactory, indexManifestEntry); + indexPathFactory( + targetFileStorePathFactory, indexManifestEntry, dataEvolutionIndex); Path indexFilePath = sourceIndexPathFactory.toPath(fileMeta); Path targetIndexFilePath = targetIndexPathFactory.newPath(); IndexFileMeta targetFileMeta = - CopyFilesUtil.toNewIndexFileMeta( - fileMeta, targetIndexFilePath.getName(), targetSchemaId); + primaryKeyPayload + ? CopyFilesUtil.toNewPrimaryKeyIndexFileMeta( + fileMeta, targetIndexFilePath.getName(), dataFileNameMapping) + : CopyFilesUtil.toNewIndexFileMeta( + fileMeta, targetIndexFilePath.getName(), targetSchemaId); return new CopyFileInfo( indexFilePath.toString(), targetIndexFilePath.toString(), @@ -122,9 +143,32 @@ private CopyFileInfo pickIndexFiles( } private static IndexPathFactory indexPathFactory( - FileStorePathFactory pathFactory, IndexManifestEntry entry) { - return entry.indexFile().globalIndexMeta() == null - ? pathFactory.indexFileFactory(entry.partition(), entry.bucket()) - : pathFactory.globalIndexFileFactory(); + FileStorePathFactory pathFactory, + IndexManifestEntry entry, + boolean dataEvolutionIndex) { + return dataEvolutionIndex + ? pathFactory.globalIndexFileFactory() + : pathFactory.indexFileFactory(entry.partition(), entry.bucket()); + } + + private static boolean isDataEvolutionIndex(FileStoreTable table, IndexManifestEntry entry) { + return table.coreOptions().dataEvolutionEnabled() + && entry.indexFile().globalIndexMeta() != null; + } + + private static boolean isPrimaryKeyPayload(FileStoreTable table, IndexManifestEntry entry) { + return !table.schema().primaryKeys().isEmpty() + && entry.indexFile().globalIndexMeta() != null + && entry.indexFile().globalIndexMeta().sourceMeta() != null; + } + + private static Map dataFileNameMapping(List dataFiles) { + Map mapping = new HashMap<>(); + for (CopyFileInfo dataFile : dataFiles) { + mapping.put( + new Path(dataFile.sourceFilePath()).getName(), + new Path(dataFile.targetFilePath()).getName()); + } + return mapping; } } diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/DefaultGlobalIndexTopoBuilder.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/DefaultGlobalIndexTopoBuilder.java index 7e5a401df302..f70a6c779320 100644 --- a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/DefaultGlobalIndexTopoBuilder.java +++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/DefaultGlobalIndexTopoBuilder.java @@ -24,6 +24,7 @@ import org.apache.paimon.fs.Path; import org.apache.paimon.globalindex.DataEvolutionGlobalIndexRefreshPlanner; import org.apache.paimon.globalindex.GlobalIndexBuilderUtils; +import org.apache.paimon.globalindex.GlobalIndexSchemaCompatibility; import org.apache.paimon.globalindex.IndexedSplit; import org.apache.paimon.index.DataEvolutionIndexSourceMeta; import org.apache.paimon.io.CompactIncrement; @@ -115,9 +116,16 @@ public List buildIndex( List currentIndexes = GlobalIndexBuilderUtils.currentIndexEntries( table, snapshot, indexType, indexFields, partitionPredicate); + GlobalIndexSchemaCompatibility.CompatibilityResult compatibility = + GlobalIndexSchemaCompatibility.partitionByCompatibility(table, currentIndexes); + List compatibleIndexes = compatibility.compatible(); List rowRangesToBuild = new ArrayList<>( - GlobalIndexBuilderUtils.unindexedRowRanges(snapshot, currentIndexes)); + GlobalIndexBuilderUtils.unindexedRowRanges(snapshot, compatibleIndexes)); + List indexesToReplace = new ArrayList<>(compatibility.incompatible()); + for (IndexManifestEntry index : compatibility.incompatible()) { + rowRangesToBuild.add(index.indexFile().globalIndexMeta().rowRange()); + } byte[] sourceMeta = new DataEvolutionIndexSourceMeta(snapshot.id()).serialize(); boolean detectDataFileChange = new Options(table.options(), options.toMap()).get(GLOBAL_INDEX_COLUMN_UPDATE_ACTION) @@ -132,11 +140,11 @@ public List buildIndex( .withPartitionFilter(partitionPredicate) .plan() .files(); - List indexesToRefresh = Collections.emptyList(); if (detectDataFileChange) { - indexesToRefresh = + List indexesToRefresh = DataEvolutionGlobalIndexRefreshPlanner.findIndexesToRefresh( - table.schemaManager(), entries, currentIndexes, indexFields); + table.schemaManager(), entries, compatibleIndexes, indexFields); + indexesToReplace.addAll(indexesToRefresh); for (IndexManifestEntry index : indexesToRefresh) { rowRangesToBuild.add(index.indexFile().globalIndexMeta().rowRange()); } @@ -182,7 +190,7 @@ public List buildIndex( .collect(); commitMessages.addAll(CommitMessageSerializer.deserializeAll(commitMessageBytes)); } - for (IndexManifestEntry index : indexesToRefresh) { + for (IndexManifestEntry index : indexesToReplace) { commitMessages.add( new CommitMessageImpl( index.partition(), diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CopyFilesProcedure.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CopyFilesProcedure.java index 8b29dabb2dda..fe9b901e00ba 100644 --- a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CopyFilesProcedure.java +++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CopyFilesProcedure.java @@ -175,7 +175,11 @@ private void doCopy( sourceTableIdentifier, targetTableIdentifier, snapshot, partitionPredicate); List indexFilesRdd = listIndexFilesOperator.execute( - sourceTableIdentifier, targetTableIdentifier, snapshot, partitionPredicate); + sourceTableIdentifier, + targetTableIdentifier, + snapshot, + partitionPredicate, + dataFilesRdd); // 3. copy data and index files JavaRDD dataCopyFileInfoRdd = 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 dc8fd97e302f..a9b516910a42 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 @@ -18,6 +18,8 @@ package org.apache.paimon.spark.procedure +import org.apache.paimon.globalindex.testvector.TestVectorGlobalIndexerFactory +import org.apache.paimon.index.pk.PrimaryKeyIndexSourceMeta import org.apache.paimon.spark.PaimonSparkTestBase import org.apache.spark.sql.Row @@ -138,6 +140,129 @@ class CopyFilesProcedureTest extends PaimonSparkTestBase { } } + test("Paimon copy files procedure: primary-key index payloads") { + val random = ThreadLocalRandom.current().nextInt(100000) + val source = s"source_tbl$random" + val target = s"target_tbl$random" + withTable(source, target) { + sql( + s""" + |CREATE TABLE $source ( + | id INT, + | score INT, + | content STRING, + | embedding ARRAY) + |TBLPROPERTIES ( + | 'primary-key' = 'id', + | 'bucket' = '1', + | 'deletion-vectors.enabled' = 'true', + | 'deletion-vectors.merge-on-read' = 'false', + | 'index-file-in-data-file-dir' = 'true', + | 'pk-btree.index.columns' = 'score', + | 'pk-full-text.index.columns' = 'content', + | 'pk-vector.index.columns' = 'embedding', + | 'fields.embedding.pk-vector.index.type' = '${TestVectorGlobalIndexerFactory.IDENTIFIER}', + | 'fields.embedding.pk-vector.distance.metric' = 'l2', + | 'vector-field' = 'embedding', + | 'field.embedding.vector-dim' = '2', + | 'test.vector.dimension' = '2', + | 'test.vector.metric' = 'l2') + |""".stripMargin) + + sql(s""" + |INSERT INTO $source VALUES + | (1, 10, 'paimon lake format', array(1.0f, 0.0f)), + | (2, 20, 'apache paimon storage', array(2.0f, 0.0f)), + | (3, 30, 'other engine', array(3.0f, 0.0f)) + |""".stripMargin) + sql(s"CALL sys.compact(table => '$source')") + + val sourcePayloads = loadTable(source) + .store() + .newIndexFileHandler() + .scanEntries() + .asScala + .filter( + entry => + entry.indexFile().globalIndexMeta() != null && + entry.indexFile().globalIndexMeta().sourceMeta() != null) + val expectedIndexTypes = + Set("btree", "full-text", TestVectorGlobalIndexerFactory.IDENTIFIER) + assert(sourcePayloads.map(_.indexFile().indexType()).toSet == expectedIndexTypes) + + checkAnswer( + sql(s"CALL sys.copy(source_table => '$source', target_table => '$target')"), + Row(true) :: Nil + ) + + val targetTable = loadTable(target) + val targetPayloads = targetTable + .store() + .newIndexFileHandler() + .scanEntries() + .asScala + .filter( + entry => + entry.indexFile().globalIndexMeta() != null && + entry.indexFile().globalIndexMeta().sourceMeta() != null) + assert(targetPayloads.map(_.indexFile().indexType()).toSet == expectedIndexTypes) + assert(targetPayloads.size == sourcePayloads.size) + targetPayloads.foreach( + entry => assert(targetTable.store().newIndexFileHandler().existsIndexFile(entry))) + val targetDataFiles = targetTable + .newSnapshotReader() + .readFileIterator() + .asScala + .map(_.file().fileName()) + .toSet + targetPayloads.foreach { + entry => + val sourceFiles = PrimaryKeyIndexSourceMeta + .fromIndexFile(entry.indexFile()) + .sourceFiles() + .asScala + assert(sourceFiles.forall(file => targetDataFiles.contains(file.fileName()))) + } + + checkAnswer( + sql(s"SELECT id FROM $target WHERE score = 20"), + sql(s"SELECT id FROM $source WHERE score = 20") + ) + checkAnswer( + sql(s""" + |SELECT id + |FROM full_text_search( + | '$target', + | 'content', + | '{"match":{"column":"content","terms":"paimon"}}', + | 10) + |ORDER BY id + |""".stripMargin), + sql(s""" + |SELECT id + |FROM full_text_search( + | '$source', + | 'content', + | '{"match":{"column":"content","terms":"paimon"}}', + | 10) + |ORDER BY id + |""".stripMargin) + ) + checkAnswer( + sql(s""" + |SELECT id + |FROM vector_search('$target', 'embedding', array(0.0f, 0.0f), 2) + |ORDER BY id + |""".stripMargin), + sql(s""" + |SELECT id + |FROM vector_search('$source', 'embedding', array(0.0f, 0.0f), 2) + |ORDER BY id + |""".stripMargin) + ) + } + } + test("Paimon copy files procedure: schema change") { val random = ThreadLocalRandom.current().nextInt(100000); withTable(s"tbl$random") { @@ -194,7 +319,7 @@ class CopyFilesProcedureTest extends PaimonSparkTestBase { assert(sourceEntries.nonEmpty) val buildSchemaId = sourceEntries.head.schemaId().longValue() - sql(s"ALTER TABLE $source RENAME COLUMN payload_at_build TO payload_after_build") + sql(s"ALTER TABLE $source ADD COLUMN added_after_build STRING") assert(loadTable(source).schema().id() != buildSchemaId) checkAnswer(