From 15606ce454d86a7f3387dd5d21583de297a9287d Mon Sep 17 00:00:00 2001 From: "wenchao.wu" Date: Mon, 26 Oct 2026 18:49:21 +0800 Subject: [PATCH] [core] Fix lookup returning rows deleted by deletion vectors Lookup ignored deletion vectors after compaction dropped the delete record, so partial lookup and the query service returned deleted rows. Deletion vectors are now loaded from the snapshot that produced the installed files. A newer commit can no longer hide or reveal rows in a file plan that has not been applied yet. If that snapshot has expired, refresh fails with OutOfRangeException instead of serving the files without their vectors. --- .../paimon/table/query/LocalTableQuery.java | 276 ++++++++++--- .../table/PrimaryKeySimpleTableTest.java | 35 +- .../LocalTableQueryDeletionVectorTest.java | 379 ++++++++++++++++++ .../lookup/PrimaryKeyPartialLookupTable.java | 8 +- .../flink/service/QueryExecutorOperator.java | 4 +- .../paimon/flink/RemoteLookupJoinITCase.java | 3 +- .../paimon/service/KvQueryTableTest.java | 5 +- 7 files changed, 642 insertions(+), 68 deletions(-) create mode 100644 paimon-core/src/test/java/org/apache/paimon/table/query/LocalTableQueryDeletionVectorTest.java diff --git a/paimon-core/src/main/java/org/apache/paimon/table/query/LocalTableQuery.java b/paimon-core/src/main/java/org/apache/paimon/table/query/LocalTableQuery.java index 6f355d6602e5..66c4a93d1ed9 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/query/LocalTableQuery.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/query/LocalTableQuery.java @@ -22,38 +22,51 @@ import org.apache.paimon.FileStore; import org.apache.paimon.KeyValue; import org.apache.paimon.KeyValueFileStore; +import org.apache.paimon.Snapshot; import org.apache.paimon.data.BinaryRow; import org.apache.paimon.data.InternalRow; import org.apache.paimon.data.serializer.InternalRowSerializer; import org.apache.paimon.data.serializer.InternalSerializers; import org.apache.paimon.data.serializer.RowCompactedSerializer; import org.apache.paimon.deletionvectors.DeletionVector; +import org.apache.paimon.deletionvectors.DeletionVectorsIndexFile; import org.apache.paimon.disk.IOManager; import org.apache.paimon.fs.FileIO; +import org.apache.paimon.index.IndexFileHandler; +import org.apache.paimon.index.IndexFileMeta; import org.apache.paimon.io.DataFileMeta; import org.apache.paimon.io.KeyValueFileReaderFactory; import org.apache.paimon.io.cache.CacheManager; import org.apache.paimon.lookup.LookupStoreFactory; +import org.apache.paimon.manifest.IndexManifestEntry; import org.apache.paimon.mergetree.Levels; import org.apache.paimon.mergetree.LookupFile; import org.apache.paimon.mergetree.LookupLevels; import org.apache.paimon.mergetree.lookup.LookupSerializerFactory; +import org.apache.paimon.mergetree.lookup.PersistProcessor; +import org.apache.paimon.mergetree.lookup.PersistValueAndPosProcessor; import org.apache.paimon.mergetree.lookup.PersistValueProcessor; +import org.apache.paimon.mergetree.lookup.PositionedKeyValue; import org.apache.paimon.mergetree.lookup.RemoteLookupFileManager; import org.apache.paimon.operation.metrics.PartialLookupMetrics; import org.apache.paimon.options.Options; import org.apache.paimon.reader.RecordReader; import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.source.OutOfRangeException; import org.apache.paimon.types.RowType; import org.apache.paimon.utils.Filter; import org.apache.paimon.utils.KeyComparatorSupplier; import org.apache.paimon.utils.Preconditions; +import org.apache.paimon.utils.SnapshotManager; import org.apache.paimon.shade.caffeine2.com.github.benmanes.caffeine.cache.Cache; import javax.annotation.Nullable; +import java.io.FileNotFoundException; import java.io.IOException; +import java.util.ArrayList; +import java.util.Collections; import java.util.Comparator; import java.util.List; import java.util.Map; @@ -80,6 +93,18 @@ public class LocalTableQuery implements TableQuery { private final int startLevel; + private final boolean deletionVectorsEnabled; + + @Nullable private final IndexFileHandler indexFileHandler; + + @Nullable private final SnapshotManager snapshotManager; + + private final Object deletionVectorCacheLock = new Object(); + + private long cachedDeletionVectorSnapshotId = Long.MIN_VALUE; + + @Nullable private List cachedDeletionVectorEntries; + private IOManager ioManager; @Nullable private volatile Cache lookupFileCache; @@ -121,14 +146,42 @@ public LocalTableQuery(FileStoreTable table) { options.lookupCacheMaxMemory(), options.lookupCacheHighPrioPoolRatio()), new RowCompactedSerializer(keyType).createSliceComparator()); - startLevel = options.needLookup() ? 1 : 0; + // needLookup() is also true for deletion vectors, but that flag is the compaction lookup + // which starts above the level being merged. A point query has to see level 0, where an + // uncompacted delete still lives, only when that level-0 row is already the merged value. + // Partial-update, aggregation and sequence fields keep the historical level: a level-0 + // record is unmerged, and returning it skips the merge function. + this.deletionVectorsEnabled = options.deletionVectorsEnabled(); + if (deletionVectorsEnabled) { + this.indexFileHandler = table.store().newIndexFileHandler(); + this.snapshotManager = table.snapshotManager(); + } else { + this.indexFileHandler = null; + this.snapshotManager = null; + } + boolean level0RowIsComplete = + options.mergeEngine() == CoreOptions.MergeEngine.DEDUPLICATE + && options.sequenceField().isEmpty(); + if (deletionVectorsEnabled && level0RowIsComplete) { + this.startLevel = 0; + } else { + this.startLevel = options.needLookup() ? 1 : 0; + } } + /** + * Installs {@code dataFiles} and drops {@code beforeFiles} for one bucket. + * + * @param snapshotId snapshot that produced these files. Deletion vectors are loaded from this + * snapshot so a newer commit cannot hide or reveal rows in the files being installed. A + * negative id means the files are not committed yet and no deletion vector is applied. + */ public void refreshFiles( BinaryRow partition, int bucket, List beforeFiles, - List dataFiles) { + List dataFiles, + long snapshotId) { // Both tableView and its nested bucket maps are ConcurrentHashMaps; this nested // computeIfAbsent pattern relies on each map providing atomic insertion. BucketLookupState state = @@ -137,68 +190,117 @@ public void refreshFiles( .computeIfAbsent(bucket, k -> new BucketLookupState()); state.lock.writeLock().lock(); try { + // Load before touching the levels so a failure leaves the previous view intact. + // Compaction can mark a row deleted without rewriting its data file. Streaming + // plans leave deletion files empty (SnapshotReaderImpl fills them for batch plans + // only), and the updated file may not appear in this split at all, so the bucket + // index is read from snapshotId rather than the latest snapshot. Unchanged index + // files are not read again. + Map deletionVectors = + loadDeletionVectors(partition, bucket, snapshotId, state); if (state.lookupLevels == null) { // Initial phase: ignore beforeFiles as they represent deletions from previous state state.lookupLevels = createLookupLevels(partition, bucket, dataFiles); } else { state.lookupLevels.getLevels().update(beforeFiles, dataFiles); } + state.deletionVectors = deletionVectors; } finally { state.lock.writeLock().unlock(); } } - private LookupLevels createLookupLevels( + private Map loadDeletionVectors( + BinaryRow partition, int bucket, long snapshotId, BucketLookupState state) { + if (!deletionVectorsEnabled) { + return Collections.emptyMap(); + } + List fileMetas = new ArrayList<>(); + for (IndexManifestEntry entry : deletionVectorEntries(snapshotId)) { + if (entry.bucket() == bucket && entry.partition().equals(partition)) { + fileMetas.add(entry.indexFile()); + } + } + if (fileMetas.equals(state.deletionVectorFiles)) { + return state.deletionVectors; + } + if (fileMetas.isEmpty()) { + state.deletionVectorFiles = fileMetas; + return Collections.emptyMap(); + } + // Record the file list only after the read succeeds. A failure leaves the previous + // list and vectors in place, so the next refresh retries instead of trusting them. + Map loaded = + indexFileHandler.readAllDeletionVectors(partition, bucket, fileMetas); + state.deletionVectorFiles = fileMetas; + return loaded; + } + + private List deletionVectorEntries(long snapshotId) { + synchronized (deletionVectorCacheLock) { + if (snapshotId != cachedDeletionVectorSnapshotId) { + cachedDeletionVectorEntries = + snapshotId < 0 + ? Collections.emptyList() + : indexFileHandler.scan( + snapshotOf(snapshotId), + DeletionVectorsIndexFile.DELETION_VECTORS_INDEX); + cachedDeletionVectorSnapshotId = snapshotId; + } + return cachedDeletionVectorEntries; + } + } + + private Snapshot snapshotOf(long snapshotId) { + try { + return snapshotManager.tryGetSnapshot(snapshotId); + } catch (FileNotFoundException e) { + // The files were planned from a snapshot that has since expired. Serving them without + // their deletion vectors would resurrect deleted rows, so fail with the exception the + // lookup function already handles by reopening against the current snapshot. + throw new OutOfRangeException( + String.format( + "The snapshot %d that produced these files has expired, so its " + + "deletion vectors can no longer be read. Lookup has to " + + "restart from a live snapshot.", + snapshotId)); + } + } + + private LookupLevels createLookupLevels( BinaryRow partition, int bucket, List dataFiles) { Levels levels = new Levels(keyComparatorSupplier.get(), dataFiles, options.numLevels()); - // TODO pass DeletionVector factory + // Do not apply the deletion vector while building the lookup file. Dropping the key here + // would make lookup continue into an older level and return a stale value. The vector is + // checked against the stored row position at lookup time, including after a later refresh. KeyValueFileReaderFactory factory = readerFactoryBuilder.build(partition, bucket, DeletionVector.emptyFactory()); - Options options = this.options.toConfiguration(); + if (deletionVectorsEnabled) { + return newLookupLevels( + partition, + bucket, + levels, + factory, + PersistValueAndPosProcessor.factory(readerFactoryBuilder.readValueType())); + } - RowType readValueType = readerFactoryBuilder.readValueType(); LookupLevels lookupLevels = - new LookupLevels<>( - schemaId -> readValueType, - 0L, + newLookupLevels( + partition, + bucket, levels, - keyComparatorSupplier.get(), - readerFactoryBuilder.keyType(), - PersistValueProcessor.factory(readValueType), - LookupSerializerFactory.INSTANCE.get(), - file -> { - RecordReader reader = factory.createRecordReader(file); - if (cacheRowFilter != null) { - reader = - reader.filter( - keyValue -> cacheRowFilter.test(keyValue.value())); - } - return reader; - }, - file -> - Preconditions.checkNotNull(ioManager, "IOManager is required.") - .createChannel( - localFilePrefix( - partitionType, partition, bucket, file)) - .getPathFile(), - lookupStoreFactory, - bloomFilterBuilderFactory(options), - lookupFileCache(options)); - - // Optimization - download lookup files if already persisted to object store - // We download these files if three conditions are met - // 1) lookup.remote-file.enabled is true - files are persisted in the first place - // 2) deletion-vectors.enabled is false - SSTables only contain row positions, not values, - // when DVs are enabled - // 3) The client is accessing the full data row, as opposed to a projection - // - The persisted remote SSTable files are created during compaction and hold the entire - // data row value - // - We could deserialize and project in memory, but we'll have to read much more data, - // not as clear of a win + factory, + PersistValueProcessor.factory(readerFactoryBuilder.readValueType())); + + // Optimization - download lookup files if already persisted to object store. + // This branch builds value-only files, so remote files are used when: + // 1. lookup.remote-file.enabled is true, so compaction persisted them + // 2. deletion vectors are disabled. Their SSTables store a row position beside the + // value, which these remote files do not contain + // 3. the client reads the full row. Remote files are written at compaction time with + // the whole value; projecting after a full download reads more data than it saves boolean fullValueRead = readerFactoryBuilder.readValueType().equals(rowType); - if (this.options.lookupRemoteFileEnabled() - && !this.options.deletionVectorsEnabled() - && fullValueRead) { + if (this.options.lookupRemoteFileEnabled() && fullValueRead) { // Calling the constructor tells `lookupLevels` to load remote files new RemoteLookupFileManager<>( fileIO, @@ -210,6 +312,39 @@ private LookupLevels createLookupLevels( return lookupLevels; } + private LookupLevels newLookupLevels( + BinaryRow partition, + int bucket, + Levels levels, + KeyValueFileReaderFactory factory, + PersistProcessor.Factory processorFactory) { + Options lookupOptions = this.options.toConfiguration(); + RowType readValueType = readerFactoryBuilder.readValueType(); + return new LookupLevels<>( + schemaId -> readValueType, + 0L, + levels, + keyComparatorSupplier.get(), + readerFactoryBuilder.keyType(), + processorFactory, + LookupSerializerFactory.INSTANCE.get(), + file -> { + RecordReader reader = factory.createRecordReader(file); + if (cacheRowFilter != null) { + reader = reader.filter(keyValue -> cacheRowFilter.test(keyValue.value())); + } + return reader; + }, + file -> + Preconditions.checkNotNull(ioManager, "IOManager is required.") + .createChannel( + localFilePrefix(partitionType, partition, bucket, file)) + .getPathFile(), + lookupStoreFactory, + bloomFilterBuilderFactory(lookupOptions), + lookupFileCache(lookupOptions)); + } + private Cache lookupFileCache(Options options) { Cache cache = lookupFileCache; if (cache == null) { @@ -260,22 +395,52 @@ private InternalRow lookup( state.lock.readLock().lock(); try { - LookupLevels lookupLevels = state.lookupLevels; + LookupLevels lookupLevels = state.lookupLevels; if (lookupLevels == null) { return null; } - KeyValue kv = lookupLevels.lookup(key, startLevel, context); - if (kv == null || kv.valueKind().isRetract()) { - return null; - } else { - return kv.value(); - } + Object lookedUp = lookup(lookupLevels, key, context); + return visibleValue(lookedUp, state.deletionVectors); } finally { state.lock.readLock().unlock(); } } + @SuppressWarnings({"unchecked", "rawtypes"}) + private Object lookup( + LookupLevels lookupLevels, + InternalRow key, + @Nullable LookupLevels.LookupContext context) + throws IOException { + return ((LookupLevels) lookupLevels).lookup(key, startLevel, context); + } + + @Nullable + private InternalRow visibleValue( + @Nullable Object lookedUp, Map deletionVectors) { + if (lookedUp == null) { + return null; + } + KeyValue kv; + if (lookedUp instanceof PositionedKeyValue) { + PositionedKeyValue positioned = (PositionedKeyValue) lookedUp; + DeletionVector deletionVector = deletionVectors.get(positioned.fileName()); + // A hit inside the deletion vector is the newest record and is gone. Older levels are + // older versions of the same key, so do not continue the search. + if (deletionVector != null && deletionVector.isDeleted(positioned.rowPosition())) { + return null; + } + kv = positioned.keyValue(); + } else { + kv = (KeyValue) lookedUp; + } + if (kv.valueKind().isRetract()) { + return null; + } + return kv.value(); + } + @Override public LocalTableQuery withValueProjection(int[] projection) { this.readerFactoryBuilder.withReadValueType(rowType.project(projection)); @@ -329,6 +494,11 @@ private static class BucketLookupState { private final ReadWriteLock lock = new ReentrantReadWriteLock(); - @Nullable private LookupLevels lookupLevels; + @Nullable private LookupLevels lookupLevels; + + private Map deletionVectors = Collections.emptyMap(); + + /** Index files that produced {@link #deletionVectors}. Null until the first load. */ + @Nullable private List deletionVectorFiles; } } diff --git a/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeySimpleTableTest.java b/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeySimpleTableTest.java index f6bc8fcc423c..416d541cb003 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeySimpleTableTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeySimpleTableTest.java @@ -2717,7 +2717,11 @@ public void testTableQueryDownloadsRemoteLookupFile() throws Exception { table.newLocalTableQuery().withIOManager(ioManager).withMetrics(metrics); for (DataSplit split : dataSplits) { query.refreshFiles( - split.partition(), split.bucket(), Collections.emptyList(), split.dataFiles()); + split.partition(), + split.bucket(), + Collections.emptyList(), + split.dataFiles(), + split.snapshotId()); } InternalRow value = query.lookup(row(1), 0, row(10)); assertThat(value).isNotNull(); @@ -2739,7 +2743,11 @@ public void testTableQueryDownloadsRemoteLookupFile() throws Exception { .withIOManager(ioManager); for (DataSplit split : dataSplits) { projected.refreshFiles( - split.partition(), split.bucket(), Collections.emptyList(), split.dataFiles()); + split.partition(), + split.bucket(), + Collections.emptyList(), + split.dataFiles(), + split.snapshotId()); } assertThatThrownBy(() -> projected.lookup(row(1), 0, row(10))) .isInstanceOf(Exception.class); @@ -2756,7 +2764,11 @@ public void testTableQueryDownloadsRemoteLookupFile() throws Exception { .withIOManager(ioManager); for (DataSplit split : dataSplits) { rebuilt.refreshFiles( - split.partition(), split.bucket(), Collections.emptyList(), split.dataFiles()); + split.partition(), + split.bucket(), + Collections.emptyList(), + split.dataFiles(), + split.snapshotId()); } InternalRow projectedValue = rebuilt.lookup(row(1), 0, row(10)); assertThat(projectedValue).isNotNull(); @@ -2980,7 +2992,7 @@ private void innerTestTableQuery(FileStoreTable table) throws Exception { LocalTableQuery query = table.newLocalTableQuery(); query.withIOManager(ioManager); - refreshTableService(query, commitMessages1); + refreshTableService(query, commitMessages1, table.snapshotManager().latestSnapshot().id()); InternalRow value = query.lookup(row(1), 0, row(10)); assertThat(value).isNotNull(); assertThat(BATCH_ROW_TO_STRING.apply(value)) @@ -2993,7 +3005,7 @@ private void innerTestTableQuery(FileStoreTable table) throws Exception { List commitMessages2 = write.prepareCommit(true, 0); commit.commit(0, commitMessages2); - refreshTableService(query, commitMessages2); + refreshTableService(query, commitMessages2, table.snapshotManager().latestSnapshot().id()); value = query.lookup(row(1), 0, row(10)); assertThat(value).isNotNull(); @@ -3014,8 +3026,8 @@ private void innerTestTableQuery(FileStoreTable table) throws Exception { query.close(); query.withValueProjection(new int[] {2, 1, 0}); - refreshTableService(query, commitMessages1); - refreshTableService(query, commitMessages2); + refreshTableService(query, commitMessages1, table.snapshotManager().latestSnapshot().id()); + refreshTableService(query, commitMessages2, table.snapshotManager().latestSnapshot().id()); value = query.lookup(row(1), 0, row(10)); assertThat(value).isNotNull(); Function projectToString = @@ -3027,19 +3039,22 @@ private void innerTestTableQuery(FileStoreTable table) throws Exception { commit.close(); } - private void refreshTableService(LocalTableQuery query, List commitMessages) { + private void refreshTableService( + LocalTableQuery query, List commitMessages, long snapshotId) { for (CommitMessage m : commitMessages) { CommitMessageImpl msg = (CommitMessageImpl) m; query.refreshFiles( msg.partition(), msg.bucket(), Collections.emptyList(), - msg.newFilesIncrement().newFiles()); + msg.newFilesIncrement().newFiles(), + snapshotId); query.refreshFiles( msg.partition(), msg.bucket(), msg.compactIncrement().compactBefore(), - msg.compactIncrement().compactAfter()); + msg.compactIncrement().compactAfter(), + snapshotId); } } diff --git a/paimon-core/src/test/java/org/apache/paimon/table/query/LocalTableQueryDeletionVectorTest.java b/paimon-core/src/test/java/org/apache/paimon/table/query/LocalTableQueryDeletionVectorTest.java new file mode 100644 index 000000000000..3eac175a3dde --- /dev/null +++ b/paimon-core/src/test/java/org/apache/paimon/table/query/LocalTableQueryDeletionVectorTest.java @@ -0,0 +1,379 @@ +/* + * 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.table.query; + +import org.apache.paimon.CoreOptions; +import org.apache.paimon.catalog.Catalog; +import org.apache.paimon.catalog.CatalogContext; +import org.apache.paimon.catalog.CatalogFactory; +import org.apache.paimon.catalog.Identifier; +import org.apache.paimon.data.BinaryRow; +import org.apache.paimon.data.GenericRow; +import org.apache.paimon.data.InternalRow; +import org.apache.paimon.deletionvectors.DeletionVectorsIndexFile; +import org.apache.paimon.disk.IOManager; +import org.apache.paimon.fs.Path; +import org.apache.paimon.io.DataFileMeta; +import org.apache.paimon.manifest.IndexManifestEntry; +import org.apache.paimon.options.Options; +import org.apache.paimon.schema.Schema; +import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.sink.BatchTableCommit; +import org.apache.paimon.table.sink.BatchTableWrite; +import org.apache.paimon.table.sink.BatchWriteBuilder; +import org.apache.paimon.table.source.DataSplit; +import org.apache.paimon.table.source.IncrementalSplit; +import org.apache.paimon.table.source.OutOfRangeException; +import org.apache.paimon.table.source.Split; +import org.apache.paimon.table.source.StreamTableScan; +import org.apache.paimon.types.DataTypes; +import org.apache.paimon.types.RowKind; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.util.Collections; +import java.util.List; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Lookup must not return a row whose only deletion record is a deletion vector. */ +public class LocalTableQueryDeletionVectorTest { + + @TempDir java.nio.file.Path tempDir; + + private Catalog catalog; + private final Identifier identifier = Identifier.create("my_db", "T"); + + @BeforeEach + public void before() throws Exception { + Options options = new Options(); + options.set("warehouse", new Path(tempDir.toString() + "/warehouse").toUri().toString()); + catalog = CatalogFactory.createCatalog(CatalogContext.create(options)); + catalog.createDatabase("my_db", true); + } + + @Test + public void testLookupHonorsDeletionVectorAfterCompaction() throws Exception { + FileStoreTable table = createTable(); + IOManager ioManager = IOManager.create(tempDir.toString() + "/io"); + LocalTableQuery query = table.newLocalTableQuery().withIOManager(ioManager); + // Flink lookup follows every snapshot, including compaction, so removed files are dropped + // from the levels. The default stream scan skips compact commits. + StreamTableScan scan = fileMonitorScan(table); + try { + for (int key = 1; key <= 20; key++) { + write(table, GenericRow.of(key, key * 10)); + } + compact(table, ioManager, true); + refresh(query, scan); + assertThat(value(query, 1)).isEqualTo(10); + assertThat(value(query, 2)).isEqualTo(20); + + // The delete is still a level-0 record. Streaming lookup has to see it; a batch scan + // skips level 0 until compaction. + write(table, GenericRow.ofKind(RowKind.DELETE, 1, 10)); + refresh(query, scan); + assertThat(value(query, 1)).isNull(); + assertThat(value(query, 2)).isEqualTo(20); + + // Compaction drops the delete record and marks the old file position instead. + compact(table, ioManager, false); + refresh(query, scan); + assertThat(deletionVectorIndex(table)).isNotEmpty(); + assertThat(value(query, 1)).isNull(); + assertThat(value(query, 2)).isEqualTo(20); + assertThat(highLevelRowCount(table)).isGreaterThan(1); + + write(table, GenericRow.of(1, 99)); + refresh(query, scan); + assertThat(value(query, 1)).isEqualTo(99); + assertThat(value(query, 2)).isEqualTo(20); + } finally { + query.close(); + ioManager.close(); + } + } + + @Test + public void testDeletionVectorUsesTheFileSnapshot() throws Exception { + FileStoreTable table = createTable(); + IOManager ioManager = IOManager.create(tempDir.toString() + "/io"); + LocalTableQuery query = table.newLocalTableQuery().withIOManager(ioManager); + try { + for (int key = 1; key <= 20; key++) { + write(table, GenericRow.of(key, key * 10)); + } + compact(table, ioManager, true); + + // Hold the compacted file plan, then let a later commit mark one of its positions. + List held = fileMonitorScan(table).plan().splits(); + assertThat(held).isNotEmpty(); + long heldSnapshotId = snapshotId(held.get(0)); + + write(table, GenericRow.of(1, 99)); + compact(table, ioManager, false); + assertThat(table.snapshotManager().latestSnapshot().id()).isGreaterThan(heldSnapshotId); + assertThat(deletionVectorIndex(table)).isNotEmpty(); + + refreshSplits(query, held); + assertThat(value(query, 1)).isEqualTo(10); + assertThat(value(query, 2)).isEqualTo(20); + } finally { + query.close(); + ioManager.close(); + } + } + + @Test + public void testExpiredFileSnapshotIsReportedAsOutOfRange() throws Exception { + FileStoreTable table = createTable(); + IOManager ioManager = IOManager.create(tempDir.toString() + "/io"); + LocalTableQuery query = table.newLocalTableQuery().withIOManager(ioManager); + try { + for (int key = 1; key <= 20; key++) { + write(table, GenericRow.of(key, key * 10)); + } + compact(table, ioManager, true); + write(table, GenericRow.ofKind(RowKind.DELETE, 1, 10)); + compact(table, ioManager, false); + assertThat(deletionVectorIndex(table)).isNotEmpty(); + + // Hold a plan, then let the snapshot it came from expire. Serving these files + // without their deletion vectors would resurrect the deleted row, so refresh has + // to fail with the exception the lookup function reopens on. + List held = fileMonitorScan(table).plan().splits(); + assertThat(held).isNotEmpty(); + long heldSnapshotId = snapshotId(held.get(0)); + expireSnapshot(table, heldSnapshotId); + + assertThatThrownBy(() -> refreshSplits(query, held)) + .isInstanceOf(OutOfRangeException.class) + .hasMessageContaining("has expired"); + } finally { + query.close(); + ioManager.close(); + } + } + + private void expireSnapshot(FileStoreTable table, long snapshotId) throws Exception { + table.fileIO() + .deleteQuietly( + new Path( + table.location(), + String.format("snapshot/snapshot-%d", snapshotId))); + // The manager caches snapshots by path, so drop the cached copy too. + table.snapshotManager().invalidateCache(); + } + + @Test + public void testPartialUpdateKeepsMergedColumns() throws Exception { + FileStoreTable table = + createTable( + Schema.newBuilder() + .column("k", DataTypes.INT()) + .column("a", DataTypes.INT()) + .column("b", DataTypes.INT()) + .primaryKey("k") + .option(CoreOptions.BUCKET.key(), "1") + .option(CoreOptions.DELETION_VECTORS_ENABLED.key(), "true") + .option(CoreOptions.WRITE_ONLY.key(), "true") + .option(CoreOptions.MERGE_ENGINE.key(), "partial-update") + .build()); + IOManager ioManager = IOManager.create(tempDir.toString() + "/io"); + LocalTableQuery query = table.newLocalTableQuery().withIOManager(ioManager); + StreamTableScan scan = fileMonitorScan(table); + try { + write(table, GenericRow.of(1, 10, 20)); + compact(table, ioManager, true); + refresh(query, scan); + + // The level-0 record carries a null for the untouched column. Lookup has to keep + // the merged value from the compacted file. + write(table, GenericRow.of(1, null, 99)); + refresh(query, scan); + InternalRow row = lookup(query, 1); + assertThat(row).isNotNull(); + assertThat(row.isNullAt(1)).isFalse(); + assertThat(row.getInt(1)).isEqualTo(10); + assertThat(row.getInt(2)).isEqualTo(20); + } finally { + query.close(); + ioManager.close(); + } + } + + @Test + public void testSequenceFieldKeepsNewerValue() throws Exception { + FileStoreTable table = + createTable( + Schema.newBuilder() + .column("k", DataTypes.INT()) + .column("seq", DataTypes.INT()) + .column("v", DataTypes.INT()) + .primaryKey("k") + .option(CoreOptions.BUCKET.key(), "1") + .option(CoreOptions.DELETION_VECTORS_ENABLED.key(), "true") + .option(CoreOptions.WRITE_ONLY.key(), "true") + .option(CoreOptions.SEQUENCE_FIELD.key(), "seq") + .build()); + IOManager ioManager = IOManager.create(tempDir.toString() + "/io"); + LocalTableQuery query = table.newLocalTableQuery().withIOManager(ioManager); + StreamTableScan scan = fileMonitorScan(table); + try { + write(table, GenericRow.of(1, 100, 10)); + compact(table, ioManager, true); + refresh(query, scan); + + write(table, GenericRow.of(1, 50, 999)); + refresh(query, scan); + InternalRow row = lookup(query, 1); + assertThat(row).isNotNull(); + assertThat(row.getInt(1)).isEqualTo(100); + assertThat(row.getInt(2)).isEqualTo(10); + } finally { + query.close(); + ioManager.close(); + } + } + + private FileStoreTable createTable() throws Exception { + return createTable( + Schema.newBuilder() + .column("k", DataTypes.INT()) + .column("v", DataTypes.INT()) + .primaryKey("k") + .option(CoreOptions.BUCKET.key(), "1") + .option(CoreOptions.DELETION_VECTORS_ENABLED.key(), "true") + .option(CoreOptions.WRITE_ONLY.key(), "true") + .build()); + } + + private FileStoreTable createTable(Schema schema) throws Exception { + catalog.createTable(identifier, schema, true); + return (FileStoreTable) catalog.getTable(identifier); + } + + private StreamTableScan fileMonitorScan(FileStoreTable table) { + return table.copy( + Collections.singletonMap( + CoreOptions.STREAM_SCAN_MODE.key(), + CoreOptions.StreamScanMode.FILE_MONITOR.toString())) + .newReadBuilder() + .newStreamScan(); + } + + private long snapshotId(Split split) { + if (split instanceof DataSplit) { + return ((DataSplit) split).snapshotId(); + } else if (split instanceof IncrementalSplit) { + return ((IncrementalSplit) split).snapshotId(); + } + throw new IllegalStateException(split.getClass().getName()); + } + + private InternalRow lookup(LocalTableQuery query, int key) throws Exception { + return query.lookup(BinaryRow.EMPTY_ROW, 0, GenericRow.of(key)); + } + + private void write(FileStoreTable table, GenericRow row) throws Exception { + BatchWriteBuilder writeBuilder = table.newBatchWriteBuilder(); + BatchTableWrite write = writeBuilder.newWrite(); + try { + write.write(row); + writeBuilder.newCommit().commit(write.prepareCommit()); + } finally { + write.close(); + } + } + + private void compact(FileStoreTable table, IOManager ioManager, boolean fullCompaction) + throws Exception { + FileStoreTable compactTable = + table.copy(Collections.singletonMap(CoreOptions.WRITE_ONLY.key(), "false")); + BatchWriteBuilder writeBuilder = compactTable.newBatchWriteBuilder(); + try (BatchTableWrite write = writeBuilder.newWrite(); + BatchTableCommit commit = writeBuilder.newCommit()) { + write.withIOManager(ioManager); + write.compact(BinaryRow.EMPTY_ROW, 0, fullCompaction); + commit.commit(write.prepareCommit()); + } + } + + private void refresh(LocalTableQuery query, StreamTableScan scan) { + while (true) { + List splits = scan.plan().splits(); + if (splits.isEmpty()) { + return; + } + refreshSplits(query, splits); + } + } + + private void refreshSplits(LocalTableQuery query, List splits) { + for (Split split : splits) { + if (split instanceof DataSplit) { + DataSplit dataSplit = (DataSplit) split; + query.refreshFiles( + dataSplit.partition(), + dataSplit.bucket(), + Collections.emptyList(), + dataSplit.dataFiles(), + dataSplit.snapshotId()); + } else if (split instanceof IncrementalSplit) { + IncrementalSplit incrementalSplit = (IncrementalSplit) split; + query.refreshFiles( + incrementalSplit.partition(), + incrementalSplit.bucket(), + incrementalSplit.beforeFiles(), + incrementalSplit.afterFiles(), + incrementalSplit.snapshotId()); + } else { + throw new IllegalStateException(split.getClass().getName()); + } + } + } + + private Integer value(LocalTableQuery query, int key) throws Exception { + InternalRow row = lookup(query, key); + if (row == null) { + return null; + } + return row.getInt(1); + } + + private List deletionVectorIndex(FileStoreTable table) { + return table.store() + .newIndexFileHandler() + .scanEntries(DeletionVectorsIndexFile.DELETION_VECTORS_INDEX, BinaryRow.EMPTY_ROW); + } + + private long highLevelRowCount(FileStoreTable table) { + long rowCount = 0L; + for (Split split : table.newReadBuilder().newScan().plan().splits()) { + DataSplit dataSplit = (DataSplit) split; + for (DataFileMeta file : dataSplit.dataFiles()) { + rowCount += file.rowCount(); + } + } + return rowCount; + } +} diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/PrimaryKeyPartialLookupTable.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/PrimaryKeyPartialLookupTable.java index c88112844f0e..1ed22cb1b71b 100644 --- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/PrimaryKeyPartialLookupTable.java +++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/PrimaryKeyPartialLookupTable.java @@ -355,7 +355,7 @@ private void refreshSplit(DataSplit split) { int bucket = split.bucket(); List after = split.dataFiles(); - tableQuery.refreshFiles(partition, bucket, emptyList(), after); + tableQuery.refreshFiles(partition, bucket, emptyList(), after, split.snapshotId()); Integer totalBuckets = split.totalBuckets(); if (totalBuckets == null) { // Just for compatibility with older versions @@ -370,7 +370,11 @@ private void refreshSplit(DataSplit split) { private void refreshSplit(IncrementalSplit split) { BinaryRow partition = split.partition(); tableQuery.refreshFiles( - partition, split.bucket(), split.beforeFiles(), split.afterFiles()); + partition, + split.bucket(), + split.beforeFiles(), + split.afterFiles(), + split.snapshotId()); numBuckets.put(partition, split.totalBuckets()); } diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/service/QueryExecutorOperator.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/service/QueryExecutorOperator.java index ed4758e211f8..75821cf8a3af 100644 --- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/service/QueryExecutorOperator.java +++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/service/QueryExecutorOperator.java @@ -119,12 +119,14 @@ public void initializeState(StateInitializationContext context) throws Exception @Override public void processElement(StreamRecord streamRecord) throws Exception { InternalRow row = streamRecord.getValue(); + // FileMonitorTable row: snapshot id, partition, bucket, before files, data files. + long snapshotId = row.getLong(0); BinaryRow partition = deserializeBinaryRow(row.getBinary(1)); int bucket = row.getInt(2); DataFileMetaSerializer fileMetaSerializer = new DataFileMetaSerializer(); List beforeFiles = fileMetaSerializer.deserializeList(row.getBinary(3)); List dataFiles = fileMetaSerializer.deserializeList(row.getBinary(4)); - query.refreshFiles(partition, bucket, beforeFiles, dataFiles); + query.refreshFiles(partition, bucket, beforeFiles, dataFiles, snapshotId); } @Override diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/RemoteLookupJoinITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/RemoteLookupJoinITCase.java index 0854272e46dd..aede56bad873 100644 --- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/RemoteLookupJoinITCase.java +++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/RemoteLookupJoinITCase.java @@ -274,7 +274,8 @@ public void write(InternalRow... rows) throws Exception { message.partition(), message.bucket(), Collections.emptyList(), - message.newFilesIncrement().newFiles()); + message.newFilesIncrement().newFiles(), + table.snapshotManager().latestSnapshot().id()); } @Override diff --git a/paimon-service/paimon-service-runtime/src/test/java/org/apache/paimon/service/KvQueryTableTest.java b/paimon-service/paimon-service-runtime/src/test/java/org/apache/paimon/service/KvQueryTableTest.java index e882b6607c39..a9800cd4b662 100644 --- a/paimon-service/paimon-service-runtime/src/test/java/org/apache/paimon/service/KvQueryTableTest.java +++ b/paimon-service/paimon-service-runtime/src/test/java/org/apache/paimon/service/KvQueryTableTest.java @@ -18,6 +18,7 @@ package org.apache.paimon.service; +import org.apache.paimon.Snapshot; import org.apache.paimon.catalog.PrimaryKeyTableTestBase; import org.apache.paimon.data.BinaryRow; import org.apache.paimon.data.GenericRow; @@ -216,11 +217,13 @@ private void write(int partition, int key, int value) throws Exception { BatchTableWrite write = table.newBatchWriteBuilder().newWrite(); write.write(row(partition, key, value), bucket); CommitMessageImpl message = (CommitMessageImpl) write.prepareCommit().get(0); + Snapshot latest = table.snapshotManager().latestSnapshot(); query.refreshFiles( message.partition(), message.bucket(), Collections.emptyList(), - message.newFilesIncrement().newFiles()); + message.newFilesIncrement().newFiles(), + latest == null ? -1L : latest.id()); } private int computeBucket(int partition, int key, int value) throws Exception {