add parquet writer

Author: ZENOTME

Cargo.toml

@@ -36,6 +36,7 @@ arrow-schema = { version = ">=46" }
 async-trait = "0.1"
 bimap = "0.6"
 bitvec = "1.0.1"
+bytes = "1.5"
 chrono = "0.4"
 derive_builder = "0.12.0"
 either = "1"
@@ -53,6 +54,7 @@ opendal = "0.44"
 ordered-float = "4.0.0"
 pretty_assertions = "1.4.0"
 port_scanner = "0.1.5"
+parquet = { version = ">=46", features = ["async"] }
 reqwest = { version = "^0.11", features = ["json"] }
 rust_decimal = "1.31.0"
 serde = { version = "^1.0", features = ["rc"] }

crates/iceberg/Cargo.toml

@@ -36,6 +36,7 @@ arrow-schema = { workspace = true }
 async-trait = { workspace = true }
 bimap = { workspace = true }
 bitvec = { workspace = true }
+bytes ={ workspace = true }
 chrono = { workspace = true }
 derive_builder = { workspace = true }
 either = { workspace = true }
@@ -59,9 +60,10 @@ typed-builder = { workspace = true }
 url = { workspace = true }
 urlencoding = { workspace = true }
 uuid = { workspace = true }
+parquet ={ workspace = true }
+tokio = { workspace = true }
 
 [dev-dependencies]
 pretty_assertions = { workspace = true }
 tempfile = { workspace = true }
 tera = { workspace = true }
-tokio = { workspace = true }

crates/iceberg/src/io.rs

@@ -53,7 +53,7 @@ use std::{collections::HashMap, sync::Arc};
 use crate::{error::Result, Error, ErrorKind};
 use futures::{AsyncRead, AsyncSeek, AsyncWrite};
 use once_cell::sync::Lazy;
-use opendal::{Operator, Scheme};
+use opendal::{Operator, Scheme, Writer};
 use url::Url;
 
 /// Following are arguments for [s3 file io](https://py.iceberg.apache.org/configuration/#s3).
@@ -245,7 +245,7 @@ pub trait FileWrite: AsyncWrite {}
 impl<T> FileWrite for T where T: AsyncWrite {}
 
 /// Output file is used for writing to files..
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 pub struct OutputFile {
     op: Operator,
     // Absolution path of file.
@@ -268,6 +268,16 @@ impl OutputFile {
             .await?)
     }
 
+    /// Delete file.
+    pub async fn delete(&self) -> Result<()> {
+        // #TODO
+        // Do we need to check if file exists?
+        if self.exists().await? {
+            self.op.delete(&self.path[self.relative_path_pos..]).await?
+        }
+        Ok(())
+    }
+
     /// Converts into [`InputFile`].
     pub fn to_input_file(self) -> InputFile {
         InputFile {
@@ -278,7 +288,7 @@ impl OutputFile {
     }
 
     /// Creates output file for writing.
-    pub async fn writer(&self) -> Result<impl FileWrite> {
+    pub async fn writer(&self) -> Result<Writer> {
         Ok(self.op.writer(&self.path[self.relative_path_pos..]).await?)
     }
 }

crates/iceberg/src/scan.rs

@@ -188,7 +188,7 @@ impl FileScanTask {
 mod tests {
     use crate::io::{FileIO, OutputFile};
     use crate::spec::{
-        DataContentType, DataFile, DataFileFormat, FormatVersion, Literal, Manifest,
+        DataContentType, DataFileBuilder, DataFileFormat, FormatVersion, Literal, Manifest,
         ManifestContentType, ManifestEntry, ManifestListWriter, ManifestMetadata, ManifestStatus,
         ManifestWriter, Struct, TableMetadata, EMPTY_SNAPSHOT_ID,
     };
@@ -351,14 +351,15 @@ mod tests {
                 ManifestEntry::builder()
                     .status(ManifestStatus::Added)
                     .data_file(
-                        DataFile::builder()
+                        DataFileBuilder::default()
                             .content(DataContentType::Data)
                             .file_path(format!("{}/1.parquet", &fixture.table_location))
                             .file_format(DataFileFormat::Parquet)
                             .file_size_in_bytes(100)
                             .record_count(1)
                             .partition(Struct::from_iter([Some(Literal::long(100))]))
-                            .build(),
+                            .build()
+                            .unwrap(),
                     )
                     .build(),
                 ManifestEntry::builder()
@@ -367,14 +368,15 @@ mod tests {
                     .sequence_number(parent_snapshot.sequence_number())
                     .file_sequence_number(parent_snapshot.sequence_number())
                     .data_file(
-                        DataFile::builder()
+                        DataFileBuilder::default()
                             .content(DataContentType::Data)
                             .file_path(format!("{}/2.parquet", &fixture.table_location))
                             .file_format(DataFileFormat::Parquet)
                             .file_size_in_bytes(100)
                             .record_count(1)
                             .partition(Struct::from_iter([Some(Literal::long(200))]))
-                            .build(),
+                            .build()
+                            .unwrap(),
                     )
                     .build(),
                 ManifestEntry::builder()
@@ -383,14 +385,15 @@ mod tests {
                     .sequence_number(parent_snapshot.sequence_number())
                     .file_sequence_number(parent_snapshot.sequence_number())
                     .data_file(
-                        DataFile::builder()
+                        DataFileBuilder::default()
                             .content(DataContentType::Data)
                             .file_path(format!("{}/3.parquet", &fixture.table_location))
                             .file_format(DataFileFormat::Parquet)
                             .file_size_in_bytes(100)
                             .record_count(1)
                             .partition(Struct::from_iter([Some(Literal::long(300))]))
-                            .build(),
+                            .build()
+                            .unwrap(),
                     )
                     .build(),
             ],

crates/iceberg/src/spec/manifest.rs

@@ -924,7 +924,7 @@ impl TryFrom<i32> for ManifestStatus {
 }
 
 /// Data file carries data file path, partition tuple, metrics, …
-#[derive(Debug, PartialEq, Clone, Eq, TypedBuilder)]
+#[derive(Debug, PartialEq, Clone, Eq, Builder)]
 pub struct DataFile {
     /// field id: 134
     ///

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

@@ -22,6 +22,9 @@ use crate::Result;
 use arrow_array::RecordBatch;
 use futures::Future;
 
+pub mod parquet_writer;
+mod track_writer;
+
 /// File writer builder trait.
 pub trait FileWriterBuilder<O = DefaultOutput>: Send + Clone + 'static {
     /// The associated file writer type.