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 {