feat(io): UnpartitionedWriter + TaskWriter

[CTTY]

Which issue does this PR close?

• Closes Add TaskWriter #1770

What changes are included in this PR?

Are these changes tested?

[CTTY]

It may make more sense if we just construct partition_splitter within new?

Looking at Splitter::new, it takes a input_schema: ArrowSchemaRef, but in reality we can get the input_schema directly from the RecordBatch. If we initialize the partition splitter lazily then we don't need to ask users to build a partition_splitter themselves

[CTTY]

maybe it's a good idea to change RecordBatchPartitionSplitter to PartitionSplitter<I = RecordBatch>? This way the TaskWriter implementation can also be generic. Haven't explored too much on that front tho. What I have in mind rn:

trait PartitionSplitter<I = DefaultInput> {
     fn split(&self, input: I) -> Result<Vec<(PartitionKey, I)>>
}

impl<I: PositionalDeletes> PartitionSplitter<I> for PositionalDeletePartitionSplitter

cc @ZENOTME

[CTTY]

Remove this

[CTTY]

Remove this

[liurenjie1024]

I think we are on the right track. I left some comments, and we need to split them into smaller prs.

[liurenjie1024]

This change is somehow ugly. The splitter could be split into two parts:

1. Calculate partition value.
2. Split record batch according to partition value.

We could abstract out the process of calculating partition value.

[CTTY]

I think we can reuse the PartititonValueCalculator here.

Are you suggesting that we should have "calculate" and "split" two functions within the splitter? I think that way it would be hard for people to tell when should they call calculate before calling split.

Maybe this would look better?

impl RecordBatchPartitionSplitter {
 pub fn new(
        iceberg_schema: SchemaRef,
        partition_spec: PartitionSpecRef,
       // if some, then calculate the partition value, otherwise use `_partition`
        calculator: Option<PartitionValueCalculator>, 
    )
}
 

[liurenjie1024]

We could simplify this as

Partitioned {
   splitter: RecordBatchSplitter,
   partitioned_writer: Arc<dyn PartitionedWriter>
}