return Optional.of(result) when non DataSplit

Author: xiaohongbo

paimon-core/src/main/java/org/apache/paimon/table/source/DataTableBatchScan.java

@@ -138,7 +138,7 @@ private Optional<StartingScanner.Result> applyPushDownLimit() {
         SnapshotReader.Plan plan = ((ScannedResult) result).plan();
         List<Split> planSplits = plan.splits();
         if (planSplits.stream().anyMatch(s -> !(s instanceof DataSplit))) {
-            return Optional.empty();
+            return Optional.of(result);
         }
         @SuppressWarnings("unchecked")
         List<DataSplit> splits = (List<DataSplit>) (List<?>) planSplits;
@@ -202,7 +202,7 @@ private Optional<StartingScanner.Result> applyPushDownTopN() {
         List<Split> planSplits = plan.splits();
         // TopN pushdown only supports DataSplit. Skip for IncrementalSplit.
         if (planSplits.stream().anyMatch(s -> !(s instanceof DataSplit))) {
-            return Optional.empty();
+            return Optional.of(result);
         }
         @SuppressWarnings("unchecked")
         List<DataSplit> splits = (List<DataSplit>) (List<?>) planSplits;

paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/TableValuedFunctionsTest.scala

@@ -361,7 +361,7 @@ class TableValuedFunctionsTest extends PaimonHiveTestBase {
 
       checkAnswer(
         spark.sql(
-          "SELECT * FROM paimon_incremental_query('t', 'tag1', 'tag2') LIMIT 5 ORDER BY a, b"),
+          "SELECT * FROM paimon_incremental_query('t', 'tag1', 'tag2') ORDER BY a, b LIMIT 5"),
         Seq(Row(1, 3, "3"), Row(2, 4, "4")))
     }
   }