[flink][lookup] Check max pending snapshot count for lookup join

FangYongs · FangYongs · commit 1e7acadf88aa · 2024-05-22T11:20:01.000+08:00
diff --git a/docs/layouts/shortcodes/generated/flink_connector_configuration.html b/docs/layouts/shortcodes/generated/flink_connector_configuration.html
@@ -68,6 +68,12 @@
             <td>Duration</td>
             <td>Specific dynamic partition refresh interval for lookup, scan all partitions and obtain corresponding partition.</td>
         </tr>
+        <tr>
+            <td><h5>lookup.max-pending-snapshot.count</h5></td>
+            <td style="word-wrap: break-word;">5</td>
+            <td>Integer</td>
+            <td>The max pending snapshot count between the lookup snapshot and the latest snapshot.</td>
+        </tr>
         <tr>
             <td><h5>lookup.refresh.async</h5></td>
             <td style="word-wrap: break-word;">false</td>
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkConnectorOptions.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkConnectorOptions.java
@@ -277,6 +277,13 @@ public class FlinkConnectorOptions {
                     .defaultValue(false)
                     .withDescription("Whether to enable async lookup join.");
 
+    public static final ConfigOption<Integer> LOOKUP_MAX_PENDING_SNAPSHOT_COUNT =
+            ConfigOptions.key("lookup.max-pending-snapshot.count")
+                    .intType()
+                    .defaultValue(5)
+                    .withDescription(
+                            "The max pending snapshot count between the lookup snapshot and the latest snapshot.");
+
     public static final ConfigOption<Integer> LOOKUP_BOOTSTRAP_PARALLELISM =
             ConfigOptions.key("lookup.bootstrap-parallelism")
                     .intType()
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/FullCacheLookupTable.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/FullCacheLookupTable.java
@@ -33,6 +33,7 @@
 import org.apache.paimon.sort.BinaryExternalSortBuffer;
 import org.apache.paimon.table.FileStoreTable;
 import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.ExecutorThreadFactory;
 import org.apache.paimon.utils.FieldsComparator;
 import org.apache.paimon.utils.FileIOUtils;
 import org.apache.paimon.utils.MutableObjectIterator;
@@ -57,6 +58,7 @@
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 
+import static org.apache.paimon.flink.FlinkConnectorOptions.LOOKUP_MAX_PENDING_SNAPSHOT_COUNT;
 import static org.apache.paimon.flink.FlinkConnectorOptions.LOOKUP_REFRESH_ASYNC;
 
 /** Lookup table of full cache. */
@@ -74,12 +76,14 @@ public abstract class FullCacheLookupTable implements LookupTable {
     protected RocksDBStateFactory stateFactory;
     private final ExecutorService refreshExecutor;
     private final AtomicReference<Exception> cachedException;
+    private final int maxPendingSnapshotCount;
+    private final FileStoreTable table;
     private LookupStreamingReader reader;
     private Predicate specificPartition;
 
     public FullCacheLookupTable(Context context) {
         this.context = context;
-        FileStoreTable table = context.table;
+        this.table = context.table;
         List<String> sequenceFields = new ArrayList<>();
         if (table.primaryKeys().size() > 0) {
             sequenceFields = new CoreOptions(table.options()).sequenceField();
@@ -106,10 +110,20 @@ public FullCacheLookupTable(Context context) {
             this.userDefinedSeqComparator = null;
             this.appendUdsFieldNumber = 0;
         }
+
+        Options options = Options.fromMap(context.table.options());
         this.projectedType = projectedType;
-        this.refreshAsync = Options.fromMap(context.table.options()).get(LOOKUP_REFRESH_ASYNC);
-        this.refreshExecutor = this.refreshAsync ? Executors.newSingleThreadExecutor() : null;
+        this.refreshAsync = options.get(LOOKUP_REFRESH_ASYNC);
+        this.refreshExecutor =
+                this.refreshAsync
+                        ? Executors.newSingleThreadExecutor(
+                                new ExecutorThreadFactory(
+                                        String.format(
+                                                "%s-lookup-refresh",
+                                                Thread.currentThread().getName())))
+                        : null;
         this.cachedException = new AtomicReference<>();
+        this.maxPendingSnapshotCount = options.get(LOOKUP_MAX_PENDING_SNAPSHOT_COUNT);
     }
 
     @Override
@@ -184,6 +198,18 @@ public void refresh() throws Exception {
     }
 
     private void doRefresh() throws Exception {
+        Long latestSnapshotId = table.snapshotManager().latestSnapshotId();
+        Long nextSnapshotId = reader.nextSnapshotId();
+        if (latestSnapshotId != null
+                && nextSnapshotId != null
+                && latestSnapshotId - nextSnapshotId > maxPendingSnapshotCount) {
+            throw new Exception(
+                    String.format(
+                            "The latest snapshot id %s is much greater than the next snapshot id %s for %s, "
+                                    + "you may need to increase the parallelism of lookup operator.",
+                            latestSnapshotId, nextSnapshotId, maxPendingSnapshotCount));
+        }
+
         while (true) {
             try (RecordReaderIterator<InternalRow> batch =
                     new RecordReaderIterator<>(reader.nextBatch(false))) {
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/LookupStreamingReader.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/LookupStreamingReader.java
@@ -144,4 +144,9 @@ public RecordReader<InternalRow> nextBatch(boolean useParallelism) throws Except
         }
         return reader;
     }
+
+    @Nullable
+    public Long nextSnapshotId() {
+        return scan.checkpoint();
+    }
 }
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lookup/LookupTableTest.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lookup/LookupTableTest.java
@@ -656,6 +656,71 @@ public void testPKLookupTableRefreshAsync(boolean refreshAsync) throws Exception
         table.close();
     }
 
+    @Test
+    public void testRefreshMaxPendingSnapshot() throws Exception {
+        FileStoreTable storeTable = createTable(singletonList("f0"), new Options());
+        FullCacheLookupTable.Context context =
+                new FullCacheLookupTable.Context(
+                        storeTable,
+                        new int[] {0, 1, 2},
+                        null,
+                        null,
+                        tempDir.toFile(),
+                        singletonList("f0"));
+        table = FullCacheLookupTable.create(context, ThreadLocalRandom.current().nextInt(2) * 10);
+        table.open();
+
+        // test bulk load 100_000 records
+        List<Pair<byte[], byte[]>> records = new ArrayList<>();
+        for (int i = 1; i <= 100_000; i++) {
+            InternalRow row = row(i, 11 * i, 111 * i);
+            records.add(Pair.of(table.toKeyBytes(row), table.toValueBytes(row)));
+        }
+        records.sort((o1, o2) -> SortUtil.compareBinary(o1.getKey(), o2.getKey()));
+        TableBulkLoader bulkLoader = table.createBulkLoader();
+        for (Pair<byte[], byte[]> kv : records) {
+            bulkLoader.write(kv.getKey(), kv.getValue());
+        }
+        bulkLoader.finish();
+
+        for (int i = 1; i <= 100_000; i++) {
+            List<InternalRow> result = table.get(row(i));
+            assertThat(result).hasSize(1);
+            assertRow(result.get(0), i, 11 * i, 111 * i);
+        }
+
+        // Add 10 snapshots in the table
+        BatchWriteBuilder writeBuilder = storeTable.newBatchWriteBuilder();
+        try (BatchTableWrite write = writeBuilder.newWrite()) {
+            for (int i = 1; i <= 100; i++) {
+                write.write(row(i, 11 * i, 111 * i), 0);
+            }
+            try (BatchTableCommit commit = writeBuilder.newCommit()) {
+                commit.commit(write.prepareCommit());
+            }
+        }
+
+        // test refresh to update
+        table.refresh();
+
+        for (int k = 0; k < 10; k++) {
+            try (BatchTableWrite write = writeBuilder.newWrite()) {
+                for (int i = 1; i <= 100; i++) {
+                    write.write(row(i, 11 * i, 111 * i), 0);
+                }
+                try (BatchTableCommit commit = writeBuilder.newCommit()) {
+                    commit.commit(write.prepareCommit());
+                }
+            }
+        }
+
+        assertThatThrownBy(() -> table.refresh())
+                .hasMessageContaining(
+                        "you may need to increase the parallelism of lookup operator");
+
+        table.close();
+    }
+
     private FileStoreTable createDimTable() throws Exception {
         FileIO fileIO = LocalFileIO.create();
         org.apache.paimon.fs.Path tablePath =

Original file line number	Diff line number	Diff line change
`@@ -144,4 +144,9 @@ public RecordReader<InternalRow> nextBatch(boolean useParallelism) throws Except`
`144`	`144`	`}`
`145`	`145`	`return reader;`
`146`	`146`	`}`
	`147`	`+`
	`148`	`+ @Nullable`
	`149`	`+ public Long nextSnapshotId() {`
	`150`	`+ return scan.checkpoint();`
	`151`	`+ }`
`147`	`152`	`}`