From 0a2930f786e8a5332dd71c057c4f90d021a4ecd9 Mon Sep 17 00:00:00 2001 From: QuakeWang Date: Wed, 26 Aug 2026 14:20:23 +0800 Subject: [PATCH] [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 =