feat: skip reading empty files when load/query location. (#17522)

feat: skip reading empty files.

Author: youngsofun

src/query/storages/parquet/src/parquet_rs/copy_into_table/table.rs

@@ -60,6 +60,7 @@ impl ParquetTableForCopy {
         );
         let file_infos = files
             .iter()
+            .filter(|f| f.size > 0)
             .map(|f| (f.path.clone(), f.size))
             .collect::<Vec<_>>();
         let total_size = file_infos.iter().map(|(_, size)| *size as usize).sum();

src/query/storages/parquet/src/parquet_rs/parquet_table/partition.rs

@@ -56,6 +56,7 @@ impl ParquetRSTable {
             match &self.files_to_read {
                 Some(files) => files
                     .iter()
+                    .filter(|f| f.size > 0)
                     .map(|f| (f.path.clone(), f.size))
                     .collect::<Vec<_>>(),
                 None => self
@@ -75,7 +76,7 @@ impl ParquetRSTable {
         };
 
         // If a file size is less than `parquet_fast_read_bytes`,
-        // we treat it as a small file and it will be totally loaded into memory.
+        // we treat it as a small file, and it will be totally loaded into memory.
         let fast_read_bytes = ctx.get_settings().get_parquet_fast_read_bytes()?;
         let mut large_files = vec![];
         let mut large_file_indices = vec![];
@@ -85,7 +86,7 @@ impl ParquetRSTable {
             if size > fast_read_bytes {
                 large_files.push((location, size));
                 large_file_indices.push(index);
-            } else {
+            } else if size > 0 {
                 small_files.push((location, size));
                 small_file_indices.push(index);
             }

src/query/storages/parquet/src/parquet_rs/parquet_table/table.rs

@@ -291,6 +291,7 @@ impl Table for ParquetRSTable {
         let file_locations = match &self.files_to_read {
             Some(files) => files
                 .iter()
+                .filter(|f| f.size > 0)
                 .map(|f| (f.path.clone(), f.size))
                 .collect::<Vec<_>>(),
             None => self

src/query/storages/stage/src/stage_table.rs

@@ -109,6 +109,7 @@ impl StageTable {
 
         let partitions = files
             .into_iter()
+            .filter(|f| f.size > 0)
             .map(|v| {
                 let part = SingleFilePartition {
                     path: v.path.clone(),

tests/data/empty.txt

@@ -0,0 +1,47 @@
+query ok
+select $1 from @data (files=>( 'empty.txt'), file_format=>'csv')
+----
+
+query ok
+select $1 from @data (files=>('csv/it.csv', 'empty.txt'), file_format=>'csv')
+----
+1
+2
+
+query ok
+select $1 from @data (files=>('parquet/ii/f2.parquet', 'empty.txt'), file_format=>'parquet');
+----
+3
+4
+
+statement ok
+create or replace table t (c1 int, id string);
+
+query ok
+copy into t(c1) from  @data files=('parquet/ii/f2.parquet', 'empty.txt') file_format=(type='parquet')
+----
+parquet/ii/f2.parquet 2 0 NULL NULL
+
+query ok
+select * from t
+----
+3 NULL
+4 NULL
+
+query ok
+copy into t from  @data files=('csv/it.csv', 'empty.txt') file_format=(type='csv')
+----
+csv/it.csv 2 0 NULL NULL
+
+query ok
+select * from t order by c1
+----
+1 b
+2 d
+3 NULL
+4 NULL
+
+query ok
+select name from list_stage(location=>'@data', pattern=>'emp.*')
+----
+empty.txt