feat: init iceberg writer

[ZENOTME]

This PR init the iceberg writer and implement the data file writer. It's the final part of #135.

[ZENOTME]

sad message: we can't use async_fn_in_trait here because it does not allow object safety so we can't construct something like Box<dyn IcebergWriter>. It's important to erase the type for complex writer.

[ZENOTME]

cc @Fokko @Xuanwo @liurenjie1024 @JanKaul

[liurenjie1024]

Thanks @ZENOTME for this pr, we are close to have writing support. Left some questions about the implementation.

[liurenjie1024]

Why we need this?

[ZENOTME]

oh, I think builder also can use #[async_trait::async_trait]🤔

[liurenjie1024]

Should we follow the pattern in FileWriterBuilder to return impl Future?

[liurenjie1024]

Instead of returning a builder, could we return DataFile by storing necessary fields in writer?

[ZENOTME]

Indeed it's reasonable to return DataFile here. The IcebergWriter return DataFile which represent that this DataFile is complete. It guarantee that it can used to commit directly but also means that it attach more optional message from other writer.

[ZENOTME]

I find that this will cause more complexity for the interface. The necessary fields for writer can't be store in the builder because a builder can used to build multiple writers so it means that the fields should be pass when we create the writer. So we need interface like following so that we can pass different config.

pub trait IcebergWriterBuilder<I = DefaultInput, O = DefaultOutput>:
    Send + Clone + 'static
{
    /// The associated writer type.
    type R: IcebergWriter<I, O>;
    type C;
    /// Build the iceberg writer.
    async fn build(self,config: Self::C) -> Result<Self::R>;
}

This interface can let us create the writer in a more flexible way but incurs more complexity. So I'm concerned is that worth.🤔

[liurenjie1024]

Sorry, I don't get your point. We don't need to expose configs, it's builder specific stuff. For DataFileWriterBuilder, we can create DataFileWriter with same partition value, and clone this partition value for every DataFileWriter

[ZENOTME]

The partition value may only be known when running. e.g. PartitionWriterBuilder<DataFileWriterBuilder>, the partition value is only knowing when partition writer accept the value. And we can't create the DataFileWriter with partition value in advance.

[liurenjie1024]

That makes sense to me.

[liurenjie1024]

Why the field id is "-1"?

[ZENOTME]

For the inner nest field, set the field id as -1. It coming from #176 (comment).

[viirya]

I'm also wondering why there are field ids -1? Took a look at the comment, but it only said to assign the field ids. Is -1 valid field id?

[liurenjie1024]

The iceberg spec doesn't forbid -1 as field id, but it requires that all field ids should be unique, so assigning different fields to same field id would be illegal.

[ZENOTME]

As spec, we should assign a unique id for each field in a struct. It's wrong here. I have fixed the field id.

[liurenjie1024]

cc @Xuanwo @Fokko PTAL

[marvinlanhenke]

@liurenjie1024 @ZENOTME
What's the current status on this PR - as it looks very promising as well as the outlined framework in #34 ?

