[flink] Disable count push down for lake enabled table and catch exception when create lake source (#1679)

Author: caozhen1937

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlinkTableSource.java

@@ -185,7 +185,10 @@ public FlinkTableSource(
         this.mergeEngineType = mergeEngineType;
         this.tableOptions = tableOptions;
         if (isDataLakeEnabled) {
-            this.lakeSource = createLakeSource(tablePath, tableOptions);
+            this.lakeSource =
+                    checkNotNull(
+                            createLakeSource(tablePath, tableOptions),
+                            "LakeSource must not be null if enable datalake");
         }
     }
 
@@ -567,7 +570,11 @@ public boolean applyAggregates(
                 || aggregateExpressions.size() != 1
                 || hasPrimaryKey()
                 || groupingSets.size() > 1
-                || (groupingSets.size() == 1 && groupingSets.get(0).length > 0)) {
+                || (groupingSets.size() == 1 && groupingSets.get(0).length > 0)
+                // The count pushdown feature is not supported when the data lake is enabled.
+                // Otherwise, it'll cause miss count data in lake. But In the future, we can push
+                // down count into lake.
+                || isDataLakeEnabled) {
             return false;
         }
 

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/LakeSourceUtils.java

@@ -60,19 +60,20 @@ public static LakeSource<LakeSplit> createLakeSource(
         try {
             lakeStoragePlugin = LakeStoragePluginSetUp.fromDataLakeFormat(dataLake, null);
         } catch (UnsupportedOperationException e) {
-            LOG.info(
-                    "No LakeStoragePlugin can be found for datalake format: {}, return null to disable reading from lake source.",
-                    dataLake);
-            return null;
+            throw new UnsupportedOperationException(
+                    String.format(
+                            "No LakeStoragePlugin available for data lake format: %s. "
+                                    + "To resolve this, ensure fluss-lake-%s.jar is in the classpath.",
+                            dataLake, dataLake.toLowerCase()));
         }
         LakeStorage lakeStorage = checkNotNull(lakeStoragePlugin).createLakeStorage(lakeConfig);
         try {
             return (LakeSource<LakeSplit>) lakeStorage.createLakeSource(tablePath);
         } catch (UnsupportedOperationException e) {
-            LOG.info(
-                    "method createLakeSource throw UnsupportedOperationException for datalake format {}, return null as lakeSource to disable reading from lake source.",
-                    dataLake);
-            return null;
+            throw new UnsupportedOperationException(
+                    String.format(
+                            "Table using '%s' data lake format cannot be used as historical data in Fluss.",
+                            dataLake));
         }
     }
 }