[flink] support computing the changelog generated by compact during r…

…ead time.

This is used when changelog producer is none, but CoreOptions#needLookup is true and the table is used as a dim table.

paimon-core/src/main/java/org/apache/paimon/table/source/DataTableStreamScan.java

@@ -219,7 +219,7 @@ private boolean shouldDelaySnapshot(long snapshotId) {
         return false;
     }
 
-    private FollowUpScanner createFollowUpScanner() {
+    protected FollowUpScanner createFollowUpScanner() {
         CoreOptions.StreamScanMode type =
                 options.toConfiguration().get(CoreOptions.STREAM_SCAN_MODE);
         switch (type) {
@@ -249,7 +249,7 @@ private FollowUpScanner createFollowUpScanner() {
         return followUpScanner;
     }
 
-    private BoundedChecker createBoundedChecker() {
+    protected BoundedChecker createBoundedChecker() {
         Long boundedWatermark = options.scanBoundedWatermark();
         return boundedWatermark != null
                 ? BoundedChecker.watermark(boundedWatermark)

paimon-core/src/test/java/org/apache/paimon/table/TableTestBase.java

@@ -127,10 +127,21 @@ protected void write(Table table, IOManager ioManager, InternalRow... rows) thro
     }
 
     protected void compact(Table table, BinaryRow partition, int bucket) throws Exception {
+        compact(table, partition, bucket, null, true);
+    }
+
+    protected void compact(
+            Table table,
+            BinaryRow partition,
+            int bucket,
+            IOManager ioManager,
+            boolean fullCompaction)
+            throws Exception {
         BatchWriteBuilder writeBuilder = table.newBatchWriteBuilder();
         try (BatchTableWrite write = writeBuilder.newWrite();
                 BatchTableCommit commit = writeBuilder.newCommit()) {
-            write.compact(partition, bucket, true);
+            write.withIOManager(ioManager);
+            write.compact(partition, bucket, fullCompaction);
             commit.commit(write.prepareCommit());
         }
     }

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/CompactionFollowUpScanner.java

@@ -0,0 +1,51 @@
+/*
+ * 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.flink.lookup;
+
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.table.source.ScanMode;
+import org.apache.paimon.table.source.snapshot.FollowUpScanner;
+import org.apache.paimon.table.source.snapshot.SnapshotReader;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/** {@link FollowUpScanner} for read all changed files after compact. */
+public class CompactionFollowUpScanner implements FollowUpScanner {
+
+    private static final Logger LOG = LoggerFactory.getLogger(CompactionFollowUpScanner.class);
+
+    @Override
+    public boolean shouldScanSnapshot(Snapshot snapshot) {
+        if (snapshot.commitKind() == Snapshot.CommitKind.COMPACT) {
+            return true;
+        }
+
+        LOG.debug(
+                "Next snapshot id {} is not COMPACT, but is {}, check next one.",
+                snapshot.id(),
+                snapshot.commitKind());
+        return false;
+    }
+
+    @Override
+    public SnapshotReader.Plan scan(Snapshot snapshot, SnapshotReader snapshotReader) {
+        return snapshotReader.withMode(ScanMode.DELTA).withSnapshot(snapshot).readChanges();
+    }
+}

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/FileStoreLookupFunction.java

@@ -98,7 +98,9 @@ public class FileStoreLookupFunction implements Serializable, Closeable {
 
     public FileStoreLookupFunction(
             Table table, int[] projection, int[] joinKeyIndex, @Nullable Predicate predicate) {
-        TableScanUtils.streamingReadingValidate(table);
+        if (!TableScanUtils.supportCompactDiffStreamingReading(table)) {
+            TableScanUtils.streamingReadingValidate(table);
+        }
 
         this.table = table;
         this.partitionLoader = DynamicPartitionLoader.of(table);

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/FullCacheLookupTable.java

@@ -350,7 +350,7 @@ public Context(
                 File tempPath,
                 List<String> joinKey,
                 @Nullable Set<Integer> requiredCachedBucketIds) {
-            this.table = table;
+            this.table = new LookupFileStoreTable(table, joinKey);
             this.projection = projection;
             this.tablePredicate = tablePredicate;
             this.projectedPredicate = projectedPredicate;

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/IncrementalCompactDiffSplitRead.java

@@ -0,0 +1,77 @@
+/*
+ * 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.flink.lookup;
+
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.operation.MergeFileSplitRead;
+import org.apache.paimon.operation.SplitRead;
+import org.apache.paimon.reader.EmptyRecordReader;
+import org.apache.paimon.reader.RecordReader;
+import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.table.source.splitread.IncrementalDiffSplitRead;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.stream.Collectors;
+
+/** A {@link SplitRead} for streaming incremental diff after compaction. */
+public class IncrementalCompactDiffSplitRead extends IncrementalDiffSplitRead {
+
+    public IncrementalCompactDiffSplitRead(MergeFileSplitRead mergeRead) {
+        super(mergeRead);
+    }
+
+    @Override
+    public RecordReader<InternalRow> createReader(DataSplit split) throws IOException {
+        if (split.beforeFiles().stream().noneMatch(file -> file.level() == 0)) {
+            return new EmptyRecordReader<>();
+        }
+        return super.createReader(filterLevel0Files(split));
+    }
+
+    private DataSplit filterLevel0Files(DataSplit split) {
+        List<DataFileMeta> beforeFiles =
+                split.beforeFiles().stream()
+                        .filter(file -> file.level() > 0)
+                        .collect(Collectors.toList());
+        List<DataFileMeta> afterFiles =
+                split.dataFiles().stream()
+                        .filter(file -> file.level() > 0)
+                        .collect(Collectors.toList());
+        DataSplit.Builder builder =
+                new DataSplit.Builder()
+                        .withSnapshot(split.snapshotId())
+                        .withPartition(split.partition())
+                        .withBucket(split.bucket())
+                        .withBucketPath(split.bucketPath())
+                        .withBeforeFiles(beforeFiles)
+                        .withDataFiles(afterFiles)
+                        .isStreaming(split.isStreaming())
+                        .rawConvertible(split.rawConvertible());
+
+        if (split.beforeDeletionFiles().isPresent()) {
+            builder.withBeforeDeletionFiles(split.beforeDeletionFiles().get());
+        }
+        if (split.deletionFiles().isPresent()) {
+            builder.withDataDeletionFiles(split.deletionFiles().get());
+        }
+        return builder.build();
+    }
+}

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/LookupDataTableScan.java

@@ -0,0 +1,87 @@
+/*
+ * 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.flink.lookup;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.operation.DefaultValueAssigner;
+import org.apache.paimon.table.source.DataTableStreamScan;
+import org.apache.paimon.table.source.snapshot.AllDeltaFollowUpScanner;
+import org.apache.paimon.table.source.snapshot.BoundedChecker;
+import org.apache.paimon.table.source.snapshot.FollowUpScanner;
+import org.apache.paimon.table.source.snapshot.FullStartingScanner;
+import org.apache.paimon.table.source.snapshot.SnapshotReader;
+import org.apache.paimon.table.source.snapshot.StartingScanner;
+import org.apache.paimon.utils.SnapshotManager;
+
+import static org.apache.paimon.CoreOptions.StartupMode;
+import static org.apache.paimon.flink.lookup.LookupFileStoreTable.LookupStreamScanMode;
+
+/**
+ * {@link org.apache.paimon.table.source.StreamTableScan} implementation for lookup streaming
+ * planning.
+ */
+public class LookupDataTableScan extends DataTableStreamScan {
+
+    private StartupMode startupMode;
+    private LookupStreamScanMode scanMode;
+
+    public LookupDataTableScan(
+            CoreOptions options,
+            SnapshotReader snapshotReader,
+            SnapshotManager snapshotManager,
+            boolean supportStreamingReadOverwrite,
+            DefaultValueAssigner defaultValueAssigner,
+            LookupStreamScanMode scanMode) {
+        super(
+                options,
+                snapshotReader,
+                snapshotManager,
+                supportStreamingReadOverwrite,
+                defaultValueAssigner);
+        this.startupMode = options.startupMode();
+        this.scanMode = scanMode;
+    }
+
+    @Override
+    protected StartingScanner createStartingScanner(boolean isStreaming) {
+        return startupMode != CoreOptions.StartupMode.COMPACTED_FULL
+                ? new FullStartingScanner(snapshotReader.snapshotManager())
+                : super.createStartingScanner(isStreaming);
+    }
+
+    @Override
+    protected FollowUpScanner createFollowUpScanner() {
+        switch (scanMode) {
+            case NORMAL:
+                return super.createFollowUpScanner();
+            case FILE_MONITOR:
+                return new AllDeltaFollowUpScanner();
+            case COMPACT_DELTA_MONITOR:
+                return new CompactionFollowUpScanner();
+            default:
+                throw new UnsupportedOperationException(
+                        "Unknown lookup stream scan mode: " + scanMode.name());
+        }
+    }
+
+    @Override
+    protected BoundedChecker createBoundedChecker() {
+        return BoundedChecker.neverEnd(); // dim table should never end
+    }
+}