[core] Support changelog-on-read for primary-key tables

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

docs/generated/core_configuration.html

@@ -212,29 +212,35 @@
             <td>String</td>
             <td>Changelog file metadata stats collection. none, counts, truncate(16), full is available.</td>
         </tr>
+        <tr>
+            <td><h5>changelog-on-read.lookup-state.max-size</h5></td>
+            <td style="word-wrap: break-word;">2048</td>
+            <td>Integer</td>
+            <td>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.</td>
+        </tr>
         <tr>
             <td><h5>changelog-producer</h5></td>
             <td style="word-wrap: break-word;">none</td>
             <td><p>Enum</p></td>
-            <td>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. <br /><br />Possible values:<ul><li>"none": No changelog file.</li><li>"input": Double write to a changelog file when flushing memory table, the changelog is from input.</li><li>"full-compaction": Generate changelog files with each full compaction.</li><li>"lookup": Generate changelog files through 'lookup' compaction.</li></ul></td>
+            <td>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.<br /><br />Possible values:<ul><li>"none": No changelog file.</li><li>"input": Double write to a changelog file when flushing memory table, the changelog is from input.</li><li>"full-compaction": Generate changelog files with each full compaction.</li><li>"lookup": Generate changelog files through 'lookup' compaction.</li><li>"on-read": Generate changelog during reads without writing changelog files.</li></ul></td>
         </tr>
         <tr>
             <td><h5>changelog-producer.ignore-delete</h5></td>
             <td style="word-wrap: break-word;">false</td>
             <td>Boolean</td>
-            <td>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.</td>
+            <td>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.</td>
         </tr>
         <tr>
             <td><h5>changelog-producer.ignore-update-before</h5></td>
             <td style="word-wrap: break-word;">false</td>
             <td>Boolean</td>
-            <td>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.</td>
+            <td>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.</td>
         </tr>
         <tr>
             <td><h5>changelog-producer.row-deduplicate</h5></td>
             <td style="word-wrap: break-word;">false</td>
             <td>Boolean</td>
-            <td>Whether to generate -U, +U changelog for the same record. This configuration is only valid for the changelog-producer is lookup or full-compaction.</td>
+            <td>Whether to suppress -U, +U changelog for equal records. Valid for lookup, full-compaction and on-read.</td>
         </tr>
         <tr>
             <td><h5>changelog-producer.row-deduplicate-ignore-fields</h5></td>

paimon-api/src/main/java/org/apache/paimon/CoreOptions.java

@@ -1161,34 +1161,45 @@ 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<Integer> 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<Boolean> CHANGELOG_PRODUCER_IGNORE_UPDATE_BEFORE =
             key("changelog-producer.ignore-update-before")
                     .booleanType()
                     .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<Boolean> CHANGELOG_PRODUCER_IGNORE_DELETE =
             key("changelog-producer.ignore-delete")
                     .booleanType()
                     .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<Boolean> 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<String> 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;

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());

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<T> lookupAll(InternalRow key) throws IOException {
+        List<T> 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<DataFileMeta> 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<DataFileMeta> 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);
         }
     }
 

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<T> 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}. */

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<KeyValue> reader, @Nullable Throwable collected) {
+            AutoCloseable reader, @Nullable Throwable collected) {
         try {
             reader.close();
         } catch (Throwable t) {