This is an automated email from the ASF dual-hosted git repository.
fresh-borzoni pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git
The following commit(s) were added to refs/heads/main by this push:
new af0e1bd0d [flink] Fix batch query on empty datalake-enabled table to
return 0 rows instead of failing (#3208)
af0e1bd0d is described below
commit af0e1bd0dfd6db8399e5d7a7160f52397d638b50
Author: Howie Wang <[email protected]>
AuthorDate: Sat May 9 00:04:58 2026 +0800
[flink] Fix batch query on empty datalake-enabled table to return 0 rows
instead of failing (#3208)
* [flink] Fix batch query on empty datalake-enabled table to fall back to
fluss splits
* [flink] Add info log when falling back to Fluss-only splits
---
.../source/enumerator/FlinkSourceEnumerator.java | 21 +++++++++++++++++++--
1 file changed, 19 insertions(+), 2 deletions(-)
diff --git
a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumerator.java
b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumerator.java
index 3ad234aa0..1a6521f57 100644
---
a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumerator.java
+++
b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumerator.java
@@ -361,9 +361,26 @@ public class FlinkSourceEnumerator
context.callAsync(
() -> {
List<SourceSplitBase> splits =
generateHybridLakeFlussSplits();
+ // No lake snapshot exists, fall back to Fluss-only
splits
if (splits == null) {
- throw new UnsupportedOperationException(
- "Currently, Batch mode can only be
supported if one lake snapshot exists for the table.");
+ LOG.info(
+ "No lake snapshot found for table {},"
+ + " falling back to Fluss-only
splits.",
+ tablePath);
+ if (isPartitioned) {
+ Set<PartitionInfo> partitionInfos =
listPartitions();
+ Collection<Partition> partitions =
+ partitionInfos.stream()
+ .map(
+ p ->
+ new Partition(
+
p.getPartitionId(),
+
p.getPartitionName()))
+ .collect(Collectors.toList());
+ splits =
this.initPartitionedSplits(partitions);
+ } else {
+ splits = this.initNonPartitionedSplits();
+ }
}
return splits;
},
