You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "CTTY (via GitHub)" <gi...@apache.org> on 2025/09/07 01:58:38 UTC
[PR] refactor(writer): Refactor writers for the future partitioning writers [iceberg-rust]
CTTY opened a new pull request, #1657:
URL: https://github.com/apache/iceberg-rust/pull/1657
## Which issue does this PR close?
<!--
We generally require a GitHub issue to be filed for all bug fixes and enhancements and this helps us generate change logs for our releases. You can link an issue to this PR using the GitHub syntax. For example `Closes #123` indicates that this PR will close issue #123.
-->
- Closes #1650
## What changes are included in this PR?
Refactored the writer layers; from a bird’s-eye view, the structure now looks like this:
```mermaid
flowchart TD
subgraph PartitioningWriter
PW[PartitioningWriter]
subgraph RollingWriter
RW[RollingWriter]
subgraph DataFileWriter
DFW[DataFileWriter]
subgraph FileWriter
FW[FileWriter]
end
DFW --> FW
end
RW --> DFW
end
PW --> RW
end
```
- __Modified Writer Interfaces__:
- Updated the `FileWriterBuilder` and `IcebergWriterBuilder` interfaces to accept an `OutputFile` parameter in their `build` methods
- Added a new `PartitioningWriter` trait in `writer/mod.rs` with methods for writing to multiple partitions (not implemented yet, we can use a separate PR to add this trait if needed)
- __Transformed RollingFileWriter to RollingWriter__:
- Renamed `RollingFileWriter` to `RollingWriter` and significantly enhanced its capabilities
- Changed from implementing `FileWriter` to being a standalone writer that uses `IcebergWriterBuilder`
- Added support for location generation and file naming within the writer itself
- Made it generic over `IcebergWriterBuilder`, `LocationGenerator`, and `FileNameGenerator`
- Added methods to create new writers for partitions
- __Updated ParquetWriter__:
- Removed direct dependencies on `LocationGenerator` and `FileNameGenerator`
- Renamed `out_file` to `output_file` for consistency
- Simplified the interface to focus on writing to a provided output file
- __Updated DataFileWriter and EqualityDeleteWriter__:
- Modified to pass the `OutputFile` parameter to their inner writers
- __Updated DataFusion Integration__:
- Modified `IcebergWriteExec` to use the new `RollingWriter` directly instead of using builder patterns
- Simplified the writer creation process in the execution plan
<!--
Provide a summary of the modifications in this PR. List the main changes such as new features, bug fixes, refactoring, or any other updates.
-->
## Are these changes tested?
Not yet, but changing the existing tests accordingly should be enough
<!--
Specify what test covers (unit test, integration test, etc.).
If tests are not included in your PR, please explain why (for example, are they covered by existing tests)?
-->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] refactor(writer): Refactor writers for the future partitioning writers [iceberg-rust]
Posted by "ZENOTME (via GitHub)" <gi...@apache.org>.
ZENOTME commented on code in PR #1657:
URL: https://github.com/apache/iceberg-rust/pull/1657#discussion_r2328690254
##########
crates/iceberg/src/writer/file_writer/rolling_writer.rs:
##########
@@ -17,65 +17,75 @@
use arrow_array::RecordBatch;
-use crate::spec::DataFileBuilder;
-use crate::writer::CurrentFileStatus;
-use crate::writer::file_writer::{FileWriter, FileWriterBuilder};
+use crate::io::FileIO;
+use crate::spec::{DataFile, PartitionKey};
+use crate::writer::file_writer::location_generator::{FileNameGenerator, LocationGenerator};
+use crate::writer::{CurrentFileStatus, IcebergWriter, IcebergWriterBuilder};
use crate::{Error, ErrorKind, Result};
-/// Builder for creating a `RollingFileWriter` that rolls over to a new file
-/// when the data size exceeds a target threshold.
-#[derive(Clone)]
-pub struct RollingFileWriterBuilder<B: FileWriterBuilder> {
+/// A writer that automatically rolls over to a new file when the data size
+/// exceeds a target threshold.
+///
+/// This writer wraps another writer that tracks the amount of data written.
+/// When the data size exceeds the target size, it closes the current file and
+/// starts writing to a new one.
+pub struct RollingWriter<B, L, F>
+where
+ B: IcebergWriterBuilder,
+ L: LocationGenerator,
+ F: FileNameGenerator,
+{
+ inner: Option<B::R>,
inner_builder: B,
target_file_size: usize,
+ location_generator: L,
+ file_name_generator: F,
+ file_io: FileIO,
+ partition_key: Option<PartitionKey>,
+ data_files: Vec<DataFile>, // todo this should be B::R::O? DefaultOutput?
}
-impl<B: FileWriterBuilder> RollingFileWriterBuilder<B> {
- /// Creates a new `RollingFileWriterBuilder` with the specified inner builder and target size.
+impl<B, L, F> RollingWriter<B, L, F>
+where
+ B: IcebergWriterBuilder,
Review Comment:
One thing need to noticed is that following is what IcebergWriterBuilder looks like.
```
#[async_trait::async_trait]
pub trait IcebergWriterBuilder<I = DefaultInput, O = DefaultOutput>:
Send + Clone + 'static
{
/// The associated writer type.
type R: IcebergWriter<I, O>;
/// Build the iceberg writer.
async fn build(self) -> Result<Self::R>;
}
```
For writer like position delete writer, it has different input like following, see: https://github.com/apache/iceberg-rust/pull/704
```
#[async_trait::async_trait]
impl<B: FileWriterBuilder> IcebergWriterBuilder<Vec<PositionDeleteInput>>
for PositionDeleteWriterBuilder<B>
{
type R = PositionDeleteWriter<B>;
async fn build(self) -> Result<Self::R> {
Ok(PositionDeleteWriter {
inner_writer: Some(self.inner.build().await?),
partition_value: self.partition_value.unwrap_or(Struct::empty()),
})
}
}
```
And that's why rolling writer is a FileWriter at first. After we adopt this design, how can we something like
```
RollingWriter<PostitionDeletWriter>
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] refactor(writer): Refactor writers for the future partitioning writers [iceberg-rust]
Posted by "CTTY (via GitHub)" <gi...@apache.org>.
CTTY commented on code in PR #1657:
URL: https://github.com/apache/iceberg-rust/pull/1657#discussion_r2384579825
##########
crates/iceberg/src/writer/base_writer/data_file_writer.rs:
##########
@@ -20,74 +20,101 @@
use arrow_array::RecordBatch;
use itertools::Itertools;
-use crate::Result;
-use crate::spec::{DataContentType, DataFile, Struct};
-use crate::writer::file_writer::{FileWriter, FileWriterBuilder};
+use crate::spec::{DEFAULT_PARTITION_SPEC_ID, DataContentType, DataFile, PartitionKey, Struct};
+use crate::writer::file_writer::FileWriterBuilder;
+use crate::writer::file_writer::location_generator::{FileNameGenerator, LocationGenerator};
+use crate::writer::file_writer::rolling_writer::RollingFileWriter;
use crate::writer::{CurrentFileStatus, IcebergWriter, IcebergWriterBuilder};
+use crate::{Error, ErrorKind, Result};
/// Builder for `DataFileWriter`.
#[derive(Clone, Debug)]
-pub struct DataFileWriterBuilder<B: FileWriterBuilder> {
- inner: B,
- partition_value: Option<Struct>,
- partition_spec_id: i32,
+pub struct DataFileWriterBuilder<B: FileWriterBuilder, L: LocationGenerator, F: FileNameGenerator> {
+ inner_writer: RollingFileWriter<B, L, F>,
+ partition_key: Option<PartitionKey>,
}
-impl<B: FileWriterBuilder> DataFileWriterBuilder<B> {
+impl<B: FileWriterBuilder, L: LocationGenerator, F: FileNameGenerator>
+ DataFileWriterBuilder<B, L, F>
+{
/// Create a new `DataFileWriterBuilder` using a `FileWriterBuilder`.
- pub fn new(inner: B, partition_value: Option<Struct>, partition_spec_id: i32) -> Self {
+ pub fn new(
+ inner_writer: RollingFileWriter<B, L, F>,
+ partition_key: Option<PartitionKey>,
+ ) -> Self {
Self {
- inner,
- partition_value,
- partition_spec_id,
+ inner_writer,
+ partition_key,
}
}
}
#[async_trait::async_trait]
-impl<B: FileWriterBuilder> IcebergWriterBuilder for DataFileWriterBuilder<B> {
- type R = DataFileWriter<B>;
+impl<B: FileWriterBuilder, L: LocationGenerator, F: FileNameGenerator> IcebergWriterBuilder
+ for DataFileWriterBuilder<B, L, F>
+{
+ type R = DataFileWriter<B, L, F>;
async fn build(self) -> Result<Self::R> {
Ok(DataFileWriter {
- inner_writer: Some(self.inner.clone().build().await?),
- partition_value: self.partition_value.unwrap_or(Struct::empty()),
- partition_spec_id: self.partition_spec_id,
+ inner_writer: Some(self.inner_writer),
+ partition_key: self.partition_key,
})
}
}
/// A writer write data is within one spec/partition.
#[derive(Debug)]
-pub struct DataFileWriter<B: FileWriterBuilder> {
- inner_writer: Option<B::R>,
- partition_value: Struct,
- partition_spec_id: i32,
+pub struct DataFileWriter<B: FileWriterBuilder, L: LocationGenerator, F: FileNameGenerator> {
+ inner_writer: Option<RollingFileWriter<B, L, F>>,
+ partition_key: Option<PartitionKey>,
}
#[async_trait::async_trait]
-impl<B: FileWriterBuilder> IcebergWriter for DataFileWriter<B> {
+impl<B: FileWriterBuilder, L: LocationGenerator, F: FileNameGenerator> IcebergWriter
+ for DataFileWriter<B, L, F>
+{
async fn write(&mut self, batch: RecordBatch) -> Result<()> {
- self.inner_writer.as_mut().unwrap().write(&batch).await
+ self.inner_writer
+ .as_mut()
+ .unwrap()
+ .write(&self.partition_key, &batch)
+ .await
}
async fn close(&mut self) -> Result<Vec<DataFile>> {
- let writer = self.inner_writer.take().unwrap();
- Ok(writer
- .close()
- .await?
- .into_iter()
- .map(|mut res| {
- res.content(DataContentType::Data);
- res.partition(self.partition_value.clone());
- res.partition_spec_id(self.partition_spec_id);
- res.build().expect("Guaranteed to be valid")
- })
- .collect_vec())
+ if let Some(writer) = self.inner_writer.take() {
+ Ok(writer
+ .close()
+ .await?
+ .into_iter()
+ .map(|mut res| {
+ res.content(DataContentType::Data);
+ res.partition(
+ self.partition_key
+ .as_ref()
+ .map_or(Struct::empty(), |pk| pk.data().clone()),
+ );
+ res.partition_spec_id(
+ self.partition_key
+ .as_ref()
+ .map_or(DEFAULT_PARTITION_SPEC_ID, |pk| pk.spec().spec_id()),
+ );
Review Comment:
I've added
```
#[builder(default = "Struct::empty()")]
pub(crate) partition: Struct,
```
and
```
#[builder(default = "DEFAULT_PARTITION_SPEC_ID")]
pub(crate) partition_spec_id: i32,
```
to make building data files easier, it's following the same logic tho
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] refactor(writer): Refactor writers for the future partitioning writers [iceberg-rust]
Posted by "ZENOTME (via GitHub)" <gi...@apache.org>.
ZENOTME commented on code in PR #1657:
URL: https://github.com/apache/iceberg-rust/pull/1657#discussion_r2337086685
##########
crates/iceberg/src/writer/file_writer/rolling_writer.rs:
##########
@@ -17,65 +17,75 @@
use arrow_array::RecordBatch;
-use crate::spec::DataFileBuilder;
-use crate::writer::CurrentFileStatus;
-use crate::writer::file_writer::{FileWriter, FileWriterBuilder};
+use crate::io::FileIO;
+use crate::spec::{DataFile, PartitionKey};
+use crate::writer::file_writer::location_generator::{FileNameGenerator, LocationGenerator};
+use crate::writer::{CurrentFileStatus, IcebergWriter, IcebergWriterBuilder};
use crate::{Error, ErrorKind, Result};
-/// Builder for creating a `RollingFileWriter` that rolls over to a new file
-/// when the data size exceeds a target threshold.
-#[derive(Clone)]
-pub struct RollingFileWriterBuilder<B: FileWriterBuilder> {
+/// A writer that automatically rolls over to a new file when the data size
+/// exceeds a target threshold.
+///
+/// This writer wraps another writer that tracks the amount of data written.
+/// When the data size exceeds the target size, it closes the current file and
+/// starts writing to a new one.
+pub struct RollingWriter<B, L, F>
+where
+ B: IcebergWriterBuilder,
+ L: LocationGenerator,
+ F: FileNameGenerator,
+{
+ inner: Option<B::R>,
inner_builder: B,
target_file_size: usize,
+ location_generator: L,
+ file_name_generator: F,
+ file_io: FileIO,
+ partition_key: Option<PartitionKey>,
+ data_files: Vec<DataFile>, // todo this should be B::R::O? DefaultOutput?
}
-impl<B: FileWriterBuilder> RollingFileWriterBuilder<B> {
- /// Creates a new `RollingFileWriterBuilder` with the specified inner builder and target size.
+impl<B, L, F> RollingWriter<B, L, F>
+where
+ B: IcebergWriterBuilder,
Review Comment:
> Meanwhile I've been wondering how useful is the abstraction of IcebergWriter
E.g the user want to custom their own writer with to track some metrics like following:
```
RollingWriter<TrackPositionalDeletesWriter>
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] refactor(writer): Refactor writers for the future partitioning writers [iceberg-rust]
Posted by "CTTY (via GitHub)" <gi...@apache.org>.
CTTY commented on code in PR #1657:
URL: https://github.com/apache/iceberg-rust/pull/1657#discussion_r2371143498
##########
crates/iceberg/src/writer/base_writer/data_file_writer.rs:
##########
@@ -21,54 +21,65 @@ use arrow_array::RecordBatch;
use itertools::Itertools;
use crate::Result;
-use crate::spec::{DataContentType, DataFile, Struct};
-use crate::writer::file_writer::{FileWriter, FileWriterBuilder};
+use crate::spec::{DEFAULT_PARTITION_SPEC_ID, DataContentType, DataFile, PartitionKey, Struct};
+use crate::writer::file_writer::FileWriterBuilder;
+use crate::writer::file_writer::location_generator::{FileNameGenerator, LocationGenerator};
+use crate::writer::file_writer::rolling_writer::RollingFileWriter;
use crate::writer::{CurrentFileStatus, IcebergWriter, IcebergWriterBuilder};
/// Builder for `DataFileWriter`.
#[derive(Clone, Debug)]
-pub struct DataFileWriterBuilder<B: FileWriterBuilder> {
- inner: B,
- partition_value: Option<Struct>,
- partition_spec_id: i32,
+pub struct DataFileWriterBuilder<B: FileWriterBuilder, L: LocationGenerator, F: FileNameGenerator> {
+ inner_writer: RollingFileWriter<B, L, F>,
+ partition_key: Option<PartitionKey>,
}
-impl<B: FileWriterBuilder> DataFileWriterBuilder<B> {
+impl<B: FileWriterBuilder, L: LocationGenerator, F: FileNameGenerator>
+ DataFileWriterBuilder<B, L, F>
+{
/// Create a new `DataFileWriterBuilder` using a `FileWriterBuilder`.
- pub fn new(inner: B, partition_value: Option<Struct>, partition_spec_id: i32) -> Self {
+ pub fn new(
+ inner_writer: RollingFileWriter<B, L, F>,
+ partition_key: Option<PartitionKey>,
+ ) -> Self {
Self {
- inner,
- partition_value,
- partition_spec_id,
+ inner_writer,
+ partition_key,
}
}
}
#[async_trait::async_trait]
-impl<B: FileWriterBuilder> IcebergWriterBuilder for DataFileWriterBuilder<B> {
- type R = DataFileWriter<B>;
+impl<B: FileWriterBuilder, L: LocationGenerator, F: FileNameGenerator> IcebergWriterBuilder
+ for DataFileWriterBuilder<B, L, F>
+{
+ type R = DataFileWriter<B, L, F>;
async fn build(self) -> Result<Self::R> {
Ok(DataFileWriter {
- inner_writer: Some(self.inner.clone().build().await?),
- partition_value: self.partition_value.unwrap_or(Struct::empty()),
- partition_spec_id: self.partition_spec_id,
+ inner_writer: Some(self.inner_writer), // todo revisit this, probably still need a builder for rolling writer
+ partition_key: self.partition_key,
})
}
}
/// A writer write data is within one spec/partition.
#[derive(Debug)]
-pub struct DataFileWriter<B: FileWriterBuilder> {
- inner_writer: Option<B::R>,
- partition_value: Struct,
- partition_spec_id: i32,
+pub struct DataFileWriter<B: FileWriterBuilder, L: LocationGenerator, F: FileNameGenerator> {
+ inner_writer: Option<RollingFileWriter<B, L, F>>,
Review Comment:
Yeah I agree, it's that we need to `take` the rolling writer when closing the `DataFileWriter`, and I haven't found a nice solution to do it without `Option`, and I also don't think it makes much sense to implement `Default` for `RollingFileWriter`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] refactor(writer): Refactor writers for the future partitioning writers [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 commented on code in PR #1657:
URL: https://github.com/apache/iceberg-rust/pull/1657#discussion_r2389936071
##########
crates/iceberg/src/writer/file_writer/rolling_writer.rs:
##########
@@ -15,67 +15,121 @@
// specific language governing permissions and limitations
// under the License.
+use std::fmt::{Debug, Formatter};
+
use arrow_array::RecordBatch;
-use crate::spec::DataFileBuilder;
+use crate::io::{FileIO, OutputFile};
+use crate::spec::{DataFileBuilder, PROPERTY_WRITE_TARGET_FILE_SIZE_BYTES_DEFAULT, PartitionKey};
use crate::writer::CurrentFileStatus;
+use crate::writer::file_writer::location_generator::{FileNameGenerator, LocationGenerator};
use crate::writer::file_writer::{FileWriter, FileWriterBuilder};
use crate::{Error, ErrorKind, Result};
-/// Builder for creating a `RollingFileWriter` that rolls over to a new file
-/// when the data size exceeds a target threshold.
-#[derive(Clone)]
-pub struct RollingFileWriterBuilder<B: FileWriterBuilder> {
+/// A writer that automatically rolls over to a new file when the data size
+/// exceeds a target threshold.
+///
+/// This writer wraps another file writer that tracks the amount of data written.
+/// When the data size exceeds the target size, it closes the current file and
+/// starts writing to a new one.
+pub struct RollingFileWriter<B: FileWriterBuilder, L: LocationGenerator, F: FileNameGenerator> {
+ inner: Option<B::R>,
inner_builder: B,
target_file_size: usize,
+ data_file_builders: Vec<DataFileBuilder>,
+ file_io: FileIO,
+ location_generator: L,
+ file_name_generator: F,
+}
+
+impl<B: FileWriterBuilder, L: LocationGenerator, F: FileNameGenerator> Clone
Review Comment:
This is incorrect, clone means creating a new one with exactly same data. But you are creating a new IcebergWriterBuilder with following changes:
1. Partition key change
2. Rolling file writer changed
I don't think it's reasonable to put it in `Clone`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] refactor(writer): Refactor writers for the future partitioning writers [iceberg-rust]
Posted by "CTTY (via GitHub)" <gi...@apache.org>.
CTTY commented on code in PR #1657:
URL: https://github.com/apache/iceberg-rust/pull/1657#discussion_r2379638740
##########
crates/iceberg/src/writer/mod.rs:
##########
@@ -229,20 +229,30 @@ pub mod file_writer;
use arrow_array::RecordBatch;
use crate::Result;
-use crate::spec::DataFile;
+use crate::io::OutputFile;
+use crate::spec::{DataFile, PartitionKey};
type DefaultInput = RecordBatch;
type DefaultOutput = Vec<DataFile>;
+/// The partitioning writer used to write data to multiple partitions.
+pub trait PartitioningWriter {
+ /// Write a record batch, all rows from this record batch should come from one partition
+ fn write(&mut self, partition_key: PartitionKey, batch: RecordBatch) -> Result<()>;
+
+ /// Close all writers and return the data files.
+ fn close(&mut self) -> Result<Vec<DataFile>>;
+}
+
/// The builder for iceberg writer.
#[async_trait::async_trait]
pub trait IcebergWriterBuilder<I = DefaultInput, O = DefaultOutput>:
Review Comment:
not needed as per https://github.com/apache/iceberg-rust/pull/1657#issuecomment-3306441725
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] refactor(writer): Refactor writers for the future partitioning writers [iceberg-rust]
Posted by "CTTY (via GitHub)" <gi...@apache.org>.
CTTY commented on code in PR #1657:
URL: https://github.com/apache/iceberg-rust/pull/1657#discussion_r2365926310
##########
crates/iceberg/src/writer/partitioning/clustered.rs:
##########
@@ -0,0 +1,251 @@
+// // Licensed to the Apache Software Foundation (ASF) under one
+// // or more contributor license agreements. See the NOTICE file
+// // distributed with this work for additional information
+// // regarding copyright ownership. The ASF licenses this file
+// // to you under the Apache License, Version 2.0 (the
+// // "License"); you may not use this file except in compliance
+// // with the License. You may obtain a copy of the License at
+// //
+// // http://www.apache.org/licenses/LICENSE-2.0
+// //
+// // Unless required by applicable law or agreed to in writing,
+// // software distributed under the License is distributed on an
+// // "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// // KIND, either express or implied. See the License for the
+// // specific language governing permissions and limitations
+// // under the License.
+//
+// //! This module provides the `ClusteredWriter` implementation.
+//
+// use std::collections::HashSet;
+//
+// use arrow_array::RecordBatch;
+// use async_trait::async_trait;
+//
+// use crate::spec::{PartitionKey, Struct};
+// use crate::writer::{DefaultInput, DefaultOutput, IcebergWriter, IcebergWriterBuilder};
+// use crate::writer::file_writer::location_generator::{FileNameGenerator, LocationGenerator};
+// use crate::{Error, ErrorKind, Result};
+// use crate::writer::partitioning::PartitioningWriter;
+//
+// /// A writer that writes data to a single partition at a time.
+// ///
+// /// When data for a new partition arrives, it closes the current writer
+// /// and creates a new one for the new partition.
+// ///
+// /// Once a partition has been written to and closed, any further attempts
+// /// to write to that partition will result in an error.
+// pub struct ClusteredWriter<B: IcebergWriterBuilder, I: Default + Send = DefaultInput, O: Default + Send = DefaultOutput>
Review Comment:
Please ignore this for now, I think it's better to keep this draft/round of changes focused on the interfaces changes with existing writer
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] refactor(writer): Refactor writers for the future partitioning writers [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 commented on code in PR #1657:
URL: https://github.com/apache/iceberg-rust/pull/1657#discussion_r2387417043
##########
crates/iceberg/src/writer/base_writer/equality_delete_writer.rs:
##########
@@ -168,17 +168,12 @@ impl<B: FileWriterBuilder, L: LocationGenerator, F: FileNameGenerator> IcebergWr
.map(|mut res| {
res.content(crate::spec::DataContentType::EqualityDeletes);
res.equality_ids(Some(self.equality_ids.iter().copied().collect_vec()));
- res.partition(
- self.partition_key
- .as_ref()
- .map_or(Struct::empty(), |pk| pk.data().clone()),
- );
- res.partition_spec_id(
- self.partition_key
- .as_ref()
- .map_or(DEFAULT_PARTITION_SPEC_ID, |pk| pk.spec().spec_id()),
- );
- res.build().expect("msg")
+ if let Some(pk) = self.partition_key.as_ref() {
+ res.partition(pk.data().clone());
+ res.partition_spec_id(pk.spec().spec_id());
+ }
+ res.build()
Review Comment:
Ditto.
##########
crates/iceberg/src/writer/base_writer/data_file_writer.rs:
##########
@@ -90,17 +93,12 @@ impl<B: FileWriterBuilder, L: LocationGenerator, F: FileNameGenerator> IcebergWr
.into_iter()
.map(|mut res| {
res.content(DataContentType::Data);
- res.partition(
- self.partition_key
- .as_ref()
- .map_or(Struct::empty(), |pk| pk.data().clone()),
- );
- res.partition_spec_id(
- self.partition_key
- .as_ref()
- .map_or(DEFAULT_PARTITION_SPEC_ID, |pk| pk.spec().spec_id()),
- );
- res.build().expect("Guaranteed to be valid")
+ if let Some(pk) = self.partition_key.as_ref() {
+ res.partition(pk.data().clone());
+ res.partition_spec_id(pk.spec().spec_id());
+ }
+ res.build()
+ .expect("DataFileBuilder is guaranteed to be valid")
Review Comment:
We should not panic here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] refactor(writer): Refactor writers for the future partitioning writers [iceberg-rust]
Posted by "CTTY (via GitHub)" <gi...@apache.org>.
CTTY commented on code in PR #1657:
URL: https://github.com/apache/iceberg-rust/pull/1657#discussion_r2389958571
##########
crates/iceberg/src/writer/file_writer/rolling_writer.rs:
##########
@@ -15,67 +15,121 @@
// specific language governing permissions and limitations
// under the License.
+use std::fmt::{Debug, Formatter};
+
use arrow_array::RecordBatch;
-use crate::spec::DataFileBuilder;
+use crate::io::{FileIO, OutputFile};
+use crate::spec::{DataFileBuilder, PROPERTY_WRITE_TARGET_FILE_SIZE_BYTES_DEFAULT, PartitionKey};
use crate::writer::CurrentFileStatus;
+use crate::writer::file_writer::location_generator::{FileNameGenerator, LocationGenerator};
use crate::writer::file_writer::{FileWriter, FileWriterBuilder};
use crate::{Error, ErrorKind, Result};
-/// Builder for creating a `RollingFileWriter` that rolls over to a new file
-/// when the data size exceeds a target threshold.
-#[derive(Clone)]
-pub struct RollingFileWriterBuilder<B: FileWriterBuilder> {
+/// A writer that automatically rolls over to a new file when the data size
+/// exceeds a target threshold.
+///
+/// This writer wraps another file writer that tracks the amount of data written.
+/// When the data size exceeds the target size, it closes the current file and
+/// starts writing to a new one.
+pub struct RollingFileWriter<B: FileWriterBuilder, L: LocationGenerator, F: FileNameGenerator> {
+ inner: Option<B::R>,
inner_builder: B,
target_file_size: usize,
+ data_file_builders: Vec<DataFileBuilder>,
+ file_io: FileIO,
+ location_generator: L,
+ file_name_generator: F,
+}
+
+impl<B: FileWriterBuilder, L: LocationGenerator, F: FileNameGenerator> Clone
Review Comment:
Yes I agree, I think bringing the RollingFileWriterBuilder back and removing this Clone implementation is the correct way to go. I can make the change tomorrow and we can review that
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] refactor(writer): Refactor writers for the future partitioning writers [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 commented on code in PR #1657:
URL: https://github.com/apache/iceberg-rust/pull/1657#discussion_r2351756821
##########
crates/iceberg/src/writer/mod.rs:
##########
@@ -229,20 +229,30 @@ pub mod file_writer;
use arrow_array::RecordBatch;
use crate::Result;
-use crate::spec::DataFile;
+use crate::io::OutputFile;
+use crate::spec::{DataFile, PartitionKey};
type DefaultInput = RecordBatch;
type DefaultOutput = Vec<DataFile>;
+/// The partitioning writer used to write data to multiple partitions.
+pub trait PartitioningWriter {
Review Comment:
Add `#[async_trait]` annotation?
##########
crates/iceberg/src/writer/mod.rs:
##########
@@ -229,20 +229,30 @@ pub mod file_writer;
use arrow_array::RecordBatch;
use crate::Result;
-use crate::spec::DataFile;
+use crate::io::OutputFile;
+use crate::spec::{DataFile, PartitionKey};
type DefaultInput = RecordBatch;
type DefaultOutput = Vec<DataFile>;
+/// The partitioning writer used to write data to multiple partitions.
+pub trait PartitioningWriter {
+ /// Write a record batch, all rows from this record batch should come from one partition
+ fn write(&mut self, partition_key: PartitionKey, batch: RecordBatch) -> Result<()>;
+
+ /// Close all writers and return the data files.
+ fn close(&mut self) -> Result<Vec<DataFile>>;
+}
+
/// The builder for iceberg writer.
#[async_trait::async_trait]
pub trait IcebergWriterBuilder<I = DefaultInput, O = DefaultOutput>:
Send + Clone + 'static
{
/// The associated writer type.
type R: IcebergWriter<I, O>;
- /// Build the iceberg writer.
- async fn build(self) -> Result<Self::R>;
+ /// Build the iceberg writer with the provided output file.
+ async fn build(self, output_file: OutputFile) -> Result<Self::R>;
Review Comment:
We don't need this change per our discussion?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] refactor(writer): Refactor writers for the future partitioning writers [iceberg-rust]
Posted by "CTTY (via GitHub)" <gi...@apache.org>.
CTTY commented on code in PR #1657:
URL: https://github.com/apache/iceberg-rust/pull/1657#discussion_r2388942753
##########
crates/iceberg/src/writer/file_writer/rolling_writer.rs:
##########
@@ -15,67 +15,121 @@
// specific language governing permissions and limitations
// under the License.
+use std::fmt::{Debug, Formatter};
+
use arrow_array::RecordBatch;
-use crate::spec::DataFileBuilder;
+use crate::io::{FileIO, OutputFile};
+use crate::spec::{DataFileBuilder, PROPERTY_WRITE_TARGET_FILE_SIZE_BYTES_DEFAULT, PartitionKey};
use crate::writer::CurrentFileStatus;
+use crate::writer::file_writer::location_generator::{FileNameGenerator, LocationGenerator};
use crate::writer::file_writer::{FileWriter, FileWriterBuilder};
use crate::{Error, ErrorKind, Result};
-/// Builder for creating a `RollingFileWriter` that rolls over to a new file
-/// when the data size exceeds a target threshold.
-#[derive(Clone)]
-pub struct RollingFileWriterBuilder<B: FileWriterBuilder> {
+/// A writer that automatically rolls over to a new file when the data size
+/// exceeds a target threshold.
+///
+/// This writer wraps another file writer that tracks the amount of data written.
+/// When the data size exceeds the target size, it closes the current file and
+/// starts writing to a new one.
+pub struct RollingFileWriter<B: FileWriterBuilder, L: LocationGenerator, F: FileNameGenerator> {
+ inner: Option<B::R>,
inner_builder: B,
target_file_size: usize,
+ data_file_builders: Vec<DataFileBuilder>,
+ file_io: FileIO,
+ location_generator: L,
+ file_name_generator: F,
+}
+
+impl<B: FileWriterBuilder, L: LocationGenerator, F: FileNameGenerator> Clone
Review Comment:
Having `Clone` can be useful in the upcoming `PartitioningWriter` level when we need to spawn new writers with the same configuration. for example, if we have a fanout partitioning writer, and we will need to create a new writer whenever there is a new partition coming in.
With `Clone`, it would be simple:
```rust
let new_writer = self.iceberg_writer_builder.clone().build()?; // iceberg_writer can be generic type
```
Without `Clone`, we will need to re-populate the `IcebergWriterBuilder` and the inner writers all over again:
```rust
let parquet_writer = ...;
let new_rolling_writer = RollingFileWriter::new(parquet_writer, /* pass down objs like file_io, target_file_size... */);
let iceberg_writer_builder = DataFileWriterBuilder::new(...); // this has to be concrete type
let new_writer = iceberg_writer_builder.build()?;
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] refactor(writer): Refactor writers for the future partitioning writers [iceberg-rust]
Posted by "CTTY (via GitHub)" <gi...@apache.org>.
CTTY commented on code in PR #1657:
URL: https://github.com/apache/iceberg-rust/pull/1657#discussion_r2383825905
##########
crates/iceberg/src/writer/base_writer/data_file_writer.rs:
##########
@@ -20,74 +20,101 @@
use arrow_array::RecordBatch;
use itertools::Itertools;
-use crate::Result;
-use crate::spec::{DataContentType, DataFile, Struct};
-use crate::writer::file_writer::{FileWriter, FileWriterBuilder};
+use crate::spec::{DEFAULT_PARTITION_SPEC_ID, DataContentType, DataFile, PartitionKey, Struct};
+use crate::writer::file_writer::FileWriterBuilder;
+use crate::writer::file_writer::location_generator::{FileNameGenerator, LocationGenerator};
+use crate::writer::file_writer::rolling_writer::RollingFileWriter;
use crate::writer::{CurrentFileStatus, IcebergWriter, IcebergWriterBuilder};
+use crate::{Error, ErrorKind, Result};
/// Builder for `DataFileWriter`.
#[derive(Clone, Debug)]
-pub struct DataFileWriterBuilder<B: FileWriterBuilder> {
- inner: B,
- partition_value: Option<Struct>,
- partition_spec_id: i32,
+pub struct DataFileWriterBuilder<B: FileWriterBuilder, L: LocationGenerator, F: FileNameGenerator> {
+ inner_writer: RollingFileWriter<B, L, F>,
+ partition_key: Option<PartitionKey>,
}
-impl<B: FileWriterBuilder> DataFileWriterBuilder<B> {
+impl<B: FileWriterBuilder, L: LocationGenerator, F: FileNameGenerator>
+ DataFileWriterBuilder<B, L, F>
+{
/// Create a new `DataFileWriterBuilder` using a `FileWriterBuilder`.
- pub fn new(inner: B, partition_value: Option<Struct>, partition_spec_id: i32) -> Self {
+ pub fn new(
+ inner_writer: RollingFileWriter<B, L, F>,
+ partition_key: Option<PartitionKey>,
+ ) -> Self {
Self {
- inner,
- partition_value,
- partition_spec_id,
+ inner_writer,
+ partition_key,
}
}
}
#[async_trait::async_trait]
-impl<B: FileWriterBuilder> IcebergWriterBuilder for DataFileWriterBuilder<B> {
- type R = DataFileWriter<B>;
+impl<B: FileWriterBuilder, L: LocationGenerator, F: FileNameGenerator> IcebergWriterBuilder
+ for DataFileWriterBuilder<B, L, F>
+{
+ type R = DataFileWriter<B, L, F>;
async fn build(self) -> Result<Self::R> {
Ok(DataFileWriter {
- inner_writer: Some(self.inner.clone().build().await?),
- partition_value: self.partition_value.unwrap_or(Struct::empty()),
- partition_spec_id: self.partition_spec_id,
+ inner_writer: Some(self.inner_writer),
+ partition_key: self.partition_key,
})
}
}
/// A writer write data is within one spec/partition.
#[derive(Debug)]
-pub struct DataFileWriter<B: FileWriterBuilder> {
- inner_writer: Option<B::R>,
- partition_value: Struct,
- partition_spec_id: i32,
+pub struct DataFileWriter<B: FileWriterBuilder, L: LocationGenerator, F: FileNameGenerator> {
+ inner_writer: Option<RollingFileWriter<B, L, F>>,
+ partition_key: Option<PartitionKey>,
}
#[async_trait::async_trait]
-impl<B: FileWriterBuilder> IcebergWriter for DataFileWriter<B> {
+impl<B: FileWriterBuilder, L: LocationGenerator, F: FileNameGenerator> IcebergWriter
+ for DataFileWriter<B, L, F>
+{
async fn write(&mut self, batch: RecordBatch) -> Result<()> {
- self.inner_writer.as_mut().unwrap().write(&batch).await
+ self.inner_writer
+ .as_mut()
+ .unwrap()
+ .write(&self.partition_key, &batch)
+ .await
}
async fn close(&mut self) -> Result<Vec<DataFile>> {
- let writer = self.inner_writer.take().unwrap();
- Ok(writer
- .close()
- .await?
- .into_iter()
- .map(|mut res| {
- res.content(DataContentType::Data);
- res.partition(self.partition_value.clone());
- res.partition_spec_id(self.partition_spec_id);
- res.build().expect("Guaranteed to be valid")
- })
- .collect_vec())
+ if let Some(writer) = self.inner_writer.take() {
+ Ok(writer
+ .close()
+ .await?
+ .into_iter()
+ .map(|mut res| {
+ res.content(DataContentType::Data);
+ res.partition(
+ self.partition_key
+ .as_ref()
+ .map_or(Struct::empty(), |pk| pk.data().clone()),
+ );
+ res.partition_spec_id(
+ self.partition_key
+ .as_ref()
+ .map_or(DEFAULT_PARTITION_SPEC_ID, |pk| pk.spec().spec_id()),
+ );
Review Comment:
I think `partition` is a required field in spec: https://iceberg.apache.org/spec/#manifest-entry-fields:~:text=102-,partition,-struct%3C...%3E
We can ignore `partition_spec_id` but the `DataFileBuilder` will fail since macro builder will require it
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] refactor(writer): Refactor writers for the future partitioning writers [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 merged PR #1657:
URL: https://github.com/apache/iceberg-rust/pull/1657
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] refactor(writer): Refactor writers for the future partitioning writers [iceberg-rust]
Posted by "CTTY (via GitHub)" <gi...@apache.org>.
CTTY commented on code in PR #1657:
URL: https://github.com/apache/iceberg-rust/pull/1657#discussion_r2365926622
##########
crates/iceberg/src/writer/file_writer/rolling_writer.rs:
##########
@@ -15,67 +15,138 @@
// specific language governing permissions and limitations
// under the License.
+use std::fmt::{Debug, Formatter};
+
use arrow_array::RecordBatch;
-use crate::spec::DataFileBuilder;
+use crate::io::{FileIO, OutputFile};
+use crate::spec::{DataFileBuilder, PartitionKey};
use crate::writer::CurrentFileStatus;
+use crate::writer::file_writer::location_generator::{FileNameGenerator, LocationGenerator};
use crate::writer::file_writer::{FileWriter, FileWriterBuilder};
use crate::{Error, ErrorKind, Result};
+//
+// /// Builder for creating a `RollingFileWriter` that rolls over to a new file
+// /// when the data size exceeds a target threshold.
+// #[derive(Clone)]
+// pub struct RollingFileWriterBuilder<
Review Comment:
I think we will no longer need a builder for RollingFileWriter since it will be injected into IcebergWriter directly
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] refactor(writer): Refactor writers for the future partitioning writers [iceberg-rust]
Posted by "CTTY (via GitHub)" <gi...@apache.org>.
CTTY commented on code in PR #1657:
URL: https://github.com/apache/iceberg-rust/pull/1657#discussion_r2389958571
##########
crates/iceberg/src/writer/file_writer/rolling_writer.rs:
##########
@@ -15,67 +15,121 @@
// specific language governing permissions and limitations
// under the License.
+use std::fmt::{Debug, Formatter};
+
use arrow_array::RecordBatch;
-use crate::spec::DataFileBuilder;
+use crate::io::{FileIO, OutputFile};
+use crate::spec::{DataFileBuilder, PROPERTY_WRITE_TARGET_FILE_SIZE_BYTES_DEFAULT, PartitionKey};
use crate::writer::CurrentFileStatus;
+use crate::writer::file_writer::location_generator::{FileNameGenerator, LocationGenerator};
use crate::writer::file_writer::{FileWriter, FileWriterBuilder};
use crate::{Error, ErrorKind, Result};
-/// Builder for creating a `RollingFileWriter` that rolls over to a new file
-/// when the data size exceeds a target threshold.
-#[derive(Clone)]
-pub struct RollingFileWriterBuilder<B: FileWriterBuilder> {
+/// A writer that automatically rolls over to a new file when the data size
+/// exceeds a target threshold.
+///
+/// This writer wraps another file writer that tracks the amount of data written.
+/// When the data size exceeds the target size, it closes the current file and
+/// starts writing to a new one.
+pub struct RollingFileWriter<B: FileWriterBuilder, L: LocationGenerator, F: FileNameGenerator> {
+ inner: Option<B::R>,
inner_builder: B,
target_file_size: usize,
+ data_file_builders: Vec<DataFileBuilder>,
+ file_io: FileIO,
+ location_generator: L,
+ file_name_generator: F,
+}
+
+impl<B: FileWriterBuilder, L: LocationGenerator, F: FileNameGenerator> Clone
Review Comment:
Yes I agree, the point is we still have to be able to clone all the writers via builders and I think bringing the RollingFileWriterBuilder back and removing this Clone implementation is the correct way to go. I can make the change tomorrow and we can review that
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] refactor(writer): Refactor writers for the future partitioning writers [iceberg-rust]
Posted by "CTTY (via GitHub)" <gi...@apache.org>.
CTTY commented on code in PR #1657:
URL: https://github.com/apache/iceberg-rust/pull/1657#discussion_r2331461550
##########
crates/iceberg/src/writer/file_writer/rolling_writer.rs:
##########
@@ -17,65 +17,75 @@
use arrow_array::RecordBatch;
-use crate::spec::DataFileBuilder;
-use crate::writer::CurrentFileStatus;
-use crate::writer::file_writer::{FileWriter, FileWriterBuilder};
+use crate::io::FileIO;
+use crate::spec::{DataFile, PartitionKey};
+use crate::writer::file_writer::location_generator::{FileNameGenerator, LocationGenerator};
+use crate::writer::{CurrentFileStatus, IcebergWriter, IcebergWriterBuilder};
use crate::{Error, ErrorKind, Result};
-/// Builder for creating a `RollingFileWriter` that rolls over to a new file
-/// when the data size exceeds a target threshold.
-#[derive(Clone)]
-pub struct RollingFileWriterBuilder<B: FileWriterBuilder> {
+/// A writer that automatically rolls over to a new file when the data size
+/// exceeds a target threshold.
+///
+/// This writer wraps another writer that tracks the amount of data written.
+/// When the data size exceeds the target size, it closes the current file and
+/// starts writing to a new one.
+pub struct RollingWriter<B, L, F>
+where
+ B: IcebergWriterBuilder,
+ L: LocationGenerator,
+ F: FileNameGenerator,
+{
+ inner: Option<B::R>,
inner_builder: B,
target_file_size: usize,
+ location_generator: L,
+ file_name_generator: F,
+ file_io: FileIO,
+ partition_key: Option<PartitionKey>,
+ data_files: Vec<DataFile>, // todo this should be B::R::O? DefaultOutput?
}
-impl<B: FileWriterBuilder> RollingFileWriterBuilder<B> {
- /// Creates a new `RollingFileWriterBuilder` with the specified inner builder and target size.
+impl<B, L, F> RollingWriter<B, L, F>
+where
+ B: IcebergWriterBuilder,
Review Comment:
I think your concern is valid, we may need to expose `I` and `O` in the `RollingWriter` as well, and that should solve this problem?
Meanwhile I've been wondering how useful is the abstraction of `IcebergWriter`... If we separate `RollingWriter` into `RollingPositionalDeletesWriter` and `RollingXXXWriter` and have them use concrete types then this would be a lot easier
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] refactor(writer): Refactor writers for the future partitioning writers [iceberg-rust]
Posted by "CTTY (via GitHub)" <gi...@apache.org>.
CTTY commented on code in PR #1657:
URL: https://github.com/apache/iceberg-rust/pull/1657#discussion_r2334450044
##########
crates/iceberg/src/writer/mod.rs:
##########
@@ -229,20 +229,30 @@ pub mod file_writer;
use arrow_array::RecordBatch;
use crate::Result;
-use crate::spec::DataFile;
+use crate::io::OutputFile;
+use crate::spec::{DataFile, PartitionKey};
type DefaultInput = RecordBatch;
type DefaultOutput = Vec<DataFile>;
+/// The partitioning writer used to write data to multiple partitions.
+pub trait PartitioningWriter {
+ /// Write a record batch, all rows from this record batch should come from one partition
+ fn write(&mut self, partition_key: PartitionKey, batch: RecordBatch) -> Result<()>;
+
+ /// Close all writers and return the data files.
+ fn close(&mut self) -> Result<Vec<DataFile>>;
+}
+
/// The builder for iceberg writer.
#[async_trait::async_trait]
pub trait IcebergWriterBuilder<I = DefaultInput, O = DefaultOutput>:
Review Comment:
I believe we will also need to change the DefaultOutput for `IcebergWriter` from `Vec<DataFile>` to `Vec<DataFileBuilder>` since `IcebergWriter` is no longer the outermost writer
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] refactor(writer): Refactor writers for the future partitioning writers [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 commented on code in PR #1657:
URL: https://github.com/apache/iceberg-rust/pull/1657#discussion_r2367361410
##########
crates/iceberg/src/writer/partitioning/mod.rs:
##########
@@ -0,0 +1,40 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+mod clustered;
+
+use crate::Result;
+use crate::spec::PartitionKey;
+use crate::writer::{DefaultInput, DefaultOutput};
+
+#[async_trait::async_trait]
+pub trait PartitioningWriterBuilder<I = DefaultInput, O = DefaultOutput>:
Review Comment:
I don't think we need a builder for partition writer?
##########
crates/iceberg/src/writer/base_writer/data_file_writer.rs:
##########
@@ -21,54 +21,65 @@ use arrow_array::RecordBatch;
use itertools::Itertools;
use crate::Result;
-use crate::spec::{DataContentType, DataFile, Struct};
-use crate::writer::file_writer::{FileWriter, FileWriterBuilder};
+use crate::spec::{DEFAULT_PARTITION_SPEC_ID, DataContentType, DataFile, PartitionKey, Struct};
+use crate::writer::file_writer::FileWriterBuilder;
+use crate::writer::file_writer::location_generator::{FileNameGenerator, LocationGenerator};
+use crate::writer::file_writer::rolling_writer::RollingFileWriter;
use crate::writer::{CurrentFileStatus, IcebergWriter, IcebergWriterBuilder};
/// Builder for `DataFileWriter`.
#[derive(Clone, Debug)]
-pub struct DataFileWriterBuilder<B: FileWriterBuilder> {
- inner: B,
- partition_value: Option<Struct>,
- partition_spec_id: i32,
+pub struct DataFileWriterBuilder<B: FileWriterBuilder, L: LocationGenerator, F: FileNameGenerator> {
+ inner_writer: RollingFileWriter<B, L, F>,
+ partition_key: Option<PartitionKey>,
}
-impl<B: FileWriterBuilder> DataFileWriterBuilder<B> {
+impl<B: FileWriterBuilder, L: LocationGenerator, F: FileNameGenerator>
+ DataFileWriterBuilder<B, L, F>
+{
/// Create a new `DataFileWriterBuilder` using a `FileWriterBuilder`.
- pub fn new(inner: B, partition_value: Option<Struct>, partition_spec_id: i32) -> Self {
+ pub fn new(
+ inner_writer: RollingFileWriter<B, L, F>,
+ partition_key: Option<PartitionKey>,
+ ) -> Self {
Self {
- inner,
- partition_value,
- partition_spec_id,
+ inner_writer,
+ partition_key,
}
}
}
#[async_trait::async_trait]
-impl<B: FileWriterBuilder> IcebergWriterBuilder for DataFileWriterBuilder<B> {
- type R = DataFileWriter<B>;
+impl<B: FileWriterBuilder, L: LocationGenerator, F: FileNameGenerator> IcebergWriterBuilder
+ for DataFileWriterBuilder<B, L, F>
+{
+ type R = DataFileWriter<B, L, F>;
async fn build(self) -> Result<Self::R> {
Ok(DataFileWriter {
- inner_writer: Some(self.inner.clone().build().await?),
- partition_value: self.partition_value.unwrap_or(Struct::empty()),
- partition_spec_id: self.partition_spec_id,
+ inner_writer: Some(self.inner_writer), // todo revisit this, probably still need a builder for rolling writer
+ partition_key: self.partition_key,
})
}
}
/// A writer write data is within one spec/partition.
#[derive(Debug)]
-pub struct DataFileWriter<B: FileWriterBuilder> {
- inner_writer: Option<B::R>,
- partition_value: Struct,
- partition_spec_id: i32,
+pub struct DataFileWriter<B: FileWriterBuilder, L: LocationGenerator, F: FileNameGenerator> {
+ inner_writer: Option<RollingFileWriter<B, L, F>>,
Review Comment:
This doesn't need to be `Option`?
##########
crates/iceberg/src/writer/file_writer/rolling_writer.rs:
##########
@@ -15,67 +15,138 @@
// specific language governing permissions and limitations
// under the License.
+use std::fmt::{Debug, Formatter};
+
use arrow_array::RecordBatch;
-use crate::spec::DataFileBuilder;
+use crate::io::{FileIO, OutputFile};
+use crate::spec::{DataFileBuilder, PartitionKey};
use crate::writer::CurrentFileStatus;
+use crate::writer::file_writer::location_generator::{FileNameGenerator, LocationGenerator};
use crate::writer::file_writer::{FileWriter, FileWriterBuilder};
use crate::{Error, ErrorKind, Result};
+//
+// /// Builder for creating a `RollingFileWriter` that rolls over to a new file
+// /// when the data size exceeds a target threshold.
+// #[derive(Clone)]
+// pub struct RollingFileWriterBuilder<
Review Comment:
Reasonable to me.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] refactor(writer): Refactor writers for the future partitioning writers [iceberg-rust]
Posted by "CTTY (via GitHub)" <gi...@apache.org>.
CTTY commented on code in PR #1657:
URL: https://github.com/apache/iceberg-rust/pull/1657#discussion_r2383592758
##########
crates/iceberg/src/writer/file_writer/rolling_writer.rs:
##########
@@ -15,67 +15,121 @@
// specific language governing permissions and limitations
// under the License.
+use std::fmt::{Debug, Formatter};
+
use arrow_array::RecordBatch;
-use crate::spec::DataFileBuilder;
+use crate::io::{FileIO, OutputFile};
+use crate::spec::{DataFileBuilder, PROPERTY_WRITE_TARGET_FILE_SIZE_BYTES_DEFAULT, PartitionKey};
use crate::writer::CurrentFileStatus;
+use crate::writer::file_writer::location_generator::{FileNameGenerator, LocationGenerator};
use crate::writer::file_writer::{FileWriter, FileWriterBuilder};
use crate::{Error, ErrorKind, Result};
-/// Builder for creating a `RollingFileWriter` that rolls over to a new file
-/// when the data size exceeds a target threshold.
-#[derive(Clone)]
-pub struct RollingFileWriterBuilder<B: FileWriterBuilder> {
+/// A writer that automatically rolls over to a new file when the data size
+/// exceeds a target threshold.
+///
+/// This writer wraps another file writer that tracks the amount of data written.
+/// When the data size exceeds the target size, it closes the current file and
+/// starts writing to a new one.
+pub struct RollingFileWriter<B: FileWriterBuilder, L: LocationGenerator, F: FileNameGenerator> {
+ inner: Option<B::R>,
inner_builder: B,
target_file_size: usize,
+ data_file_builders: Vec<DataFileBuilder>,
+ file_io: FileIO,
+ location_generator: L,
+ file_name_generator: F,
+}
+
+impl<B: FileWriterBuilder, L: LocationGenerator, F: FileNameGenerator> Clone
Review Comment:
This is because `IcebergWriterBuilder` trait requires `Clone`, so `DataFileWriterBuilder` and the inner `RollingFileWriter` have to derive `Clone`. The `Clone` here is not cloning the writer with its states like data files, but rather a helper to make building new data file writers easier.
I'm thinking maybe it's better to bring back the `RollingFileWriterBuilder`, so we can have something like:
```
pub struct DataFileWriterBuilder<B: FileWriterBuilder, L: LocationGenerator, F: FileNameGenerator> {
// RollingFileWriter won't need to implement Clone because it's Option
inner: Option<RollingFileWriter>,
// builder derives Clone and only contains stateless objs like target_file_size, file_io, location_gen, etc.
inner_builder: RollingFileWriterBuilder<B, L, F>,
partition_key: Option<PartitionKey>,
}
```
wdyt?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] refactor(writer): Refactor writers for the future partitioning writers [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 commented on code in PR #1657:
URL: https://github.com/apache/iceberg-rust/pull/1657#discussion_r2387404485
##########
crates/iceberg/src/writer/file_writer/rolling_writer.rs:
##########
@@ -15,67 +15,121 @@
// specific language governing permissions and limitations
// under the License.
+use std::fmt::{Debug, Formatter};
+
use arrow_array::RecordBatch;
-use crate::spec::DataFileBuilder;
+use crate::io::{FileIO, OutputFile};
+use crate::spec::{DataFileBuilder, PROPERTY_WRITE_TARGET_FILE_SIZE_BYTES_DEFAULT, PartitionKey};
use crate::writer::CurrentFileStatus;
+use crate::writer::file_writer::location_generator::{FileNameGenerator, LocationGenerator};
use crate::writer::file_writer::{FileWriter, FileWriterBuilder};
use crate::{Error, ErrorKind, Result};
-/// Builder for creating a `RollingFileWriter` that rolls over to a new file
-/// when the data size exceeds a target threshold.
-#[derive(Clone)]
-pub struct RollingFileWriterBuilder<B: FileWriterBuilder> {
+/// A writer that automatically rolls over to a new file when the data size
+/// exceeds a target threshold.
+///
+/// This writer wraps another file writer that tracks the amount of data written.
+/// When the data size exceeds the target size, it closes the current file and
+/// starts writing to a new one.
+pub struct RollingFileWriter<B: FileWriterBuilder, L: LocationGenerator, F: FileNameGenerator> {
+ inner: Option<B::R>,
inner_builder: B,
target_file_size: usize,
+ data_file_builders: Vec<DataFileBuilder>,
+ file_io: FileIO,
+ location_generator: L,
+ file_name_generator: F,
+}
+
+impl<B: FileWriterBuilder, L: LocationGenerator, F: FileNameGenerator> Clone
Review Comment:
Or another option is to remove `Clone` in `IcebergWriterBuilder`? I don't see why we need to clone `IcebergWriterBuilder`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] refactor(writer): Refactor writers for the future partitioning writers [iceberg-rust]
Posted by "ZENOTME (via GitHub)" <gi...@apache.org>.
ZENOTME commented on code in PR #1657:
URL: https://github.com/apache/iceberg-rust/pull/1657#discussion_r2352879253
##########
crates/iceberg/src/writer/file_writer/rolling_writer.rs:
##########
@@ -17,65 +17,75 @@
use arrow_array::RecordBatch;
-use crate::spec::DataFileBuilder;
-use crate::writer::CurrentFileStatus;
-use crate::writer::file_writer::{FileWriter, FileWriterBuilder};
+use crate::io::FileIO;
+use crate::spec::{DataFile, PartitionKey};
+use crate::writer::file_writer::location_generator::{FileNameGenerator, LocationGenerator};
+use crate::writer::{CurrentFileStatus, IcebergWriter, IcebergWriterBuilder};
use crate::{Error, ErrorKind, Result};
-/// Builder for creating a `RollingFileWriter` that rolls over to a new file
-/// when the data size exceeds a target threshold.
-#[derive(Clone)]
-pub struct RollingFileWriterBuilder<B: FileWriterBuilder> {
+/// A writer that automatically rolls over to a new file when the data size
+/// exceeds a target threshold.
+///
+/// This writer wraps another writer that tracks the amount of data written.
+/// When the data size exceeds the target size, it closes the current file and
+/// starts writing to a new one.
+pub struct RollingWriter<B, L, F>
+where
+ B: IcebergWriterBuilder,
+ L: LocationGenerator,
+ F: FileNameGenerator,
+{
+ inner: Option<B::R>,
inner_builder: B,
target_file_size: usize,
+ location_generator: L,
+ file_name_generator: F,
+ file_io: FileIO,
+ partition_key: Option<PartitionKey>,
+ data_files: Vec<DataFile>, // todo this should be B::R::O? DefaultOutput?
}
-impl<B: FileWriterBuilder> RollingFileWriterBuilder<B> {
- /// Creates a new `RollingFileWriterBuilder` with the specified inner builder and target size.
+impl<B, L, F> RollingWriter<B, L, F>
+where
+ B: IcebergWriterBuilder,
Review Comment:
> I think custom writers can either implement FileWriter (lightweighted, file-level customization) or PartitioningWriter (heavier, customization across multiple partitions).
E.g. user want to access PositionDeleteInput directly.
```
pub struct RollingWriter<B, L, F, I, O>
where
B: IcebergWriterBuilder<I, O>,
L: LocationGenerator,
F: FileNameGenerator,
```
I think this way can be easier to extend in the future and give more feasible to let user custom. But both way looks good to me if it introduce too much unnecessary complication.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] refactor(writer): Refactor writers for the future partitioning writers [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 commented on PR #1657:
URL: https://github.com/apache/iceberg-rust/pull/1657#issuecomment-3297297522
Hi, @CTTY Seems this is not updated following our discussion?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] refactor(writer): Refactor writers for the future partitioning writers [iceberg-rust]
Posted by "CTTY (via GitHub)" <gi...@apache.org>.
CTTY commented on PR #1657:
URL: https://github.com/apache/iceberg-rust/pull/1657#issuecomment-3300729462
Hi @liurenjie1024 , do you mean that we should also include `TaskWriter` and have `TaskWriter` to split batches by partition? This draft mainly focuses on refactoring the existing layers and have RollingWriter to become the top-level writer as of now, and I haven't incoporated this with an actual partitioning writer or task writer yet. Or do you think it's better to have everything in one draft?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] refactor(writer): Refactor writers for the future partitioning writers [iceberg-rust]
Posted by "CTTY (via GitHub)" <gi...@apache.org>.
CTTY commented on code in PR #1657:
URL: https://github.com/apache/iceberg-rust/pull/1657#discussion_r2345429910
##########
crates/iceberg/src/writer/file_writer/rolling_writer.rs:
##########
@@ -17,65 +17,75 @@
use arrow_array::RecordBatch;
-use crate::spec::DataFileBuilder;
-use crate::writer::CurrentFileStatus;
-use crate::writer::file_writer::{FileWriter, FileWriterBuilder};
+use crate::io::FileIO;
+use crate::spec::{DataFile, PartitionKey};
+use crate::writer::file_writer::location_generator::{FileNameGenerator, LocationGenerator};
+use crate::writer::{CurrentFileStatus, IcebergWriter, IcebergWriterBuilder};
use crate::{Error, ErrorKind, Result};
-/// Builder for creating a `RollingFileWriter` that rolls over to a new file
-/// when the data size exceeds a target threshold.
-#[derive(Clone)]
-pub struct RollingFileWriterBuilder<B: FileWriterBuilder> {
+/// A writer that automatically rolls over to a new file when the data size
+/// exceeds a target threshold.
+///
+/// This writer wraps another writer that tracks the amount of data written.
+/// When the data size exceeds the target size, it closes the current file and
+/// starts writing to a new one.
+pub struct RollingWriter<B, L, F>
+where
+ B: IcebergWriterBuilder,
+ L: LocationGenerator,
+ F: FileNameGenerator,
+{
+ inner: Option<B::R>,
inner_builder: B,
target_file_size: usize,
+ location_generator: L,
+ file_name_generator: F,
+ file_io: FileIO,
+ partition_key: Option<PartitionKey>,
+ data_files: Vec<DataFile>, // todo this should be B::R::O? DefaultOutput?
}
-impl<B: FileWriterBuilder> RollingFileWriterBuilder<B> {
- /// Creates a new `RollingFileWriterBuilder` with the specified inner builder and target size.
+impl<B, L, F> RollingWriter<B, L, F>
+where
+ B: IcebergWriterBuilder,
Review Comment:
I think custom writers can either implement `FileWriter` (lightweighted, file-level customization) or `PartitioningWriter` (heavier, customization across multiple partitions).
In your example, the custom writer can implement FileWriter and be used like this:
```
RollingPositionalDeletesWriter<TrackWriter>
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] refactor(writer): Refactor writers for the future partitioning writers [iceberg-rust]
Posted by "CTTY (via GitHub)" <gi...@apache.org>.
CTTY commented on code in PR #1657:
URL: https://github.com/apache/iceberg-rust/pull/1657#discussion_r2331461550
##########
crates/iceberg/src/writer/file_writer/rolling_writer.rs:
##########
@@ -17,65 +17,75 @@
use arrow_array::RecordBatch;
-use crate::spec::DataFileBuilder;
-use crate::writer::CurrentFileStatus;
-use crate::writer::file_writer::{FileWriter, FileWriterBuilder};
+use crate::io::FileIO;
+use crate::spec::{DataFile, PartitionKey};
+use crate::writer::file_writer::location_generator::{FileNameGenerator, LocationGenerator};
+use crate::writer::{CurrentFileStatus, IcebergWriter, IcebergWriterBuilder};
use crate::{Error, ErrorKind, Result};
-/// Builder for creating a `RollingFileWriter` that rolls over to a new file
-/// when the data size exceeds a target threshold.
-#[derive(Clone)]
-pub struct RollingFileWriterBuilder<B: FileWriterBuilder> {
+/// A writer that automatically rolls over to a new file when the data size
+/// exceeds a target threshold.
+///
+/// This writer wraps another writer that tracks the amount of data written.
+/// When the data size exceeds the target size, it closes the current file and
+/// starts writing to a new one.
+pub struct RollingWriter<B, L, F>
+where
+ B: IcebergWriterBuilder,
+ L: LocationGenerator,
+ F: FileNameGenerator,
+{
+ inner: Option<B::R>,
inner_builder: B,
target_file_size: usize,
+ location_generator: L,
+ file_name_generator: F,
+ file_io: FileIO,
+ partition_key: Option<PartitionKey>,
+ data_files: Vec<DataFile>, // todo this should be B::R::O? DefaultOutput?
}
-impl<B: FileWriterBuilder> RollingFileWriterBuilder<B> {
- /// Creates a new `RollingFileWriterBuilder` with the specified inner builder and target size.
+impl<B, L, F> RollingWriter<B, L, F>
+where
+ B: IcebergWriterBuilder,
Review Comment:
I think your concern is valid, we may need to expose `I` and `O` in the `RollingWriter` as well, and that should solve this problem?
```
pub struct RollingWriter<B, L, F, I, O>
where
B: IcebergWriterBuilder<I, O>,
L: LocationGenerator,
F: FileNameGenerator,
```
Meanwhile I've been wondering how useful is the abstraction of `IcebergWriter`... If we separate `RollingWriter` into `RollingPositionalDeletesWriter` and `RollingXXXWriter` and have them use concrete types then this would be a lot easier
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] refactor(writer): Refactor writers for the future partitioning writers [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 commented on code in PR #1657:
URL: https://github.com/apache/iceberg-rust/pull/1657#discussion_r2381465307
##########
crates/iceberg/src/writer/file_writer/rolling_writer.rs:
##########
@@ -15,67 +15,121 @@
// specific language governing permissions and limitations
// under the License.
+use std::fmt::{Debug, Formatter};
+
use arrow_array::RecordBatch;
-use crate::spec::DataFileBuilder;
+use crate::io::{FileIO, OutputFile};
+use crate::spec::{DataFileBuilder, PROPERTY_WRITE_TARGET_FILE_SIZE_BYTES_DEFAULT, PartitionKey};
use crate::writer::CurrentFileStatus;
+use crate::writer::file_writer::location_generator::{FileNameGenerator, LocationGenerator};
use crate::writer::file_writer::{FileWriter, FileWriterBuilder};
use crate::{Error, ErrorKind, Result};
-/// Builder for creating a `RollingFileWriter` that rolls over to a new file
-/// when the data size exceeds a target threshold.
-#[derive(Clone)]
-pub struct RollingFileWriterBuilder<B: FileWriterBuilder> {
+/// A writer that automatically rolls over to a new file when the data size
+/// exceeds a target threshold.
+///
+/// This writer wraps another file writer that tracks the amount of data written.
+/// When the data size exceeds the target size, it closes the current file and
+/// starts writing to a new one.
+pub struct RollingFileWriter<B: FileWriterBuilder, L: LocationGenerator, F: FileNameGenerator> {
+ inner: Option<B::R>,
inner_builder: B,
target_file_size: usize,
+ data_file_builders: Vec<DataFileBuilder>,
+ file_io: FileIO,
+ location_generator: L,
+ file_name_generator: F,
+}
+
+impl<B: FileWriterBuilder, L: LocationGenerator, F: FileNameGenerator> Clone
Review Comment:
Why we need to implement clone? This is a stateful struct since it contains generated data files, the semantics of clone is unclear.
##########
crates/iceberg/src/writer/base_writer/data_file_writer.rs:
##########
@@ -20,74 +20,101 @@
use arrow_array::RecordBatch;
use itertools::Itertools;
-use crate::Result;
-use crate::spec::{DataContentType, DataFile, Struct};
-use crate::writer::file_writer::{FileWriter, FileWriterBuilder};
+use crate::spec::{DEFAULT_PARTITION_SPEC_ID, DataContentType, DataFile, PartitionKey, Struct};
+use crate::writer::file_writer::FileWriterBuilder;
+use crate::writer::file_writer::location_generator::{FileNameGenerator, LocationGenerator};
+use crate::writer::file_writer::rolling_writer::RollingFileWriter;
use crate::writer::{CurrentFileStatus, IcebergWriter, IcebergWriterBuilder};
+use crate::{Error, ErrorKind, Result};
/// Builder for `DataFileWriter`.
#[derive(Clone, Debug)]
-pub struct DataFileWriterBuilder<B: FileWriterBuilder> {
- inner: B,
- partition_value: Option<Struct>,
- partition_spec_id: i32,
+pub struct DataFileWriterBuilder<B: FileWriterBuilder, L: LocationGenerator, F: FileNameGenerator> {
+ inner_writer: RollingFileWriter<B, L, F>,
+ partition_key: Option<PartitionKey>,
}
-impl<B: FileWriterBuilder> DataFileWriterBuilder<B> {
+impl<B: FileWriterBuilder, L: LocationGenerator, F: FileNameGenerator>
+ DataFileWriterBuilder<B, L, F>
+{
/// Create a new `DataFileWriterBuilder` using a `FileWriterBuilder`.
- pub fn new(inner: B, partition_value: Option<Struct>, partition_spec_id: i32) -> Self {
+ pub fn new(
+ inner_writer: RollingFileWriter<B, L, F>,
+ partition_key: Option<PartitionKey>,
+ ) -> Self {
Self {
- inner,
- partition_value,
- partition_spec_id,
+ inner_writer,
+ partition_key,
}
}
}
#[async_trait::async_trait]
-impl<B: FileWriterBuilder> IcebergWriterBuilder for DataFileWriterBuilder<B> {
- type R = DataFileWriter<B>;
+impl<B: FileWriterBuilder, L: LocationGenerator, F: FileNameGenerator> IcebergWriterBuilder
+ for DataFileWriterBuilder<B, L, F>
+{
+ type R = DataFileWriter<B, L, F>;
async fn build(self) -> Result<Self::R> {
Ok(DataFileWriter {
- inner_writer: Some(self.inner.clone().build().await?),
- partition_value: self.partition_value.unwrap_or(Struct::empty()),
- partition_spec_id: self.partition_spec_id,
+ inner_writer: Some(self.inner_writer),
+ partition_key: self.partition_key,
})
}
}
/// A writer write data is within one spec/partition.
#[derive(Debug)]
-pub struct DataFileWriter<B: FileWriterBuilder> {
- inner_writer: Option<B::R>,
- partition_value: Struct,
- partition_spec_id: i32,
+pub struct DataFileWriter<B: FileWriterBuilder, L: LocationGenerator, F: FileNameGenerator> {
+ inner_writer: Option<RollingFileWriter<B, L, F>>,
+ partition_key: Option<PartitionKey>,
}
#[async_trait::async_trait]
-impl<B: FileWriterBuilder> IcebergWriter for DataFileWriter<B> {
+impl<B: FileWriterBuilder, L: LocationGenerator, F: FileNameGenerator> IcebergWriter
+ for DataFileWriter<B, L, F>
+{
async fn write(&mut self, batch: RecordBatch) -> Result<()> {
- self.inner_writer.as_mut().unwrap().write(&batch).await
+ self.inner_writer
+ .as_mut()
+ .unwrap()
Review Comment:
We should not use `unwrap` here. We could use `if let Some(_)` to make it easier to read.
##########
crates/iceberg/src/writer/base_writer/equality_delete_writer.rs:
##########
@@ -157,8 +168,16 @@ impl<B: FileWriterBuilder> IcebergWriter for EqualityDeleteFileWriter<B> {
.map(|mut res| {
res.content(crate::spec::DataContentType::EqualityDeletes);
res.equality_ids(Some(self.equality_ids.iter().copied().collect_vec()));
- res.partition(self.partition_value.clone());
- res.partition_spec_id(self.partition_spec_id);
+ res.partition(
Review Comment:
Similar question.
##########
crates/iceberg/src/writer/base_writer/data_file_writer.rs:
##########
@@ -20,74 +20,101 @@
use arrow_array::RecordBatch;
use itertools::Itertools;
-use crate::Result;
-use crate::spec::{DataContentType, DataFile, Struct};
-use crate::writer::file_writer::{FileWriter, FileWriterBuilder};
+use crate::spec::{DEFAULT_PARTITION_SPEC_ID, DataContentType, DataFile, PartitionKey, Struct};
+use crate::writer::file_writer::FileWriterBuilder;
+use crate::writer::file_writer::location_generator::{FileNameGenerator, LocationGenerator};
+use crate::writer::file_writer::rolling_writer::RollingFileWriter;
use crate::writer::{CurrentFileStatus, IcebergWriter, IcebergWriterBuilder};
+use crate::{Error, ErrorKind, Result};
/// Builder for `DataFileWriter`.
#[derive(Clone, Debug)]
-pub struct DataFileWriterBuilder<B: FileWriterBuilder> {
- inner: B,
- partition_value: Option<Struct>,
- partition_spec_id: i32,
+pub struct DataFileWriterBuilder<B: FileWriterBuilder, L: LocationGenerator, F: FileNameGenerator> {
+ inner_writer: RollingFileWriter<B, L, F>,
+ partition_key: Option<PartitionKey>,
}
-impl<B: FileWriterBuilder> DataFileWriterBuilder<B> {
+impl<B: FileWriterBuilder, L: LocationGenerator, F: FileNameGenerator>
+ DataFileWriterBuilder<B, L, F>
+{
/// Create a new `DataFileWriterBuilder` using a `FileWriterBuilder`.
- pub fn new(inner: B, partition_value: Option<Struct>, partition_spec_id: i32) -> Self {
+ pub fn new(
+ inner_writer: RollingFileWriter<B, L, F>,
+ partition_key: Option<PartitionKey>,
+ ) -> Self {
Self {
- inner,
- partition_value,
- partition_spec_id,
+ inner_writer,
+ partition_key,
}
}
}
#[async_trait::async_trait]
-impl<B: FileWriterBuilder> IcebergWriterBuilder for DataFileWriterBuilder<B> {
- type R = DataFileWriter<B>;
+impl<B: FileWriterBuilder, L: LocationGenerator, F: FileNameGenerator> IcebergWriterBuilder
+ for DataFileWriterBuilder<B, L, F>
+{
+ type R = DataFileWriter<B, L, F>;
async fn build(self) -> Result<Self::R> {
Ok(DataFileWriter {
- inner_writer: Some(self.inner.clone().build().await?),
- partition_value: self.partition_value.unwrap_or(Struct::empty()),
- partition_spec_id: self.partition_spec_id,
+ inner_writer: Some(self.inner_writer),
+ partition_key: self.partition_key,
})
}
}
/// A writer write data is within one spec/partition.
#[derive(Debug)]
-pub struct DataFileWriter<B: FileWriterBuilder> {
- inner_writer: Option<B::R>,
- partition_value: Struct,
- partition_spec_id: i32,
+pub struct DataFileWriter<B: FileWriterBuilder, L: LocationGenerator, F: FileNameGenerator> {
+ inner_writer: Option<RollingFileWriter<B, L, F>>,
+ partition_key: Option<PartitionKey>,
}
#[async_trait::async_trait]
-impl<B: FileWriterBuilder> IcebergWriter for DataFileWriter<B> {
+impl<B: FileWriterBuilder, L: LocationGenerator, F: FileNameGenerator> IcebergWriter
+ for DataFileWriter<B, L, F>
+{
async fn write(&mut self, batch: RecordBatch) -> Result<()> {
- self.inner_writer.as_mut().unwrap().write(&batch).await
+ self.inner_writer
+ .as_mut()
+ .unwrap()
+ .write(&self.partition_key, &batch)
+ .await
}
async fn close(&mut self) -> Result<Vec<DataFile>> {
- let writer = self.inner_writer.take().unwrap();
- Ok(writer
- .close()
- .await?
- .into_iter()
- .map(|mut res| {
- res.content(DataContentType::Data);
- res.partition(self.partition_value.clone());
- res.partition_spec_id(self.partition_spec_id);
- res.build().expect("Guaranteed to be valid")
- })
- .collect_vec())
+ if let Some(writer) = self.inner_writer.take() {
+ Ok(writer
+ .close()
+ .await?
+ .into_iter()
+ .map(|mut res| {
+ res.content(DataContentType::Data);
+ res.partition(
+ self.partition_key
+ .as_ref()
+ .map_or(Struct::empty(), |pk| pk.data().clone()),
+ );
+ res.partition_spec_id(
+ self.partition_key
+ .as_ref()
+ .map_or(DEFAULT_PARTITION_SPEC_ID, |pk| pk.spec().spec_id()),
+ );
Review Comment:
Do we have to do this? I assume we could skip this for unpartitioned data?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] refactor(writer): Refactor writers for the future partitioning writers [iceberg-rust]
Posted by "CTTY (via GitHub)" <gi...@apache.org>.
CTTY commented on code in PR #1657:
URL: https://github.com/apache/iceberg-rust/pull/1657#discussion_r2400176020
##########
crates/iceberg/src/writer/file_writer/rolling_writer.rs:
##########
@@ -15,67 +15,121 @@
// specific language governing permissions and limitations
// under the License.
+use std::fmt::{Debug, Formatter};
+
use arrow_array::RecordBatch;
-use crate::spec::DataFileBuilder;
+use crate::io::{FileIO, OutputFile};
+use crate::spec::{DataFileBuilder, PROPERTY_WRITE_TARGET_FILE_SIZE_BYTES_DEFAULT, PartitionKey};
use crate::writer::CurrentFileStatus;
+use crate::writer::file_writer::location_generator::{FileNameGenerator, LocationGenerator};
use crate::writer::file_writer::{FileWriter, FileWriterBuilder};
use crate::{Error, ErrorKind, Result};
-/// Builder for creating a `RollingFileWriter` that rolls over to a new file
-/// when the data size exceeds a target threshold.
-#[derive(Clone)]
-pub struct RollingFileWriterBuilder<B: FileWriterBuilder> {
+/// A writer that automatically rolls over to a new file when the data size
+/// exceeds a target threshold.
+///
+/// This writer wraps another file writer that tracks the amount of data written.
+/// When the data size exceeds the target size, it closes the current file and
+/// starts writing to a new one.
+pub struct RollingFileWriter<B: FileWriterBuilder, L: LocationGenerator, F: FileNameGenerator> {
+ inner: Option<B::R>,
inner_builder: B,
target_file_size: usize,
+ data_file_builders: Vec<DataFileBuilder>,
+ file_io: FileIO,
+ location_generator: L,
+ file_name_generator: F,
+}
+
+impl<B: FileWriterBuilder, L: LocationGenerator, F: FileNameGenerator> Clone
Review Comment:
I have updated the code to bring back the `RollingFileWriterBuilder` and removed the `Clone` impl, it should make much more sense now :D
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] refactor(writer): Refactor writers for the future partitioning writers [iceberg-rust]
Posted by "CTTY (via GitHub)" <gi...@apache.org>.
CTTY commented on code in PR #1657:
URL: https://github.com/apache/iceberg-rust/pull/1657#discussion_r2389958571
##########
crates/iceberg/src/writer/file_writer/rolling_writer.rs:
##########
@@ -15,67 +15,121 @@
// specific language governing permissions and limitations
// under the License.
+use std::fmt::{Debug, Formatter};
+
use arrow_array::RecordBatch;
-use crate::spec::DataFileBuilder;
+use crate::io::{FileIO, OutputFile};
+use crate::spec::{DataFileBuilder, PROPERTY_WRITE_TARGET_FILE_SIZE_BYTES_DEFAULT, PartitionKey};
use crate::writer::CurrentFileStatus;
+use crate::writer::file_writer::location_generator::{FileNameGenerator, LocationGenerator};
use crate::writer::file_writer::{FileWriter, FileWriterBuilder};
use crate::{Error, ErrorKind, Result};
-/// Builder for creating a `RollingFileWriter` that rolls over to a new file
-/// when the data size exceeds a target threshold.
-#[derive(Clone)]
-pub struct RollingFileWriterBuilder<B: FileWriterBuilder> {
+/// A writer that automatically rolls over to a new file when the data size
+/// exceeds a target threshold.
+///
+/// This writer wraps another file writer that tracks the amount of data written.
+/// When the data size exceeds the target size, it closes the current file and
+/// starts writing to a new one.
+pub struct RollingFileWriter<B: FileWriterBuilder, L: LocationGenerator, F: FileNameGenerator> {
+ inner: Option<B::R>,
inner_builder: B,
target_file_size: usize,
+ data_file_builders: Vec<DataFileBuilder>,
+ file_io: FileIO,
+ location_generator: L,
+ file_name_generator: F,
+}
+
+impl<B: FileWriterBuilder, L: LocationGenerator, F: FileNameGenerator> Clone
Review Comment:
Yes, I'm thinking of bringing the RollingFileWriterBuilder back and remove this Clone impl
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] refactor(writer): Refactor writers for the future partitioning writers [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 commented on PR #1657:
URL: https://github.com/apache/iceberg-rust/pull/1657#issuecomment-3306441725
> Hi @liurenjie1024 , do you mean that we should also include `TaskWriter` and have `TaskWriter` to split batches by partition? This draft mainly focuses on refactoring the existing layers and have RollingWriter to become the top-level writer as of now, and I haven't incoporated this with an actual partitioning writer or task writer yet. Or do you think it's better to have everything in one draft?
Hi, @CTTY I'm not saying we should include `TaskWriter`. Per our discussion, we should have following dependency:
```
PartitionedWriter
|
DataFileWriter(EqDeleateWriter, PositionDeleteWriter) -> This layer is IcebergWriter
|
RollingFileWriter
|
FileWriter(Parquet, ORC) -> This layer is file format writer
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org