Skip producing empty parquet files (#1230)

## Which issue does this PR close?


- Closes #1224.

## What changes are included in this PR?

Skip generating empty parquet files.

## Are these changes tested?

UT.

Author: liurenjie1024

crates/iceberg/src/io/file_io.rs

@@ -331,10 +331,17 @@ impl OutputFile {
     }
 
     /// Checks if file exists.
-    pub async fn exists(&self) -> crate::Result<bool> {
+    pub async fn exists(&self) -> Result<bool> {
         Ok(self.op.exists(&self.path[self.relative_path_pos..]).await?)
     }
 
+    /// Deletes file.
+    ///
+    /// If the file does not exist, it will not return error.
+    pub async fn delete(&self) -> Result<()> {
+        Ok(self.op.delete(&self.path[self.relative_path_pos..]).await?)
+    }
+
     /// Converts into [`InputFile`].
     pub fn to_input_file(self) -> InputFile {
         InputFile {

crates/iceberg/src/writer/file_writer/parquet_writer.rs

@@ -554,7 +554,7 @@ impl FileWriter for ParquetWriter {
         Ok(())
     }
 
-    async fn close(mut self) -> crate::Result<Vec<crate::spec::DataFileBuilder>> {
+    async fn close(mut self) -> Result<Vec<DataFileBuilder>> {
         let writer = match self.inner_writer.take() {
             Some(writer) => writer,
             None => return Ok(vec![]),
@@ -566,22 +566,33 @@ impl FileWriter for ParquetWriter {
 
         let written_size = self.written_size.load(std::sync::atomic::Ordering::Relaxed);
 
-        let parquet_metadata =
-            Arc::new(self.thrift_to_parquet_metadata(metadata).map_err(|err| {
+        if self.current_row_num == 0 {
+            self.out_file.delete().await.map_err(|err| {
                 Error::new(
                     ErrorKind::Unexpected,
-                    "Failed to convert metadata from thrift to parquet.",
+                    "Failed to delete empty parquet file.",
                 )
                 .with_source(err)
-            })?);
-
-        Ok(vec![Self::parquet_to_data_file_builder(
-            self.schema,
-            parquet_metadata,
-            written_size as usize,
-            self.out_file.location().to_string(),
-            self.nan_value_count_visitor.nan_value_counts,
-        )?])
+            })?;
+            Ok(vec![])
+        } else {
+            let parquet_metadata =
+                Arc::new(self.thrift_to_parquet_metadata(metadata).map_err(|err| {
+                    Error::new(
+                        ErrorKind::Unexpected,
+                        "Failed to convert metadata from thrift to parquet.",
+                    )
+                    .with_source(err)
+                })?);
+
+            Ok(vec![Self::parquet_to_data_file_builder(
+                self.schema,
+                parquet_metadata,
+                written_size as usize,
+                self.out_file.location().to_string(),
+                self.nan_value_count_visitor.nan_value_counts,
+            )?])
+        }
     }
 }
 
@@ -2218,4 +2229,44 @@ mod tests {
 
         Ok(())
     }
+
+    #[tokio::test]
+    async fn test_write_empty_parquet_file() {
+        let temp_dir = TempDir::new().unwrap();
+        let file_io = FileIOBuilder::new_fs_io().build().unwrap();
+        let location_gen =
+            MockLocationGenerator::new(temp_dir.path().to_str().unwrap().to_string());
+        let file_name_gen =
+            DefaultFileNameGenerator::new("test".to_string(), None, DataFileFormat::Parquet);
+
+        // write data
+        let pw = ParquetWriterBuilder::new(
+            WriterProperties::builder().build(),
+            Arc::new(
+                Schema::builder()
+                    .with_schema_id(1)
+                    .with_fields(vec![NestedField::required(
+                        0,
+                        "col",
+                        Type::Primitive(PrimitiveType::Long),
+                    )
+                    .with_id(0)
+                    .into()])
+                    .build()
+                    .expect("Failed to create schema"),
+            ),
+            file_io.clone(),
+            location_gen,
+            file_name_gen,
+        )
+        .build()
+        .await
+        .unwrap();
+
+        let res = pw.close().await.unwrap();
+        assert_eq!(res.len(), 0);
+
+        // Check that file should have been deleted.
+        assert_eq!(std::fs::read_dir(temp_dir.path()).unwrap().count(), 0);
+    }
 }