[flink] refactor the code of the lookup and support computing the cha…

…ngelog generated by compact during read time.

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/CompactionDiffFollowUpScanner.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 CompactionDiffFollowUpScanner implements FollowUpScanner {
+
+    private static final Logger LOG = LoggerFactory.getLogger(CompactionDiffFollowUpScanner.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/LookupCompactDiffRead.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.KeyValue;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.disk.IOManager;
+import org.apache.paimon.operation.MergeFileSplitRead;
+import org.apache.paimon.operation.SplitRead;
+import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.reader.RecordReader;
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.table.source.AbstractDataTableRead;
+import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.table.source.InnerTableRead;
+import org.apache.paimon.table.source.Split;
+import org.apache.paimon.table.source.TableRead;
+
+import java.io.IOException;
+
+import static org.apache.paimon.table.source.KeyValueTableRead.unwrap;
+
+/** An {@link InnerTableRead} that reads the data changed before and after compaction. */
+public class LookupCompactDiffRead extends AbstractDataTableRead<KeyValue> {
+    private final SplitRead<InternalRow> fullPhaseMergeRead;
+    private final SplitRead<InternalRow> incrementalDiffRead;
+
+    public LookupCompactDiffRead(MergeFileSplitRead mergeRead, TableSchema schema) {
+        super(schema);
+        this.incrementalDiffRead = new IncrementalCompactDiffSplitRead(mergeRead);
+        this.fullPhaseMergeRead =
+                SplitRead.convert(mergeRead, split -> unwrap(mergeRead.createReader(split)));
+    }
+
+    @Override
+    public void projection(int[][] projection) {
+        fullPhaseMergeRead.withProjection(projection);
+        incrementalDiffRead.withProjection(projection);
+    }
+
+    @Override
+    public RecordReader<InternalRow> reader(Split split) throws IOException {
+        DataSplit dataSplit = (DataSplit) split;
+        if (dataSplit.beforeFiles().isEmpty()) {
+            return fullPhaseMergeRead.createReader(dataSplit); // full reading phase
+        } else {
+            return incrementalDiffRead.createReader((DataSplit) split);
+        }
+    }
+
+    @Override
+    protected InnerTableRead innerWithFilter(Predicate predicate) {
+        fullPhaseMergeRead.withFilter(predicate);
+        incrementalDiffRead.withFilter(predicate);
+        return this;
+    }
+
+    @Override
+    public InnerTableRead forceKeepDelete() {
+        fullPhaseMergeRead.forceKeepDelete();
+        incrementalDiffRead.forceKeepDelete();
+        return this;
+    }
+
+    @Override
+    public TableRead withIOManager(IOManager ioManager) {
+        fullPhaseMergeRead.withIOManager(ioManager);
+        incrementalDiffRead.withIOManager(ioManager);
+        return this;
+    }
+}

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 lookupScanMode;
+
+    public LookupDataTableScan(
+            CoreOptions options,
+            SnapshotReader snapshotReader,
+            SnapshotManager snapshotManager,
+            boolean supportStreamingReadOverwrite,
+            DefaultValueAssigner defaultValueAssigner,
+            LookupStreamScanMode lookupScanMode) {
+        super(
+                options,
+                snapshotReader,
+                snapshotManager,
+                supportStreamingReadOverwrite,
+                defaultValueAssigner);
+        this.startupMode = options.startupMode();
+        this.lookupScanMode = lookupScanMode;
+    }
+
+    @Override
+    protected StartingScanner createStartingScanner(boolean isStreaming) {
+        return startupMode != CoreOptions.StartupMode.COMPACTED_FULL
+                ? new FullStartingScanner(snapshotReader.snapshotManager())
+                : super.createStartingScanner(isStreaming);
+    }
+
+    @Override
+    protected FollowUpScanner createFollowUpScanner() {
+        switch (lookupScanMode) {
+            case CHANGELOG:
+                return super.createFollowUpScanner();
+            case FILE_MONITOR:
+                return new AllDeltaFollowUpScanner();
+            case COMPACT_DELTA_MONITOR:
+                return new CompactionDiffFollowUpScanner();
+            default:
+                throw new UnsupportedOperationException(
+                        "Unknown lookup stream scan mode: " + lookupScanMode.name());
+        }
+    }
+
+    @Override
+    protected BoundedChecker createBoundedChecker() {
+        return BoundedChecker.neverEnd(); // dim table should never end
+    }
+}