diff --git a/docs/docs/primary-key-table/changelog-producer.md b/docs/docs/primary-key-table/changelog-producer.md index ecbc2755a363..b57cc26245bc 100644 --- a/docs/docs/primary-key-table/changelog-producer.md +++ b/docs/docs/primary-key-table/changelog-producer.md @@ -39,6 +39,7 @@ only `5` requires the consumer to remember the previous value. | `none` (default) | Consumer state, if needed | Incremental snapshot reads | The consumer accepts upserts or can normalize them | | `input` | Complete changelog supplied by the source | Committed input changelog files | The upstream already supplies the required before/after records | | `lookup` | Lookup of existing rows during compaction | After lookup compaction is committed | The input has no before images and consumers need complete changes | +| `on-read` | Lookup of historical rows during streaming reads | Incremental snapshot reads | Readers reconstruct complete changes without stored changelog files | | `full-compaction` | Difference between full-compaction results | After a full compaction is committed | Consumers can wait for periodic full compactions | The diagram follows one existing key whose value changes from `4` to `5`. `-U` is @@ -109,6 +110,17 @@ checkpoint completion waits for compaction. Tune it with checkpoint duration and `lookup` is incompatible with `full-compaction.delta-commits`. For periodic full compaction with changelog generation, use `full-compaction` instead. +## On Read + +With `changelog-producer = on-read`, streaming readers reconstruct changes from new data files +and historical records. Each reader caches partition-bucket lookup states for reuse across splits. + +`changelog-on-read.lookup-state.max-size` limits the number of cached states per reader. It defaults +to `2048` and must be greater than zero. When the limit is exceeded, the least recently used state +and its local lookup files are evicted. Reading that partition and bucket again rebuilds the state +from the split's historical files and recreates lookup files as needed. This option limits the +number of states, not their size in bytes. + ## Full Compaction ```sql diff --git a/docs/generated/core_configuration.html b/docs/generated/core_configuration.html index 0fc8c42fd261..8bfd57d12cb3 100644 --- a/docs/generated/core_configuration.html +++ b/docs/generated/core_configuration.html @@ -212,29 +212,35 @@ String Changelog file metadata stats collection. none, counts, truncate(16), full is available. + +
changelog-on-read.lookup-state.max-size
+ 2048 + Integer + Maximum number of partition-bucket lookup states cached by each reader with changelog-producer=on-read. Must be greater than 0. Least recently used states and their local lookup files are evicted when the limit is exceeded. +
changelog-producer
none

Enum

- Whether to double write to a changelog file. This changelog file keeps the details of data changes, it can be read directly during stream reads. This can be applied to tables with primary keys.

Possible values: + How to produce changelog for primary-key tables. Input, lookup and full-compaction produce changelog files; on-read reconstructs changes from data files during streaming reads.

