fix(table): keep every split of a format table under a positive LIMIT

crates/integrations/datafusion/tests/format_table_statistics.rs

@@ -197,3 +197,17 @@ async fn test_format_table_with_empty_file_counts_zero() {
         0
     );
 }
+
+/// A positive `LIMIT` must not be answered from a file count. A format table has no
+/// row counts, so keeping only the first `limit` files is a guess, and an empty data
+/// file — ordinary output from an engine that writes one file per task — makes the
+/// guess wrong.
+#[tokio::test]
+async fn test_format_table_limit_is_not_answered_from_a_file_count() {
+    let (_tmp, context, table_dir) = setup_format_table().await;
+    write_parquet(&table_dir.join("part-0.parquet"), &[]);
+    write_parquet(&table_dir.join("part-1.parquet"), &[1, 2, 3]);
+
+    let scanned = scanned_rows(&context, "SELECT id FROM paimon.test_db.events LIMIT 1").await;
+    assert_eq!(scanned, 1, "LIMIT 1 over 3 rows must return a row");
+}

crates/paimon/src/table/format_table_scan.rs

@@ -272,13 +272,17 @@ impl<'a> FormatTableScan<'a> {
         }
     }
 
+    /// A limit of zero needs no split at all. A positive one keeps every split: a format
+    /// table has no row counts, so the number of files says nothing about the number of
+    /// rows, and dropping files would answer the query from a guess.
+    ///
+    /// Mirrors Java `FormatTableScan.FormatTableScanPlan.splits`.
     pub(crate) fn apply_limit_pushdown(
         &self,
         splits: Vec<crate::DataSplit>,
     ) -> Vec<crate::DataSplit> {
         match self.limit {
             Some(0) => Vec::new(),
-            Some(limit) if splits.len() > limit => splits.into_iter().take(limit).collect(),
             _ => splits,
         }
     }