This is an automated email from the ASF dual-hosted git repository. leaves12138 pushed a commit to branch codex/de-vector-index-refresh in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 86c6eb840991982bfc72fce05e4a81e832b001f8 Author: 仟弋 <[email protected]> AuthorDate: Thu Jul 23 20:57:09 2026 +0800 [core][flink][spark] Refresh data evolution vector indexes incrementally --- docs/generated/core_configuration.html | 6 + .../main/java/org/apache/paimon/CoreOptions.java | 11 + .../DataEvolutionRowIdReassigner.java | 3 +- .../DataEvolutionGlobalIndexRefreshPlanner.java | 230 +++++++++++ .../globalindex/GlobalIndexBuilderUtils.java | 108 ++--- .../paimon/index/DataEvolutionIndexSourceMeta.java | 101 +++++ .../apache/paimon/manifest/IndexManifestFile.java | 18 +- .../paimon/manifest/IndexManifestFileHandler.java | 33 +- .../operation/DataEvolutionFileStoreScan.java | 24 +- .../paimon/operation/FileStoreCommitImpl.java | 6 +- .../apache/paimon/utils/DataEvolutionUtils.java | 22 + .../java/org/apache/paimon/CoreOptionsTest.java | 9 + .../DataEvolutionRowIdReassignerTest.java | 56 +++ ...DataEvolutionGlobalIndexRefreshPlannerTest.java | 454 +++++++++++++++++++++ .../globalindex/GlobalIndexBuilderUtilsTest.java | 47 ++- .../index/DataEvolutionIndexSourceMetaTest.java | 100 +++++ .../manifest/IndexManifestFileHandlerTest.java | 91 +++++ .../paimon/operation/FileStoreCommitTest.java | 55 +++ .../table/source/FullTextSearchBuilderTest.java | 9 +- .../table/source/VectorSearchBuilderTest.java | 3 +- .../paimon/utils/DataEvolutionUtilsTest.java | 83 +++- .../flink/globalindex/GenericIndexTopoBuilder.java | 78 +++- .../flink/LuminaVectorGlobalIndexITCase.java | 92 +++++ .../globalindex/GenericIndexTopoBuilderTest.java | 87 +++- .../procedure/FullTextSearchProcedureITCase.java | 34 ++ .../globalindex/DefaultGlobalIndexBuilder.java | 32 +- .../globalindex/DefaultGlobalIndexTopoBuilder.java | 73 +++- .../paimon/spark/sql/FullTextSearchTest.scala | 7 + .../paimon/spark/sql/LuminaVectorIndexTest.scala | 135 ++++++ 29 files changed, 1835 insertions(+), 172 deletions(-) diff --git a/docs/generated/core_configuration.html b/docs/generated/core_configuration.html index 0776eaf9e3..432edf85c5 100644 --- a/docs/generated/core_configuration.html +++ b/docs/generated/core_configuration.html @@ -782,6 +782,12 @@ under the License. <td>String</td> <td>Global index root directory, if not set, the global index files will be stored under the <table-root-directory>/index.</td> </tr> + <tr> + <td><h5>global-index.ignore-missing-delete</h5></td> + <td style="word-wrap: break-word;">false</td> + <td>Boolean</td> + <td>Whether to ignore deleting a global index file which does not exist in the previous index manifest.</td> + </tr> <tr> <td><h5>global-index.row-count-per-shard</h5></td> <td style="word-wrap: break-word;">100000</td> diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java index ec6a2296bf..f0039d3852 100644 --- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java +++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java @@ -2507,6 +2507,13 @@ public class CoreOptions implements Serializable { "Defines the action to take when an update modifies columns that are covered by a global index. " + "IGNORE leaves existing index files unchanged and may make the index stale."); + public static final ConfigOption<Boolean> GLOBAL_INDEX_IGNORE_MISSING_DELETE = + key("global-index.ignore-missing-delete") + .booleanType() + .defaultValue(false) + .withDescription( + "Whether to ignore deleting a global index file which does not exist in the previous index manifest."); + public static final ConfigOption<MemorySize> LOOKUP_MERGE_BUFFER_SIZE = key("lookup.merge-buffer-size") .memoryType() @@ -3559,6 +3566,10 @@ public class CoreOptions implements Serializable { return options.get(GLOBAL_INDEX_COLUMN_UPDATE_ACTION); } + public boolean globalIndexIgnoreMissingDelete() { + return options.get(GLOBAL_INDEX_IGNORE_MISSING_DELETE); + } + public LookupStrategy lookupStrategy() { return LookupStrategy.from( mergeEngine().equals(MergeEngine.FIRST_ROW), 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 d81d98ed70..c8daa1fe2d 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 @@ -857,7 +857,8 @@ public class DataEvolutionRowIdReassigner { rewrittenRange.to, globalIndex.indexFieldId(), globalIndex.extraFieldIds(), - globalIndex.indexMeta()); + globalIndex.indexMeta(), + globalIndex.sourceMeta()); IndexFileMeta newIndexFile = new IndexFileMeta( indexFile.indexType(), diff --git a/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionGlobalIndexRefreshPlanner.java b/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionGlobalIndexRefreshPlanner.java new file mode 100644 index 0000000000..44ed39685d --- /dev/null +++ b/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionGlobalIndexRefreshPlanner.java @@ -0,0 +1,230 @@ +/* + * 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.data.BinaryRow; +import org.apache.paimon.index.DataEvolutionIndexSourceMeta; +import org.apache.paimon.index.GlobalIndexMeta; +import org.apache.paimon.io.DataFileMeta; +import org.apache.paimon.manifest.FileKind; +import org.apache.paimon.manifest.IndexManifestEntry; +import org.apache.paimon.manifest.ManifestEntry; +import org.apache.paimon.schema.SchemaManager; +import org.apache.paimon.types.DataField; +import org.apache.paimon.utils.Pair; +import org.apache.paimon.utils.Range; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Comparator; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.NavigableMap; +import java.util.Set; +import java.util.TreeMap; + +import static org.apache.paimon.utils.DataEvolutionUtils.fileFieldIds; + +/** Plans existing global index files which need refresh after data-evolution updates. */ +public final class DataEvolutionGlobalIndexRefreshPlanner { + + private DataEvolutionGlobalIndexRefreshPlanner() {} + + public static List<IndexManifestEntry> findIndexesToRefresh( + SchemaManager schemaManager, + List<ManifestEntry> dataEntries, + List<IndexManifestEntry> indexEntries, + List<DataField> indexedFields) { + Set<Integer> indexedFieldIds = new HashSet<>(); + for (DataField field : indexedFields) { + indexedFieldIds.add(field.id()); + } + + Map<Pair<BinaryRow, Integer>, RefreshGroup> groups = new HashMap<>(); + for (int i = 0; i < indexEntries.size(); i++) { + IndexManifestEntry indexEntry = indexEntries.get(i); + GlobalIndexMeta indexMeta = indexEntry.indexFile().globalIndexMeta(); + if (indexEntry.kind() != FileKind.ADD + || indexMeta == null + || !matchesFields(indexMeta, indexedFields)) { + continue; + } + + byte[] sourceMeta = indexMeta.sourceMeta(); + if (!DataEvolutionIndexSourceMeta.isDataEvolutionMeta(sourceMeta)) { + // Legacy indexes have no trustworthy scan baseline and require an explicit rebuild. + continue; + } + long scanSnapshotId = + DataEvolutionIndexSourceMeta.deserialize(sourceMeta).scanSnapshotId(); + groups.computeIfAbsent( + Pair.of(indexEntry.partition(), indexEntry.bucket()), + key -> new RefreshGroup()) + .addIndex(i, indexMeta.rowRange(), scanSnapshotId); + } + + Map<Pair<Long, List<String>>, Set<Integer>> fileFieldIdsCache = new HashMap<>(); + for (ManifestEntry dataEntry : dataEntries) { + DataFileMeta file = dataEntry.file(); + if (dataEntry.kind() != FileKind.ADD || file.firstRowId() == null) { + continue; + } + + RefreshGroup group = groups.get(Pair.of(dataEntry.partition(), dataEntry.bucket())); + if (group == null || !group.mayContainUpdate(file)) { + continue; + } + + Set<Integer> physicalFieldIds = + fileFieldIdsCache.computeIfAbsent( + Pair.of(file.schemaId(), file.writeCols()), + key -> fileFieldIds(schemaManager::schema, file)); + if (!disjoint(indexedFieldIds, physicalFieldIds)) { + group.addDataFile(file); + } + } + + boolean[] indexesToRefresh = new boolean[indexEntries.size()]; + for (RefreshGroup group : groups.values()) { + group.markIndexesToRefresh(indexesToRefresh); + } + + List<IndexManifestEntry> result = new ArrayList<>(); + for (int i = 0; i < indexEntries.size(); i++) { + if (indexesToRefresh[i]) { + result.add(indexEntries.get(i)); + } + } + return result; + } + + private static final class RefreshGroup { + + private final List<IndexQuery> indexes = new ArrayList<>(); + private final List<DataFileMeta> dataFiles = new ArrayList<>(); + private final MergedRanges indexedRanges = new MergedRanges(); + private long minScanSnapshotId = Long.MAX_VALUE; + + private void addIndex(int ordinal, Range rowRange, long scanSnapshotId) { + indexes.add(new IndexQuery(ordinal, rowRange, scanSnapshotId)); + indexedRanges.add(rowRange); + minScanSnapshotId = Math.min(minScanSnapshotId, scanSnapshotId); + } + + private boolean mayContainUpdate(DataFileMeta file) { + return file.maxSequenceNumber() > minScanSnapshotId + && indexedRanges.intersects(file.nonNullRowIdRange()); + } + + private void addDataFile(DataFileMeta file) { + dataFiles.add(file); + } + + private void markIndexesToRefresh(boolean[] result) { + // As scan watermarks decrease, eligible data files only grow. + dataFiles.sort(Comparator.comparingLong(DataFileMeta::maxSequenceNumber).reversed()); + indexes.sort((left, right) -> Long.compare(right.scanSnapshotId, left.scanSnapshotId)); + + MergedRanges updatedRanges = new MergedRanges(); + int nextFile = 0; + for (IndexQuery index : indexes) { + while (nextFile < dataFiles.size() + && dataFiles.get(nextFile).maxSequenceNumber() > index.scanSnapshotId) { + updatedRanges.add(dataFiles.get(nextFile).nonNullRowIdRange()); + nextFile++; + } + if (updatedRanges.intersects(index.rowRange)) { + result[index.ordinal] = true; + } + } + } + } + + private static final class IndexQuery { + + private final int ordinal; + private final Range rowRange; + private final long scanSnapshotId; + + private IndexQuery(int ordinal, Range rowRange, long scanSnapshotId) { + this.ordinal = ordinal; + this.rowRange = rowRange; + this.scanSnapshotId = scanSnapshotId; + } + } + + /** Dynamically merged inclusive ranges supporting logarithmic intersection checks. */ + private static final class MergedRanges { + + private final NavigableMap<Long, Long> ranges = new TreeMap<>(); + + private void add(Range range) { + long from = range.from; + long to = range.to; + + Map.Entry<Long, Long> floor = ranges.floorEntry(from); + if (floor != null && floor.getValue() >= from) { + from = floor.getKey(); + to = Math.max(to, floor.getValue()); + ranges.remove(floor.getKey()); + } + + Map.Entry<Long, Long> next = ranges.ceilingEntry(from); + while (next != null && next.getKey() <= to) { + to = Math.max(to, next.getValue()); + ranges.remove(next.getKey()); + next = ranges.ceilingEntry(from); + } + ranges.put(from, to); + } + + private boolean intersects(Range range) { + Map.Entry<Long, Long> floor = ranges.floorEntry(range.to); + return floor != null && floor.getValue() >= range.from; + } + } + + private static boolean matchesFields(GlobalIndexMeta meta, List<DataField> fields) { + if (fields.isEmpty() || meta.indexFieldId() != fields.get(0).id()) { + return false; + } + int[] expectedExtraFields = + fields.size() == 1 + ? null + : fields.subList(1, fields.size()).stream() + .mapToInt(DataField::id) + .toArray(); + int[] actualExtraFields = meta.extraFieldIds(); + if (actualExtraFields == null || actualExtraFields.length == 0) { + return expectedExtraFields == null || expectedExtraFields.length == 0; + } + return expectedExtraFields != null && Arrays.equals(actualExtraFields, expectedExtraFields); + } + + private static boolean disjoint(Set<Integer> left, Set<Integer> right) { + for (Integer value : left) { + if (right.contains(value)) { + return false; + } + } + return true; + } +} 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 0537e50114..f79d10b660 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 @@ -31,7 +31,6 @@ import org.apache.paimon.manifest.IndexManifestEntry; import org.apache.paimon.manifest.ManifestEntry; import org.apache.paimon.options.Options; import org.apache.paimon.partition.PartitionPredicate; -import org.apache.paimon.schema.SchemaManager; import org.apache.paimon.table.FileStoreTable; import org.apache.paimon.table.source.DataSplit; import org.apache.paimon.types.DataField; @@ -47,7 +46,6 @@ import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; import java.util.Comparator; -import java.util.HashMap; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; @@ -70,7 +68,15 @@ public class GlobalIndexBuilderUtils { List<ResultEntry> entries) throws IOException { return toIndexFileMetas( - fileIO, indexPathFactory, options, range, indexFieldId, null, indexType, entries); + fileIO, + indexPathFactory, + options, + range, + indexFieldId, + null, + indexType, + entries, + null); } /** @@ -86,21 +92,19 @@ public class GlobalIndexBuilderUtils { Range range, List<DataField> fields, String indexType, - List<ResultEntry> entries) + List<ResultEntry> entries, + @Nullable byte[] sourceMeta) throws IOException { - // The first column is the primary index column and is stored as indexFieldId; the - // remaining columns (if any) go into extraFieldIds. - int indexFieldId = fields.get(0).id(); - int[] extraFieldIds = extraFieldIds(fields); return toIndexFileMetas( fileIO, indexPathFactory, options, range, - indexFieldId, - extraFieldIds, + fields.get(0).id(), + extraFieldIds(fields), indexType, - entries); + entries, + sourceMeta); } public static List<Range> unindexedRowRanges( @@ -349,7 +353,8 @@ public class GlobalIndexBuilderUtils { int indexFieldId, @Nullable int[] extraFieldIds, String indexType, - List<ResultEntry> entries) + List<ResultEntry> entries, + @Nullable byte[] sourceMeta) throws IOException { List<IndexFileMeta> results = new ArrayList<>(); for (ResultEntry entry : entries) { @@ -357,7 +362,12 @@ public class GlobalIndexBuilderUtils { long fileSize = fileIO.getFileSize(indexPathFactory.toPath(fileName)); GlobalIndexMeta globalIndexMeta = new GlobalIndexMeta( - range.from, range.to, indexFieldId, extraFieldIds, entry.meta()); + range.from, + range.to, + indexFieldId, + extraFieldIds, + entry.meta(), + sourceMeta); Path externalPathDir = options.globalIndexExternalPath(); String externalPathString = null; @@ -397,68 +407,16 @@ public class GlobalIndexBuilderUtils { return globalIndexer.createWriter(createGlobalIndexFileReadWrite(table)); } - /** - * Find the minimum firstRowId among files whose schema does not contain all index columns. - * Files at or beyond this rowId cannot be indexed because the column was added later via ALTER - * TABLE. - * - * @return the boundary rowId, or {@link Long#MAX_VALUE} if all files contain the columns - */ - public static long findMinNonIndexableRowId( - SchemaManager schemaManager, List<ManifestEntry> entries, List<String> indexColumns) { - Map<Long, Boolean> schemaContainsColumns = new HashMap<>(); - long minRowId = Long.MAX_VALUE; - long minSchemaId = -1; - for (ManifestEntry entry : entries) { - long sid = entry.file().schemaId(); - boolean contains = - schemaContainsColumns.computeIfAbsent( - sid, - id -> schemaManager.schema(id).fieldNames().containsAll(indexColumns)); - if (!contains && entry.file().firstRowId() != null) { - long rowId = entry.file().nonNullFirstRowId(); - if (rowId < minRowId) { - minRowId = rowId; - minSchemaId = sid; - } - } - } - if (minRowId != Long.MAX_VALUE) { - List<String> schemaFields = schemaManager.schema(minSchemaId).fieldNames(); - List<String> missingColumns = new ArrayList<>(); - for (String col : indexColumns) { - if (!schemaFields.contains(col)) { - missingColumns.add(col); - } - } - LOG.info( - "Found non-indexable files: schemaId={} missing columns {}, boundaryRowId={}.", - minSchemaId, - missingColumns, - minRowId); - } - return minRowId; - } - - /** Keep only entries whose firstRowId is strictly less than the given boundary. */ - public static List<ManifestEntry> filterEntriesBefore( - List<ManifestEntry> entries, long boundaryRowId) { - if (boundaryRowId == Long.MAX_VALUE) { - return entries; - } - List<ManifestEntry> result = new ArrayList<>(); - for (ManifestEntry entry : entries) { - if (entry.file().firstRowId() != null - && entry.file().nonNullFirstRowId() < boundaryRowId) { - result.add(entry); - } - } - LOG.info( - "Filtered {} files to {} indexable files (boundaryRowId={}).", - entries.size(), - result.size(), - boundaryRowId); - return result; + /** Whether a global index should be refreshed automatically for data-evolution updates. */ + public static boolean shouldRefreshDataEvolutionIndex( + FileStoreTable table, + String indexType, + DataField indexField, + List<DataField> extraFields, + Options options) { + return table.coreOptions().dataEvolutionEnabled() + && GlobalIndexer.create(indexType, indexField, extraFields, options) + instanceof VectorGlobalIndexer; } private static GlobalIndexFileReadWrite createGlobalIndexFileReadWrite(FileStoreTable table) { diff --git a/paimon-core/src/main/java/org/apache/paimon/index/DataEvolutionIndexSourceMeta.java b/paimon-core/src/main/java/org/apache/paimon/index/DataEvolutionIndexSourceMeta.java new file mode 100644 index 0000000000..fb39bc2f8e --- /dev/null +++ b/paimon-core/src/main/java/org/apache/paimon/index/DataEvolutionIndexSourceMeta.java @@ -0,0 +1,101 @@ +/* + * 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.io.DataInputDeserializer; +import org.apache.paimon.io.DataOutputSerializer; + +import javax.annotation.Nullable; + +import java.io.IOException; + +import static org.apache.paimon.utils.Preconditions.checkArgument; + +/** Data snapshot scanned to build a global index for a data-evolution table. */ +public final class DataEvolutionIndexSourceMeta { + + // "DEIX". The marker distinguishes this metadata from primary-key index source metadata. + private static final int MAGIC = 0x44454958; + private static final int VERSION = 1; + + private final long scanSnapshotId; + + public DataEvolutionIndexSourceMeta(long scanSnapshotId) { + checkArgument(scanSnapshotId > 0, "Scan snapshot id must be positive."); + this.scanSnapshotId = scanSnapshotId; + } + + public long scanSnapshotId() { + return scanSnapshotId; + } + + public byte[] serialize() { + try { + DataOutputSerializer output = new DataOutputSerializer(16); + output.writeInt(MAGIC); + output.writeInt(VERSION); + output.writeLong(scanSnapshotId); + return output.getCopyOfBuffer(); + } catch (IOException e) { + throw new RuntimeException( + "Failed to serialize data-evolution index source metadata.", e); + } + } + + public static boolean isDataEvolutionMeta(@Nullable byte[] bytes) { + if (bytes == null || bytes.length < Integer.BYTES) { + return false; + } + try { + return new DataInputDeserializer(bytes).readInt() == MAGIC; + } catch (IOException e) { + return false; + } + } + + public static DataEvolutionIndexSourceMeta deserialize(byte[] bytes) { + try { + DataInputDeserializer input = new DataInputDeserializer(bytes); + int magic = input.readInt(); + checkArgument(magic == MAGIC, "Not data-evolution index source metadata."); + int version = input.readInt(); + checkArgument( + version == VERSION, + "Unsupported data-evolution index source version: %s.", + version); + long scanSnapshotId = input.readLong(); + checkArgument( + input.available() == 0, + "Unexpected trailing bytes in data-evolution index source metadata."); + return new DataEvolutionIndexSourceMeta(scanSnapshotId); + } catch (IOException e) { + throw new IllegalArgumentException( + "Failed to deserialize data-evolution index source metadata.", e); + } + } + + public static DataEvolutionIndexSourceMeta fromIndexFile(IndexFileMeta indexFile) { + GlobalIndexMeta globalIndexMeta = indexFile.globalIndexMeta(); + checkArgument( + globalIndexMeta != null && isDataEvolutionMeta(globalIndexMeta.sourceMeta()), + "Index file %s has no data-evolution source metadata.", + indexFile.fileName()); + return deserialize(globalIndexMeta.sourceMeta()); + } +} diff --git a/paimon-core/src/main/java/org/apache/paimon/manifest/IndexManifestFile.java b/paimon-core/src/main/java/org/apache/paimon/manifest/IndexManifestFile.java index a46762c1af..366de511d7 100644 --- a/paimon-core/src/main/java/org/apache/paimon/manifest/IndexManifestFile.java +++ b/paimon-core/src/main/java/org/apache/paimon/manifest/IndexManifestFile.java @@ -68,10 +68,26 @@ public class IndexManifestFile extends ObjectsFile<IndexManifestEntry> { @Nullable String previousIndexManifest, List<IndexManifestEntry> newIndexFiles, BucketMode bucketMode) { + return writeIndexFiles(previousIndexManifest, newIndexFiles, bucketMode, false); + } + + /** + * Write new index files to index manifest. + * + * @param ignoreMissingGlobalIndexDelete whether to ignore deleting a global index file which + * does not exist in the previous index manifest + */ + @Nullable + public String writeIndexFiles( + @Nullable String previousIndexManifest, + List<IndexManifestEntry> newIndexFiles, + BucketMode bucketMode, + boolean ignoreMissingGlobalIndexDelete) { if (newIndexFiles.isEmpty()) { return previousIndexManifest; } - IndexManifestFileHandler handler = new IndexManifestFileHandler(this, bucketMode); + IndexManifestFileHandler handler = + new IndexManifestFileHandler(this, bucketMode, ignoreMissingGlobalIndexDelete); return handler.write(previousIndexManifest, newIndexFiles); } diff --git a/paimon-core/src/main/java/org/apache/paimon/manifest/IndexManifestFileHandler.java b/paimon-core/src/main/java/org/apache/paimon/manifest/IndexManifestFileHandler.java index 486f39fef4..45a8705f33 100644 --- a/paimon-core/src/main/java/org/apache/paimon/manifest/IndexManifestFileHandler.java +++ b/paimon-core/src/main/java/org/apache/paimon/manifest/IndexManifestFileHandler.java @@ -19,6 +19,7 @@ package org.apache.paimon.manifest; import org.apache.paimon.data.BinaryRow; +import org.apache.paimon.index.DataEvolutionIndexSourceMeta; import org.apache.paimon.index.DeletionVectorMeta; import org.apache.paimon.index.GlobalIndexMeta; import org.apache.paimon.index.IndexFileMeta; @@ -51,9 +52,19 @@ public class IndexManifestFileHandler { private final BucketMode bucketMode; + private final boolean ignoreMissingGlobalIndexDelete; + IndexManifestFileHandler(IndexManifestFile indexManifestFile, BucketMode bucketMode) { + this(indexManifestFile, bucketMode, false); + } + + IndexManifestFileHandler( + IndexManifestFile indexManifestFile, + BucketMode bucketMode, + boolean ignoreMissingGlobalIndexDelete) { this.indexManifestFile = indexManifestFile; this.bucketMode = bucketMode; + this.ignoreMissingGlobalIndexDelete = ignoreMissingGlobalIndexDelete; } String write(@Nullable String previousIndexManifest, List<IndexManifestEntry> newIndexFiles) { @@ -96,7 +107,7 @@ public class IndexManifestFileHandler { private IndexManifestFileCombiner getIndexManifestFileCombine(String indexType) { if (!DELETION_VECTORS_INDEX.equals(indexType) && !HASH_INDEX.equals(indexType)) { - return new GlobalIndexCombiner(); + return new GlobalIndexCombiner(ignoreMissingGlobalIndexDelete); } if (DELETION_VECTORS_INDEX.equals(indexType) && BucketMode.BUCKET_UNAWARE == bucketMode) { @@ -202,6 +213,12 @@ public class IndexManifestFileHandler { /** We combine the previous and new index files by file name. */ static class GlobalIndexCombiner implements IndexManifestFileCombiner { + private final boolean ignoreMissingDelete; + + GlobalIndexCombiner(boolean ignoreMissingDelete) { + this.ignoreMissingDelete = ignoreMissingDelete; + } + @Override public List<IndexManifestEntry> combine( List<IndexManifestEntry> prevIndexFiles, List<IndexManifestEntry> newIndexFiles) { @@ -220,7 +237,12 @@ public class IndexManifestFileHandler { .filter(f -> f.kind() == FileKind.ADD) .collect(Collectors.toList()); for (IndexManifestEntry entry : removed) { - indexEntries.remove(entry.indexFile().fileName()); + String fileName = entry.indexFile().fileName(); + checkState( + ignoreMissingDelete || indexEntries.containsKey(fileName), + "Trying to delete global index file %s which does not exist.", + fileName); + indexEntries.remove(fileName); } validateRetainedIndexFiles(indexEntries.values(), added); for (IndexManifestEntry entry : added) { @@ -241,7 +263,12 @@ public class IndexManifestFileHandler { for (IndexManifestEntry added : addedIndexFiles) { GlobalIndexMeta addedMeta = added.indexFile().globalIndexMeta(); if (addedMeta == null - || (retainedMeta.sourceMeta() != null && addedMeta.sourceMeta() != null) + || (retainedMeta.sourceMeta() != null + && addedMeta.sourceMeta() != null + && !DataEvolutionIndexSourceMeta.isDataEvolutionMeta( + retainedMeta.sourceMeta()) + && !DataEvolutionIndexSourceMeta.isDataEvolutionMeta( + addedMeta.sourceMeta())) || retainedMeta.indexFieldId() != addedMeta.indexFieldId() || (Arrays.equals( retainedMeta.extraFieldIds(), addedMeta.extraFieldIds()) diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionFileStoreScan.java b/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionFileStoreScan.java index e097b3be53..eef706c368 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionFileStoreScan.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionFileStoreScan.java @@ -62,6 +62,7 @@ import java.util.stream.Collectors; import static org.apache.paimon.format.blob.BlobFileFormat.isBlobFile; import static org.apache.paimon.manifest.ManifestFileMeta.allContainsRowId; import static org.apache.paimon.types.VectorType.isVectorStoreFile; +import static org.apache.paimon.utils.DataEvolutionUtils.fileFieldIds; import static org.apache.paimon.utils.DataEvolutionUtils.retrieveAnchorFile; /** {@link FileStoreScan} for data-evolution enabled table. */ @@ -251,24 +252,7 @@ public class DataEvolutionFileStoreScan extends AppendOnlyFileStoreScan { private Set<Integer> fileFieldIdsForEntry(ManifestEntry entry) { return fileFieldIdsCache.computeIfAbsent( Pair.of(entry.file().schemaId(), entry.file().writeCols()), - pair -> computeFileFieldIds(this::scanTableSchema, entry.file())); - } - - /** - * Field ids of the columns physically present in {@code file}, resolved through the file's own - * schema (i.e. the schema the file was written under). Field id, not field name, is the stable - * identity across schemas — necessary so a renamed column matches an old file written under the - * pre-rename name. - */ - @VisibleForTesting - static Set<Integer> computeFileFieldIds( - Function<Long, TableSchema> scanTableSchema, DataFileMeta file) { - Set<Integer> ids = new HashSet<>(); - for (DataField f : - scanTableSchema.apply(file.schemaId()).project(file.writeCols()).fields()) { - ids.add(f.id()); - } - return ids; + pair -> fileFieldIds(this::scanTableSchema, entry.file())); } /** TODO: Optimize implementation of this method. */ @@ -283,9 +267,7 @@ public class DataEvolutionFileStoreScan extends AppendOnlyFileStoreScan { entry -> isBlobFile(entry.file().fileName()) || isVectorStoreFile(entry.file().fileName())) - .flatMap( - entry -> - computeFileFieldIds(scanTableSchema, entry.file()).stream()) + .flatMap(entry -> fileFieldIds(scanTableSchema, entry.file()).stream()) .collect(Collectors.toSet()); // exclude blob and vector-store files, useless for predicate eval metas = diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java index b440320212..eada47b1b4 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java @@ -1105,7 +1105,11 @@ public class FileStoreCommitImpl implements FileStoreCommit { } indexManifest = - indexManifestFile.writeIndexFiles(oldIndexManifest, indexFiles, bucketMode); + indexManifestFile.writeIndexFiles( + oldIndexManifest, + indexFiles, + bucketMode, + options.globalIndexIgnoreMissingDelete()); long latestSchemaId = schemaManager diff --git a/paimon-core/src/main/java/org/apache/paimon/utils/DataEvolutionUtils.java b/paimon-core/src/main/java/org/apache/paimon/utils/DataEvolutionUtils.java index 57ba93ae19..c1294f3946 100644 --- a/paimon-core/src/main/java/org/apache/paimon/utils/DataEvolutionUtils.java +++ b/paimon-core/src/main/java/org/apache/paimon/utils/DataEvolutionUtils.java @@ -19,10 +19,14 @@ package org.apache.paimon.utils; import org.apache.paimon.io.DataFileMeta; +import org.apache.paimon.schema.TableSchema; +import org.apache.paimon.types.DataField; import java.util.Collection; import java.util.Comparator; +import java.util.HashSet; import java.util.List; +import java.util.Set; import java.util.function.Function; import java.util.stream.Collectors; @@ -34,6 +38,24 @@ import static org.apache.paimon.utils.Preconditions.checkState; /** Util class for data evolution. */ public class DataEvolutionUtils { + /** + * Table field ids physically present in a file, resolved through the schema used to write it. + */ + public static Set<Integer> fileFieldIds( + Function<Long, TableSchema> scanTableSchema, DataFileMeta file) { + TableSchema schema = scanTableSchema.apply(file.schemaId()); + List<String> writeCols = file.writeCols(); + Set<String> writeColNames = writeCols == null ? null : new HashSet<>(writeCols); + Set<Integer> ids = new HashSet<>(); + for (DataField field : schema.fields()) { + // writeCols may also contain physical row-tracking fields outside the table schema. + if (writeColNames == null || writeColNames.contains(field.name())) { + ids.add(field.id()); + } + } + return ids; + } + /** * Retrieve the anchor file of a row range group. Always the oldest normal file. Files are * compared by (max_seq, fileName) pairs. diff --git a/paimon-core/src/test/java/org/apache/paimon/CoreOptionsTest.java b/paimon-core/src/test/java/org/apache/paimon/CoreOptionsTest.java index b3e91a3a4b..c0629b19aa 100644 --- a/paimon-core/src/test/java/org/apache/paimon/CoreOptionsTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/CoreOptionsTest.java @@ -107,6 +107,15 @@ public class CoreOptionsTest { .isEqualTo(CoreOptions.GlobalIndexColumnUpdateAction.IGNORE); } + @Test + public void testGlobalIndexIgnoreMissingDelete() { + Options conf = new Options(); + assertThat(new CoreOptions(conf).globalIndexIgnoreMissingDelete()).isFalse(); + + conf.set(CoreOptions.GLOBAL_INDEX_IGNORE_MISSING_DELETE, true); + assertThat(new CoreOptions(conf).globalIndexIgnoreMissingDelete()).isTrue(); + } + @Test public void testBlobSplitByFileSizeDefault() { Options conf = new Options(); 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 77cb9802f3..daa5942c5c 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 @@ -29,6 +29,7 @@ import org.apache.paimon.data.serializer.InternalRowSerializer; import org.apache.paimon.fs.Path; import org.apache.paimon.globalindex.btree.BTreeIndexOptions; import org.apache.paimon.globalindex.sorted.SortedGlobalIndexBuilder; +import org.apache.paimon.index.DataEvolutionIndexSourceMeta; import org.apache.paimon.index.GlobalIndexMeta; import org.apache.paimon.index.IndexFileMeta; import org.apache.paimon.io.CompactIncrement; @@ -887,6 +888,27 @@ public class DataEvolutionRowIdReassignerTest extends TableTestBase { assertThat(readPayloads(table, predicate)).containsExactly("v4"); } + @Test + public void testReassignPreservesGlobalIndexSourceMeta() throws Exception { + FileStoreTable table = createTableWithInterleavedPartitions(); + createBTreeIndex(table); + long scanSnapshotId = table.snapshotManager().latestSnapshot().id(); + setGlobalIndexSourceMeta(table, scanSnapshotId); + + new DataEvolutionRowIdReassigner(table).reassign("test-preserve-index-source-meta"); + + List<IndexManifestEntry> entries = table.store().newIndexFileHandler().scanEntries(); + assertThat(entries).isNotEmpty(); + assertThat(entries) + .allSatisfy( + entry -> + assertThat( + DataEvolutionIndexSourceMeta.fromIndexFile( + entry.indexFile()) + .scanSnapshotId()) + .isEqualTo(scanSnapshotId)); + } + @Test public void testReassignKeepsConcurrentDisjointGlobalIndexRange() throws Exception { FileStoreTable table = createTableWithInterleavedPartitions(); @@ -1932,6 +1954,40 @@ public class DataEvolutionRowIdReassignerTest extends TableTestBase { } } + private void setGlobalIndexSourceMeta(FileStoreTable table, long scanSnapshotId) + throws Exception { + Snapshot latest = table.snapshotManager().latestSnapshot(); + IndexManifestFile indexManifestFile = table.store().indexManifestFileFactory().create(); + byte[] sourceMeta = new DataEvolutionIndexSourceMeta(scanSnapshotId).serialize(); + List<IndexManifestEntry> rewritten = new ArrayList<>(); + for (IndexManifestEntry entry : indexManifestFile.read(latest.indexManifest())) { + IndexFileMeta indexFile = entry.indexFile(); + GlobalIndexMeta globalIndex = indexFile.globalIndexMeta(); + assertThat(globalIndex).isNotNull(); + rewritten.add( + new IndexManifestEntry( + entry.kind(), + entry.partition(), + entry.bucket(), + 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(), + sourceMeta)))); + } + replaceLatestSnapshotIndexManifest( + table, latest, indexManifestFile.writeWithoutRolling(rewritten)); + } + private void replaceGlobalIndexRangesWithPartitionSpanningRanges(FileStoreTable table) throws Exception { Snapshot latest = table.snapshotManager().latestSnapshot(); diff --git a/paimon-core/src/test/java/org/apache/paimon/globalindex/DataEvolutionGlobalIndexRefreshPlannerTest.java b/paimon-core/src/test/java/org/apache/paimon/globalindex/DataEvolutionGlobalIndexRefreshPlannerTest.java new file mode 100644 index 0000000000..1eb4632453 --- /dev/null +++ b/paimon-core/src/test/java/org/apache/paimon/globalindex/DataEvolutionGlobalIndexRefreshPlannerTest.java @@ -0,0 +1,454 @@ +/* + * 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.data.BinaryRow; +import org.apache.paimon.index.DataEvolutionIndexSourceMeta; +import org.apache.paimon.index.GlobalIndexMeta; +import org.apache.paimon.index.IndexFileMeta; +import org.apache.paimon.io.DataFileMeta; +import org.apache.paimon.manifest.FileKind; +import org.apache.paimon.manifest.IndexManifestEntry; +import org.apache.paimon.manifest.ManifestEntry; +import org.apache.paimon.schema.SchemaManager; +import org.apache.paimon.schema.TableSchema; +import org.apache.paimon.stats.SimpleStats; +import org.apache.paimon.table.SpecialFields; +import org.apache.paimon.types.ArrayType; +import org.apache.paimon.types.DataField; +import org.apache.paimon.types.FloatType; +import org.apache.paimon.types.IntType; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Random; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +/** Tests for {@link DataEvolutionGlobalIndexRefreshPlanner}. */ +class DataEvolutionGlobalIndexRefreshPlannerTest { + + private static final DataField VECTOR_FIELD = + new DataField(1, "vector", new ArrayType(new FloatType())); + private static final DataField OTHER_FIELD = new DataField(2, "other", new IntType()); + private static final DataField UNRELATED_FIELD = new DataField(3, "unrelated", new IntType()); + + private SchemaManager schemaManager; + + @BeforeEach + void beforeEach() { + schemaManager = mock(SchemaManager.class); + when(schemaManager.schema(0L)) + .thenReturn( + new TableSchema( + 0L, + Collections.singletonList(OTHER_FIELD), + 1, + Collections.emptyList(), + Collections.emptyList(), + new HashMap<>(), + "")); + when(schemaManager.schema(1L)) + .thenReturn( + new TableSchema( + 1L, + Arrays.asList(VECTOR_FIELD, OTHER_FIELD, UNRELATED_FIELD), + 3, + Collections.emptyList(), + Collections.emptyList(), + new HashMap<>(), + "")); + when(schemaManager.schema(2L)) + .thenReturn( + new TableSchema( + 2L, + Arrays.asList( + new DataField( + 1, + "renamed_vector", + new ArrayType(new FloatType())), + OTHER_FIELD, + UNRELATED_FIELD), + 3, + Collections.emptyList(), + Collections.emptyList(), + new HashMap<>(), + "")); + } + + @Test + void testRefreshesOnlyForNewPhysicalIndexColumnFile() { + IndexManifestEntry index = index("index", 0, 99, 5L, BinaryRow.EMPTY_ROW, 0); + + assertThat( + plan( + Collections.singletonList( + data("vector-update", 0, 100, 6, 1, "vector")), + index)) + .containsExactly(index); + assertThat( + plan( + Collections.singletonList( + data("old-vector", 0, 100, 5, 1, "vector")), + index)) + .isEmpty(); + assertThat( + plan( + Collections.singletonList( + data("other-update", 0, 100, 6, 1, "other")), + index)) + .isEmpty(); + assertThat( + plan( + Collections.singletonList( + data("outside", 100, 100, 6, 1, "vector")), + index)) + .isEmpty(); + } + + @Test + void testLegacyIndexIsNotRefreshedAutomatically() { + IndexManifestEntry legacy = index("legacy", 0, 99, null, BinaryRow.EMPTY_ROW, 0); + + assertThat(plan(Collections.singletonList(data("base", 0, 100, 0, 1, "vector")), legacy)) + .isEmpty(); + } + + @Test + void testUsesStableFieldIdAcrossRenameAndFullWrites() { + IndexManifestEntry index = index("index", 0, 99, 5L, BinaryRow.EMPTY_ROW, 0); + + assertThat( + plan( + Collections.singletonList( + data("renamed", 0, 100, 6, 2, "renamed_vector")), + index)) + .containsExactly(index); + assertThat(plan(Collections.singletonList(data("full", 0, 100, 6, 1)), index)) + .containsExactly(index); + } + + @Test + void testRefreshesFromUpdateLayerOverBaseSchemaWithoutIndexColumn() { + IndexManifestEntry index = index("index", 0, 99, 5L, BinaryRow.EMPTY_ROW, 0); + ManifestEntry baseWithoutVector = data("base", 0, 100, 1, 0); + ManifestEntry vectorUpdate = data("vector-update", 0, 100, 6, 1, "vector"); + + assertThat(plan(Arrays.asList(baseWithoutVector, vectorUpdate), index)) + .containsExactly(index); + } + + @Test + void testHandlesSystemAndEmptyPhysicalColumns() { + IndexManifestEntry index = index("index", 0, 99, 5L, BinaryRow.EMPTY_ROW, 0); + + assertThat( + plan( + Collections.singletonList( + data( + "vector-with-system-fields", + 0, + 100, + 6, + 1, + SpecialFields.ROW_ID.name(), + "vector", + SpecialFields.SEQUENCE_NUMBER.name())), + index)) + .containsExactly(index); + assertThat( + plan( + Collections.singletonList( + data( + "system-only", + 0, + 100, + 6, + 1, + SpecialFields.ROW_ID.name(), + SpecialFields.SEQUENCE_NUMBER.name())), + index)) + .isEmpty(); + assertThat( + plan( + Collections.singletonList( + dataWithWriteCols( + "empty", 0, 100, 6, 1, Collections.emptyList())), + index)) + .isEmpty(); + } + + @Test + void testMultiColumnIndexRefreshesForEitherIndexedField() { + List<DataField> indexedFields = Arrays.asList(VECTOR_FIELD, OTHER_FIELD); + IndexManifestEntry index = + index( + "multi-column", + 0, + 99, + 5L, + BinaryRow.EMPTY_ROW, + 0, + new int[] {OTHER_FIELD.id()}); + + assertThat( + plan( + Collections.singletonList( + data("vector-update", 0, 100, 6, 1, "vector")), + Collections.singletonList(index), + indexedFields)) + .containsExactly(index); + assertThat( + plan( + Collections.singletonList( + data("extra-field-update", 0, 100, 6, 1, "other")), + Collections.singletonList(index), + indexedFields)) + .containsExactly(index); + assertThat( + plan( + Collections.singletonList( + data("unrelated-update", 0, 100, 6, 1, "unrelated")), + Collections.singletonList(index), + indexedFields)) + .isEmpty(); + } + + @Test + void testIsolatesPartitionAndBucket() { + BinaryRow partition = BinaryRow.singleColumn(1); + IndexManifestEntry index = index("index", 0, 99, 5L, partition, 2); + + ManifestEntry wrongPartition = + data("wrong-partition", 0, 100, 6, 1, BinaryRow.singleColumn(2), 2, "vector"); + ManifestEntry wrongBucket = data("wrong-bucket", 0, 100, 6, 1, partition, 1, "vector"); + assertThat(plan(Arrays.asList(wrongPartition, wrongBucket), index)).isEmpty(); + + ManifestEntry matching = data("matching", 0, 100, 6, 1, partition, 2, "vector"); + assertThat(plan(Collections.singletonList(matching), index)).containsExactly(index); + } + + @Test + void testSequenceSweepHandlesOverlappingRangesAndPreservesOrder() { + IndexManifestEntry equalSequence = index("equal", 20, 29, 5L, BinaryRow.EMPTY_ROW, 0); + IndexManifestEntry boundary = index("boundary", 10, 19, 6L, BinaryRow.EMPTY_ROW, 0); + IndexManifestEntry first = index("first", 0, 9, 8L, BinaryRow.EMPTY_ROW, 0); + + List<ManifestEntry> dataEntries = + Arrays.asList( + data("wide-equal", 0, 30, 5, 1, "vector"), + data("high-outside", 30, 10, 100, 1, "vector"), + data("boundary-update", 9, 2, 7, 1, "vector"), + data("first-update", 0, 10, 9, 1, "vector")); + + assertThat(plan(dataEntries, Arrays.asList(equalSequence, boundary, first))) + .containsExactly(boundary, first); + } + + @Test + void testSequenceSweepMatchesBruteForceForOverlappingRanges() { + Random random = new Random(123456L); + for (int round = 0; round < 10; round++) { + List<ManifestEntry> dataEntries = new ArrayList<>(); + for (int i = 0; i < 100; i++) { + dataEntries.add( + data( + "data-" + round + "-" + i, + random.nextInt(1000), + random.nextInt(100) + 1, + random.nextInt(25), + 1, + "vector")); + } + + List<IndexManifestEntry> indexEntries = new ArrayList<>(); + for (int i = 0; i < 100; i++) { + long from = random.nextInt(1000); + indexEntries.add( + index( + "index-" + round + "-" + i, + from, + from + random.nextInt(100), + (long) random.nextInt(24) + 1, + BinaryRow.EMPTY_ROW, + 0)); + } + Collections.shuffle(dataEntries, random); + Collections.shuffle(indexEntries, random); + + assertThat(plan(dataEntries, indexEntries)) + .containsExactlyElementsOf(bruteForcePlan(dataEntries, indexEntries)); + } + } + + private List<IndexManifestEntry> plan( + List<ManifestEntry> dataEntries, IndexManifestEntry indexEntry) { + return plan(dataEntries, Collections.singletonList(indexEntry)); + } + + private List<IndexManifestEntry> plan( + List<ManifestEntry> dataEntries, List<IndexManifestEntry> indexEntries) { + return plan(dataEntries, indexEntries, Collections.singletonList(VECTOR_FIELD)); + } + + private List<IndexManifestEntry> plan( + List<ManifestEntry> dataEntries, + List<IndexManifestEntry> indexEntries, + List<DataField> indexedFields) { + return DataEvolutionGlobalIndexRefreshPlanner.findIndexesToRefresh( + schemaManager, dataEntries, indexEntries, indexedFields); + } + + private List<IndexManifestEntry> bruteForcePlan( + List<ManifestEntry> dataEntries, List<IndexManifestEntry> indexEntries) { + List<IndexManifestEntry> result = new ArrayList<>(); + for (IndexManifestEntry indexEntry : indexEntries) { + GlobalIndexMeta indexMeta = indexEntry.indexFile().globalIndexMeta(); + long scanSnapshotId = + DataEvolutionIndexSourceMeta.deserialize(indexMeta.sourceMeta()) + .scanSnapshotId(); + for (ManifestEntry dataEntry : dataEntries) { + DataFileMeta file = dataEntry.file(); + if (file.maxSequenceNumber() > scanSnapshotId + && file.nonNullRowIdRange().hasIntersection(indexMeta.rowRange())) { + result.add(indexEntry); + break; + } + } + } + return result; + } + + private IndexManifestEntry index( + String fileName, + long from, + long to, + Long scanSnapshotId, + BinaryRow partition, + int bucket) { + return index(fileName, from, to, scanSnapshotId, partition, bucket, null); + } + + private IndexManifestEntry index( + String fileName, + long from, + long to, + Long scanSnapshotId, + BinaryRow partition, + int bucket, + int[] extraFieldIds) { + byte[] sourceMeta = + scanSnapshotId == null + ? null + : new DataEvolutionIndexSourceMeta(scanSnapshotId).serialize(); + return new IndexManifestEntry( + FileKind.ADD, + partition, + bucket, + new IndexFileMeta( + "lumina", + fileName, + 1L, + to - from + 1, + new GlobalIndexMeta( + from, to, VECTOR_FIELD.id(), extraFieldIds, null, sourceMeta), + null)); + } + + private ManifestEntry data( + String fileName, + long firstRowId, + long rowCount, + long maxSequenceNumber, + long schemaId, + String... writeCols) { + return data( + fileName, + firstRowId, + rowCount, + maxSequenceNumber, + schemaId, + BinaryRow.EMPTY_ROW, + 0, + writeCols); + } + + private ManifestEntry data( + String fileName, + long firstRowId, + long rowCount, + long maxSequenceNumber, + long schemaId, + BinaryRow partition, + int bucket, + String... writeCols) { + List<String> physicalColumns = writeCols.length == 0 ? null : Arrays.asList(writeCols); + DataFileMeta file = + DataFileMeta.forAppend( + fileName, + 1L, + rowCount, + SimpleStats.EMPTY_STATS, + maxSequenceNumber, + maxSequenceNumber, + schemaId, + Collections.emptyList(), + null, + null, + null, + null, + firstRowId, + physicalColumns); + return ManifestEntry.create(FileKind.ADD, partition, bucket, 1, file); + } + + private ManifestEntry dataWithWriteCols( + String fileName, + long firstRowId, + long rowCount, + long maxSequenceNumber, + long schemaId, + List<String> writeCols) { + DataFileMeta file = + DataFileMeta.forAppend( + fileName, + 1L, + rowCount, + SimpleStats.EMPTY_STATS, + maxSequenceNumber, + maxSequenceNumber, + schemaId, + Collections.emptyList(), + null, + null, + null, + null, + firstRowId, + writeCols); + return ManifestEntry.create(FileKind.ADD, BinaryRow.EMPTY_ROW, 0, 1, file); + } +} 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 11c8ddb64c..c05a1ee501 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 @@ -23,6 +23,7 @@ import org.apache.paimon.data.BinaryRow; import org.apache.paimon.fs.FileIO; import org.apache.paimon.fs.Path; import org.apache.paimon.fs.local.LocalFileIO; +import org.apache.paimon.index.DataEvolutionIndexSourceMeta; import org.apache.paimon.index.IndexFileMeta; import org.apache.paimon.index.IndexPathFactory; import org.apache.paimon.io.PojoDataFileMeta; @@ -94,7 +95,14 @@ class GlobalIndexBuilderUtilsTest { List<IndexFileMeta> metas = GlobalIndexBuilderUtils.toIndexFileMetas( - fileIO, indexPathFactory, coreOptions, range, fields, "test-type", entries); + fileIO, + indexPathFactory, + coreOptions, + range, + fields, + "test-type", + entries, + null); assertThat(metas).hasSize(1); assertThat(metas.get(0).globalIndexMeta().indexFieldId()).isEqualTo(1); @@ -115,13 +123,39 @@ class GlobalIndexBuilderUtilsTest { List<IndexFileMeta> metas = GlobalIndexBuilderUtils.toIndexFileMetas( - fileIO, indexPathFactory, coreOptions, range, fields, "test-type", entries); + fileIO, + indexPathFactory, + coreOptions, + range, + fields, + "test-type", + entries, + null); assertThat(metas).hasSize(1); assertThat(metas.get(0).globalIndexMeta().indexFieldId()).isEqualTo(1); assertThat(metas.get(0).globalIndexMeta().extraFieldIds()).isNull(); } + @Test + void testToIndexFileMetasWithSourceMeta() throws IOException { + DataField field = new DataField(1, "vec", new ArrayType(new FloatType())); + byte[] sourceMeta = new DataEvolutionIndexSourceMeta(7L).serialize(); + + List<IndexFileMeta> metas = + GlobalIndexBuilderUtils.toIndexFileMetas( + fileIO, + indexPathFactory, + coreOptions, + new Range(0, 9), + Collections.singletonList(field), + "lumina", + createDummyResultEntries(), + sourceMeta); + + assertThat(metas.get(0).globalIndexMeta().sourceMeta()).containsExactly(sourceMeta); + } + // Test: 3 columns (title + vec + id), primary column title is indexFieldId, rest in // extraFieldIds @Test @@ -136,7 +170,14 @@ class GlobalIndexBuilderUtilsTest { List<IndexFileMeta> metas = GlobalIndexBuilderUtils.toIndexFileMetas( - fileIO, indexPathFactory, coreOptions, range, fields, "test-type", entries); + fileIO, + indexPathFactory, + coreOptions, + range, + fields, + "test-type", + entries, + null); assertThat(metas).hasSize(1); assertThat(metas.get(0).globalIndexMeta().indexFieldId()).isEqualTo(1); diff --git a/paimon-core/src/test/java/org/apache/paimon/index/DataEvolutionIndexSourceMetaTest.java b/paimon-core/src/test/java/org/apache/paimon/index/DataEvolutionIndexSourceMetaTest.java new file mode 100644 index 0000000000..c281ebdd2f --- /dev/null +++ b/paimon-core/src/test/java/org/apache/paimon/index/DataEvolutionIndexSourceMetaTest.java @@ -0,0 +1,100 @@ +/* + * 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.io.DataOutputSerializer; + +import org.junit.jupiter.api.Test; + +import java.util.Arrays; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Tests for {@link DataEvolutionIndexSourceMeta}. */ +class DataEvolutionIndexSourceMetaTest { + + @Test + void testRoundTripAndDetection() { + DataEvolutionIndexSourceMeta sourceMeta = new DataEvolutionIndexSourceMeta(42L); + + byte[] serialized = sourceMeta.serialize(); + + assertThat(DataEvolutionIndexSourceMeta.isDataEvolutionMeta(serialized)).isTrue(); + assertThat(DataEvolutionIndexSourceMeta.deserialize(serialized).scanSnapshotId()) + .isEqualTo(42L); + assertThat(DataEvolutionIndexSourceMeta.isDataEvolutionMeta(null)).isFalse(); + assertThat(DataEvolutionIndexSourceMeta.isDataEvolutionMeta(new byte[] {1})).isFalse(); + } + + @Test + void testRejectsInvalidSnapshotId() { + assertThatThrownBy(() -> new DataEvolutionIndexSourceMeta(0L)) + .hasMessageContaining("snapshot id must be positive"); + } + + @Test + void testRejectsWrongMagicAndVersion() throws Exception { + DataOutputSerializer wrongMagic = new DataOutputSerializer(16); + wrongMagic.writeInt(1); + wrongMagic.writeInt(1); + wrongMagic.writeLong(1L); + assertThatThrownBy( + () -> + DataEvolutionIndexSourceMeta.deserialize( + wrongMagic.getCopyOfBuffer())) + .hasMessageContaining("Not data-evolution index source metadata"); + + byte[] wrongVersion = new DataEvolutionIndexSourceMeta(1L).serialize(); + wrongVersion[7] = 2; + assertThatThrownBy(() -> DataEvolutionIndexSourceMeta.deserialize(wrongVersion)) + .hasMessageContaining("Unsupported data-evolution index source version"); + } + + @Test + void testRejectsTruncatedAndTrailingBytes() { + byte[] serialized = new DataEvolutionIndexSourceMeta(1L).serialize(); + + assertThatThrownBy( + () -> + DataEvolutionIndexSourceMeta.deserialize( + Arrays.copyOf(serialized, serialized.length - 1))) + .hasMessageContaining("Failed to deserialize data-evolution index source metadata"); + + byte[] trailing = Arrays.copyOf(serialized, serialized.length + 1); + assertThatThrownBy(() -> DataEvolutionIndexSourceMeta.deserialize(trailing)) + .hasMessageContaining("Unexpected trailing bytes"); + } + + @Test + void testReadsFromIndexFile() { + DataEvolutionIndexSourceMeta sourceMeta = new DataEvolutionIndexSourceMeta(9L); + IndexFileMeta indexFile = + new IndexFileMeta( + "lumina", + "index-1", + 1L, + 10L, + new GlobalIndexMeta(0, 9, 1, null, null, sourceMeta.serialize()), + null); + + assertThat(DataEvolutionIndexSourceMeta.fromIndexFile(indexFile).scanSnapshotId()) + .isEqualTo(9L); + } +} diff --git a/paimon-core/src/test/java/org/apache/paimon/manifest/IndexManifestFileHandlerTest.java b/paimon-core/src/test/java/org/apache/paimon/manifest/IndexManifestFileHandlerTest.java index cfcc896525..4ecf0aa8c3 100644 --- a/paimon-core/src/test/java/org/apache/paimon/manifest/IndexManifestFileHandlerTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/manifest/IndexManifestFileHandlerTest.java @@ -21,6 +21,7 @@ package org.apache.paimon.manifest; import org.apache.paimon.TestAppendFileStore; import org.apache.paimon.data.BinaryRow; import org.apache.paimon.format.FileFormat; +import org.apache.paimon.index.DataEvolutionIndexSourceMeta; import org.apache.paimon.index.GlobalIndexMeta; import org.apache.paimon.index.IndexFileMeta; import org.apache.paimon.table.BucketMode; @@ -162,6 +163,71 @@ public class IndexManifestFileHandlerTest { assertThat(entries).containsExactly(added); } + @Test + public void testMissingGlobalIndexDeleteRejectedByDefault() throws Exception { + TestAppendFileStore fileStore = + TestAppendFileStore.createAppendStore(tempDir, new HashMap<>()); + IndexManifestFile indexManifestFile = createIndexManifestFile(fileStore); + + IndexManifestEntry previous = globalIndexEntry("prev-index", 0, 99, 1); + String manifest = + indexManifestFile.writeIndexFiles( + null, Arrays.asList(previous), BucketMode.BUCKET_UNAWARE); + IndexManifestEntry missing = globalIndexEntry("missing-index", 100, 199, 1); + + assertThatThrownBy( + () -> + indexManifestFile.writeIndexFiles( + manifest, + Arrays.asList(missing.toDeleteEntry()), + BucketMode.BUCKET_UNAWARE)) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining( + "Trying to delete global index file missing-index which does not exist."); + } + + @Test + public void testMissingGlobalIndexDeleteCanBeIgnoredExplicitly() throws Exception { + TestAppendFileStore fileStore = + TestAppendFileStore.createAppendStore(tempDir, new HashMap<>()); + IndexManifestFile indexManifestFile = createIndexManifestFile(fileStore); + + IndexManifestEntry previous = globalIndexEntry("prev-index", 0, 99, 1); + String manifest = + indexManifestFile.writeIndexFiles( + null, Arrays.asList(previous), BucketMode.BUCKET_UNAWARE); + IndexManifestEntry missing = globalIndexEntry("missing-index", 100, 199, 1); + + String ignored = + indexManifestFile.writeIndexFiles( + manifest, + Arrays.asList(missing.toDeleteEntry()), + BucketMode.BUCKET_UNAWARE, + true); + + assertThat(indexManifestFile.read(ignored)).containsExactly(previous); + } + + @Test + public void testDataEvolutionSourceMetaDoesNotDisableRangeValidation() throws Exception { + TestAppendFileStore fileStore = + TestAppendFileStore.createAppendStore(tempDir, new HashMap<>()); + IndexManifestFile indexManifestFile = createIndexManifestFile(fileStore); + IndexManifestFileHandler handler = + new IndexManifestFileHandler(indexManifestFile, BucketMode.BUCKET_UNAWARE); + + IndexManifestEntry previous = dataEvolutionIndexEntry("old-index", 0, 99, 1, 1); + String manifest = handler.write(null, Arrays.asList(previous)); + IndexManifestEntry replacement = dataEvolutionIndexEntry("new-index", 0, 99, 1, 2); + + assertThatThrownBy(() -> handler.write(manifest, Arrays.asList(replacement))) + .hasMessageContaining("overlapping row range"); + + String replaced = + handler.write(manifest, Arrays.asList(previous.toDeleteEntry(), replacement)); + assertThat(indexManifestFile.read(replaced)).containsExactly(replacement); + } + @Test public void testGlobalIndexOverlappingRangeAllowedForDifferentFieldId() throws Exception { TestAppendFileStore fileStore = @@ -282,4 +348,29 @@ public class IndexManifestFileHandlerTest { new GlobalIndexMeta(0, 1, 1, null, null, new byte[] {1}), null)); } + + private IndexManifestEntry dataEvolutionIndexEntry( + String fileName, + long rowRangeStart, + long rowRangeEnd, + int indexFieldId, + long scanSnapshotId) { + return new IndexManifestEntry( + FileKind.ADD, + BinaryRow.EMPTY_ROW, + 0, + new IndexFileMeta( + "lumina", + fileName, + 1L, + rowRangeEnd - rowRangeStart + 1, + new GlobalIndexMeta( + rowRangeStart, + rowRangeEnd, + indexFieldId, + null, + null, + new DataEvolutionIndexSourceMeta(scanSnapshotId).serialize()), + null)); + } } diff --git a/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java b/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java index 97f4a5b804..868eee8f82 100644 --- a/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java @@ -1360,6 +1360,39 @@ public class FileStoreCommitTest { } } + @ParameterizedTest + @ValueSource(booleans = {false, true}) + public void testGlobalIndexIgnoreMissingDelete(boolean ignoreMissingDelete) throws Exception { + Map<String, String> options = new HashMap<>(); + if (ignoreMissingDelete) { + options.put(CoreOptions.GLOBAL_INDEX_IGNORE_MISSING_DELETE.key(), "true"); + } + TestFileStore store = createStore(false, 1, CoreOptions.ChangelogProducer.NONE, options); + KeyValue record = gen.next(); + BinaryRow partition = gen.getPartition(record); + store.commitData(Collections.singletonList(record), s -> partition, kv -> 0); + + if (ignoreMissingDelete) { + try (FileStoreCommitImpl commit = store.newCommit()) { + commit.commit(deleteIndexCommittable(partition, "missing-index", 0, 0), false); + } + } else { + assertThatThrownBy( + () -> { + try (FileStoreCommitImpl commit = store.newCommit()) { + commit.commit( + deleteIndexCommittable( + partition, "missing-index", 0, 0), + false); + } + }) + .satisfies( + anyCauseMatches( + IllegalStateException.class, + "Trying to delete global index file missing-index which does not exist.")); + } + } + @Test public void testCommitTwiceWithDifferentKind() throws Exception { TestFileStore store = createStore(false); @@ -1918,6 +1951,28 @@ public class FileStoreCommitTest { return committable; } + private ManifestCommittable deleteIndexCommittable( + BinaryRow partition, String fileName, long rowRangeStart, long rowRangeEnd) { + ManifestCommittable committable = new ManifestCommittable(0); + committable.addFileCommittable( + new CommitMessageImpl( + partition, + 0, + null, + DataIncrement.deleteIndexIncrement( + Collections.singletonList( + new IndexFileMeta( + "btree", + fileName, + 1, + 1, + new GlobalIndexMeta( + rowRangeStart, rowRangeEnd, 0, null, null), + null))), + CompactIncrement.emptyIncrement())); + return committable; + } + private static List<ManifestEntry> tableFilesFrom( ManifestCommittable committable, CoreOptions options) { ManifestEntryChanges changes = new ManifestEntryChanges(options.bucket()); 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 12af712f1d..f9708e0c85 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 @@ -912,7 +912,8 @@ public class FullTextSearchBuilderTest extends TableTestBase { rowRange, indexFields, TestFullTextGlobalIndexerFactory.IDENTIFIER, - entries); + entries, + null); DataIncrement dataIncrement = DataIncrement.indexIncrement(indexFiles); CommitMessage message = @@ -950,7 +951,8 @@ public class FullTextSearchBuilderTest extends TableTestBase { new Range(0, documents.length - 1), Collections.singletonList(textField), TestFullTextGlobalIndexerFactory.IDENTIFIER, - writer.finish()); + writer.finish(), + null); byte[] sourceMeta = new PrimaryKeyIndexSourceMeta( 1, new PrimaryKeyIndexSourceFile("data-file", documents.length)) @@ -1021,7 +1023,8 @@ public class FullTextSearchBuilderTest extends TableTestBase { rowRange, indexFields, TestFullTextGlobalIndexerFactory.IDENTIFIER, - entries); + entries, + null); DataIncrement dataIncrement = DataIncrement.indexIncrement(indexFiles); CommitMessage message = 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 2153b44f2b..5afead03ee 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 @@ -1894,7 +1894,8 @@ public class VectorSearchBuilderTest extends TableTestBase { rowRange, indexFields, TestVectorGlobalIndexerFactory.IDENTIFIER, - entries); + entries, + null); DataIncrement dataIncrement = DataIncrement.indexIncrement(indexFiles); CommitMessage message = diff --git a/paimon-core/src/test/java/org/apache/paimon/utils/DataEvolutionUtilsTest.java b/paimon-core/src/test/java/org/apache/paimon/utils/DataEvolutionUtilsTest.java index 6f51235362..33feb9d850 100644 --- a/paimon-core/src/test/java/org/apache/paimon/utils/DataEvolutionUtilsTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/utils/DataEvolutionUtilsTest.java @@ -19,12 +19,18 @@ package org.apache.paimon.utils; import org.apache.paimon.io.DataFileMeta; +import org.apache.paimon.schema.TableSchema; import org.apache.paimon.stats.SimpleStats; +import org.apache.paimon.table.SpecialFields; +import org.apache.paimon.types.DataField; +import org.apache.paimon.types.IntType; import org.junit.jupiter.api.Test; import java.util.Arrays; import java.util.Collections; +import java.util.HashMap; +import java.util.List; import java.util.function.Function; import static org.assertj.core.api.Assertions.assertThat; @@ -33,6 +39,76 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy; /** Test for {@link DataEvolutionUtils}. */ public class DataEvolutionUtilsTest { + @Test + public void testFileFieldIdsIgnoresSystemFields() { + TableSchema schema = + new TableSchema( + 1L, + Arrays.asList( + new DataField(1, "indexed", new IntType()), + new DataField(2, "other", new IntType())), + 2, + Collections.emptyList(), + Collections.emptyList(), + new HashMap<>(), + ""); + + assertThat( + DataEvolutionUtils.fileFieldIds( + ignored -> schema, + dataFile( + "mixed.parquet", + 1, + Arrays.asList( + SpecialFields.ROW_ID.name(), + "indexed", + SpecialFields.SEQUENCE_NUMBER.name())))) + .containsExactly(1); + assertThat( + DataEvolutionUtils.fileFieldIds( + ignored -> schema, + dataFile( + "system-only.parquet", + 1, + Arrays.asList( + SpecialFields.ROW_ID.name(), + SpecialFields.SEQUENCE_NUMBER.name())))) + .isEmpty(); + } + + @Test + public void testFileFieldIdsHandlesFullEmptyAndUnrelatedWrites() { + TableSchema schema = + new TableSchema( + 1L, + Arrays.asList( + new DataField(1, "indexed", new IntType()), + new DataField(2, "other", new IntType())), + 2, + Collections.emptyList(), + Collections.emptyList(), + new HashMap<>(), + ""); + + assertThat( + DataEvolutionUtils.fileFieldIds( + ignored -> schema, dataFile("full.parquet", 1, null))) + .containsExactlyInAnyOrder(1, 2); + assertThat( + DataEvolutionUtils.fileFieldIds( + ignored -> schema, + dataFile("empty.parquet", 1, Collections.emptyList()))) + .isEmpty(); + assertThat( + DataEvolutionUtils.fileFieldIds( + ignored -> schema, + dataFile( + "unrelated.parquet", + 1, + Collections.singletonList("other")))) + .containsExactly(2); + } + @Test public void testRetrieveAnchorFileSkipsSpecialFiles() { DataFileMeta blobFile = dataFile("blob-file.blob", 1); @@ -74,6 +150,11 @@ public class DataEvolutionUtilsTest { } private static DataFileMeta dataFile(String fileName, long maxSequenceNumber) { + return dataFile(fileName, maxSequenceNumber, Collections.emptyList()); + } + + private static DataFileMeta dataFile( + String fileName, long maxSequenceNumber, List<String> writeCols) { return DataFileMeta.forAppend( fileName, 1L, @@ -88,6 +169,6 @@ public class DataEvolutionUtilsTest { null, null, 0L, - Collections.emptyList()); + writeCols); } } 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 34461d7f75..5f6dfc9049 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 @@ -30,11 +30,13 @@ import org.apache.paimon.flink.sink.StoreCommitter; import org.apache.paimon.flink.utils.BoundedOneInputOperator; import org.apache.paimon.flink.utils.JavaTypeInfo; import org.apache.paimon.flink.utils.StreamExecutionEnvironmentUtils; +import org.apache.paimon.globalindex.DataEvolutionGlobalIndexRefreshPlanner; import org.apache.paimon.globalindex.GlobalIndexMultiColumnWriter; import org.apache.paimon.globalindex.GlobalIndexSingleColumnWriter; import org.apache.paimon.globalindex.GlobalIndexWriter; import org.apache.paimon.globalindex.IndexedSplit; import org.apache.paimon.globalindex.ResultEntry; +import org.apache.paimon.index.DataEvolutionIndexSourceMeta; import org.apache.paimon.index.IndexFileMeta; import org.apache.paimon.manifest.IndexManifestEntry; import org.apache.paimon.manifest.ManifestEntry; @@ -74,15 +76,15 @@ import java.util.stream.Collectors; import static org.apache.paimon.globalindex.GlobalIndexBuilderUtils.createIndexWriter; import static org.apache.paimon.globalindex.GlobalIndexBuilderUtils.createShardIndexedSplits; -import static org.apache.paimon.globalindex.GlobalIndexBuilderUtils.filterEntriesBefore; -import static org.apache.paimon.globalindex.GlobalIndexBuilderUtils.findMinNonIndexableRowId; import static org.apache.paimon.globalindex.GlobalIndexBuilderUtils.rowRangesAfter; +import static org.apache.paimon.globalindex.GlobalIndexBuilderUtils.shouldRefreshDataEvolutionIndex; import static org.apache.paimon.globalindex.GlobalIndexBuilderUtils.toIndexFileMetas; import static org.apache.paimon.globalindex.GlobalIndexBuilderUtils.unindexedRowRanges; import static org.apache.paimon.io.CompactIncrement.emptyIncrement; import static org.apache.paimon.io.DataIncrement.deleteIndexIncrement; import static org.apache.paimon.io.DataIncrement.indexIncrement; import static org.apache.paimon.utils.Preconditions.checkArgument; +import static org.apache.paimon.utils.Preconditions.checkState; /** * Builds a Flink topology for creating generic (non-btree) global indexes with parallelism. Each @@ -304,6 +306,13 @@ public class GenericIndexTopoBuilder { List<ManifestEntry> entries = indexBuilder.scan(); List<IndexManifestEntry> deletedIndexEntries = indexBuilder.deletedIndexEntries(); + Snapshot scanSnapshot = indexBuilder.scanSnapshot(); + if (scanSnapshot == null) { + checkState( + entries.isEmpty() && deletedIndexEntries.isEmpty(), + "Global index builder returned index work without a scan snapshot."); + return false; + } return buildTopology( env, @@ -315,7 +324,7 @@ public class GenericIndexTopoBuilder { entries, deletedIndexEntries, partitionPredicate, - indexBuilder.scanSnapshot(), + scanSnapshot, maxIndexedRowId, autoIncremental); } @@ -336,7 +345,7 @@ public class GenericIndexTopoBuilder { List<ManifestEntry> entries, List<IndexManifestEntry> deletedIndexEntries, PartitionPredicate partitionPredicate, - @Nullable Snapshot scanSnapshot, + Snapshot scanSnapshot, long maxIndexedRowId, boolean autoIncremental) throws Exception { @@ -354,10 +363,6 @@ public class GenericIndexTopoBuilder { indexType, indexColumns); - long minNonIndexableRowId = - findMinNonIndexableRowId(table.schemaManager(), entries, indexColumns); - entries = filterEntriesBefore(entries, minNonIndexableRowId); - RowType rowType = table.rowType(); DataField indexField = rowType.getField(indexColumn); List<DataField> extraFields = @@ -371,22 +376,50 @@ public class GenericIndexTopoBuilder { RowType projectedRowType = SpecialFields.rowTypeWithRowId(rowType).project(readColumns); Options mergedOptions = new Options(table.options(), userOptions.toMap()); + boolean refreshDataEvolutionIndex = + shouldRefreshDataEvolutionIndex( + table, indexType, indexField, extraFields, mergedOptions); + byte[] sourceMeta = new DataEvolutionIndexSourceMeta(scanSnapshot.id()).serialize(); long rowsPerShard = mergedOptions.get(CoreOptions.GLOBAL_INDEX_ROW_COUNT_PER_SHARD); checkArgument( rowsPerShard > 0, "Option 'global-index.row-count-per-shard' must be greater than 0."); + deletedIndexEntries = new ArrayList<>(deletedIndexEntries); List<Range> rowRangesToBuild = null; if (deletedIndexEntries.isEmpty()) { if (autoIncremental) { - Snapshot snapshot = - scanSnapshot == null - ? table.snapshotManager().latestSnapshot() - : scanSnapshot; rowRangesToBuild = - unindexedRowRanges( - table, snapshot, indexType, indexedFields, partitionPredicate); + new ArrayList<>( + unindexedRowRanges( + table, + scanSnapshot, + indexType, + indexedFields, + partitionPredicate)); + if (refreshDataEvolutionIndex) { + List<IndexManifestEntry> currentIndexes = + table.store().newIndexFileHandler().scan(scanSnapshot, indexType) + .stream() + .filter( + entry -> + partitionPredicate == null + || partitionPredicate.test( + entry.partition())) + .collect(Collectors.toList()); + List<IndexManifestEntry> indexesToRefresh = + DataEvolutionGlobalIndexRefreshPlanner.findIndexesToRefresh( + table.schemaManager(), entries, currentIndexes, indexedFields); + deletedIndexEntries.addAll(indexesToRefresh); + for (IndexManifestEntry index : indexesToRefresh) { + rowRangesToBuild.add(index.indexFile().globalIndexMeta().rowRange()); + } + rowRangesToBuild = Range.sortAndMergeOverlap(rowRangesToBuild, true); + LOG.info( + "Selected {} data-evolution index files for refresh.", + indexesToRefresh.size()); + } LOG.info("Automatically selected unindexed row ranges: {}.", rowRangesToBuild); } else if (maxIndexedRowId != NO_MAX_INDEXED_ROW_ID) { rowRangesToBuild = rowRangesAfter(maxIndexedRowId); @@ -441,7 +474,8 @@ public class GenericIndexTopoBuilder { indexField, extraFields, projectedRowType, - mergedOptions)) + mergedOptions, + sourceMeta)) .setParallelism(parallelism); if (!deletedIndexEntries.isEmpty()) { @@ -531,6 +565,7 @@ public class GenericIndexTopoBuilder { private final List<DataField> extraFields; private final RowType projectedRowType; private final Options mergedOptions; + private final @Nullable byte[] sourceMeta; private transient TableRead tableRead; private transient List<DataField> indexedFields; @@ -546,7 +581,8 @@ public class GenericIndexTopoBuilder { DataField indexField, List<DataField> extraFields, RowType projectedRowType, - Options mergedOptions) { + Options mergedOptions, + @Nullable byte[] sourceMeta) { this.readBuilder = readBuilder; this.table = table; this.indexType = indexType; @@ -554,6 +590,7 @@ public class GenericIndexTopoBuilder { this.extraFields = extraFields; this.projectedRowType = projectedRowType; this.mergedOptions = mergedOptions; + this.sourceMeta = sourceMeta; } @Override @@ -678,7 +715,8 @@ public class GenericIndexTopoBuilder { shardRange, indexedFields, indexType, - resultEntries); + resultEntries, + sourceMeta); output.collect( new StreamRecord<>( new Committable( @@ -702,7 +740,8 @@ public class GenericIndexTopoBuilder { Range rowRange, List<DataField> indexFields, String indexType, - List<ResultEntry> resultEntries) + List<ResultEntry> resultEntries, + @Nullable byte[] sourceMeta) throws IOException { List<IndexFileMeta> indexFileMetas = toIndexFileMetas( @@ -712,7 +751,8 @@ public class GenericIndexTopoBuilder { rowRange, indexFields, indexType, - resultEntries); + resultEntries, + sourceMeta); return new CommitMessageImpl( partition, 0, null, indexIncrement(indexFileMetas), emptyIncrement()); } diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/LuminaVectorGlobalIndexITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/LuminaVectorGlobalIndexITCase.java index c7e11a5cd8..3f04e74947 100644 --- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/LuminaVectorGlobalIndexITCase.java +++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/LuminaVectorGlobalIndexITCase.java @@ -21,6 +21,7 @@ package org.apache.paimon.flink; import org.apache.paimon.catalog.Catalog; import org.apache.paimon.flink.globalindex.GenericGlobalIndexBuilder; import org.apache.paimon.flink.globalindex.GenericIndexTopoBuilder; +import org.apache.paimon.index.DataEvolutionIndexSourceMeta; import org.apache.paimon.index.IndexFileMeta; import org.apache.paimon.manifest.IndexManifestEntry; import org.apache.paimon.manifest.ManifestEntry; @@ -46,6 +47,11 @@ public class LuminaVectorGlobalIndexITCase extends CatalogITCaseBase { private static final String INDEX_TYPE = "lumina"; private static final String LEGACY_INDEX_TYPE = "lumina-vector-ann"; + @Override + protected Boolean sqlSyncMode() { + return true; + } + @BeforeAll static void checkLuminaAvailable() { try { @@ -541,6 +547,92 @@ public class LuminaVectorGlobalIndexITCase extends CatalogITCaseBase { assertThat(newFileNames).doesNotContainAnyElementsOf(oldFileNames); } + @Test + public void testDataEvolutionUpdateRefreshesIndexAtomically() throws Exception { + sql( + "CREATE TABLE T_REFRESH (id INT, v ARRAY<FLOAT>) WITH (" + + "'bucket' = '-1', " + + "'row-tracking.enabled' = 'true', " + + "'data-evolution.enabled' = 'true', " + + "'global-index.column-update-action' = 'IGNORE', " + + "'lumina.index.dimension' = '3', " + + "'lumina.distance.metric' = 'l2'" + + ")"); + String nullValues = + IntStream.range(0, 10) + .mapToObj(i -> "(" + i + ", CAST(NULL AS ARRAY<FLOAT>))") + .collect(Collectors.joining(",")); + sql("INSERT INTO T_REFRESH VALUES " + nullValues + "," + vectorValues(10, 20, 3)); + + FileStoreTable table = paimonTable("T_REFRESH"); + long firstDataSnapshotId = table.snapshotManager().latestSnapshot().id(); + sql( + "CALL sys.create_global_index(" + + "`table` => 'default.T_REFRESH', " + + "index_column => 'v', " + + "index_type => '" + + INDEX_TYPE + + "')"); + + List<IndexManifestEntry> oldEntries = table.store().newIndexFileHandler().scan(INDEX_TYPE); + assertThat(oldEntries).isNotEmpty(); + assertThat(oldEntries) + .allSatisfy( + entry -> + assertThat( + DataEvolutionIndexSourceMeta.fromIndexFile( + entry.indexFile()) + .scanSnapshotId()) + .isEqualTo(firstDataSnapshotId)); + List<String> oldFileNames = + oldEntries.stream() + .map(entry -> entry.indexFile().fileName()) + .collect(Collectors.toList()); + + sql("CREATE TABLE S_REFRESH (id INT, v ARRAY<FLOAT>)"); + sql( + "INSERT INTO S_REFRESH VALUES " + + "(1, ARRAY[CAST(0.0 AS FLOAT), CAST(1.0 AS FLOAT), " + + "CAST(2.0 AS FLOAT)])"); + sql( + "CALL sys.data_evolution_merge_into(" + + "'default.T_REFRESH', '', '', 'S_REFRESH', " + + "'T_REFRESH._ROW_ID=S_REFRESH.id', 'v=S_REFRESH.v', 2)"); + assertThat(sql("SELECT id FROM T_REFRESH WHERE id = 1 AND v IS NOT NULL")).hasSize(1); + + long updatedSnapshotId = table.snapshotManager().latestSnapshot().id(); + assertThat( + table.store().newIndexFileHandler().scan(INDEX_TYPE).stream() + .map(entry -> entry.indexFile().fileName())) + .containsExactlyInAnyOrderElementsOf(oldFileNames); + + sql( + "CALL sys.create_global_index(" + + "`table` => 'default.T_REFRESH', " + + "index_column => 'v', " + + "index_type => '" + + INDEX_TYPE + + "')"); + + assertThat(table.snapshotManager().latestSnapshot().id()).isEqualTo(updatedSnapshotId + 1); + List<IndexManifestEntry> refreshedEntries = + table.store().newIndexFileHandler().scan(INDEX_TYPE); + assertThat(refreshedEntries).isNotEmpty(); + assertThat( + refreshedEntries.stream() + .map(entry -> entry.indexFile().fileName()) + .collect(Collectors.toList())) + .doesNotContainAnyElementsOf(oldFileNames); + assertThat(refreshedEntries) + .allSatisfy( + entry -> + assertThat( + DataEvolutionIndexSourceMeta.fromIndexFile( + entry.indexFile()) + .scanSnapshotId()) + .isEqualTo(updatedSnapshotId)); + } + // -- Helpers -- private List<IndexFileMeta> getVectorIndexFiles(FileStoreTable table) { diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/globalindex/GenericIndexTopoBuilderTest.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/globalindex/GenericIndexTopoBuilderTest.java index 916d6c2578..658a25e619 100644 --- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/globalindex/GenericIndexTopoBuilderTest.java +++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/globalindex/GenericIndexTopoBuilderTest.java @@ -18,23 +18,31 @@ package org.apache.paimon.flink.globalindex; +import org.apache.paimon.CoreOptions; import org.apache.paimon.FileStore; +import org.apache.paimon.Snapshot; import org.apache.paimon.data.BinaryRow; import org.apache.paimon.data.BinaryRowWriter; import org.apache.paimon.data.BinaryString; import org.apache.paimon.fs.Path; -import org.apache.paimon.globalindex.GlobalIndexBuilderUtils; import org.apache.paimon.globalindex.IndexedSplit; +import org.apache.paimon.globalindex.testfulltext.TestFullTextGlobalIndexerFactory; import org.apache.paimon.io.PojoDataFileMeta; import org.apache.paimon.manifest.FileKind; import org.apache.paimon.manifest.ManifestEntry; +import org.apache.paimon.options.Options; import org.apache.paimon.schema.SchemaManager; import org.apache.paimon.schema.TableSchema; import org.apache.paimon.stats.SimpleStats; import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.source.ReadBuilder; +import org.apache.paimon.types.DataType; +import org.apache.paimon.types.DataTypes; +import org.apache.paimon.types.RowType; import org.apache.paimon.utils.FileStorePathFactory; import org.apache.paimon.utils.Range; +import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; @@ -46,6 +54,7 @@ import java.util.List; import java.util.UUID; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.mock; @@ -172,6 +181,28 @@ class GenericIndexTopoBuilderTest { assertThat(tasks).isEmpty(); } + @Test + void testRejectsIndexWorkWithoutScanSnapshot() { + GenericGlobalIndexBuilder indexBuilder = mock(GenericGlobalIndexBuilder.class); + when(indexBuilder.scan()) + .thenReturn(Collections.singletonList(createEntry(BinaryRow.EMPTY_ROW, 0L, 1))); + when(indexBuilder.deletedIndexEntries()).thenReturn(Collections.emptyList()); + when(indexBuilder.scanSnapshot()).thenReturn(null); + + assertThatThrownBy( + () -> + GenericIndexTopoBuilder.buildIndex( + StreamExecutionEnvironment.getExecutionEnvironment(), + () -> indexBuilder, + table, + "id", + "test-index", + null, + new Options())) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("index work without a scan snapshot"); + } + @Test void testAllFilesNullRowId() throws IOException { List<ManifestEntry> entries = new ArrayList<>(); @@ -478,10 +509,13 @@ class GenericIndexTopoBuilderTest { } @Test - void testAppendFilterOldFilesBeforeNewFiles() { - // Typical append: write file0[0,99](schema1), file1[100,199](schema1), - // then file2[200,299](schema0) arrives (old schema). - // Boundary = 200, keep files with firstRowId < 200. + void testAppendIncludesFilesAfterOldSchemaBoundary() throws Exception { + // The first file uses a schema from before "vec" was added. It must not prevent later + // files written with the new schema from participating in index topology construction. + List<ManifestEntry> entries = new ArrayList<>(); + entries.add(createEntryWithSchemaId(BinaryRow.EMPTY_ROW, 0L, 100, 0L)); + entries.add(createEntryWithSchemaId(BinaryRow.EMPTY_ROW, 100L, 100, 1L)); + SchemaManager schemaManager = mock(SchemaManager.class); TableSchema oldSchema = mock(TableSchema.class); TableSchema newSchema = mock(TableSchema.class); @@ -489,21 +523,34 @@ class GenericIndexTopoBuilderTest { when(schemaManager.schema(1L)).thenReturn(newSchema); when(oldSchema.fieldNames()).thenReturn(Arrays.asList("id", "name")); when(newSchema.fieldNames()).thenReturn(Arrays.asList("id", "name", "vec")); - - List<ManifestEntry> entries = new ArrayList<>(); - entries.add(createEntryWithSchemaId(BinaryRow.EMPTY_ROW, 0L, 100, 1L)); - entries.add(createEntryWithSchemaId(BinaryRow.EMPTY_ROW, 100L, 100, 1L)); - entries.add(createEntryWithSchemaId(BinaryRow.EMPTY_ROW, 200L, 100, 0L)); - - List<ManifestEntry> result = - GlobalIndexBuilderUtils.filterEntriesBefore( - entries, - GlobalIndexBuilderUtils.findMinNonIndexableRowId( - schemaManager, entries, Collections.singletonList("vec"))); - - assertThat(result).hasSize(2); - assertThat(result.get(0).file().nonNullFirstRowId()).isEqualTo(0L); - assertThat(result.get(1).file().nonNullFirstRowId()).isEqualTo(100L); + when(table.schemaManager()).thenReturn(schemaManager); + + GenericGlobalIndexBuilder indexBuilder = mock(GenericGlobalIndexBuilder.class); + when(indexBuilder.scan()).thenReturn(entries); + when(indexBuilder.deletedIndexEntries()).thenReturn(Collections.emptyList()); + Snapshot scanSnapshot = mock(Snapshot.class); + when(scanSnapshot.id()).thenReturn(1L); + when(indexBuilder.scanSnapshot()).thenReturn(scanSnapshot); + + RowType rowType = RowType.of(new DataType[] {DataTypes.STRING()}, new String[] {"vec"}); + when(table.rowType()).thenReturn(rowType); + when(table.options()).thenReturn(Collections.emptyMap()); + CoreOptions coreOptions = mock(CoreOptions.class); + when(coreOptions.dataEvolutionEnabled()).thenReturn(false); + when(table.coreOptions()).thenReturn(coreOptions); + when(table.newReadBuilder()).thenReturn(mock(ReadBuilder.class)); + + assertThat( + GenericIndexTopoBuilder.buildIndex( + StreamExecutionEnvironment.getExecutionEnvironment(), + () -> indexBuilder, + table, + "vec", + TestFullTextGlobalIndexerFactory.IDENTIFIER, + null, + new Options(), + 99L)) + .isTrue(); } // -- Helpers -- diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/FullTextSearchProcedureITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/FullTextSearchProcedureITCase.java index b621d0bc9f..f1783bd1e2 100644 --- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/FullTextSearchProcedureITCase.java +++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/FullTextSearchProcedureITCase.java @@ -19,7 +19,10 @@ package org.apache.paimon.flink.procedure; import org.apache.paimon.flink.CatalogITCaseBase; +import org.apache.paimon.index.DataEvolutionIndexSourceMeta; import org.apache.paimon.index.pkfulltext.PkFullTextIndexFile; +import org.apache.paimon.manifest.IndexManifestEntry; +import org.apache.paimon.table.FileStoreTable; import org.apache.flink.table.api.config.TableConfigOptions; import org.apache.flink.types.Row; @@ -34,6 +37,37 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy; /** IT cases for {@link FullTextSearchProcedure}. */ public class FullTextSearchProcedureITCase extends CatalogITCaseBase { + @Test + public void testDataEvolutionSourceMetaForGenericFullTextIndex() throws Exception { + sql( + "CREATE TABLE T_DE (id INT, content STRING) WITH (" + + "'bucket' = '-1', " + + "'row-tracking.enabled' = 'true', " + + "'data-evolution.enabled' = 'true'" + + ")"); + sql("INSERT INTO T_DE VALUES (1, 'apache paimon'), (2, 'lake format')"); + + FileStoreTable table = paimonTable("T_DE"); + long scanSnapshotId = table.snapshotManager().latestSnapshot().id(); + tEnv.getConfig().set(TableConfigOptions.TABLE_DML_SYNC, true); + sql( + "CALL sys.create_global_index(" + + "`table` => 'default.T_DE', " + + "index_column => 'content', " + + "index_type => 'full-text')"); + + List<IndexManifestEntry> entries = table.store().newIndexFileHandler().scan("full-text"); + assertThat(entries).isNotEmpty(); + assertThat(entries) + .allSatisfy( + entry -> + assertThat( + DataEvolutionIndexSourceMeta.fromIndexFile( + entry.indexFile()) + .scanSnapshotId()) + .isEqualTo(scanSnapshotId)); + } + @Test public void testPrimaryKeyFullTextSearchWithScoreProjection() throws Exception { createPrimaryKeyFullTextTable("T"); 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 fd13f5ee0a..5734cf84e3 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 @@ -39,6 +39,8 @@ import org.apache.paimon.utils.LongCounter; import org.apache.paimon.utils.ProjectedRow; import org.apache.paimon.utils.Range; +import javax.annotation.Nullable; + import java.io.IOException; import java.io.Serializable; import java.util.ArrayList; @@ -61,6 +63,7 @@ public class DefaultGlobalIndexBuilder implements Serializable { private final String indexType; private final Range rowRange; private final Options options; + private final @Nullable byte[] sourceMeta; public DefaultGlobalIndexBuilder( FileStoreTable table, @@ -78,7 +81,8 @@ public class DefaultGlobalIndexBuilder implements Serializable { Collections.emptyList(), indexType, rowRange, - options); + options, + null); } public DefaultGlobalIndexBuilder( @@ -90,6 +94,28 @@ public class DefaultGlobalIndexBuilder implements Serializable { String indexType, Range rowRange, Options options) { + this( + table, + partition, + readType, + indexField, + extraFields, + indexType, + rowRange, + options, + null); + } + + public DefaultGlobalIndexBuilder( + FileStoreTable table, + BinaryRow partition, + RowType readType, + DataField indexField, + List<DataField> extraFields, + String indexType, + Range rowRange, + Options options, + @Nullable byte[] sourceMeta) { this.table = table; this.partition = partition; this.readType = readType; @@ -102,6 +128,7 @@ public class DefaultGlobalIndexBuilder implements Serializable { this.indexType = indexType; this.rowRange = rowRange; this.options = options; + this.sourceMeta = sourceMeta; } /** The primary index column followed by the extra columns, in index order. */ @@ -131,7 +158,8 @@ public class DefaultGlobalIndexBuilder implements Serializable { rowRange, indexedFields(), indexType, - resultEntries); + resultEntries, + sourceMeta); DataIncrement dataIncrement = DataIncrement.indexIncrement(indexFileMetas); return new CommitMessageImpl( partition, 0, null, dataIncrement, CompactIncrement.emptyIncrement()); diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/DefaultGlobalIndexTopoBuilder.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/DefaultGlobalIndexTopoBuilder.java index faa85f6588..5b45607cec 100644 --- a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/DefaultGlobalIndexTopoBuilder.java +++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/DefaultGlobalIndexTopoBuilder.java @@ -22,15 +22,20 @@ import org.apache.paimon.Snapshot; import org.apache.paimon.data.BinaryRow; import org.apache.paimon.data.InternalRow; import org.apache.paimon.fs.Path; +import org.apache.paimon.globalindex.DataEvolutionGlobalIndexRefreshPlanner; import org.apache.paimon.globalindex.GlobalIndexBuilderUtils; import org.apache.paimon.globalindex.IndexedSplit; +import org.apache.paimon.index.DataEvolutionIndexSourceMeta; +import org.apache.paimon.io.CompactIncrement; +import org.apache.paimon.io.DataIncrement; +import org.apache.paimon.manifest.IndexManifestEntry; import org.apache.paimon.manifest.ManifestEntry; import org.apache.paimon.options.Options; import org.apache.paimon.partition.PartitionPredicate; import org.apache.paimon.reader.RecordReader; -import org.apache.paimon.schema.SchemaManager; import org.apache.paimon.table.FileStoreTable; import org.apache.paimon.table.sink.CommitMessage; +import org.apache.paimon.table.sink.CommitMessageImpl; import org.apache.paimon.table.sink.CommitMessageSerializer; import org.apache.paimon.table.source.ReadBuilder; import org.apache.paimon.types.DataField; @@ -112,16 +117,31 @@ public class DefaultGlobalIndexTopoBuilder implements GlobalIndexTopologyBuilder List<DataField> indexFields = new ArrayList<>(); indexFields.add(indexField); indexFields.addAll(extraFields); - List<String> indexColumns = - indexFields.stream().map(DataField::name).collect(Collectors.toList()); - SchemaManager schemaManager = new SchemaManager(table.fileIO(), table.location()); - long boundaryRowId = - GlobalIndexBuilderUtils.findMinNonIndexableRowId( - schemaManager, entries, indexColumns); - entries = GlobalIndexBuilderUtils.filterEntriesBefore(entries, boundaryRowId); List<Range> rowRangesToBuild = - GlobalIndexBuilderUtils.unindexedRowRanges( - table, snapshot, indexType, indexFields, partitionPredicate); + new ArrayList<>( + GlobalIndexBuilderUtils.unindexedRowRanges( + table, snapshot, indexType, indexFields, partitionPredicate)); + boolean refreshDataEvolutionIndex = + GlobalIndexBuilderUtils.shouldRefreshDataEvolutionIndex( + table, indexType, indexField, extraFields, options); + List<IndexManifestEntry> indexesToRefresh = Collections.emptyList(); + byte[] sourceMeta = new DataEvolutionIndexSourceMeta(snapshot.id()).serialize(); + if (refreshDataEvolutionIndex) { + List<IndexManifestEntry> currentIndexes = + table.store().newIndexFileHandler().scan(snapshot, indexType).stream() + .filter( + entry -> + partitionPredicate == null + || partitionPredicate.test(entry.partition())) + .collect(Collectors.toList()); + indexesToRefresh = + DataEvolutionGlobalIndexRefreshPlanner.findIndexesToRefresh( + table.schemaManager(), entries, currentIndexes, indexFields); + for (IndexManifestEntry index : indexesToRefresh) { + rowRangesToBuild.add(index.indexFile().globalIndexMeta().rowRange()); + } + rowRangesToBuild = Range.sortAndMergeOverlap(rowRangesToBuild, true); + } if (rowRangesToBuild.isEmpty()) { return Collections.emptyList(); } @@ -145,23 +165,34 @@ public class DefaultGlobalIndexTopoBuilder implements GlobalIndexTopologyBuilder extraFields, indexType, indexedSplit.rowRanges().get(0), - options); + options, + sourceMeta); byte[] builderBytes = InstantiationUtil.serializeObject(builder); byte[] splitBytes = InstantiationUtil.serializeObject(indexedSplit); taskList.add(Pair.of(builderBytes, splitBytes)); } - if (taskList.isEmpty()) { - return Collections.emptyList(); + List<CommitMessage> commitMessages = new ArrayList<>(); + if (!taskList.isEmpty()) { + int parallelism = parallelism(taskList.size(), options); + List<byte[]> commitMessageBytes = + javaSparkContext + .parallelize(taskList, parallelism) + .map(DefaultGlobalIndexTopoBuilder::buildIndex) + .collect(); + commitMessages.addAll(CommitMessageSerializer.deserializeAll(commitMessageBytes)); } - - int parallelism = parallelism(taskList.size(), options); - List<byte[]> commitMessageBytes = - javaSparkContext - .parallelize(taskList, parallelism) - .map(DefaultGlobalIndexTopoBuilder::buildIndex) - .collect(); - return CommitMessageSerializer.deserializeAll(commitMessageBytes); + for (IndexManifestEntry index : indexesToRefresh) { + commitMessages.add( + new CommitMessageImpl( + index.partition(), + index.bucket(), + null, + DataIncrement.deleteIndexIncrement( + Collections.singletonList(index.indexFile())), + CompactIncrement.emptyIncrement())); + } + return commitMessages; } static long rowsPerShard(Options options) { diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/FullTextSearchTest.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/FullTextSearchTest.scala index 5777f867fb..4c131e1f03 100644 --- a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/FullTextSearchTest.scala +++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/FullTextSearchTest.scala @@ -18,6 +18,7 @@ package org.apache.paimon.spark.sql +import org.apache.paimon.index.DataEvolutionIndexSourceMeta import org.apache.paimon.spark.PaimonSparkTestBase import scala.collection.JavaConverters._ @@ -44,6 +45,7 @@ class FullTextSearchTest extends PaimonSparkTestBase { .map(i => s"($i, 'document number $i about paimon lake format')") .mkString(",") spark.sql(s"INSERT INTO T VALUES $values") + val scanSnapshotId = loadTable("T").snapshotManager().latestSnapshot().id() val output = spark .sql( @@ -61,6 +63,11 @@ class FullTextSearchTest extends PaimonSparkTestBase { .filter(_.indexFile().indexType() == indexType) assert(indexEntries.nonEmpty) + assert( + indexEntries.forall( + entry => + DataEvolutionIndexSourceMeta.fromIndexFile(entry.indexFile()).scanSnapshotId() == + scanSnapshotId)) val totalRowCount = indexEntries.map(_.indexFile().rowCount()).sum assert(totalRowCount == 100L) } diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/LuminaVectorIndexTest.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/LuminaVectorIndexTest.scala index 6f079d4dcf..3cc2339704 100644 --- a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/LuminaVectorIndexTest.scala +++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/LuminaVectorIndexTest.scala @@ -18,6 +18,7 @@ package org.apache.paimon.spark.sql +import org.apache.paimon.index.DataEvolutionIndexSourceMeta import org.apache.paimon.spark.PaimonSparkTestBase import scala.collection.JavaConverters._ @@ -69,6 +70,54 @@ class LuminaVectorIndexTest extends PaimonSparkTestBase { } } + test("create lumina vector index after adding the index column") { + withTable("T") { + spark.sql(""" + |CREATE TABLE T (id INT) + |TBLPROPERTIES ( + | 'bucket' = '-1', + | 'global-index.row-count-per-shard' = '10000', + | 'row-tracking.enabled' = 'true', + | 'data-evolution.enabled' = 'true') + |""".stripMargin) + spark.sql("INSERT INTO T VALUES (0), (1), (2), (3), (4)") + + spark.sql("ALTER TABLE T ADD COLUMNS (v ARRAY<FLOAT>)") + spark.sql(""" + |INSERT INTO T VALUES + | (5, array(5.0f, 6.0f, 7.0f)), + | (6, array(6.0f, 7.0f, 8.0f)), + | (7, array(7.0f, 8.0f, 9.0f)), + | (8, array(8.0f, 9.0f, 10.0f)), + | (9, array(9.0f, 10.0f, 11.0f)) + |""".stripMargin) + + val output = spark + .sql( + s"CALL sys.create_global_index(table => 'test.T', index_column => 'v', index_type => '$indexType', options => '$defaultOptions')") + .collect() + .head + assert(output.getBoolean(0)) + + val indexEntries = + loadTable("T").store().newIndexFileHandler().scan(indexType).asScala + assert(indexEntries.size == 1) + val indexFile = indexEntries.head.indexFile() + assert(indexFile.rowCount() == 10L) + assert(indexFile.globalIndexMeta().rowRangeStart() == 0L) + assert(indexFile.globalIndexMeta().rowRangeEnd() == 9L) + + val searchResult = spark + .sql(""" + |SELECT id FROM vector_search( + | 'T', 'v', array(5.0f, 6.0f, 7.0f), 5, + | map('refine_factor', '5')) + |""".stripMargin) + .collect() + assert(searchResult.map(_.getInt(0)).toSet == (5 to 9).toSet) + } + } + test("create lumina vector index - legacy index type") { withTable("T") { spark.sql(""" @@ -406,4 +455,90 @@ class LuminaVectorIndexTest extends PaimonSparkTestBase { assert(searchResult.length == 10) } } + + test("incremental build atomically refreshes vectors updated inside an indexed range") { + withTable("T") { + spark.sql(""" + |CREATE TABLE T (id INT, v ARRAY<FLOAT>) + |TBLPROPERTIES ( + | 'bucket' = '-1', + | 'global-index.row-count-per-shard' = '10000', + | 'row-tracking.enabled' = 'true', + | 'data-evolution.enabled' = 'true', + | 'lumina.distance.metric' = 'l2', + | 'global-index.column-update-action' = 'IGNORE') + |""".stripMargin) + + val values = (0 until 200) + .map { + case i if i < 100 => s"($i, cast(NULL as ARRAY<FLOAT>))" + case i => + s"($i, array(cast($i as float), cast(${i + 1} as float), cast(${i + 2} as float)))" + } + .mkString(",") + spark.sql(s"INSERT INTO T VALUES $values") + + val table = loadTable("T") + val firstDataSnapshotId = table.snapshotManager().latestSnapshot().id() + spark.sql( + s"CALL sys.create_global_index(table => 'test.T', index_column => 'v', " + + s"index_type => '$indexType', options => '$defaultOptions')") + + val oldEntries = table.store().newIndexFileHandler().scan(indexType).asScala + assert(oldEntries.nonEmpty) + assert( + oldEntries.forall( + entry => + DataEvolutionIndexSourceMeta.fromIndexFile(entry.indexFile()).scanSnapshotId() == + firstDataSnapshotId)) + val oldFileNames = oldEntries.map(_.indexFile().fileName()).toSet + + spark.sql(""" + |MERGE INTO T + |USING (SELECT 0 AS id, + | array(cast(0 as float), cast(1 as float), cast(2 as float)) AS v) S + |ON T.id = S.id + |WHEN MATCHED THEN UPDATE SET v = S.v + |""".stripMargin) + + val updatedSnapshot = table.snapshotManager().latestSnapshot() + val stillVisible = + table.store().newIndexFileHandler().scan(updatedSnapshot, indexType).asScala + assert(stillVisible.map(_.indexFile().fileName()).toSet == oldFileNames) + // Retrieve every indexed candidate before exact reranking so ANN recall is deterministic. + val beforeRefresh = spark + .sql(""" + |SELECT id FROM vector_search( + | 'T', 'v', array(0.0f, 1.0f, 2.0f), 1, + | map('refine_factor', '200')) + |""".stripMargin) + .collect() + assert(beforeRefresh.head.getInt(0) != 0) + + spark.sql( + s"CALL sys.create_global_index(table => 'test.T', index_column => 'v', " + + s"index_type => '$indexType', options => '$defaultOptions')") + + val indexCommitSnapshot = table.snapshotManager().latestSnapshot() + assert(indexCommitSnapshot.id() == updatedSnapshot.id() + 1) + val refreshedEntries = + table.store().newIndexFileHandler().scan(indexCommitSnapshot, indexType).asScala + assert(refreshedEntries.nonEmpty) + assert(refreshedEntries.map(_.indexFile().fileName()).toSet.intersect(oldFileNames).isEmpty) + assert( + refreshedEntries.forall( + entry => + DataEvolutionIndexSourceMeta.fromIndexFile(entry.indexFile()).scanSnapshotId() == + updatedSnapshot.id())) + + val afterRefresh = spark + .sql(""" + |SELECT id FROM vector_search( + | 'T', 'v', array(0.0f, 1.0f, 2.0f), 1, + | map('refine_factor', '200')) + |""".stripMargin) + .collect() + assert(afterRefresh.head.getInt(0) == 0) + } + } }