Since we have already completed some issues (or they are in progress) for read support, I think it would be beneficial to outline the next steps for implementing write support. Perhaps in another tracking issue (I don't think we have none yet?).

I think we have most of the writers in place (when this PR is ready), but have yet to 'orchestrate' them?

[ZENOTME]

@liurenjie1024 @ZENOTME What's the current status on this PR - as it looks very promising as well as the outlined framework in #34 ?

I think this PR is ready to go.

Since we have already completed some issues (or they are in progress) for read support, I think it would be beneficial to outline the next steps for implementing write support. Perhaps in another tracking issue (I don't think we have none yet?).

I think we have most of the writers in place (when this PR is ready), but have yet to 'orchestrate' them?

Yes! Looks good to me. And after this basic framework, we can create a track for more specific writer. cc @liurenjie1024 @Xuanwo @Fokko

[Xuanwo]

Hi, I'm bit confused about this. Why we need this? What will happen if user calling flush on the same file twice?

[ZENOTME]

Because the FileWriter only provides the interface close(self) which will make it simple. To provide the interface flush(&mut self), each time flush we will create a new FileWriter.

A FileWriter will write the data into one or multiple files. There is no restriction for that.

What will happen if user calling flush on the same file twice?

It's safe. The semantic of flush is to flush data written before into the storage and generate the files.

[Xuanwo]

I'm thinking about the error handling cases. If user hits error during flush, they can't retry it and will lost all existing written data. How about pushing the writer back if it flush failed? User can decide whether abort the whole writing process or retry this flush.

[Xuanwo]

I'm thinking about the error handling cases. If user hits error during flush, they can't retry it and will lost all existing written data. How about pushing the writer back if it flush failed? User can decide whether abort the whole writing process or retry this flush.

It's fine if you think it's out of the current scope. We can create an issue for this and keep moving forward.

[ZENOTME]

I'm thinking about the error handling cases. If user hits error during flush, they can't retry it and will lost all existing written data. How about pushing the writer back if it flush failed? User can decide whether abort the whole writing process or retry this flush.

Thanks! I think you remind me that this way is not good for the error handle.
For a flush process, there are two phases:

1. close inner writer.
2. create a new writer.

Because of fn close(self), we have to replace a new writer before closing successfully. So there are two error cases that may happen:

1. create a new writer fail so that we skip to close the original writer
2. close fail but we create a new writer

I think there are two solutions:

1. Use fn close(&mut self)
2. Use Option<Writer> so that we can take the writer temporarily.

[Xuanwo]

I do prefer to use fn close(&mut self) since close also might return error.

[ZENOTME]

I'm thinking about the error handling cases. If user hits error during flush, they can't retry it and will lost all existing written data. How about pushing the writer back if it flush failed? User can decide whether abort the whole writing process or retry this flush.

I find that we can't guarantee that the user can retry the flush. E.g. the close interface for parquet will consume itself. I think a more simple semantic we can guarantee that if flush fails, the data will lose all existing written data and the writer can be used again. It is not friendly for users but it's easier to maintain its semantics.
For this semantic, I think Option<Writer> and fn close(self) may be more appropriate because it guarantees that a writer will not be used again if the close fails.

[Xuanwo]

I think a more simple semantic we can guarantee that if flush fails, the data will lose all existing written data and the writer can be used again. It is not friendly for users but it's easier to maintain its semantics.

Makes sense. By the way, opendal can retry internally, so this should not be a big issue. Let's keep moving.

[sdd]

It feels a little bit odd to pass a builder to another builder. Could we do something like this?

DataFileWriter
    .builder()
    // this would call ParquetWriter::builder() to get a `ParquetWriterBuilder`,
    // and then pass ownership of the `DataFileWriter` to the `ParquetWriterBuilder`,
    // returning the `ParquetWriterBuilder`
    .with_writer(ParquetWriter)  
    // these calls happen in the `ParquetWriterBuilder`, 
    // allowing customization of the wrapped concrete writer
    .with_foo()
    .with_bar()
    // this finalizes the `ParquetWriterBuilder`, building a
    // `ParquetWriter`, and returns the `DataFileWriterBuilder`
    // that was passed earlier, after first passing in the `ParrquetWriter`
    .build_writer()
    // these calls now happen on the `DataFileWriterBuilder`,
    // allowing further setup of the `DataFileWriter`
    .with_baz()
    .with_quux()
    // finally returns a `DataFileWriter`, or perhaps a
    // `Result<DataFileWriter>` or Future of one of those
    .build() 

[ZENOTME]

This top-down way looks good to me to avoid passing builders to each other. However, I am not sure whether this style will incur more complexity in the future.

Actually, this API for now can do this thing because it doesn't restrict how to create and use the builder like the following. So I think this is more relative to the implementation style of our builder rather than the API design.

#[derive(Clone)]
struct B;

impl B {
    pub fn new() -> Self {
        Self
    }

    pub fn with_config(&mut self, _: ()) -> &mut Self {
        self
    }
}

#[async_trait::async_trait]
impl IcebergWriterBuilder for B {
    type R = BW;
    type C = ();
    async fn build(self, _: Self::C) -> Result<Self::R> {
        Ok(BW)
    }
}

struct BW;

impl BW {
    pub fn builder() -> B {
        B::new()
    }
}

#[async_trait::async_trait]
impl IcebergWriter for BW {
    async fn write(&mut self, _input: DefaultInput) -> Result<()> {
        Ok(())
    }

    async fn flush(&mut self) -> Result<DefaultOutput> {
        Ok(vec![])
    }
}

#[derive(Clone)]
struct A<I> {
    inner: Option<I>,
}

impl<I: IcebergWriterBuilder> A<I> {
    pub fn new() -> Self {
        Self { inner: None }
    }

    pub fn with_buidler(&mut self, builder: I) -> &mut I {
        self.inner = Some(builder);
        self.inner.as_mut().unwrap()
    }

    pub fn with_config(&mut self, _: ()) -> &mut Self {
        self
    }
}

struct AW;

impl AW {
    pub fn builder<I:IcebergWriterBuilder>() -> A<I> {
        A::<I>::new()
    }
}

#[async_trait::async_trait]
impl<I: IcebergWriterBuilder> IcebergWriterBuilder for A<I> {
    type R = AW;
    type C = ();
    async fn build(self, _: Self::C) -> Result<Self::R> {
        Ok(AW)
    }
}

#[async_trait::async_trait]
impl IcebergWriter for AW {
    async fn write(&mut self, _input: DefaultInput) -> Result<()> {
        Ok(())
    }

    async fn flush(&mut self) -> Result<DefaultOutput> {
        Ok(vec![])
    }
}

async fn test() {
    let mut a = AW::builder();
    a.
        // config first A
        with_config(()).
        with_buidler(AW::builder()).
        // config second A
        with_config(()).
        with_buidler(BW::builder()).
        // config BW
        with_config(());
    let writer = a.build(()).await;
}

[ZENOTME]

Both way looks good to me. cc @Fokko @Xuanwo @liurenjie1024

[Xuanwo]

I remembered that we have discussed about this before...

[ZENOTME]

Do you mean this #135 (comment). Indeed they look similar.🥵

[Xuanwo]

Maybe we can move on first if both way looks good?

[ZENOTME]

LGTM.

[ZENOTME]

I find that flush is hard to handle the error case. To make the semantics easier to maintain. I use close(&mut) for IcebergWriter instead. The user should not use the writer again after the writer closes it, no matter successful or fail. This behavior should be guaranteed at compile time so I use panic for it rather than return an error.

Why not use close(self)? to guarantee this by compiler?
Because we need the writer can be used as a trait object, like Box<dyn IcebergWriter>. So we can't have a interface like close(self).

[ZENOTME]

I think this PR is ready to move. PTAL. I modified the interface a bit to make it easier for error handling. Feel free to let me there is still @Xuanwo @liurenjie1024 @Fokko @sdd

[Xuanwo]

Thanks! I believe this PR is in good shape.

[liurenjie1024]

LGTM, thanks @ZENOTME for this pr!