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 000000000000..b2201de23e68 Binary files /dev/null and b/paimon-core/src/test/resources/compatibility/manifest-committable-v13-global-index-v5 differ diff --git a/paimon-core/src/test/resources/compatibility/manifest-committable-v14-v5 b/paimon-core/src/test/resources/compatibility/manifest-committable-v14-v5 new file mode 100644 index 000000000000..6aeeea406b0a Binary files /dev/null and b/paimon-core/src/test/resources/compatibility/manifest-committable-v14-v5 differ diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/globalindex/GenericIndexTopoBuilder.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/globalindex/GenericIndexTopoBuilder.java index 4740a3ea805c..df23bc883ca9 100644 --- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/globalindex/GenericIndexTopoBuilder.java +++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/globalindex/GenericIndexTopoBuilder.java @@ -554,7 +554,8 @@ private static CommitMessage flushIndex( indexFields, indexType, resultEntries, - sourceMeta); + sourceMeta, + table.schema().id()); return new CommitMessageImpl( partition, 0, null, indexIncrement(indexFileMetas), emptyIncrement()); } diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/VectorSearchProcedureITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/VectorSearchProcedureITCase.java index 31324f2ec7f3..2c48bdc34a10 100644 --- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/VectorSearchProcedureITCase.java +++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/VectorSearchProcedureITCase.java @@ -1072,7 +1072,8 @@ private void buildAndCommitVectorIndex(FileStoreTable table, float[][] vectors, rowRange, vectorField.id(), TestVectorGlobalIndexerFactory.IDENTIFIER, - entries); + entries, + table.schema().id()); DataIncrement dataIncrement = DataIncrement.indexIncrement(indexFiles); CommitMessage message = @@ -1115,7 +1116,8 @@ private void buildAndCommitMultiFieldVectorIndex( Arrays.asList(vectorField, idField), TestVectorGlobalIndexerFactory.IDENTIFIER, entries, - null); + null, + table.schema().id()); DataIncrement dataIncrement = DataIncrement.indexIncrement(indexFiles); CommitMessage message = diff --git a/paimon-full-text/src/test/java/org/apache/paimon/fulltext/index/JavaPyNativeFullTextE2ETest.java b/paimon-full-text/src/test/java/org/apache/paimon/fulltext/index/JavaPyNativeFullTextE2ETest.java index 7056909d5bf3..d4971a570ca5 100644 --- a/paimon-full-text/src/test/java/org/apache/paimon/fulltext/index/JavaPyNativeFullTextE2ETest.java +++ b/paimon-full-text/src/test/java/org/apache/paimon/fulltext/index/JavaPyNativeFullTextE2ETest.java @@ -225,7 +225,8 @@ private void writeTableWithNativeFullTextIndex( rowRange, contentField.id(), NativeFullTextGlobalIndexerFactory.IDENTIFIER, - entries); + entries, + table.schema().id()); // Commit the index DataIncrement dataIncrement = DataIncrement.indexIncrement(indexFiles); diff --git a/paimon-lumina/src/test/java/org/apache/paimon/lumina/index/JavaPyLuminaE2ETest.java b/paimon-lumina/src/test/java/org/apache/paimon/lumina/index/JavaPyLuminaE2ETest.java index d402a40ac942..4c1be3f98ce3 100644 --- a/paimon-lumina/src/test/java/org/apache/paimon/lumina/index/JavaPyLuminaE2ETest.java +++ b/paimon-lumina/src/test/java/org/apache/paimon/lumina/index/JavaPyLuminaE2ETest.java @@ -194,7 +194,8 @@ public void testLuminaVectorIndexWrite(String fileFormat) throws Exception { rowRange, embeddingField.id(), LuminaVectorGlobalIndexerFactory.IDENTIFIER, - entries); + entries, + table.schema().id()); DataIncrement dataIncrement = DataIncrement.indexIncrement(indexFiles); CommitMessage message = @@ -298,7 +299,8 @@ public void testLuminaVectorWithBTreeIndexWrite() throws Exception { rowRange, embeddingField.id(), LuminaVectorGlobalIndexerFactory.IDENTIFIER, - vectorEntries); + vectorEntries, + table.schema().id()); // Build BTree global index on "id". DataField idField = table.rowType().getField("id"); @@ -318,7 +320,8 @@ public void testLuminaVectorWithBTreeIndexWrite() throws Exception { rowRange, idField.id(), BTreeGlobalIndexerFactory.IDENTIFIER, - idEntries); + idEntries, + table.schema().id()); // Commit both index sets in a single index-only commit. java.util.List allIndexFiles = new java.util.ArrayList<>(); diff --git a/paimon-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 =