From 435a9e4b5599e31d0bb3decb03680ae260ed4486 Mon Sep 17 00:00:00 2001
From: Anton Borisov <anton.borisov@fresha.com>
Date: Mon, 30 Mar 2026 02:02:42 +0100
Subject: [PATCH] [flink] Skip LogSplit for empty buckets in lake union read

---
 .../apache/fluss/flink/lake/LakeSplitGenerator.java | 13 +++++++++----
 1 file changed, 9 insertions(+), 4 deletions(-)

diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/lake/LakeSplitGenerator.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/lake/LakeSplitGenerator.java
index 9b353ccd35..f26ebddfdd 100644
--- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/lake/LakeSplitGenerator.java
+++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/lake/LakeSplitGenerator.java
@@ -235,10 +235,15 @@ private List<SourceSplitBase> generateSplit(
                 Long snapshotLogOffset = tableBucketSnapshotLogOffset.get(tableBucket);
                 Long stoppingOffset = bucketEndOffset.get(bucket);
                 if (snapshotLogOffset == null) {
-                    // no any data commit to this bucket, scan from fluss log
-                    splits.add(
-                            new LogSplit(
-                                    tableBucket, partitionName, EARLIEST_OFFSET, stoppingOffset));
+                    // no data committed to lake for this bucket, scan from fluss log
+                    if (stoppingOffset == NO_STOPPING_OFFSET || stoppingOffset > 0) {
+                        splits.add(
+                                new LogSplit(
+                                        tableBucket,
+                                        partitionName,
+                                        EARLIEST_OFFSET,
+                                        stoppingOffset));
+                    }
                 } else {
                     // need to read remain fluss log
                     if (stoppingOffset == NO_STOPPING_OFFSET