refactor: remove TrackWriter (#1575)

## Which issue does this PR close?

- Closes #1574 

## What changes are included in this PR?

1. Remove `TrackWriter`
2. impl FileWrite for Box<dyn FileWrite>

## Are these changes tested?

Covered by existing tests

Author: mapleFU

crates/iceberg/src/io/file_io.rs

@@ -390,6 +390,17 @@ impl FileWrite for opendal::Writer {
     }
 }
 
+#[async_trait::async_trait]
+impl FileWrite for Box<dyn FileWrite> {
+    async fn write(&mut self, bs: Bytes) -> crate::Result<()> {
+        self.as_mut().write(bs).await
+    }
+
+    async fn close(&mut self) -> crate::Result<()> {
+        self.as_mut().close().await
+    }
+}
+
 /// Output file is used for writing to files..
 #[derive(Debug)]
 pub struct OutputFile {

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

@@ -26,7 +26,6 @@ use crate::spec::DataFileBuilder;
 
 mod parquet_writer;
 pub use parquet_writer::{ParquetWriter, ParquetWriterBuilder};
-mod track_writer;
 
 pub mod location_generator;
 /// Module providing writers that can automatically roll over to new files based on size thresholds.

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

@@ -19,7 +19,6 @@
 
 use std::collections::HashMap;
 use std::sync::Arc;
-use std::sync::atomic::AtomicI64;
 
 use arrow_schema::SchemaRef as ArrowSchemaRef;
 use bytes::Bytes;
@@ -36,7 +35,6 @@ use parquet::thrift::{TCompactOutputProtocol, TSerializable};
 use thrift::protocol::TOutputProtocol;
 
 use super::location_generator::{FileNameGenerator, LocationGenerator};
-use super::track_writer::TrackWriter;
 use super::{FileWriter, FileWriterBuilder};
 use crate::arrow::{
     ArrowFileReader, DEFAULT_MAP_FIELD_NAME, NanValueCountVisitor, get_parquet_stat_max_as_datum,
@@ -87,7 +85,6 @@ impl<T: LocationGenerator, F: FileNameGenerator> FileWriterBuilder for ParquetWr
     type R = ParquetWriter;
 
     async fn build(self) -> Result<Self::R> {
-        let written_size = Arc::new(AtomicI64::new(0));
         let out_file = self.file_io.new_output(
             self.location_generator
                 .generate_location(&self.file_name_generator.generate_file_name()),
@@ -97,7 +94,6 @@ impl<T: LocationGenerator, F: FileNameGenerator> FileWriterBuilder for ParquetWr
             schema: self.schema.clone(),
             inner_writer: None,
             writer_properties: self.props,
-            written_size,
             current_row_num: 0,
             out_file,
             nan_value_count_visitor: NanValueCountVisitor::new(),
@@ -227,11 +223,8 @@ impl SchemaVisitor for IndexByParquetPathName {
 pub struct ParquetWriter {
     schema: SchemaRef,
     out_file: OutputFile,
-    inner_writer: Option<AsyncArrowWriter<AsyncFileWriter<TrackWriter>>>,
+    inner_writer: Option<AsyncArrowWriter<AsyncFileWriter<Box<dyn FileWrite>>>>,
     writer_properties: WriterProperties,
-    // written_size is only accurate after closing the inner writer,
-    // because the inner writer flushes data asynchronously.
-    written_size: Arc<AtomicI64>,
     current_row_num: usize,
     nan_value_count_visitor: NanValueCountVisitor,
 }
@@ -534,8 +527,7 @@ impl FileWriter for ParquetWriter {
             writer
         } else {
             let arrow_schema: ArrowSchemaRef = Arc::new(self.schema.as_ref().try_into()?);
-            let inner_writer =
-                TrackWriter::new(self.out_file.writer().await?, self.written_size.clone());
+            let inner_writer = self.out_file.writer().await?;
             let async_writer = AsyncFileWriter::new(inner_writer);
             let writer = AsyncArrowWriter::try_new(
                 async_writer,
@@ -562,16 +554,16 @@ impl FileWriter for ParquetWriter {
     }
 
     async fn close(mut self) -> Result<Vec<DataFileBuilder>> {
-        let writer = match self.inner_writer.take() {
+        let mut writer = match self.inner_writer.take() {
             Some(writer) => writer,
             None => return Ok(vec![]),
         };
 
-        let metadata = writer.close().await.map_err(|err| {
-            Error::new(ErrorKind::Unexpected, "Failed to close parquet writer.").with_source(err)
+        let metadata = writer.finish().await.map_err(|err| {
+            Error::new(ErrorKind::Unexpected, "Failed to finish parquet writer.").with_source(err)
         })?;
 
-        let written_size = self.written_size.load(std::sync::atomic::Ordering::Relaxed);
+        let written_size = writer.bytes_written();
 
         if self.current_row_num == 0 {
             self.out_file.delete().await.map_err(|err| {
@@ -595,7 +587,7 @@ impl FileWriter for ParquetWriter {
             Ok(vec![Self::parquet_to_data_file_builder(
                 self.schema,
                 parquet_metadata,
-                written_size as usize,
+                written_size,
                 self.out_file.location().to_string(),
                 self.nan_value_count_visitor.nan_value_counts,
             )?])