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 3ad234aa01..1a6521f57e 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 @@ private void startInBatchMode() {
             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;
                     },