Possible values:
changelog-producer.ignore-delete
false Boolean - Whether to ignore delete records in the changelog. When set to true, DELETE (-D) records will not be written to changelog files. This configuration is only valid for the changelog-producer is lookup or full-compaction. + Whether to ignore delete records in the changelog. When set to true, DELETE (-D) records will not be emitted. Valid for lookup, full-compaction and on-read.
changelog-producer.ignore-update-before
false Boolean - Whether to ignore update-before records in the changelog. When set to true, UPDATE_BEFORE (-U) records will not be written to changelog files. This configuration is only valid for the changelog-producer is lookup or full-compaction. + Whether to ignore update-before records in the changelog. When set to true, UPDATE_BEFORE (-U) records will not be emitted. Valid for lookup, full-compaction and on-read.
changelog-producer.row-deduplicate
false Boolean - Whether to generate -U, +U changelog for the same record. This configuration is only valid for the changelog-producer is lookup or full-compaction. + Whether to suppress -U, +U changelog for equal records. Valid for lookup, full-compaction and on-read.
changelog-producer.row-deduplicate-ignore-fields
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 c3202188dc09..00e1c52d0595 100644 --- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java +++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java @@ -1161,9 +1161,19 @@ public InlineElement getDescription() { .enumType(ChangelogProducer.class) .defaultValue(ChangelogProducer.NONE) .withDescription( - "Whether to double write to a changelog file. " - + "This changelog file keeps the details of data changes, " - + "it can be read directly during stream reads. This can be applied to tables with primary keys. "); + "How to produce changelog for primary-key tables. " + + "Input, lookup and full-compaction produce changelog files; " + + "on-read reconstructs changes from data files during streaming reads."); + + public static final ConfigOption CHANGELOG_ON_READ_LOOKUP_STATE_MAX_SIZE = + key("changelog-on-read.lookup-state.max-size") + .intType() + .defaultValue(2048) + .withDescription( + "Maximum number of partition-bucket lookup states cached by each reader " + + "with changelog-producer=on-read. Must be greater than 0. " + + "Least recently used states and their local lookup files are " + + "evicted when the limit is exceeded."); public static final ConfigOption CHANGELOG_PRODUCER_IGNORE_UPDATE_BEFORE = key("changelog-producer.ignore-update-before") @@ -1171,8 +1181,8 @@ public InlineElement getDescription() { .defaultValue(false) .withDescription( "Whether to ignore update-before records in the changelog. " - + "When set to true, UPDATE_BEFORE (-U) records will not be written to changelog files. " - + "This configuration is only valid for the changelog-producer is lookup or full-compaction."); + + "When set to true, UPDATE_BEFORE (-U) records will not be emitted. " + + "Valid for lookup, full-compaction and on-read."); public static final ConfigOption CHANGELOG_PRODUCER_IGNORE_DELETE = key("changelog-producer.ignore-delete") @@ -1180,15 +1190,16 @@ public InlineElement getDescription() { .defaultValue(false) .withDescription( "Whether to ignore delete records in the changelog. " - + "When set to true, DELETE (-D) records will not be written to changelog files. " - + "This configuration is only valid for the changelog-producer is lookup or full-compaction."); + + "When set to true, DELETE (-D) records will not be emitted. " + + "Valid for lookup, full-compaction and on-read."); public static final ConfigOption CHANGELOG_PRODUCER_ROW_DEDUPLICATE = key("changelog-producer.row-deduplicate") .booleanType() .defaultValue(false) .withDescription( - "Whether to generate -U, +U changelog for the same record. This configuration is only valid for the changelog-producer is lookup or full-compaction."); + "Whether to suppress -U, +U changelog for equal records. " + + "Valid for lookup, full-compaction and on-read."); public static final ConfigOption CHANGELOG_PRODUCER_ROW_DEDUPLICATE_IGNORE_FIELDS = key("changelog-producer.row-deduplicate-ignore-fields") @@ -4055,6 +4066,27 @@ public ChangelogProducer changelogProducer() { return options.get(CHANGELOG_PRODUCER); } + public boolean changelogOnRead() { + return changelogProducer() == ChangelogProducer.ON_READ; + } + + public int changelogOnReadLookupStateMaxSize() { + int maxSize = options.get(CHANGELOG_ON_READ_LOOKUP_STATE_MAX_SIZE); + checkArgument( + maxSize > 0, + "The option %s must be greater than 0.", + CHANGELOG_ON_READ_LOOKUP_STATE_MAX_SIZE.key()); + return maxSize; + } + + /** Whether this mode materializes changelog files, rather than reconstructing changes. */ + public boolean producesChangelogFiles() { + ChangelogProducer producer = changelogProducer(); + return producer == ChangelogProducer.INPUT + || producer == ChangelogProducer.LOOKUP + || producer == ChangelogProducer.FULL_COMPACTION; + } + public boolean needLookup() { return lookupStrategy().needLookup; } @@ -5465,7 +5497,9 @@ public enum ChangelogProducer implements DescribedEnum { FULL_COMPACTION("full-compaction", "Generate changelog files with each full compaction."), - LOOKUP("lookup", "Generate changelog files through 'lookup' compaction."); + LOOKUP("lookup", "Generate changelog files through 'lookup' compaction."), + + ON_READ("on-read", "Generate changelog during reads without writing changelog files."); private final String value; private final String description; diff --git a/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java b/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java index 866394577870..bf2f5b17799d 100644 --- a/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java +++ b/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java @@ -341,7 +341,7 @@ public SnapshotDeletion newSnapshotDeletion() { manifestListFactory().create(), newIndexFileHandler(), newStatsFileHandler(), - options.changelogProducer() != CoreOptions.ChangelogProducer.NONE, + options.producesChangelogFiles(), options.cleanEmptyDirectories(), options.fileOperationThreadNum(), options.scanManifestParallelism()); diff --git a/paimon-core/src/main/java/org/apache/paimon/mergetree/LookupLevels.java b/paimon-core/src/main/java/org/apache/paimon/mergetree/LookupLevels.java index 22fef5a1fc29..cf66e35dcdf0 100644 --- a/paimon-core/src/main/java/org/apache/paimon/mergetree/LookupLevels.java +++ b/paimon-core/src/main/java/org/apache/paimon/mergetree/LookupLevels.java @@ -43,8 +43,10 @@ import java.io.Closeable; import java.io.File; import java.io.IOException; +import java.util.ArrayList; import java.util.Comparator; import java.util.HashSet; +import java.util.List; import java.util.Map; import java.util.Optional; import java.util.Set; @@ -149,6 +151,49 @@ public T lookup(InternalRow key, int startLevel, @Nullable LookupContext context (lookupKey, level0) -> lookupLevel0(lookupKey, level0, context)); } + /** Lookup all records for a key instead of stopping at the first matching level. */ + public List lookupAll(InternalRow key) throws IOException { + List result = new ArrayList<>(); + for (DataFileMeta file : levels.level0()) { + if (keyComparator.compare(file.maxKey(), key) >= 0 + && keyComparator.compare(file.minKey(), key) <= 0) { + T value = lookup(key, file, null); + if (value != null) { + result.add(value); + } + } + } + for (int level = 1; level < levels.numberOfLevels(); level++) { + T value = lookup(key, levels.runOfLevel(level), null); + if (value != null) { + result.add(value); + } + } + return result; + } + + /** Prepare local lookup files before records are consumed. */ + public void prepareLookupFiles(List files) throws IOException { + for (DataFileMeta file : files) { + String fileName = file.fileName(); + LookupFile lookupFile = lookupFileCache.getIfPresent(fileName); + if (lookupFile != null && !lookupFile.isClosed()) { + continue; + } + + Object lock = lookupFileLock(fileName); + synchronized (lock) { + lookupFile = lookupFileCache.getIfPresent(fileName); + if (lookupFile != null && !lookupFile.isClosed()) { + continue; + } + + lookupFile = createLookupFile(file); + addLocalFile(file, lookupFile); + } + } + } + @Nullable private T lookupLevel0( InternalRow key, TreeSet level0, @Nullable LookupContext context) @@ -174,10 +219,11 @@ private T lookup(InternalRow key, DataFileMeta file, @Nullable LookupContext con return null; } + // A cached SST can outlive a file upgrade; use the level from the current read view. return readFromDisk( getOrCreateProcessor(lookupResult.schemaId, lookupResult.serVersion), key, - lookupResult.level, + file.level(), valueBytes, file.fileName()); } @@ -248,21 +294,18 @@ private LookupResult lookupCachedFile( private static class LookupResult { - private final int level; private final long schemaId; private final String serVersion; private final byte[] valueBytes; - private LookupResult(int level, long schemaId, String serVersion, byte[] valueBytes) { - this.level = level; + private LookupResult(long schemaId, String serVersion, byte[] valueBytes) { this.schemaId = schemaId; this.serVersion = serVersion; this.valueBytes = valueBytes; } private static LookupResult of(LookupFile lookupFile, byte[] valueBytes) { - return new LookupResult( - lookupFile.level(), lookupFile.schemaId(), lookupFile.serVersion(), valueBytes); + return new LookupResult(lookupFile.schemaId(), lookupFile.serVersion(), valueBytes); } } diff --git a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/SortMergeReaderWithLoserTree.java b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/SortMergeReaderWithLoserTree.java index a3b8b1e26f59..256ca4f3835b 100644 --- a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/SortMergeReaderWithLoserTree.java +++ b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/SortMergeReaderWithLoserTree.java @@ -21,7 +21,9 @@ import org.apache.paimon.KeyValue; import org.apache.paimon.data.InternalRow; import org.apache.paimon.reader.RecordReader; +import org.apache.paimon.utils.ExceptionUtils; import org.apache.paimon.utils.FieldsComparator; +import org.apache.paimon.utils.IOUtils; import org.apache.paimon.utils.Preconditions; import javax.annotation.Nullable; @@ -84,7 +86,11 @@ public RecordIterator readBatch() throws IOException { @Override public void close() throws IOException { - loserTree.close(); + try { + IOUtils.closeAll(loserTree, mergeFunctionWrapper::reset); + } catch (Exception e) { + ExceptionUtils.rethrowIOException(e); + } } /** The iterator iterates on {@link SortMergeReaderWithLoserTree}. */ diff --git a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/SortMergeReaderWithMinHeap.java b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/SortMergeReaderWithMinHeap.java index db09ce41a7dd..ddba5897dd92 100644 --- a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/SortMergeReaderWithMinHeap.java +++ b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/SortMergeReaderWithMinHeap.java @@ -119,6 +119,7 @@ public void close() throws IOException { for (Element element : polled) { collected = releaseCollecting(element, collected); } + collected = closeCollecting(mergeFunctionWrapper::reset, collected); if (collected != null) { rethrowCloseFailure(collected); } @@ -135,7 +136,7 @@ public void close() throws IOException { } private static @Nullable Throwable closeCollecting( - RecordReader reader, @Nullable Throwable collected) { + AutoCloseable reader, @Nullable Throwable collected) { try { reader.close(); } catch (Throwable t) { diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/MergeFileSplitRead.java b/paimon-core/src/main/java/org/apache/paimon/operation/MergeFileSplitRead.java index 9e66aad52fca..a58245dd0057 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/MergeFileSplitRead.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/MergeFileSplitRead.java @@ -21,8 +21,10 @@ import org.apache.paimon.CoreOptions; import org.apache.paimon.KeyValue; import org.apache.paimon.KeyValueFileStore; +import org.apache.paimon.codegen.RecordEqualiser; import org.apache.paimon.data.BinaryRow; import org.apache.paimon.data.InternalRow; +import org.apache.paimon.data.serializer.RowCompactedSerializer; import org.apache.paimon.deletionvectors.DeletionVector; import org.apache.paimon.disk.IOManager; import org.apache.paimon.fs.FileIO; @@ -30,17 +32,29 @@ import org.apache.paimon.io.ChainReadContext; 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.lookup.LookupStrategy; import org.apache.paimon.mergetree.DropDeleteReader; +import org.apache.paimon.mergetree.Levels; +import org.apache.paimon.mergetree.LookupFile; +import org.apache.paimon.mergetree.LookupLevels; import org.apache.paimon.mergetree.MergeSorter; import org.apache.paimon.mergetree.MergeTreeReaders; import org.apache.paimon.mergetree.SortBufferWriteBuffer; import org.apache.paimon.mergetree.SortedRun; +import org.apache.paimon.mergetree.compact.ChangelogResult; import org.apache.paimon.mergetree.compact.ConcatRecordReader; import org.apache.paimon.mergetree.compact.IntervalPartition; +import org.apache.paimon.mergetree.compact.LookupChangelogMergeFunctionWrapper; import org.apache.paimon.mergetree.compact.LookupMergeFunction; +import org.apache.paimon.mergetree.compact.MergeFunction; import org.apache.paimon.mergetree.compact.MergeFunctionFactory; import org.apache.paimon.mergetree.compact.MergeFunctionWrapper; import org.apache.paimon.mergetree.compact.ReducerMergeFunctionWrapper; +import org.apache.paimon.mergetree.lookup.LookupSerializerFactory; +import org.apache.paimon.mergetree.lookup.PersistValueProcessor; +import org.apache.paimon.options.Options; import org.apache.paimon.predicate.Predicate; import org.apache.paimon.reader.EmptyRecordReader; import org.apache.paimon.reader.ReadBatchSizer; @@ -51,25 +65,39 @@ import org.apache.paimon.table.source.ChainSplit; import org.apache.paimon.table.source.DataSplit; import org.apache.paimon.table.source.DeletionFile; +import org.apache.paimon.table.source.IncrementalSplit; import org.apache.paimon.table.source.Split; import org.apache.paimon.types.DataField; +import org.apache.paimon.types.RowKind; import org.apache.paimon.types.RowType; +import org.apache.paimon.utils.Pair; import org.apache.paimon.utils.ProjectedRow; import org.apache.paimon.utils.UserDefinedSeqComparator; +import org.apache.paimon.utils.ValueEqualiserSupplier; + +import org.apache.paimon.shade.caffeine2.com.github.benmanes.caffeine.cache.Cache; import javax.annotation.Nullable; import java.io.IOException; +import java.io.UncheckedIOException; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; import java.util.Comparator; +import java.util.Iterator; +import java.util.LinkedHashMap; import java.util.List; +import java.util.Map; import java.util.Set; import java.util.stream.Collectors; +import static org.apache.paimon.lookup.LookupStoreFactory.bloomFilterBuilderFactory; +import static org.apache.paimon.mergetree.LookupFile.localFilePrefix; import static org.apache.paimon.predicate.PredicateBuilder.containsFields; import static org.apache.paimon.predicate.PredicateBuilder.splitAnd; +import static org.apache.paimon.utils.Preconditions.checkArgument; +import static org.apache.paimon.utils.Preconditions.checkNotNull; /** * An implementation for {@link KeyValueFileStore}, this class handle LSM merging and changelog row @@ -99,6 +127,10 @@ public class MergeFileSplitRead implements SplitRead { private boolean forceKeepDelete = false; + @Nullable private Cache lookupFileCache; + @Nullable private LookupStoreFactory lookupStoreFactory; + private final Map, LookupLevels> lookupLevelsMap; + public MergeFileSplitRead( CoreOptions options, TableSchema schema, @@ -119,6 +151,24 @@ public MergeFileSplitRead( CoreOptions.fromMap(tableSchema.options()), keyType, valueType, null); this.sequenceFields = options.sequenceField(); this.sequenceOrder = options.sequenceFieldSortOrderIsAscending(); + int maxLookupStates = options.changelogOnReadLookupStateMaxSize(); + this.lookupLevelsMap = + new LinkedHashMap, LookupLevels>( + 16, 0.75f, true) { + @Override + protected boolean removeEldestEntry( + Map.Entry, LookupLevels> eldest) { + if (size() <= maxLookupStates) { + return false; + } + try { + eldest.getValue().close(); + } catch (IOException e) { + throw new UncheckedIOException(e); + } + return true; + } + }; } public Comparator keyComparator() { @@ -237,6 +287,259 @@ public MergeFileSplitRead withFilter(Predicate predicate) { return this; } + /** + * Creates a changelog reader. The previous on-read reader from this instance must be closed + * before calling this method again, because creating a reader may evict cached lookup states. + */ + public RecordReader createChangelogOnReadReader(IncrementalSplit split) + throws IOException { + checkArgument( + split.beforeDeletionFiles() == null && split.afterDeletionFiles() == null, + "Changelog-on-read does not support deletion vectors."); + checkArgument( + split.afterFiles().stream().allMatch(file -> file.level() == 0), + "Changelog-on-read expects only level-0 delta files."); + + LookupLevels history = + refreshLookupLevels(split.partition(), split.bucket(), split.beforeFiles()); + history.prepareLookupFiles(lookupFilesToPrepare(split.beforeFiles(), split.afterFiles())); + + UserDefinedSeqComparator sequenceComparator = createUdsComparator(); + MergeFunctionWrapper mergeFunctionWrapper = + createChangelogOnReadMergeWrapper(history, sequenceComparator); + + KeyValueFileReaderFactory readerFactory = + readerFactoryBuilder.build( + split.partition(), + split.bucket(), + DeletionVector.emptyFactory(), + false, + filtersForKeys); + RecordReader reader = + MergeTreeReaders.readerForMergeTree( + new IntervalPartition(split.afterFiles(), keyComparator).partition(), + readerFactory, + keyComparator, + sequenceComparator, + mergeFunctionWrapper, + mergeSorter); + return finishMergeReader(changelogReader(reader), true); + } + + private MergeFunctionWrapper createChangelogOnReadMergeWrapper( + LookupLevels history, @Nullable UserDefinedSeqComparator sequenceComparator) { + RowType readType = actualReadType(); + Comparator keyValueComparator = createSequenceComparator(sequenceComparator); + MergeFunction historyMerge = mfFactory.create(readType); + MergeFunctionFactory mergeFactory = unwrapLookup(mfFactory); + + // Keep one lookup layer and size its spill serializer for the projected values. + LookupMergeFunction.Factory lookupFactory = + (LookupMergeFunction.Factory) + LookupMergeFunction.wrap( + ignored -> mergeFactory.create(readType), + options, + keyType, + readType); + lookupFactory.withIOManager(ioManager); + + RowType equalType = outerReadType == null ? readType : outerReadType; + RecordEqualiser equaliser = + options.changelogRowDeduplicate() + ? ValueEqualiserSupplier.fromIgnoreFields( + equalType, options.changelogRowDeduplicateIgnoreFields()) + .get() + : null; + return new LookupChangelogMergeFunctionWrapper( + lookupFactory, + key -> { + try { + List values = history.lookupAll(key); + if (values.isEmpty()) { + return null; + } + values.sort(keyValueComparator); + historyMerge.reset(); + values.forEach(historyMerge::add); + KeyValue result = historyMerge.getResult(); + return result == null ? null : result.setLevel(1); + } catch (IOException e) { + throw new UncheckedIOException(e); + } + }, + equaliser, + LookupStrategy.from(false, true, false, false), + null, + sequenceComparator) { + @Override + public void reset() { + try { + super.reset(); + } finally { + historyMerge.reset(); + } + } + }; + } + + private List lookupFilesToPrepare( + List beforeFiles, List afterFiles) { + List result = new ArrayList<>(); + for (DataFileMeta before : beforeFiles) { + for (DataFileMeta after : afterFiles) { + if (keyComparator.compare(before.maxKey(), after.minKey()) >= 0 + && keyComparator.compare(before.minKey(), after.maxKey()) <= 0) { + result.add(before); + break; + } + } + } + return result; + } + + private LookupLevels refreshLookupLevels( + BinaryRow partition, int bucket, List files) { + Pair key = Pair.of(partition, bucket); + LookupLevels lookup = lookupLevelsMap.get(key); + if (lookup == null) { + lookup = createLookupLevels(partition, bucket, files); + lookupLevelsMap.put(key, lookup); + } else { + lookup.getLevels().update(lookup.getLevels().allFiles(), files); + } + return lookup; + } + + private LookupLevels createLookupLevels( + BinaryRow partition, int bucket, List files) { + checkNotNull(ioManager, "IOManager is required for changelog-on-read."); + KeyValueFileReaderFactory readerFactory = + readerFactoryBuilder.build( + partition, bucket, DeletionVector.emptyFactory(), false, filtersForKeys); + RowType readType = actualReadType(); + Options configuration = options.toConfiguration(); + return new LookupLevels<>( + ignored -> readType, + tableSchema.id(), + new Levels(keyComparator, files, options.numLevels()), + keyComparator, + keyType, + PersistValueProcessor.factory(readType), + LookupSerializerFactory.INSTANCE.get(), + readerFactory::createRecordReader, + file -> + checkNotNull(ioManager) + .createChannel( + localFilePrefix( + tableSchema.logicalPartitionType(), + partition, + bucket, + file)) + .getPathFile(), + lookupStoreFactory(), + bloomFilterBuilderFactory(configuration), + lookupFileCache()); + } + + private LookupStoreFactory lookupStoreFactory() { + if (lookupStoreFactory == null) { + lookupStoreFactory = + LookupStoreFactory.create( + options, + new CacheManager( + options.lookupCacheMaxMemory(), + options.lookupCacheHighPrioPoolRatio()), + new RowCompactedSerializer(keyType).createSliceComparator()); + } + return lookupStoreFactory; + } + + private Cache lookupFileCache() { + if (lookupFileCache == null) { + lookupFileCache = + LookupFile.createCache( + options.toConfiguration().get(CoreOptions.LOOKUP_CACHE_FILE_RETENTION), + options.toConfiguration().get(CoreOptions.LOOKUP_CACHE_MAX_DISK_SIZE)); + } + return lookupFileCache; + } + + private Comparator createSequenceComparator( + @Nullable UserDefinedSeqComparator comparator) { + return (left, right) -> { + if (comparator != null) { + int result = comparator.compare(left.value(), right.value()); + if (result != 0) { + return result; + } + } + int result = Long.compare(left.sequenceNumber(), right.sequenceNumber()); + if (result != 0) { + return result; + } + return Boolean.compare(left.isAdd(), right.isAdd()); + }; + } + + private RecordReader changelogReader(RecordReader reader) { + return new RecordReader() { + @Nullable + @Override + public RecordIterator readBatch() throws IOException { + RecordIterator batch = reader.readBatch(); + if (batch == null) { + return null; + } + return new RecordIterator() { + private Iterator current = + Collections.emptyList().iterator(); + + @Nullable + @Override + public KeyValue next() throws IOException { + while (true) { + while (!current.hasNext()) { + ChangelogResult result = batch.next(); + if (result == null) { + return null; + } + current = result.changelogs().iterator(); + } + KeyValue next = current.next(); + if (options.changelogProducerIgnoreUpdateBefore() + && next.valueKind() == RowKind.UPDATE_BEFORE) { + continue; + } + if (options.changelogProducerIgnoreDelete() + && next.valueKind() == RowKind.DELETE) { + continue; + } + return next; + } + } + + @Override + public void releaseBatch() { + batch.releaseBatch(); + } + }; + } + + @Override + public void close() throws IOException { + reader.close(); + } + }; + } + + public void close() { + lookupLevelsMap.clear(); + if (lookupFileCache != null) { + lookupFileCache.invalidateAll(); + lookupFileCache = null; + } + } + @Override public RecordReader createReader(Split split) throws IOException { if (split instanceof DataSplit) { diff --git a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java index e279d6479587..c7891acb5b5c 100644 --- a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java +++ b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java @@ -199,6 +199,8 @@ public static void validateTableSchema(TableSchema schema, Set dynamicOp validateMergeFunction(schema); + validateChangelogOnRead(schema, options); + ChangelogProducer changelogProducer = options.changelogProducer(); if (schema.primaryKeys().isEmpty() && changelogProducer != ChangelogProducer.NONE) { throw new UnsupportedOperationException( @@ -887,6 +889,46 @@ private static void validateMergeFunction(TableSchema schema) { createMergeFunctionFactory(schema); } + private static void validateChangelogOnRead(TableSchema schema, CoreOptions options) { + if (!options.changelogOnRead()) { + return; + } + + checkArgument( + !options.queryAuthEnabled(), + "changelog-producer=on-read does not support query authorization."); + + checkArgument( + !schema.primaryKeys().isEmpty(), + "%s requires a table with primary keys.", + CHANGELOG_PRODUCER.key()); + checkArgument( + options.mergeEngine() == MergeEngine.DEDUPLICATE + || options.mergeEngine() == MergeEngine.AGGREGATE, + "%s only supports deduplicate and aggregation merge engines.", + CHANGELOG_PRODUCER.key()); + checkArgument( + options.bucket() > 0, + "%s only supports fixed bucket tables.", + CHANGELOG_PRODUCER.key()); + checkArgument( + !options.deletionVectorsEnabled(), + "%s does not support deletion vectors.", + CHANGELOG_PRODUCER.key()); + checkArgument( + !schema.crossPartitionUpdate(), + "%s does not support cross-partition updates.", + CHANGELOG_PRODUCER.key()); + checkArgument( + !options.streamingReadOverwrite(), + "%s does not support streaming overwrite.", + CHANGELOG_PRODUCER.key()); + checkArgument( + !options.isChainTable() && !options.dataEvolutionEnabled(), + "%s does not support chain tables or data evolution.", + CHANGELOG_PRODUCER.key()); + } + private static void validateMapStorageLayout(TableSchema schema, CoreOptions options) { String layoutSuffix = "." + MAP_STORAGE_LAYOUT; Map fieldMap = new HashMap<>(); diff --git a/paimon-core/src/main/java/org/apache/paimon/table/FallbackReadFileStoreTable.java b/paimon-core/src/main/java/org/apache/paimon/table/FallbackReadFileStoreTable.java index 02624408f2fb..cc1c538987ac 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/FallbackReadFileStoreTable.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/FallbackReadFileStoreTable.java @@ -805,5 +805,14 @@ public RecordReader createReader(Split split) throws IOException { // Plain splits come from newSnapshotReader(), which plans the wrapped table. return wrappedRead.createReader(split); } + + @Override + public void close() throws IOException { + try { + mainRead.close(); + } finally { + fallbackRead.close(); + } + } } } diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractDataTableScan.java b/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractDataTableScan.java index 4ac5cda74fe0..59967e41cbc0 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractDataTableScan.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractDataTableScan.java @@ -761,9 +761,7 @@ private StartingScanner createIncrementalStartingScanner(SnapshotManager snapsho private ScanMode toSnapshotScanMode(CoreOptions.IncrementalBetweenScanMode scanMode) { switch (scanMode) { case AUTO: - return options.changelogProducer() == ChangelogProducer.NONE - ? ScanMode.DELTA - : ScanMode.CHANGELOG; + return options.producesChangelogFiles() ? ScanMode.CHANGELOG : ScanMode.DELTA; case DELTA: return ScanMode.DELTA; case CHANGELOG: diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/DataTableStreamScan.java b/paimon-core/src/main/java/org/apache/paimon/table/source/DataTableStreamScan.java index 02806cfe33fa..db36dee4fbf0 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/DataTableStreamScan.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/DataTableStreamScan.java @@ -276,6 +276,9 @@ protected FollowUpScanner createFollowUpScanner() { case NONE: followUpScanner = new DeltaFollowUpScanner(); break; + case ON_READ: + followUpScanner = new DeltaFollowUpScanner(true); + break; case INPUT: case FULL_COMPACTION: case LOOKUP: diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/IncrementalSplit.java b/paimon-core/src/main/java/org/apache/paimon/table/source/IncrementalSplit.java index 78dbfff66f43..c42d597a683f 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/IncrementalSplit.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/IncrementalSplit.java @@ -123,6 +123,10 @@ public boolean isStreaming() { return isStreaming; } + public OptionalLong earliestFileCreationEpochMillis() { + return afterFiles.stream().mapToLong(DataFileMeta::creationTimeEpochMillis).min(); + } + @Override public long rowCount() { long rowCount = 0; diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/KeyValueTableRead.java b/paimon-core/src/main/java/org/apache/paimon/table/source/KeyValueTableRead.java index 687304ba8cab..cb5573bfd1ab 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/KeyValueTableRead.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/KeyValueTableRead.java @@ -43,6 +43,7 @@ import javax.annotation.Nullable; +import java.io.Closeable; import java.io.IOException; import java.util.ArrayList; import java.util.Arrays; @@ -238,6 +239,15 @@ public RecordReader reader(Split split) throws IOException { throw new RuntimeException("Should not happen."); } + @Override + public void close() throws IOException { + for (SplitReadProvider provider : readProviders) { + if (provider instanceof Closeable) { + ((Closeable) provider).close(); + } + } + } + public static RecordReader unwrap( RecordReader reader, Map schemaOptions) { return new RecordReader() { diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/ReadBuilderImpl.java b/paimon-core/src/main/java/org/apache/paimon/table/source/ReadBuilderImpl.java index 51c76d71839b..e4870e08d42c 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/ReadBuilderImpl.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/ReadBuilderImpl.java @@ -361,6 +361,11 @@ public RecordReader createReader(TableScan.Plan plan) throws IOExce return limit(delegate.createReader(plan)); } + @Override + public void close() throws IOException { + delegate.close(); + } + private RecordReader limit(RecordReader reader) { if (filterPresent && !filterExecutedByReader) { return reader; diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/TableRead.java b/paimon-core/src/main/java/org/apache/paimon/table/source/TableRead.java index 91e0b7d0149c..642bc75eb75b 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/TableRead.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/TableRead.java @@ -28,6 +28,7 @@ import org.apache.paimon.reader.ReaderSupplier; import org.apache.paimon.reader.RecordReader; +import java.io.Closeable; import java.io.IOException; import java.util.ArrayList; import java.util.List; @@ -38,7 +39,10 @@ * @since 0.4.0 */ @Public -public interface TableRead { +public interface TableRead extends Closeable { + + @Override + default void close() throws IOException {} /** Set {@link MetricRegistry} to table read. */ TableRead withMetricRegistry(MetricRegistry registry); diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/DeltaFollowUpScanner.java b/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/DeltaFollowUpScanner.java index 5a8b4ec8c843..12d1b897cfdc 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/DeltaFollowUpScanner.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/DeltaFollowUpScanner.java @@ -29,6 +29,15 @@ public class DeltaFollowUpScanner implements FollowUpScanner { private static final Logger LOG = LoggerFactory.getLogger(DeltaFollowUpScanner.class); + private final boolean changelogOnRead; + + public DeltaFollowUpScanner() { + this(false); + } + + public DeltaFollowUpScanner(boolean changelogOnRead) { + this.changelogOnRead = changelogOnRead; + } @Override public boolean shouldScanSnapshot(Snapshot snapshot) { @@ -45,6 +54,9 @@ public boolean shouldScanSnapshot(Snapshot snapshot) { @Override public SnapshotReader.Plan scan(Snapshot snapshot, SnapshotReader snapshotReader) { + if (changelogOnRead) { + return snapshotReader.withSnapshot(snapshot).readChangelogOnRead(); + } return snapshotReader.withMode(ScanMode.DELTA).withSnapshot(snapshot).read(); } } diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReader.java b/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReader.java index 68611f009116..95957fde3e19 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReader.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReader.java @@ -143,6 +143,11 @@ public interface SnapshotReader { /** Get splits plan from file changes. */ Plan readChanges(); + /** Get delta files and their complete historical state for changelog-on-read. */ + default Plan readChangelogOnRead() { + throw new UnsupportedOperationException("Changelog-on-read is not supported."); + } + Plan readIncrementalDiff(Snapshot before); /** List partitions. */ diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReaderImpl.java b/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReaderImpl.java index 29f1ef029756..bbb906f18910 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReaderImpl.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReaderImpl.java @@ -80,7 +80,6 @@ import static org.apache.paimon.Snapshot.FIRST_SNAPSHOT_ID; import static org.apache.paimon.deletionvectors.DeletionVectorsIndexFile.DELETION_VECTORS_INDEX; import static org.apache.paimon.operation.FileStoreScan.Plan.groupByPartFiles; -import static org.apache.paimon.partition.PartitionPredicate.createPartitionPredicate; import static org.apache.paimon.partition.PartitionPredicate.splitPartitionPredicatesAndDataPredicates; /** Implementation of {@link SnapshotReader}. */ @@ -102,6 +101,7 @@ public class SnapshotReaderImpl implements SnapshotReader { private ScanMode scanMode = ScanMode.ALL; private boolean hasNonPartitionFilter; + @Nullable private PartitionPredicate partitionFilter; private RecordComparator lazyPartitionComparator; private CacheMetrics dvMetaCacheMetrics; @@ -197,32 +197,31 @@ public SnapshotReader withSnapshot(Snapshot snapshot) { @Override public SnapshotReader withPartitionFilter(Map partitionSpec) { if (partitionSpec != null) { - Predicate partitionPredicate = - createPartitionPredicate( - partitionSpec, + return setPartitionFilter( + PartitionPredicate.fromMap( tableSchema.logicalPartitionType(), - options.partitionDefaultName()); - scan.withPartitionFilter(partitionPredicate); + partitionSpec, + options.partitionDefaultName())); } return this; } @Override public SnapshotReader withPartitionFilter(Predicate predicate) { - scan.withPartitionFilter(predicate); - return this; + return setPartitionFilter( + PartitionPredicate.fromPredicate(tableSchema.logicalPartitionType(), predicate)); } @Override public SnapshotReader withPartitionFilter(List partitions) { - scan.withPartitionFilter(partitions); - return this; + return setPartitionFilter( + PartitionPredicate.fromMultiple(tableSchema.logicalPartitionType(), partitions)); } @Override public SnapshotReader withPartitionFilter(PartitionPredicate partitionPredicate) { if (partitionPredicate != null) { - scan.withPartitionFilter(partitionPredicate); + return setPartitionFilter(partitionPredicate); } return this; } @@ -230,11 +229,21 @@ public SnapshotReader withPartitionFilter(PartitionPredicate partitionPredicate) @Override public SnapshotReader withPartitionsFilter(List> partitions) { if (partitions != null) { - scan.withPartitionsFilter(partitions); + return setPartitionFilter( + PartitionPredicate.fromMaps( + tableSchema.logicalPartitionType(), + partitions, + options.partitionDefaultName())); } return this; } + private SnapshotReader setPartitionFilter(@Nullable PartitionPredicate partitionFilter) { + this.partitionFilter = partitionFilter; + scan.withPartitionFilter(partitionFilter); + return this; + } + @Override public SnapshotReader withFilter(Predicate predicate) { return withFilter(predicate, predicate); @@ -255,7 +264,7 @@ public SnapshotReader withFilter(Predicate predicate, @Nullable Predicate pushdo tableSchema.logicalRowType(), tableSchema.partitionKeys()); if (pushdownPair.getLeft().isPresent()) { - scan.withPartitionFilter(pushdownPair.getLeft().get()); + setPartitionFilter(pushdownPair.getLeft().get()); } if (!pushdownPair.getRight().isEmpty()) { this.hasNonPartitionFilter = true; @@ -502,6 +511,45 @@ public Plan readChanges() { true, beforeSnapshot, beforeFiles, plan.snapshot(), plan.watermark(), afterFiles); } + @Override + public Plan readChangelogOnRead() { + withMode(ScanMode.DELTA); + FileStoreScan.Plan deltaPlan = scan.plan(); + Snapshot snapshot = deltaPlan.snapshot(); + if (snapshot == null) { + return new PlanImpl(deltaPlan.watermark(), null, null, Collections.emptyList()); + } + + Map>> deltaFiles = + groupByPartFiles(deltaPlan.files(FileKind.ADD)); + if (deltaFiles.isEmpty()) { + return new PlanImpl( + deltaPlan.watermark(), snapshot.id(), snapshot, Collections.emptyList()); + } + + Map>> historyFiles; + try { + scan.withPartitionFilter(new ArrayList<>(deltaFiles.keySet())); + withMode(ScanMode.ALL); + historyFiles = groupByPartFiles(scan.plan().files(FileKind.ADD)); + } finally { + scan.withPartitionFilter(partitionFilter); + } + historyFiles.forEach( + (partition, buckets) -> { + Map> deltas = deltaFiles.get(partition); + buckets.keySet().retainAll(deltas.keySet()); + buckets.forEach((bucket, entries) -> entries.removeAll(deltas.get(bucket))); + }); + return toIncrementalPlan( + true, + new LazyField<>(() -> null), + historyFiles, + snapshot, + deltaPlan.watermark(), + deltaFiles); + } + private Plan toIncrementalPlan( boolean isStreaming, LazyField beforeSnapshot, diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/splitread/IncrementalChangelogReadProvider.java b/paimon-core/src/main/java/org/apache/paimon/table/source/splitread/IncrementalChangelogReadProvider.java index 56646dab845a..e3e0240d5b4b 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/splitread/IncrementalChangelogReadProvider.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/splitread/IncrementalChangelogReadProvider.java @@ -18,6 +18,7 @@ package org.apache.paimon.table.source.splitread; +import org.apache.paimon.CoreOptions; import org.apache.paimon.KeyValue; import org.apache.paimon.data.InternalRow; import org.apache.paimon.mergetree.compact.ConcatRecordReader; @@ -31,14 +32,19 @@ import org.apache.paimon.utils.IOFunction; import org.apache.paimon.utils.LazyField; +import javax.annotation.Nullable; + +import java.io.Closeable; +import java.io.IOException; import java.util.function.Supplier; import static org.apache.paimon.table.source.KeyValueTableRead.unwrap; /** A {@link SplitReadProvider} to incremental changelog read. */ -public class IncrementalChangelogReadProvider implements SplitReadProvider { +public class IncrementalChangelogReadProvider implements SplitReadProvider, Closeable { private final LazyField> splitRead; + @Nullable private MergeFileSplitRead mergeRead; public IncrementalChangelogReadProvider( Supplier supplier, SplitReadConfig splitReadConfig) { @@ -53,28 +59,35 @@ public IncrementalChangelogReadProvider( private SplitRead create(Supplier supplier) { final MergeFileSplitRead read = supplier.get().withReadKeyType(RowType.of()); + mergeRead = read; IOFunction> convertedFactory = - split -> { - IncrementalSplit incrementalSplit = (IncrementalSplit) split; - RecordReader reader = - ConcatRecordReader.create( - () -> - new ReverseReader( + CoreOptions.fromMap(read.tableSchema().options()).changelogOnRead() + ? split -> + unwrap( + read.createChangelogOnReadReader((IncrementalSplit) split), + read.tableSchema().options()) + : split -> { + IncrementalSplit incrementalSplit = (IncrementalSplit) split; + RecordReader reader = + ConcatRecordReader.create( + () -> + new ReverseReader( + read.createMergeReader( + incrementalSplit.partition(), + incrementalSplit.bucket(), + incrementalSplit.beforeFiles(), + incrementalSplit + .beforeDeletionFiles(), + false)), + () -> read.createMergeReader( incrementalSplit.partition(), incrementalSplit.bucket(), - incrementalSplit.beforeFiles(), - incrementalSplit.beforeDeletionFiles(), - false)), - () -> - read.createMergeReader( - incrementalSplit.partition(), - incrementalSplit.bucket(), - incrementalSplit.afterFiles(), - incrementalSplit.afterDeletionFiles(), - false)); - return unwrap(reader, read.tableSchema().options()); - }; + incrementalSplit.afterFiles(), + incrementalSplit.afterDeletionFiles(), + false)); + return unwrap(reader, read.tableSchema().options()); + }; return SplitRead.convert(read, convertedFactory); } @@ -91,4 +104,11 @@ public boolean match(Split split, Context context) { public LazyField> get() { return splitRead; } + + @Override + public void close() throws IOException { + if (mergeRead != null) { + mergeRead.close(); + } + } } diff --git a/paimon-core/src/test/java/org/apache/paimon/mergetree/LookupLevelsTest.java b/paimon-core/src/test/java/org/apache/paimon/mergetree/LookupLevelsTest.java index fe23828f08ff..8d5e41b1c6e1 100644 --- a/paimon-core/src/test/java/org/apache/paimon/mergetree/LookupLevelsTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/mergetree/LookupLevelsTest.java @@ -60,6 +60,8 @@ import org.junit.jupiter.api.RepeatedTest; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.CsvSource; import java.io.File; import java.io.IOException; @@ -283,6 +285,70 @@ public void testLookupLevel0() throws Exception { assertThat(kv.value().getInt(1)).isEqualTo(11); } + @ParameterizedTest + @CsvSource({"0,2", "2,3"}) + public void testCachedFileLevelAfterUpgrade(int initialLevel, int upgradedLevel) + throws IOException { + DataFileMeta file = newFile(initialLevel, kv(1, 11, 7)); + DataFileMeta upgraded = file.upgrade(upgradedLevel); + Levels levels = new Levels(comparator, Collections.singletonList(file), upgradedLevel + 1); + try (LookupLevels lookupLevels = + createLookupLevels(levels, MemorySize.ofMebiBytes(10))) { + lookupLevels.prepareLookupFiles(Collections.singletonList(file)); + LookupFile cached = lookupLevels.lookupFiles().getIfPresent(file.fileName()); + assertThat(cached).isNotNull(); + + // Revisit the old view after upgrading, as when reading an older on-read split. + for (DataFileMeta current : Arrays.asList(file, upgraded, file)) { + levels.update(levels.allFiles(), Collections.singletonList(current)); + KeyValue result = lookupLevels.lookup(row(1), 0); + assertThat(result).isNotNull(); + assertThat(result.level()).isEqualTo(current.level()); + assertThat(lookupLevels.lookupAll(row(1))) + .hasSize(1) + .allSatisfy( + kv -> { + assertThat(kv.level()).isEqualTo(current.level()); + assertThat(kv.sequenceNumber()).isEqualTo(7); + assertThat(kv.valueKind()).isEqualTo(RowKind.INSERT); + assertThat(kv.value().getInt(1)).isEqualTo(11); + }); + assertThat(lookupLevels.lookupFiles().getIfPresent(file.fileName())) + .isSameAs(cached); + } + } + } + + @Test + public void testPrepareLookupFiles() throws Exception { + DataFileMeta first = newFile(1, kv(1, 11)); + DataFileMeta second = newFile(1, kv(2, 22)); + Levels levels = new Levels(comparator, Arrays.asList(first, second), 1); + AtomicInteger localFileRequests = new AtomicInteger(); + LookupLevels lookupLevels = + createLookupLevels( + levels, + MemorySize.ofMebiBytes(10), + fileName -> { + localFileRequests.incrementAndGet(); + return new File(tempDir.toFile(), LOOKUP_FILE_PREFIX + fileName); + }); + + try { + lookupLevels.prepareLookupFiles(Arrays.asList(first, second)); + assertThat(localFileRequests.get()).isEqualTo(2); + assertThat(lookupLevels.lookupFiles().asMap().keySet()) + .containsExactlyInAnyOrder(first.fileName(), second.fileName()); + + assertThat(lookupLevels.lookup(row(1), 1).value().getInt(1)).isEqualTo(11); + assertThat(lookupLevels.lookup(row(2), 1).value().getInt(1)).isEqualTo(22); + lookupLevels.prepareLookupFiles(Arrays.asList(first, second)); + assertThat(localFileRequests.get()).isEqualTo(2); + } finally { + lookupLevels.close(); + } + } + @Test public void testConcurrentLookupCreatesSingleLookupFile() throws Exception { DataFileMeta file = newFile(1, kv(1, 11), kv(2, 22)); diff --git a/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/SortMergeReaderCleanupTest.java b/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/SortMergeReaderCleanupTest.java new file mode 100644 index 000000000000..c045f2537581 --- /dev/null +++ b/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/SortMergeReaderCleanupTest.java @@ -0,0 +1,154 @@ +/* + * 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.mergetree.compact; + +import org.apache.paimon.CoreOptions; +import org.apache.paimon.CoreOptions.SortEngine; +import org.apache.paimon.KeyValue; +import org.apache.paimon.data.BinaryString; +import org.apache.paimon.data.GenericRow; +import org.apache.paimon.disk.IOManager; +import org.apache.paimon.options.MemorySize; +import org.apache.paimon.options.Options; +import org.apache.paimon.reader.RecordReader; +import org.apache.paimon.types.DataTypes; +import org.apache.paimon.types.RowKind; +import org.apache.paimon.types.RowType; +import org.apache.paimon.utils.IteratorRecordReader; + +import org.junit.jupiter.api.io.TempDir; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.EnumSource; + +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.Comparator; +import java.util.List; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.stream.Stream; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Tests for merge state cleanup when a {@link SortMergeReader} is closed. */ +class SortMergeReaderCleanupTest { + + @TempDir Path tempDir; + + @ParameterizedTest + @EnumSource(SortEngine.class) + void testCloseReleasesLookupSpills(SortEngine sortEngine) throws Exception { + Options rawOptions = new Options(); + rawOptions.set(CoreOptions.LOOKUP_MERGE_RECORDS_THRESHOLD, 0); + rawOptions.set(CoreOptions.LOOKUP_MERGE_BUFFER_SIZE, MemorySize.parse("4 kb")); + rawOptions.set(CoreOptions.PAGE_SIZE, MemorySize.parse("4 kb")); + CoreOptions options = new CoreOptions(rawOptions); + RowType keyType = RowType.builder().field("k", DataTypes.INT()).build(); + RowType valueType = RowType.builder().field("v", DataTypes.STRING()).build(); + char[] chars = new char[1000]; + Arrays.fill(chars, 'x'); + BinaryString value = BinaryString.fromString(new String(chars)); + try (IOManager ioManager = IOManager.create(tempDir.toString())) { + MergeFunction merge = + new LookupMergeFunction( + DeduplicateMergeFunction.factory().create(), + options, + keyType, + valueType, + ioManager); + List> inputs = new ArrayList<>(); + for (int i = 0; i < 16; i++) { + KeyValue kv = + new KeyValue() + .replace(GenericRow.of(1), i, RowKind.INSERT, GenericRow.of(value)) + .setLevel(0); + inputs.add(new IteratorRecordReader<>(Collections.singletonList(kv).iterator())); + } + KeyValue result; + try (SortMergeReader reader = + SortMergeReader.createSortMergeReader( + inputs, + Comparator.comparingInt(row -> row.getInt(0)), + null, + new ReducerMergeFunctionWrapper(merge), + sortEngine)) { + RecordReader.RecordIterator batch = reader.readBatch(); + assertThat(batch).isNotNull(); + try { + result = batch.next(); + assertThat(result).isNotNull(); + assertThat(result.sequenceNumber()).isEqualTo(15); + assertThat(spillFileCount()).isPositive(); + } finally { + batch.releaseBatch(); + } + } + assertThat(spillFileCount()).isZero(); + assertThat(result.value().getString(0)).isEqualTo(value); + } + } + + @ParameterizedTest + @EnumSource(SortEngine.class) + void testCloseStillResetsMergeWhenInputCloseFails(SortEngine sortEngine) { + IOException inputFailure = new IOException("input close failed"); + RuntimeException resetFailure = new IllegalStateException("merge reset failed"); + RecordReader input = + new RecordReader() { + @Override + public RecordIterator readBatch() { + return null; + } + + @Override + public void close() throws IOException { + throw inputFailure; + } + }; + AtomicInteger resets = new AtomicInteger(); + MergeFunctionWrapper merge = + new ReducerMergeFunctionWrapper(DeduplicateMergeFunction.factory().create()) { + @Override + public void reset() { + resets.incrementAndGet(); + throw resetFailure; + } + }; + SortMergeReader reader = + SortMergeReader.createSortMergeReader( + Collections.singletonList(input), + Comparator.comparingInt(row -> row.getInt(0)), + null, + merge, + sortEngine); + assertThatThrownBy(reader::close).isSameAs(inputFailure); + assertThat(inputFailure.getSuppressed()).containsExactly(resetFailure); + assertThat(resets.get()).isEqualTo(1); + } + + private long spillFileCount() throws IOException { + try (Stream paths = Files.walk(tempDir)) { + return paths.filter(Files::isRegularFile).count(); + } + } +} diff --git a/paimon-core/src/test/java/org/apache/paimon/table/ChangelogOnReadTest.java b/paimon-core/src/test/java/org/apache/paimon/table/ChangelogOnReadTest.java new file mode 100644 index 000000000000..4d7979e46338 --- /dev/null +++ b/paimon-core/src/test/java/org/apache/paimon/table/ChangelogOnReadTest.java @@ -0,0 +1,655 @@ +/* + * 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; + +import org.apache.paimon.CoreOptions; +import org.apache.paimon.KeyValue; +import org.apache.paimon.data.BinaryRow; +import org.apache.paimon.data.BinaryString; +import org.apache.paimon.data.GenericRow; +import org.apache.paimon.data.InternalRow; +import org.apache.paimon.disk.IOManager; +import org.apache.paimon.io.DataFileMeta; +import org.apache.paimon.operation.MergeFileSplitRead; +import org.apache.paimon.reader.RecordReader; +import org.apache.paimon.schema.Schema; +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.sink.CommitMessage; +import org.apache.paimon.table.source.DataSplit; +import org.apache.paimon.table.source.IncrementalSplit; +import org.apache.paimon.table.source.ReadBuilder; +import org.apache.paimon.table.source.Split; +import org.apache.paimon.table.source.StreamTableScan; +import org.apache.paimon.table.source.TableRead; +import org.apache.paimon.table.source.TableScan; +import org.apache.paimon.types.DataTypes; +import org.apache.paimon.types.RowKind; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.CsvSource; +import org.junit.jupiter.params.provider.ValueSource; + +import java.io.IOException; +import java.lang.reflect.Field; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.stream.Collectors; +import java.util.stream.Stream; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Tests for generating changelog during reads. */ +public class ChangelogOnReadTest extends TableTestBase { + + @ParameterizedTest + @ValueSource(booleans = {false, true}) + public void testAggregationProjectionMultipleLevel0AndCacheLifecycle(boolean forceLookup) + throws Exception { + FileStoreTable table = + createTable( + CoreOptions.MergeEngine.AGGREGATE, + CoreOptions.CHANGELOG_ON_READ_LOOKUP_STATE_MAX_SIZE.defaultValue(), + forceLookup); + write(table, GenericRow.of(1, 1, 1, 1)); + + StreamTableScan scan = table.newStreamScan(); + ReadBuilder readBuilder = table.newReadBuilder().withProjection(new int[] {3}); + Path lookupDir = tempPath.resolve("lookup"); + IOManager lookupIO = IOManager.create(lookupDir.toString()); + TableRead read = readBuilder.newRead().withIOManager(lookupIO); + Path firstLookupFile = null; + try { + TableScan.Plan initial = scan.plan(); + String firstDataFile = + ((DataSplit) initial.splits().get(0)).dataFiles().get(0).fileName(); + assertThat(read(read, initial)).containsExactly("+I 1"); + + write(table, GenericRow.of(1, 1, 2, 2)); + assertThat(readAfterLookupPrepared(read, scan.plan(), lookupDir, firstDataFile)) + .containsExactly("-U 1", "+U 3"); + firstLookupFile = findLookupFile(lookupDir, firstDataFile); + long modified = Files.getLastModifiedTime(firstLookupFile).toMillis(); + + write(table, GenericRow.of(1, 1, 0, 3)); + assertThat(read(read, scan.plan())).containsExactly("-U 3", "+U 6"); + assertThat(firstLookupFile).exists(); + assertThat(Files.getLastModifiedTime(firstLookupFile).toMillis()).isEqualTo(modified); + + // The temporary history filter for partition 1 must not affect the next delta scan. + write(table, GenericRow.of(2, 1, 1, 4)); + assertThat(read(read, scan.plan())).containsExactly("+I 4"); + } finally { + read.close(); + if (firstLookupFile != null) { + assertThat(firstLookupFile).doesNotExist(); + } + lookupIO.close(); + } + } + + @Test + public void testForceLookupPreservesCompactedHistoryWithProjection() throws Exception { + FileStoreTable table = + createTable( + CoreOptions.MergeEngine.AGGREGATE, + CoreOptions.CHANGELOG_ON_READ_LOOKUP_STATE_MAX_SIZE.defaultValue(), + true) + .copy(Collections.singletonMap("lookup.merge-records-threshold", "0")); + write(table, ioManager, GenericRow.of(1, 1, 1, 10)); + DataSplit first = (DataSplit) table.newReadBuilder().newScan().plan().splits().get(0); + String olderFile = first.dataFiles().get(0).fileName(); + write(table, ioManager, GenericRow.of(1, 1, 2, 13)); + StreamTableScan scan = table.newStreamScan(); + scan.plan(); + write(table, ioManager, GenericRow.of(1, 1, 3, 4)); + IncrementalSplit delta = (IncrementalSplit) scan.plan().splits().get(0); + assertThat(delta.beforeFiles()).hasSize(2); + // Model lookup-compacted history: L2 has 10, and the newer L1 value 13 includes it. + List history = + delta.beforeFiles().stream() + .map(file -> file.upgrade(file.fileName().equals(olderFile) ? 2 : 1)) + .collect(Collectors.toList()); + IncrementalSplit split = + new IncrementalSplit( + delta.snapshotId(), + delta.partition(), + delta.bucket(), + delta.totalBuckets(), + history, + null, + delta.afterFiles(), + null, + true); + try (IOManager lookupIO = IOManager.create(tempPath.resolve("lookup").toString()); + TableRead read = + table.newReadBuilder() + .withProjection(new int[] {3}) + .newRead() + .withIOManager(lookupIO); + RecordReader reader = read.createReader(split)) { + List result = new ArrayList<>(); + reader.forEachRemaining( + row -> result.add(row.getRowKind().shortString() + " " + row.getInt(0))); + // History is 13, not 10 + 13; then apply the new delta 4. + assertThat(result).containsExactly("-U 13", "+U 17"); + } + } + + @Test + public void testForceLookupCachedHistoryAfterFileUpgrade() throws Exception { + Map options = new HashMap<>(); + options.put(CoreOptions.FILE_FORMAT.key(), "avro"); + options.put(CoreOptions.NUM_LEVELS.key(), "3"); + options.put(CoreOptions.NUM_SORTED_RUNS_COMPACTION_TRIGGER.key(), "10"); + options.put(CoreOptions.LOOKUP_COMPACT.key(), "radical"); + FileStoreTable table = + createTable( + CoreOptions.MergeEngine.AGGREGATE, + CoreOptions.CHANGELOG_ON_READ_LOOKUP_STATE_MAX_SIZE.defaultValue(), + true) + .copy(options); + // Keep the history file large enough that the next small delta compacts only to L1. + List rows = new ArrayList<>(); + for (int key = 1; key <= 2000; key++) { + rows.add(GenericRow.of(1, key, 1, 10)); + } + write(table, ioManager, rows.toArray(new InternalRow[0])); + ReadBuilder readBuilder = table.newReadBuilder().withProjection(new int[] {3}); + StreamTableScan scan = readBuilder.newStreamScan(); + DataSplit initial = (DataSplit) scan.plan().splits().get(0); + assertThat(initial.dataFiles()).hasSize(1); + String historyFile = initial.dataFiles().get(0).fileName(); + FileStoreTable compactionTable = + table.copy(Collections.singletonMap(CoreOptions.WRITE_ONLY.key(), "false")); + BatchWriteBuilder compactBuilder = compactionTable.newBatchWriteBuilder(); + try (TableRead read = readBuilder.newRead().withIOManager(ioManager); + BatchTableWrite compactWrite = compactBuilder.newWrite(); + BatchTableCommit compactCommit = compactBuilder.newCommit()) { + compactWrite.withIOManager(ioManager); + compactWrite.compact(initial.partition(), initial.bucket(), true); + List upgrade = compactWrite.prepareCommit(); + + // Cache the original file at L0 before publishing its same-name upgrade to L2. + write(table, ioManager, GenericRow.of(1, 1, 2, 3)); + assertThat(read(read, scan.plan())).containsExactly("-U 10", "+U 13"); + compactCommit.commit(upgrade); + compact(compactionTable, initial.partition(), initial.bucket(), ioManager, false); + + write(table, ioManager, GenericRow.of(1, 1, 3, 4)); + TableScan.Plan delta = scan.plan(); + IncrementalSplit split = (IncrementalSplit) delta.splits().get(0); + assertThat(split.beforeFiles()) + .extracting(DataFileMeta::level) + .containsExactlyInAnyOrder(1, 2); + assertThat(split.beforeFiles()) + .filteredOn(file -> file.fileName().equals(historyFile)) + .hasSize(1) + .allSatisfy(file -> assertThat(file.level()).isEqualTo(2)); + // L1 contains 13, including the older L2 value 10; count that history only once. + assertThat(read(read, delta)).containsExactly("-U 13", "+U 17"); + try (TableRead coldRead = readBuilder.newRead().withIOManager(ioManager)) { + assertThat(read(coldRead, delta)).containsExactly("-U 13", "+U 17"); + } + } + } + + @Test + public void testDeduplicateMultipleLevel0() throws Exception { + FileStoreTable table = createTable(CoreOptions.MergeEngine.DEDUPLICATE); + write(table, GenericRow.of(1, 1, 1, 10)); + + StreamTableScan scan = table.newStreamScan(); + ReadBuilder readBuilder = table.newReadBuilder().withProjection(new int[] {3}); + try (IOManager lookupIO = IOManager.create(tempPath.resolve("lookup").toString()); + TableRead read = readBuilder.newRead().withIOManager(lookupIO)) { + assertThat(read(read, scan.plan())).containsExactly("+I 10"); + + write(table, GenericRow.of(1, 1, 3, 30)); + assertThat(read(read, scan.plan())).containsExactly("-U 10", "+U 30"); + + write(table, GenericRow.of(1, 1, 4, 40)); + assertThat(read(read, scan.plan())).containsExactly("-U 30", "+U 40"); + } + } + + @ParameterizedTest + @CsvSource({ + "false,INSERT,DELETE", + "true,INSERT,DELETE", + "false,UPDATE_AFTER,UPDATE_BEFORE", + "true,UPDATE_AFTER,UPDATE_BEFORE" + }) + public void testHistoryWithSameSequenceAndDifferentRowKinds( + boolean withSequenceField, RowKind addKind, RowKind retractKind) throws Exception { + Schema.Builder schema = + Schema.newBuilder() + .column("k", DataTypes.INT()) + .column("seq", DataTypes.INT()) + .column("v", DataTypes.INT()) + .primaryKey("k") + .option(CoreOptions.BUCKET.key(), "1") + .option(CoreOptions.WRITE_ONLY.key(), "true") + .option(CoreOptions.CHANGELOG_PRODUCER.key(), "on-read"); + if (withSequenceField) { + schema.option(CoreOptions.SEQUENCE_FIELD.key(), "seq"); + } + catalog.createTable(identifier(), schema.build(), false); + FileStoreTable table = getTableDefault(); + BatchWriteBuilder first = table.newBatchWriteBuilder(); + BatchWriteBuilder second = table.newBatchWriteBuilder(); + try (BatchTableWrite firstWrite = first.newWrite(); + BatchTableWrite secondWrite = second.newWrite(); + BatchTableCommit firstCommit = first.newCommit(); + BatchTableCommit secondCommit = second.newCommit()) { + // Both writers start before either commit, so each key has equal internal sequences. + // Reverse the row kinds for key 2 to cover either history file iteration order. + firstWrite.write(GenericRow.ofKind(addKind, 1, 1, 10)); + firstWrite.write(GenericRow.ofKind(retractKind, 2, 1, 20)); + secondWrite.write(GenericRow.ofKind(retractKind, 1, 1, 10)); + secondWrite.write(GenericRow.ofKind(addKind, 2, 1, 20)); + List firstMessages = firstWrite.prepareCommit(); + List secondMessages = secondWrite.prepareCommit(); + firstCommit.commit(firstMessages); + secondCommit.commit(secondMessages); + } + + ReadBuilder readBuilder = table.newReadBuilder().withProjection(new int[] {2}); + StreamTableScan scan = readBuilder.newStreamScan(); + try (TableRead read = readBuilder.newRead().withIOManager(ioManager)) { + assertThat(read(read, scan.plan())) + .containsExactlyInAnyOrder( + addKind.shortString() + " 10", addKind.shortString() + " 20"); + write(table, GenericRow.of(1, 2, 11), GenericRow.of(2, 2, 21)); + TableScan.Plan delta = scan.plan(); + IncrementalSplit split = (IncrementalSplit) delta.splits().get(0); + assertThat(split.beforeFiles()) + .hasSize(2) + .allSatisfy( + file -> { + assertThat(file.minSequenceNumber()).isZero(); + assertThat(file.maxSequenceNumber()).isEqualTo(1); + }); + assertThat(read(read, delta)).containsExactly("-U 10", "+U 11", "-U 20", "+U 21"); + } + } + + @Test + public void testLookupStateEvictionAndReload() throws Exception { + FileStoreTable table = createTable(CoreOptions.MergeEngine.AGGREGATE, 2); + write( + table, + GenericRow.of(1, 1, 1, 10), + GenericRow.of(2, 1, 1, 20), + GenericRow.of(3, 1, 1, 30)); + + StreamTableScan scan = table.newStreamScan(); + Path lookupDir = tempPath.resolve("lookup"); + List cachedFiles = new ArrayList<>(); + try (IOManager lookupIO = IOManager.create(lookupDir.toString())) { + TableRead read = + table.newReadBuilder() + .withProjection(new int[] {3}) + .newRead() + .withIOManager(lookupIO); + try { + TableScan.Plan initial = scan.plan(); + Map dataFiles = new HashMap<>(); + for (Split split : initial.splits()) { + DataSplit dataSplit = (DataSplit) split; + dataFiles.put( + dataSplit.partition().getInt(0), + dataSplit.dataFiles().get(0).fileName()); + } + assertThat(read(read, initial)) + .containsExactlyInAnyOrder("+I 10", "+I 20", "+I 30"); + + // Access A, B, A, C: B is the least recently used state, not A. + write(table, GenericRow.of(1, 1, 2, 1)); + assertThat(read(read, scan.plan())).containsExactly("-U 10", "+U 11"); + Path lookupA = findLookupFile(lookupDir, dataFiles.get(1)); + cachedFiles.add(lookupA); + + write(table, GenericRow.of(2, 1, 2, 2)); + assertThat(read(read, scan.plan())).containsExactly("-U 20", "+U 22"); + Path lookupB = findLookupFile(lookupDir, dataFiles.get(2)); + cachedFiles.add(lookupB); + + write(table, GenericRow.of(1, 1, 3, 3)); + assertThat(read(read, scan.plan())).containsExactly("-U 11", "+U 14"); + assertThat(lookupA).exists(); + assertThat(lookupB).exists(); + + write(table, GenericRow.of(3, 1, 2, 4)); + assertThat(read(read, scan.plan())).containsExactly("-U 30", "+U 34"); + Path lookupC = findLookupFile(lookupDir, dataFiles.get(3)); + cachedFiles.add(lookupC); + assertThat(lookupA).exists(); + assertThat(lookupB).doesNotExist(); + assertThat(lookupC).exists(); + + // Reconstruct the evicted partition from both historical level-0 files. + write(table, GenericRow.of(2, 1, 3, 5)); + assertThat(read(read, scan.plan())).containsExactly("-U 22", "+U 27"); + Path reloadedLookupB = findLookupFile(lookupDir, dataFiles.get(2)); + cachedFiles.add(reloadedLookupB); + assertThat(reloadedLookupB).exists().isNotEqualTo(lookupB); + assertThat(lookupA).doesNotExist(); + assertThat(lookupB).doesNotExist(); + assertThat(lookupC).exists(); + } finally { + read.close(); + for (Path cachedFile : cachedFiles) { + assertThat(cachedFile).doesNotExist(); + } + } + } + } + + @Test + public void testLookupStateCountBoundedAcrossPartitionsAndBuckets() throws Exception { + int maxStates = 2; + FileStoreTable table = createTable(CoreOptions.MergeEngine.DEDUPLICATE, maxStates); + try (IOManager lookupIO = IOManager.create(tempPath.resolve("lookup").toString())) { + MergeFileSplitRead read = + ((PrimaryKeyFileStoreTable) table).store().newRead().withIOManager(lookupIO); + Field lookupLevelsField = MergeFileSplitRead.class.getDeclaredField("lookupLevelsMap"); + lookupLevelsField.setAccessible(true); + Map lookupLevels = (Map) lookupLevelsField.get(read); + try { + int statesCreated = 0; + for (int partition = 0; partition < 32; partition++) { + for (int bucket = 0; bucket < 4; bucket++) { + // Even states without cached files own locks and must be bounded. + IncrementalSplit split = + new IncrementalSplit( + 1L, + BinaryRow.singleColumn(partition), + bucket, + 4, + Collections.emptyList(), + null, + Collections.emptyList(), + null, + true); + try (RecordReader reader = + read.createChangelogOnReadReader(split)) { + assertThat(reader.readBatch()).isNull(); + } + statesCreated++; + assertThat(lookupLevels).hasSize(Math.min(maxStates, statesCreated)); + } + } + } finally { + read.close(); + } + assertThat(lookupLevels).isEmpty(); + } + } + + @ParameterizedTest + @ValueSource(ints = {0, -1}) + public void testRejectNonPositiveLookupStateMaxSize(int maxStates) throws Exception { + FileStoreTable table = createTable(CoreOptions.MergeEngine.DEDUPLICATE, maxStates); + assertThatThrownBy(() -> ((PrimaryKeyFileStoreTable) table).store().newRead()) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining(CoreOptions.CHANGELOG_ON_READ_LOOKUP_STATE_MAX_SIZE.key()); + } + + @ParameterizedTest + @CsvSource({"min-heap,false", "min-heap,true", "loser-tree,false", "loser-tree,true"}) + public void testCloseReleasesMergeSpills(String sortEngine, boolean forceLookup) + throws Exception { + testMergeSpillCleanup(sortEngine, forceLookup, false, false); + } + + @ParameterizedTest + @CsvSource({"min-heap,false", "min-heap,true", "loser-tree,false", "loser-tree,true"}) + public void testEarlyCloseReleasesMergeSpills(String sortEngine, boolean forceLookup) + throws Exception { + testMergeSpillCleanup(sortEngine, forceLookup, true, false); + } + + @ParameterizedTest + @CsvSource({"min-heap,false", "min-heap,true", "loser-tree,false", "loser-tree,true"}) + public void testProjectedReaderReleasesMergeSpills(String sortEngine, boolean forceLookup) + throws Exception { + testMergeSpillCleanup(sortEngine, forceLookup, false, true); + } + + private void testMergeSpillCleanup( + String sortEngine, boolean forceLookup, boolean earlyClose, boolean projected) + throws Exception { + Schema schema = + Schema.newBuilder() + .column("k", DataTypes.INT()) + .column("seq", DataTypes.INT()) + .column("v", DataTypes.STRING()) + .primaryKey("k") + .option("bucket", "1") + .option("write-only", "true") + .option("file.format", "avro") + .option("changelog-producer", "on-read") + .option("force-lookup", String.valueOf(forceLookup)) + .option("sequence.field", "seq") + .option("sort-engine", sortEngine) + .option("sort-spill-threshold", "1000") + .option("lookup.merge-records-threshold", "0") + .option("lookup.merge-buffer-size", "4 kb") + .option("page-size", "4 kb") + .build(); + catalog.createTable(identifier(), schema, false); + FileStoreTable table = getTableDefault(); + char[] chars = new char[1000]; + Arrays.fill(chars, 'x'); + BinaryString value = BinaryString.fromString(new String(chars)); + // Keep enough historical versions to spill historyMerge when force-lookup is enabled. + for (int i = 0; i < 12; i++) { + write(table, ioManager, GenericRow.of(1, i, value)); + } + ReadBuilder builder = table.newReadBuilder(); + if (projected) { + // Drop the primary key and reorder values so spill serialization must use readType. + builder.withProjection(new int[] {2, 1}); + } + StreamTableScan scan = builder.newStreamScan(); + DataSplit initial = (DataSplit) scan.plan().splits().get(0); + String firstDataFile = initial.dataFiles().get(0).fileName(); + Path lookupDir = tempPath.resolve("lookup"); + try (IOManager lookupIO = IOManager.create(lookupDir.toString())) { + TableRead read = builder.newRead().withIOManager(lookupIO); + Path firstLookupFile = null; + try { + for (int round = 1; round <= 2; round++) { + // Multiple writers contribute overlapping files to one APPEND snapshot. + List messages = new ArrayList<>(); + BatchWriteBuilder writeBuilder = table.newBatchWriteBuilder(); + for (int i = 0; i < 12; i++) { + try (BatchTableWrite write = writeBuilder.newWrite()) { + write.withIOManager(ioManager); + write.write(GenericRow.of(1, round * 100 + i, value)); + messages.addAll(write.prepareCommit()); + } + } + try (BatchTableCommit commit = writeBuilder.newCommit()) { + commit.commit(messages); + } + // The table's merge-function factory is shared by writes and reads. + read.withIOManager(lookupIO); + TableScan.Plan plan = scan.plan(); + assertThat(plan.splits()).hasSize(1); + List result = new ArrayList<>(); + try (RecordReader reader = read.createReader(plan)) { + RecordReader.RecordIterator batch = reader.readBatch(); + assertThat(batch).isNotNull(); + try { + InternalRow before = batch.next(); + assertThat(before).isNotNull(); + assertThat(before.getFieldCount()).isEqualTo(projected ? 2 : 3); + assertThat(before.getString(projected ? 0 : 2)).isEqualTo(value); + result.add(before.getRowKind().shortString() + " " + before.getInt(1)); + assertThat(mergeSpillFiles(lookupDir)).isNotEmpty(); + if (!earlyClose) { + InternalRow row; + while ((row = batch.next()) != null) { + result.add( + row.getRowKind().shortString() + " " + row.getInt(1)); + } + } + } finally { + batch.releaseBatch(); + } + if (!earlyClose) { + while ((batch = reader.readBatch()) != null) { + try { + InternalRow row; + while ((row = batch.next()) != null) { + result.add( + row.getRowKind().shortString() + + " " + + row.getInt(1)); + } + } finally { + batch.releaseBatch(); + } + } + } + } + int beforeSeq = (round - 1) * 100 + 11; + if (earlyClose) { + assertThat(result).containsExactly("-U " + beforeSeq); + } else { + assertThat(result) + .containsExactly("-U " + beforeSeq, "+U " + (round * 100 + 11)); + } + assertThat(mergeSpillFiles(lookupDir)).isEmpty(); + Path cachedFile = findLookupFile(lookupDir, firstDataFile); + if (firstLookupFile == null) { + firstLookupFile = cachedFile; + } else { + assertThat(cachedFile).isEqualTo(firstLookupFile); + } + assertThat(firstLookupFile).exists(); + } + } finally { + read.close(); + } + assertThat(firstLookupFile).doesNotExist(); + assertThat(mergeSpillFiles(lookupDir)).isEmpty(); + } + } + + private List mergeSpillFiles(Path lookupDir) throws IOException { + try (Stream paths = Files.walk(lookupDir)) { + // Lookup files have a data-file prefix; only merge buffers use unnamed channels. + return paths.filter(Files::isRegularFile) + .filter(path -> path.getFileName().toString().matches("[0-9a-f]{32}[.]channel")) + .collect(Collectors.toList()); + } + } + + private FileStoreTable createTable(CoreOptions.MergeEngine mergeEngine) throws Exception { + return createTable( + mergeEngine, CoreOptions.CHANGELOG_ON_READ_LOOKUP_STATE_MAX_SIZE.defaultValue()); + } + + private FileStoreTable createTable(CoreOptions.MergeEngine mergeEngine, int maxStates) + throws Exception { + return createTable(mergeEngine, maxStates, false); + } + + private FileStoreTable createTable( + CoreOptions.MergeEngine mergeEngine, int maxStates, boolean forceLookup) + throws Exception { + Schema.Builder builder = + Schema.newBuilder() + .column("p", DataTypes.INT()) + .column("k", DataTypes.INT()) + .column("seq", DataTypes.INT()) + .column("v", DataTypes.INT()) + .primaryKey("p", "k") + .partitionKeys("p") + .option(CoreOptions.BUCKET.key(), "1") + .option(CoreOptions.WRITE_ONLY.key(), "true") + .option(CoreOptions.CHANGELOG_PRODUCER.key(), "on-read") + .option(CoreOptions.MERGE_ENGINE.key(), mergeEngine.toString()) + .option(CoreOptions.FORCE_LOOKUP.key(), String.valueOf(forceLookup)) + .option(CoreOptions.SEQUENCE_FIELD.key(), "seq") + .option( + CoreOptions.CHANGELOG_ON_READ_LOOKUP_STATE_MAX_SIZE.key(), + String.valueOf(maxStates)); + if (mergeEngine == CoreOptions.MergeEngine.AGGREGATE) { + builder.option("fields.v.aggregate-function", "sum"); + } + catalog.createTable(identifier(), builder.build(), false); + return getTableDefault(); + } + + private List read(TableRead read, TableScan.Plan plan) throws IOException { + List result = new ArrayList<>(); + try (RecordReader reader = read.createReader(plan)) { + reader.forEachRemaining( + row -> result.add(row.getRowKind().shortString() + " " + row.getInt(0))); + } + return result; + } + + private List readAfterLookupPrepared( + TableRead read, TableScan.Plan plan, Path lookupDir, String dataFile) + throws IOException { + List result = new ArrayList<>(); + try (RecordReader reader = read.createReader(plan)) { + RecordReader.RecordIterator batch = reader.readBatch(); + assertThat(batch).isNotNull(); + assertThat(findLookupFile(lookupDir, dataFile)).exists(); + consume(batch, result); + while ((batch = reader.readBatch()) != null) { + consume(batch, result); + } + } + return result; + } + + private void consume(RecordReader.RecordIterator batch, List result) + throws IOException { + InternalRow row; + while ((row = batch.next()) != null) { + result.add(row.getRowKind().shortString() + " " + row.getInt(0)); + } + batch.releaseBatch(); + } + + private Path findLookupFile(Path lookupDir, String dataFile) throws IOException { + try (Stream paths = Files.walk(lookupDir)) { + return paths.filter(Files::isRegularFile) + .filter(path -> path.getFileName().toString().contains(dataFile)) + .findFirst() + .orElseThrow(() -> new AssertionError("Lookup file was not cached.")); + } + } +} 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..55569b97826f 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 @@ -2862,6 +2862,43 @@ public void testRollbackToTagWithChangelogDecoupled(String changelogProducer) th assertThat(changelogManager.latestLongLivedChangelogId()).isNull(); } + @ParameterizedTest + @EnumSource(CoreOptions.SortEngine.class) + public void testForceLookupWithProjectedBatchRead(CoreOptions.SortEngine sortEngine) + throws Exception { + FileStoreTable table = + createFileStoreTable( + options -> { + options.set(CoreOptions.FORCE_LOOKUP, true); + options.set(CoreOptions.WRITE_ONLY, true); + options.set(CoreOptions.LOOKUP_MERGE_RECORDS_THRESHOLD, 0); + options.set(CoreOptions.SORT_ENGINE, sortEngine); + }); + try (StreamTableWrite write = table.newWrite(commitUser); + StreamTableCommit commit = table.newCommit(commitUser)) { + write.write(rowData(1, 10, 100L)); + commit.commit(0, write.prepareCommit(false, 0)); + write.write(rowData(1, 10, 200L)); + commit.commit(1, write.prepareCommit(false, 1)); + } + ReadBuilder builder = table.newReadBuilder().withProjection(new int[] {2, 1}); + List splits = builder.newScan().plan().splits(); + assertThat(splits).hasSize(1); + assertThat(((DataSplit) splits.get(0)).dataFiles()).hasSize(2); + try (IOManager ioManager = IOManager.create(tempDir.resolve("lookup").toString()); + TableRead read = builder.newRead().withIOManager(ioManager)) { + assertThat( + getResult( + read, + splits, + row -> { + assertThat(row.getFieldCount()).isEqualTo(2); + return row.getLong(0) + "|" + row.getInt(1); + })) + .containsExactly("200|10"); + } + } + @ParameterizedTest @EnumSource(CoreOptions.MergeEngine.class) public void testForceLookupCompaction(CoreOptions.MergeEngine mergeEngine) throws Exception { diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/BaseDataTableSource.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/BaseDataTableSource.java index 9bdbf0e7dc72..675581bfb4df 100644 --- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/BaseDataTableSource.java +++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/BaseDataTableSource.java @@ -274,6 +274,11 @@ public LookupRuntimeProvider getLookupRuntimeProvider(LookupContext context) { "Option 'postpone.merge-on-read' is not supported for lookup reads."); } + if (CoreOptions.fromMap(table.options()).changelogOnRead()) { + throw new UnsupportedOperationException( + "Lookup join is not supported for tables with changelog-producer=on-read."); + } + if (limit != null) { throw new RuntimeException( "Limit push down should not happen in Lookup source, but it is " + limit); diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FileStoreSourceReader.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FileStoreSourceReader.java index 1e61126fac73..18e852b7fc57 100644 --- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FileStoreSourceReader.java +++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FileStoreSourceReader.java @@ -41,6 +41,7 @@ public class FileStoreSourceReader extends SingleThreadMultiplexSourceReaderBase< RecordIterator, RowData, FileStoreSourceSplit, FileStoreSourceSplitState> { + private final TableRead tableRead; private final IOManager ioManager; private long lastConsumeSnapshotId = Long.MIN_VALUE; @@ -68,6 +69,7 @@ public FileStoreSourceReader( readerContext, element, output, state, metrics, rowData), readerContext.getConfiguration(), readerContext); + this.tableRead = tableRead; this.ioManager = ioManager; } @@ -116,7 +118,14 @@ protected FileStoreSourceSplit toSplitType( @Override public void close() throws Exception { - super.close(); - ioManager.close(); + try { + super.close(); + } finally { + try { + tableRead.close(); + } finally { + ioManager.close(); + } + } } } diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FileStoreSourceSplitReader.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FileStoreSourceSplitReader.java index 1c230ab59b1b..782f85b92bbc 100644 --- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FileStoreSourceSplitReader.java +++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FileStoreSourceSplitReader.java @@ -25,6 +25,7 @@ import org.apache.paimon.reader.RecordReader; import org.apache.paimon.reader.RecordReader.RecordIterator; import org.apache.paimon.table.source.DataSplit; +import org.apache.paimon.table.source.IncrementalSplit; import org.apache.paimon.table.source.Split; import org.apache.paimon.table.source.Splits; import org.apache.paimon.table.source.TableRead; @@ -227,6 +228,11 @@ private void checkSplitOrStartNext() throws IOException { .earliestFileCreationEpochMillis() .orElse(FileStoreSourceReaderMetrics.UNDEFINED); metrics.recordSnapshotUpdate(eventTime); + } else if (inner instanceof IncrementalSplit) { + metrics.recordSnapshotUpdate( + ((IncrementalSplit) inner) + .earliestFileCreationEpochMillis() + .orElse(FileStoreSourceReaderMetrics.UNDEFINED)); } currentSplitId = nextSplit.splitId(); diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/align/AlignedContinuousFileSplitEnumerator.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/align/AlignedContinuousFileSplitEnumerator.java index 90164b5d3680..574147c5224e 100644 --- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/align/AlignedContinuousFileSplitEnumerator.java +++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/align/AlignedContinuousFileSplitEnumerator.java @@ -23,10 +23,9 @@ import org.apache.paimon.flink.source.PendingSplitsCheckpoint; import org.apache.paimon.flink.source.assigners.AlignedSplitAssigner; import org.apache.paimon.flink.source.assigners.SplitAssigner; -import org.apache.paimon.table.source.DataSplit; +import org.apache.paimon.flink.utils.TableScanUtils; import org.apache.paimon.table.source.EndOfScanException; import org.apache.paimon.table.source.SnapshotNotExistPlan; -import org.apache.paimon.table.source.Splits; import org.apache.paimon.table.source.StreamTableScan; import org.apache.paimon.table.source.TableScan; import org.apache.paimon.utils.Preconditions; @@ -124,7 +123,12 @@ protected void addSplits(Collection splits) { Map> splitsBySnapshot = new TreeMap<>(); for (FileStoreSourceSplit split : splits) { - long snapshotId = ((DataSplit) Splits.underlying(split.split())).snapshotId(); + long snapshotId = + TableScanUtils.getSnapshotId(split) + .orElseThrow( + () -> + new IllegalArgumentException( + "Snapshot id is required in aligned mode.")); splitsBySnapshot.computeIfAbsent(snapshotId, snapshot -> new ArrayList<>()).add(split); } diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/assigners/AlignedSplitAssigner.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/assigners/AlignedSplitAssigner.java index 5f5686093b96..7667179ae9a2 100644 --- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/assigners/AlignedSplitAssigner.java +++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/assigners/AlignedSplitAssigner.java @@ -20,7 +20,7 @@ import org.apache.paimon.flink.source.FileStoreSourceSplit; import org.apache.paimon.flink.source.align.PlaceholderSplit; -import org.apache.paimon.table.source.DataSplit; +import org.apache.paimon.flink.utils.TableScanUtils; import org.apache.paimon.table.source.Splits; import org.apache.paimon.utils.Preconditions; @@ -67,9 +67,9 @@ public List getNext(int subtask, @Nullable String hostname @Override public void addSplit(int subtask, FileStoreSourceSplit splits) { - long snapshotId = dataSplit(splits).snapshotId(); + long snapshotId = snapshotId(splits); PendingSnapshot last = pendingSplitAssignment.peekLast(); - boolean isPlaceholder = dataSplit(splits) instanceof PlaceholderSplit; + boolean isPlaceholder = Splits.underlying(splits.split()) instanceof PlaceholderSplit; if (last == null || last.snapshotId != snapshotId) { last = new PendingSnapshot(snapshotId, isPlaceholder, new HashMap<>()); last.add(subtask, splits); @@ -86,8 +86,9 @@ public void addSplitsBack(int suggestedTask, List splits) return; } - long snapshotId = dataSplit(splits.get(0)).snapshotId(); - boolean isPlaceholder = dataSplit(splits.get(0)) instanceof PlaceholderSplit; + long snapshotId = snapshotId(splits.get(0)); + boolean isPlaceholder = + Splits.underlying(splits.get(0).split()) instanceof PlaceholderSplit; PendingSnapshot head = pendingSplitAssignment.peek(); if (head == null || snapshotId != head.snapshotId) { head = new PendingSnapshot(snapshotId, isPlaceholder, new HashMap<>()); @@ -155,7 +156,7 @@ public List remove(int subtask) { public void add(int subtask, FileStoreSourceSplit split) { Preconditions.checkArgument( - dataSplit(split).snapshotId() == snapshotId, + snapshotId(split) == snapshotId, "SnapshotId not equal. This is a bug, please file an issue."); subtaskSplits.computeIfAbsent(subtask, id -> new ArrayList<>()).add(split); } @@ -167,8 +168,7 @@ public void addAll(int subtask, List splits) { splits.forEach( split -> Preconditions.checkArgument( - dataSplit(split).snapshotId() == snapshotId, - "SnapshotId not equal")); + snapshotId(split) == snapshotId, "SnapshotId not equal")); subtaskSplits.put(subtask, splits); } @@ -178,7 +178,11 @@ public boolean empty() { } /** Reads the underlying split, assigns the original. */ - private static DataSplit dataSplit(FileStoreSourceSplit split) { - return (DataSplit) Splits.underlying(split.split()); + private static long snapshotId(FileStoreSourceSplit split) { + return TableScanUtils.getSnapshotId(split) + .orElseThrow( + () -> + new IllegalArgumentException( + "Snapshot id is required in aligned mode.")); } } diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/ReadOperator.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/ReadOperator.java index 5004058561f9..e6fbcb5b34af 100644 --- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/ReadOperator.java +++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/ReadOperator.java @@ -26,6 +26,7 @@ import org.apache.paimon.flink.source.RecordLimiter; import org.apache.paimon.flink.source.metrics.FileStoreSourceReaderMetrics; import org.apache.paimon.table.source.DataSplit; +import org.apache.paimon.table.source.IncrementalSplit; import org.apache.paimon.table.source.Split; import org.apache.paimon.table.source.Splits; import org.apache.paimon.table.source.TableRead; @@ -152,7 +153,11 @@ public void processElement(StreamRecord record) throws Exception { ? ((DataSplit) inner) .earliestFileCreationEpochMillis() .orElse(FileStoreSourceReaderMetrics.UNDEFINED) - : FileStoreSourceReaderMetrics.UNDEFINED; + : inner instanceof IncrementalSplit + ? ((IncrementalSplit) inner) + .earliestFileCreationEpochMillis() + .orElse(FileStoreSourceReaderMetrics.UNDEFINED) + : FileStoreSourceReaderMetrics.UNDEFINED; sourceReaderMetrics.recordSnapshotUpdate(eventTime); // update idleStartTime when reading a new split idleStartTime = FileStoreSourceReaderMetrics.ACTIVE; @@ -194,9 +199,18 @@ public void processElement(StreamRecord record) throws Exception { @Override public void close() throws Exception { - super.close(); - if (ioManager != null) { - ioManager.close(); + try { + super.close(); + } finally { + try { + if (read != null) { + read.close(); + } + } finally { + if (ioManager != null) { + ioManager.close(); + } + } } } diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/utils/TableScanUtils.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/utils/TableScanUtils.java index d756ab8bd297..05c5eb9fb546 100644 --- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/utils/TableScanUtils.java +++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/utils/TableScanUtils.java @@ -23,6 +23,7 @@ import org.apache.paimon.options.Options; import org.apache.paimon.table.Table; import org.apache.paimon.table.source.DataSplit; +import org.apache.paimon.table.source.IncrementalSplit; import org.apache.paimon.table.source.Split; import org.apache.paimon.table.source.Splits; import org.apache.paimon.table.source.TableScan; @@ -73,6 +74,8 @@ public static Optional getSnapshotId(FileStoreSourceSplit split) { Split inner = Splits.underlying(split.split()); if (inner instanceof DataSplit) { return Optional.of(((DataSplit) inner).snapshotId()); + } else if (inner instanceof IncrementalSplit) { + return Optional.of(((IncrementalSplit) inner).snapshotId()); } return Optional.empty(); } diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/ChangelogModeTest.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/ChangelogModeTest.java index a652efa63b1d..bd8b13e4570a 100644 --- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/ChangelogModeTest.java +++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/ChangelogModeTest.java @@ -40,9 +40,11 @@ import java.util.Collections; +import static org.apache.paimon.CoreOptions.BUCKET; import static org.apache.paimon.CoreOptions.CHANGELOG_PRODUCER; import static org.apache.paimon.CoreOptions.ChangelogProducer.INPUT; import static org.apache.paimon.CoreOptions.ChangelogProducer.LOOKUP; +import static org.apache.paimon.CoreOptions.ChangelogProducer.ON_READ; import static org.apache.paimon.CoreOptions.PRIMARY_KEY_NULLABLE; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; @@ -139,6 +141,21 @@ public void testKeyOnlyDeletesIgnoredWithInputChangelogProducer() throws Excepti test(options, ChangelogMode.all(), ChangelogMode.all()); } + @Test + public void testChangelogOnRead() throws Exception { + Options options = new Options(); + options.set(BUCKET, 1); + options.set(CHANGELOG_PRODUCER, ON_READ); + test( + options, + ChangelogMode.all(), + ChangelogMode.newBuilder() + .addContainedKind(RowKind.INSERT) + .addContainedKind(RowKind.UPDATE_AFTER) + .addContainedKind(RowKind.DELETE) + .build()); + } + @Test public void testLookupChangelogProducer() throws Exception { Options options = new Options(); diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/LookupChangelogWithAggITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/LookupChangelogWithAggITCase.java index 289c857ae770..44f7eed7bbe1 100644 --- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/LookupChangelogWithAggITCase.java +++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/LookupChangelogWithAggITCase.java @@ -33,6 +33,36 @@ /** Test Lookup changelog producer with aggregation tables. */ public class LookupChangelogWithAggITCase extends CatalogITCaseBase { + @Test + public void testChangelogOnRead() throws Exception { + sql( + "CREATE TABLE T (k INT PRIMARY KEY NOT ENFORCED, seq INT, v INT) WITH (" + + "'bucket'='1', " + + "'write-only'='true', " + + "'changelog-producer'='on-read', " + + "'merge-engine'='aggregation', " + + "'sequence.field'='seq', " + + "'fields.v.aggregate-function'='sum')"); + BlockingIterator iterator = + streamSqlBlockIter( + "SELECT v FROM T /*+ OPTIONS('consumer-id'='on-read', " + + "'consumer.expiration-time'='3h') */"); + + sql("INSERT INTO T VALUES (1, 1, 1)"); + assertThat(iterator.collect(1)).containsExactly(Row.of(1)); + + sql("INSERT INTO T VALUES (1, 2, 2)"); + assertThat(iterator.collect(2)) + .containsExactly( + Row.ofKind(RowKind.UPDATE_BEFORE, 1), Row.ofKind(RowKind.UPDATE_AFTER, 3)); + + sql("INSERT INTO T VALUES (1, 0, 3)"); + assertThat(iterator.collect(2)) + .containsExactly( + Row.ofKind(RowKind.UPDATE_BEFORE, 3), Row.ofKind(RowKind.UPDATE_AFTER, 6)); + iterator.close(); + } + @ParameterizedTest @ValueSource(booleans = {false, true}) public void testMultipleCompaction(boolean changelogRowDeduplicate) throws Exception { diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/DataTableSourceTest.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/DataTableSourceTest.java index 0aca92741e75..0e349738ee56 100644 --- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/DataTableSourceTest.java +++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/DataTableSourceTest.java @@ -57,6 +57,8 @@ import org.apache.flink.table.types.logical.LogicalType; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.CsvSource; import java.util.Collections; import java.util.HashMap; @@ -165,6 +167,44 @@ void testInferPostponeMergeParallelism() throws Exception { assertThat(sourceStream.getParallelism()).isEqualTo(3); } + @ParameterizedTest + @CsvSource({ + "auto,false", + "auto,true", + "full,false", + "full,true", + "memory,false", + "memory,true" + }) + void testChangelogOnReadRejectsLookup(String cacheMode, boolean async) throws Exception { + FileIO fileIO = LocalFileIO.create(); + Path tablePath = new Path(path.toString()); + TableSchema tableSchema = + new FileSystemSchemaManager(fileIO, tablePath) + .createTable( + Schema.newBuilder() + .column("a", DataTypes.INT()) + .column("b", DataTypes.BIGINT()) + .primaryKey("b") + .option("bucket", "1") + .option("changelog-producer", "on-read") + .option("lookup.cache", cacheMode) + .option("lookup.async", Boolean.toString(async)) + .build()); + FileStoreTable fileStoreTable = + FileStoreTableFactory.create(fileIO, tablePath, tableSchema); + DataTableSource tableSource = + new DataTableSource( + ObjectIdentifier.of("cat", "db", "table"), fileStoreTable, true, null); + + assertThatThrownBy(() -> tableSource.getLookupRuntimeProvider(lookupContext())) + .isInstanceOf(UnsupportedOperationException.class) + .hasMessage( + "Lookup join is not supported for tables with changelog-producer=on-read."); + // The restriction must not prevent ordinary streaming scans. + assertThat(runtimeProvider(tableSource)).isNotNull(); + } + @Test void testPostponeMergeRejectsLookupAndDynamicFiltering() throws Exception { FileStoreTable fileStoreTable = createPostponeTable(true); diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/FileStoreSourceReaderTest.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/FileStoreSourceReaderTest.java index 77119b583db8..c99e1c4da02b 100644 --- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/FileStoreSourceReaderTest.java +++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/FileStoreSourceReaderTest.java @@ -18,18 +18,24 @@ package org.apache.paimon.flink.source; +import org.apache.paimon.data.GenericRow; +import org.apache.paimon.data.InternalRow; import org.apache.paimon.disk.IOManager; import org.apache.paimon.flink.source.metrics.FileStoreSourceReaderMetrics; import org.apache.paimon.fs.Path; import org.apache.paimon.fs.local.LocalFileIO; +import org.apache.paimon.metrics.MetricRegistry; +import org.apache.paimon.reader.RecordReader; import org.apache.paimon.schema.FileSystemSchemaManager; import org.apache.paimon.schema.Schema; import org.apache.paimon.schema.SchemaManager; import org.apache.paimon.table.source.KeyValueTableRead; +import org.apache.paimon.table.source.Split; import org.apache.paimon.table.source.TableRead; import org.apache.paimon.types.BigIntType; import org.apache.paimon.types.DataField; import org.apache.paimon.types.IntType; +import org.apache.paimon.types.RowKind; import org.apache.paimon.types.RowType; import org.apache.flink.api.connector.source.SourceEvent; @@ -46,10 +52,17 @@ import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; +import java.io.IOException; +import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; import java.util.List; import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; import static org.apache.paimon.flink.source.FileStoreSourceSplitSerializerTest.newSourceSplit; import static org.apache.paimon.mergetree.compact.MergeTreeCompactManagerTest.row; @@ -152,6 +165,88 @@ public void testIOManagerIsSet() throws Exception { assertThat(tableRead.ioManager()).isNotNull(); } + @Test + public void testCloseClosesTableRead() throws Exception { + TestingReaderContext context = new TestingReaderContext(); + CloseTrackingTableRead tableRead = + new CloseTrackingTableRead( + new TestChangelogDataReadWrite(tempDir.toString()).createReadWithKey()); + FileStoreSourceReader reader = createReader(context, tableRead); + + reader.close(); + + assertThat(tableRead.closeCount()).isEqualTo(1); + } + + @Test + public void testCheckpointDuringFetchAndRestore() throws Exception { + List> batches = + Arrays.asList( + Arrays.asList( + changelogRow(RowKind.UPDATE_BEFORE, 1L), + changelogRow(RowKind.UPDATE_AFTER, 2L)), + Collections.singletonList(changelogRow(RowKind.INSERT, 3L))); + CountDownLatch preparationStarted = new CountDownLatch(1); + CountDownLatch finishPreparation = new CountDownLatch(1); + ExecutorService checkpointExecutor = Executors.newSingleThreadExecutor(); + FileStoreSourceReader reader = + createReader( + new TestingReaderContext(), + new BlockingTableRead(batches, preparationStarted, finishPreparation)); + FileStoreSourceReader restoredReader = null; + + try { + reader.addSplits(Collections.singletonList(createTestFileSplit("id1"))); + reader.start(); + assertThat(preparationStarted.await(10, TimeUnit.SECONDS)).isTrue(); + + FileStoreSourceReader checkpointReader = reader; + Future> checkpointDuringPreparation = + checkpointExecutor.submit(() -> checkpointReader.snapshotState(1L)); + List unconsumedSplits = + checkpointDuringPreparation.get(1, TimeUnit.SECONDS); + assertThat(unconsumedSplits) + .singleElement() + .extracting(FileStoreSourceSplit::recordsToSkip) + .isEqualTo(0L); + + finishPreparation.countDown(); + TestingReaderOutput outputBeforeCheckpoint = new TestingReaderOutput<>(); + pollUntilRecords(reader, outputBeforeCheckpoint, 2); + assertThat(format(outputBeforeCheckpoint.getEmittedRecords())) + .containsExactly("-U 1", "+U 2"); + + List checkpoint = reader.snapshotState(2L); + assertThat(checkpoint) + .singleElement() + .extracting(FileStoreSourceSplit::recordsToSkip) + .isEqualTo(2L); + reader.close(); + reader = null; + + restoredReader = + createReader(new TestingReaderContext(), new BlockingTableRead(batches)); + restoredReader.addSplits(checkpoint); + restoredReader.start(); + TestingReaderOutput outputAfterRestore = new TestingReaderOutput<>(); + drain(restoredReader, outputAfterRestore); + + List allRecords = new ArrayList<>(); + allRecords.addAll(format(outputBeforeCheckpoint.getEmittedRecords())); + allRecords.addAll(format(outputAfterRestore.getEmittedRecords())); + assertThat(allRecords).containsExactly("-U 1", "+U 2", "+I 3"); + } finally { + finishPreparation.countDown(); + if (reader != null) { + reader.close(); + } + if (restoredReader != null) { + restoredReader.close(); + } + checkpointExecutor.shutdownNow(); + } + } + protected FileStoreSourceReader createReader(TestingReaderContext context) { return createReader( context, new TestChangelogDataReadWrite(tempDir.toString()).createReadWithKey()); @@ -174,6 +269,168 @@ protected static FileStoreSourceSplit createTestFileSplit(String id) { return newSourceSplit(id, row(1), 0, Collections.emptyList()); } + private static GenericRow changelogRow(RowKind kind, long value) { + GenericRow row = GenericRow.of(value); + row.setRowKind(kind); + return row; + } + + private static void pollUntilRecords( + FileStoreSourceReader reader, TestingReaderOutput output, int expected) + throws Exception { + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(10); + while (output.getEmittedRecords().size() < expected && System.nanoTime() < deadline) { + reader.pollNext(output); + Thread.sleep(10); + } + assertThat(output.getEmittedRecords()).hasSize(expected); + } + + private static void drain(FileStoreSourceReader reader, TestingReaderOutput output) + throws Exception { + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(10); + while (reader.getNumberOfCurrentlyAssignedSplits() > 0 && System.nanoTime() < deadline) { + reader.pollNext(output); + Thread.sleep(10); + } + assertThat(reader.getNumberOfCurrentlyAssignedSplits()).isZero(); + } + + private static List format(List records) { + List result = new ArrayList<>(); + for (RowData row : records) { + result.add(row.getRowKind().shortString() + " " + row.getLong(0)); + } + return result; + } + + private static class BlockingTableRead implements TableRead { + + private final List> batches; + private final CountDownLatch preparationStarted; + private final CountDownLatch finishPreparation; + + private BlockingTableRead(List> batches) { + this(batches, null, null); + } + + private BlockingTableRead( + List> batches, + CountDownLatch preparationStarted, + CountDownLatch finishPreparation) { + this.batches = batches; + this.preparationStarted = preparationStarted; + this.finishPreparation = finishPreparation; + } + + @Override + public TableRead withMetricRegistry(MetricRegistry registry) { + return this; + } + + @Override + public TableRead executeFilter() { + return this; + } + + @Override + public TableRead withIOManager(IOManager ioManager) { + return this; + } + + @Override + public RecordReader createReader(Split split) throws IOException { + if (preparationStarted != null) { + preparationStarted.countDown(); + try { + if (!finishPreparation.await(10, TimeUnit.SECONDS)) { + throw new IOException("Timed out waiting for lookup preparation."); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IOException(e); + } + } + return new BatchedRowsRecordReader(batches); + } + } + + private static class BatchedRowsRecordReader implements RecordReader { + + private final List> batches; + private int nextBatch; + + private BatchedRowsRecordReader(List> batches) { + this.batches = batches; + } + + @Override + public RecordIterator readBatch() { + if (nextBatch >= batches.size()) { + return null; + } + + List batch = batches.get(nextBatch++); + return new RecordIterator() { + private int nextRow; + + @Override + public InternalRow next() { + return nextRow < batch.size() ? batch.get(nextRow++) : null; + } + + @Override + public void releaseBatch() {} + }; + } + + @Override + public void close() {} + } + + private static class CloseTrackingTableRead implements TableRead { + + private final TableRead delegate; + private int closeCount; + + private CloseTrackingTableRead(TableRead delegate) { + this.delegate = delegate; + } + + private int closeCount() { + return closeCount; + } + + @Override + public TableRead withMetricRegistry(MetricRegistry registry) { + delegate.withMetricRegistry(registry); + return this; + } + + @Override + public TableRead executeFilter() { + delegate.executeFilter(); + return this; + } + + @Override + public TableRead withIOManager(IOManager ioManager) { + delegate.withIOManager(ioManager); + return this; + } + + @Override + public RecordReader createReader(Split split) throws IOException { + return delegate.createReader(split); + } + + @Override + public void close() throws IOException { + closeCount++; + delegate.close(); + } + } + /** A {@link MetricGroup} for testing. */ public static class DummyMetricGroup implements MetricGroup { public DummyMetricGroup() {} diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/FileStoreSourceSplitReaderTest.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/FileStoreSourceSplitReaderTest.java index 42e3c1406e04..bc389bcae431 100644 --- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/FileStoreSourceSplitReaderTest.java +++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/FileStoreSourceSplitReaderTest.java @@ -430,6 +430,17 @@ public void testCloseReleasesCurrentFirstBatch() throws Exception { assertThat(iterator.released()).isTrue(); } + @Test + public void testCloseDoesNotCloseTableRead() throws Exception { + TestingTableRead tableRead = + new TestingTableRead(new SingleBatchRecordReader(new TrackingRecordIterator())); + FileStoreSourceSplitReader reader = createReader(tableRead, null); + + reader.close(); + + assertThat(tableRead.closed()).isFalse(); + } + @Test public void testLimit() throws Exception { TestChangelogDataReadWrite rw = new TestChangelogDataReadWrite(tempDir.toString()); @@ -603,6 +614,7 @@ private RecordsWithSplitIds> fetchWithoutWaitingForPool( private static class TestingTableRead implements TableRead { private final RecordReader recordReader; + private boolean closed; private TestingTableRead(RecordReader recordReader) { this.recordReader = recordReader; @@ -627,6 +639,15 @@ public TableRead withIOManager(IOManager ioManager) { public RecordReader createReader(Split split) { return recordReader; } + + @Override + public void close() { + closed = true; + } + + private boolean closed() { + return closed; + } } private static class SingleBatchRecordReader implements RecordReader { diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/operator/OperatorSourceTest.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/operator/OperatorSourceTest.java index 4446278be1d0..e9ba0e9604bc 100644 --- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/operator/OperatorSourceTest.java +++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/operator/OperatorSourceTest.java @@ -31,6 +31,8 @@ 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.Split; import org.apache.paimon.table.source.TableRead; import org.apache.paimon.types.DataTypes; @@ -69,6 +71,7 @@ import java.nio.file.Path; import java.util.ArrayList; import java.util.Arrays; +import java.util.Collections; import java.util.List; import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.TimeUnit; @@ -380,6 +383,18 @@ public void testReadOperatorMetricsRegisterAndUpdate() throws Exception { writeToTable(2, 2, 2); List splits = table.newReadBuilder().newScan().plan().splits(); assertThat(splits.size()).isGreaterThan(0); + DataSplit dataSplit = (DataSplit) splits.get(0); + Split incrementalSplit = + new IncrementalSplit( + dataSplit.snapshotId(), + dataSplit.partition(), + dataSplit.bucket(), + dataSplit.totalBuckets(), + Collections.emptyList(), + null, + dataSplit.dataFiles(), + null, + true); MetricGroup readerOperatorMetricGroup = readOperator.getMetricGroup(); harness.open(); assertThat( @@ -401,7 +416,7 @@ public void testReadOperatorMetricsRegisterAndUpdate() throws Exception { .getValue()) .isGreaterThan(299L); - harness.processElement(new StreamRecord<>(splits.get(0))); + harness.processElement(new StreamRecord<>(incrementalSplit)); assertThat( (Long) TestingMetricUtils.getGauge( diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonV2Write.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonV2Write.scala index a6b49a79180f..a3bf5276f3ec 100644 --- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonV2Write.scala +++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonV2Write.scala @@ -18,7 +18,6 @@ package org.apache.paimon.spark.write -import org.apache.paimon.CoreOptions.ChangelogProducer import org.apache.paimon.Snapshot import org.apache.paimon.options.Options import org.apache.paimon.spark._ @@ -95,7 +94,7 @@ class PaimonV2Write( buffer += PaimonDeletedRecordsMetric() } } - if (!coreOptions.changelogProducer().equals(ChangelogProducer.NONE)) { + if (coreOptions.producesChangelogFiles()) { buffer += PaimonAppendedChangelogFilesMetric() } if (!table.partitionKeys().isEmpty) {