support delete if empty for parquet writer

Author: ZENOTME

crates/iceberg/src/io/file_io.rs

@@ -334,6 +334,11 @@ impl OutputFile {
         Ok(self.op.exists(&self.path[self.relative_path_pos..]).await?)
     }
 
+    /// Delete the file if it exists.
+    pub async fn delete(&self) -> crate::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/base_writer/data_file_writer.rs

@@ -140,6 +140,7 @@ mod test {
             file_io.clone(),
             location_gen,
             file_name_gen,
+            false,
         );
 
         let mut data_file_writer = DataFileWriterBuilder::new(pw, None).build().await.unwrap();
@@ -198,6 +199,7 @@ mod test {
             file_io.clone(),
             location_gen,
             file_name_gen,
+            false,
         );
 
         let mut data_file_writer =

crates/iceberg/src/writer/base_writer/equality_delete_writer.rs

@@ -397,6 +397,7 @@ mod test {
             file_io.clone(),
             location_gen,
             file_name_gen,
+            false,
         );
         let mut equality_delete_writer = EqualityDeleteFileWriterBuilder::new(pb, equality_config)
             .build()
@@ -561,6 +562,7 @@ mod test {
             file_io.clone(),
             location_gen,
             file_name_gen,
+            false,
         );
         let mut equality_delete_writer = EqualityDeleteFileWriterBuilder::new(pb, config)
             .build()

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

@@ -54,6 +54,8 @@ pub struct ParquetWriterBuilder<T: LocationGenerator, F: FileNameGenerator> {
     file_io: FileIO,
     location_generator: T,
     file_name_generator: F,
+
+    delete_if_empty: bool,
 }
 
 impl<T: LocationGenerator, F: FileNameGenerator> ParquetWriterBuilder<T, F> {
@@ -65,13 +67,15 @@ impl<T: LocationGenerator, F: FileNameGenerator> ParquetWriterBuilder<T, F> {
         file_io: FileIO,
         location_generator: T,
         file_name_generator: F,
+        delete_if_empty: bool,
     ) -> Self {
         Self {
             props,
             schema,
             file_io,
             location_generator,
             file_name_generator,
+            delete_if_empty,
         }
     }
 }
@@ -101,6 +105,7 @@ impl<T: LocationGenerator, F: FileNameGenerator> FileWriterBuilder for ParquetWr
             written_size,
             current_row_num: 0,
             out_file,
+            delete_if_empty: self.delete_if_empty,
         })
     }
 }
@@ -216,6 +221,7 @@ pub struct ParquetWriter {
     writer: AsyncArrowWriter<AsyncFileWriter<TrackWriter>>,
     written_size: Arc<AtomicI64>,
     current_row_num: usize,
+    delete_if_empty: bool,
 }
 
 /// Used to aggregate min and max value of each column.
@@ -411,6 +417,11 @@ impl FileWriter for ParquetWriter {
             Error::new(ErrorKind::Unexpected, "Failed to close parquet writer.").with_source(err)
         })?;
 
+        if self.delete_if_empty && self.current_row_num == 0 {
+            self.out_file.delete().await?;
+            return Ok(vec![]);
+        }
+
         let written_size = self.written_size.load(std::sync::atomic::Ordering::Relaxed);
 
         Ok(vec![Self::to_data_file_builder(
@@ -671,6 +682,7 @@ mod tests {
             file_io.clone(),
             location_gen,
             file_name_gen,
+            false,
         )
         .build()
         .await?;
@@ -867,6 +879,7 @@ mod tests {
             file_io.clone(),
             location_gen,
             file_name_gen,
+            false,
         )
         .build()
         .await?;
@@ -1057,6 +1070,7 @@ mod tests {
             file_io.clone(),
             loccation_gen,
             file_name_gen,
+            false,
         )
         .build()
         .await?;

crates/integration_tests/tests/append_data_file_test.rs

@@ -96,6 +96,7 @@ async fn test_append_data_file() {
         table.file_io().clone(),
         location_generator.clone(),
         file_name_generator.clone(),
+        false,
     );
     let data_file_writer_builder = DataFileWriterBuilder::new(parquet_writer_builder, None);
     let mut data_file_writer = data_file_writer_builder.build().await.unwrap();

crates/integration_tests/tests/append_partition_data_file_test.rs

@@ -113,6 +113,7 @@ async fn test_append_partition_data_file() {
         table.file_io().clone(),
         location_generator.clone(),
         file_name_generator.clone(),
+        false,
     );
 
     let mut data_file_writer_valid = DataFileWriterBuilder::new(

crates/integration_tests/tests/conflict_commit_test.rs

@@ -95,6 +95,7 @@ async fn test_append_data_file_conflict() {
         table.file_io().clone(),
         location_generator.clone(),
         file_name_generator.clone(),
+        false,
     );
     let data_file_writer_builder = DataFileWriterBuilder::new(parquet_writer_builder, None);
     let mut data_file_writer = data_file_writer_builder.build().await.unwrap();