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/07/23 00:51:17 UTC
[PR] feat(core): Implement `RollingFileWriter` to help split data into multiple files [iceberg-rust]
CTTY opened a new pull request, #1547:
URL: https://github.com/apache/iceberg-rust/pull/1547
## 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 #1541
## What changes are included in this PR?
- Added `RollingFileWriter`
- Fix some minor typos in writer mod
<!--
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?
added unit tests
<!--
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] feat(core): Implement `RollingFileWriter` to help split data into multiple files [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 commented on code in PR #1547:
URL: https://github.com/apache/iceberg-rust/pull/1547#discussion_r2228119893
##########
crates/iceberg/src/writer/file_writer/rolling_writer.rs:
##########
@@ -0,0 +1,331 @@
+// 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.
+
+use arrow_array::RecordBatch;
+use futures::future::try_join_all;
+
+use crate::runtime::{JoinHandle, spawn};
+use crate::spec::DataFileBuilder;
+use crate::writer::CurrentFileStatus;
+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> {
+ inner_builder: B,
+ target_size: usize,
+}
+
+impl<B: FileWriterBuilder> RollingFileWriterBuilder<B> {
+ /// Creates a new `RollingFileWriterBuilder` with the specified inner builder and target size.
+ ///
+ /// # Arguments
+ ///
+ /// * `inner_builder` - The builder for the underlying file writer
+ /// * `target_size` - The target size in bytes before rolling over to a new file
+ pub fn new(inner_builder: B, target_size: usize) -> Self {
+ Self {
+ inner_builder,
+ target_size,
+ }
+ }
+}
+
+impl<B: FileWriterBuilder> FileWriterBuilder for RollingFileWriterBuilder<B> {
+ type R = RollingFileWriter<B>;
+
+ async fn build(self) -> Result<Self::R> {
+ Ok(RollingFileWriter {
+ inner: None,
+ inner_builder: self.inner_builder,
+ target_size: self.target_size,
+ close_handles: vec![],
+ })
+ }
+}
+
+/// 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> {
+ inner: Option<B::R>,
+ inner_builder: B,
+ target_size: usize,
+ close_handles: Vec<JoinHandle<Result<Vec<DataFileBuilder>>>>,
+}
+
+impl<B: FileWriterBuilder> RollingFileWriter<B> {
+ /// Determines if the writer should roll over to a new file.
+ ///
+ /// # Arguments
+ ///
+ /// * `input_size` - The size in bytes of the incoming data
+ ///
+ /// # Returns
+ ///
+ /// `true` if a new file should be started, `false` otherwise
+ pub fn should_roll(&self, input_size: usize) -> bool {
+ self.current_written_size() + input_size > self.target_size
+ }
+}
+
+impl<B: FileWriterBuilder> FileWriter for RollingFileWriter<B> {
+ async fn write(&mut self, input: &RecordBatch) -> Result<()> {
+ let input_size = input.get_array_memory_size();
Review Comment:
This is incorrect, usually parquet files written size is much smaller than arrow's in memory array size since parquet will do a lot of compression. The `target_size` is not for exact control, so it's fine to write file a little larger that this size.
##########
crates/iceberg/src/writer/file_writer/rolling_writer.rs:
##########
@@ -0,0 +1,331 @@
+// 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.
+
+use arrow_array::RecordBatch;
+use futures::future::try_join_all;
+
+use crate::runtime::{JoinHandle, spawn};
+use crate::spec::DataFileBuilder;
+use crate::writer::CurrentFileStatus;
+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> {
+ inner_builder: B,
+ target_size: usize,
Review Comment:
```suggestion
target_file_size: usize,
```
##########
crates/iceberg/src/writer/file_writer/rolling_writer.rs:
##########
@@ -0,0 +1,331 @@
+// 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.
+
+use arrow_array::RecordBatch;
+use futures::future::try_join_all;
+
+use crate::runtime::{JoinHandle, spawn};
+use crate::spec::DataFileBuilder;
+use crate::writer::CurrentFileStatus;
+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> {
+ inner_builder: B,
+ target_size: usize,
+}
+
+impl<B: FileWriterBuilder> RollingFileWriterBuilder<B> {
+ /// Creates a new `RollingFileWriterBuilder` with the specified inner builder and target size.
+ ///
+ /// # Arguments
+ ///
+ /// * `inner_builder` - The builder for the underlying file writer
+ /// * `target_size` - The target size in bytes before rolling over to a new file
+ pub fn new(inner_builder: B, target_size: usize) -> Self {
+ Self {
+ inner_builder,
+ target_size,
+ }
+ }
+}
+
+impl<B: FileWriterBuilder> FileWriterBuilder for RollingFileWriterBuilder<B> {
+ type R = RollingFileWriter<B>;
+
+ async fn build(self) -> Result<Self::R> {
+ Ok(RollingFileWriter {
+ inner: None,
+ inner_builder: self.inner_builder,
+ target_size: self.target_size,
+ close_handles: vec![],
+ })
+ }
+}
+
+/// 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> {
+ inner: Option<B::R>,
+ inner_builder: B,
+ target_size: usize,
+ close_handles: Vec<JoinHandle<Result<Vec<DataFileBuilder>>>>,
+}
+
+impl<B: FileWriterBuilder> RollingFileWriter<B> {
+ /// Determines if the writer should roll over to a new file.
+ ///
+ /// # Arguments
+ ///
+ /// * `input_size` - The size in bytes of the incoming data
+ ///
+ /// # Returns
+ ///
+ /// `true` if a new file should be started, `false` otherwise
+ pub fn should_roll(&self, input_size: usize) -> bool {
+ self.current_written_size() + input_size > self.target_size
+ }
+}
+
+impl<B: FileWriterBuilder> FileWriter for RollingFileWriter<B> {
+ async fn write(&mut self, input: &RecordBatch) -> Result<()> {
+ let input_size = input.get_array_memory_size();
+
+ if self.inner.is_none() {
+ // initialize inner writer
+ self.inner = Some(self.inner_builder.clone().build().await?);
+ }
+
+ if self.should_roll(input_size) {
+ if let Some(inner) = self.inner.take() {
+ // close the current writer, roll to a new file
+ let handle = spawn(async move { inner.close().await });
Review Comment:
This is an interesting optimization, but I would suggest not do it for now. A writer usually consumes some resources like memory, connections etc. Closing them in an async approach make things difficult to understand in production, for example, we may have a lot of unclosed writers which consumes a lot memory and leading to oom.
--
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] feat(core): Implement `RollingFileWriter` to help split data into multiple files [iceberg-rust]
Posted by "CTTY (via GitHub)" <gi...@apache.org>.
CTTY commented on code in PR #1547:
URL: https://github.com/apache/iceberg-rust/pull/1547#discussion_r2229380369
##########
crates/iceberg/src/writer/file_writer/rolling_writer.rs:
##########
@@ -0,0 +1,331 @@
+// 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.
+
+use arrow_array::RecordBatch;
+use futures::future::try_join_all;
+
+use crate::runtime::{JoinHandle, spawn};
+use crate::spec::DataFileBuilder;
+use crate::writer::CurrentFileStatus;
+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> {
+ inner_builder: B,
+ target_size: usize,
+}
+
+impl<B: FileWriterBuilder> RollingFileWriterBuilder<B> {
+ /// Creates a new `RollingFileWriterBuilder` with the specified inner builder and target size.
+ ///
+ /// # Arguments
+ ///
+ /// * `inner_builder` - The builder for the underlying file writer
+ /// * `target_size` - The target size in bytes before rolling over to a new file
+ pub fn new(inner_builder: B, target_size: usize) -> Self {
+ Self {
+ inner_builder,
+ target_size,
+ }
+ }
+}
+
+impl<B: FileWriterBuilder> FileWriterBuilder for RollingFileWriterBuilder<B> {
+ type R = RollingFileWriter<B>;
+
+ async fn build(self) -> Result<Self::R> {
+ Ok(RollingFileWriter {
+ inner: None,
+ inner_builder: self.inner_builder,
+ target_size: self.target_size,
+ close_handles: vec![],
+ })
+ }
+}
+
+/// 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> {
+ inner: Option<B::R>,
+ inner_builder: B,
+ target_size: usize,
+ close_handles: Vec<JoinHandle<Result<Vec<DataFileBuilder>>>>,
+}
+
+impl<B: FileWriterBuilder> RollingFileWriter<B> {
+ /// Determines if the writer should roll over to a new file.
+ ///
+ /// # Arguments
+ ///
+ /// * `input_size` - The size in bytes of the incoming data
+ ///
+ /// # Returns
+ ///
+ /// `true` if a new file should be started, `false` otherwise
+ pub fn should_roll(&self, input_size: usize) -> bool {
+ self.current_written_size() + input_size > self.target_size
+ }
+}
+
+impl<B: FileWriterBuilder> FileWriter for RollingFileWriter<B> {
+ async fn write(&mut self, input: &RecordBatch) -> Result<()> {
+ let input_size = input.get_array_memory_size();
Review Comment:
You are right, I added some comment to explain this
--
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] feat(core): Implement `RollingFileWriter` to help split data into multiple files [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 commented on code in PR #1547:
URL: https://github.com/apache/iceberg-rust/pull/1547#discussion_r2233975036
##########
crates/iceberg/src/writer/file_writer/rolling_writer.rs:
##########
@@ -0,0 +1,331 @@
+// 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.
+
+use arrow_array::RecordBatch;
+use futures::future::try_join_all;
+
+use crate::runtime::{JoinHandle, spawn};
+use crate::spec::DataFileBuilder;
+use crate::writer::CurrentFileStatus;
+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> {
+ inner_builder: B,
+ target_size: usize,
+}
+
+impl<B: FileWriterBuilder> RollingFileWriterBuilder<B> {
+ /// Creates a new `RollingFileWriterBuilder` with the specified inner builder and target size.
+ ///
+ /// # Arguments
+ ///
+ /// * `inner_builder` - The builder for the underlying file writer
+ /// * `target_size` - The target size in bytes before rolling over to a new file
+ pub fn new(inner_builder: B, target_size: usize) -> Self {
+ Self {
+ inner_builder,
+ target_size,
+ }
+ }
+}
+
+impl<B: FileWriterBuilder> FileWriterBuilder for RollingFileWriterBuilder<B> {
+ type R = RollingFileWriter<B>;
+
+ async fn build(self) -> Result<Self::R> {
+ Ok(RollingFileWriter {
+ inner: None,
+ inner_builder: self.inner_builder,
+ target_size: self.target_size,
+ close_handles: vec![],
+ })
+ }
+}
+
+/// 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> {
+ inner: Option<B::R>,
+ inner_builder: B,
+ target_size: usize,
+ close_handles: Vec<JoinHandle<Result<Vec<DataFileBuilder>>>>,
+}
+
+impl<B: FileWriterBuilder> RollingFileWriter<B> {
+ /// Determines if the writer should roll over to a new file.
+ ///
+ /// # Arguments
+ ///
+ /// * `input_size` - The size in bytes of the incoming data
+ ///
+ /// # Returns
+ ///
+ /// `true` if a new file should be started, `false` otherwise
+ pub fn should_roll(&self, input_size: usize) -> bool {
+ self.current_written_size() + input_size > self.target_size
+ }
+}
+
+impl<B: FileWriterBuilder> FileWriter for RollingFileWriter<B> {
+ async fn write(&mut self, input: &RecordBatch) -> Result<()> {
+ let input_size = input.get_array_memory_size();
Review Comment:
>To fix this, I think we'll need to change the ParquetWriter's implementation of current_file_size() and use AsyncArrowWrite's in_progress_size to take buffered data into account. But again, in_progress_size is the in-memory size, not the physical size
This sounds reasonable to me. According to [the doc](https://docs.rs/parquet/latest/parquet/arrow/arrow_writer/struct.ArrowWriter.html#method.in_progress_size) , `in_progress_size` + [`bytes_written`](https://docs.rs/parquet/latest/parquet/arrow/arrow_writer/struct.ArrowWriter.html#method.bytes_written) seems a better estimation of the current file size. Due to the complex encoding of parquet, it's hard to get accurate file size before finishing one row group, so an estimation is good enough.
--
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] feat(core): Implement `RollingFileWriter` to help split data into multiple files [iceberg-rust]
Posted by "CTTY (via GitHub)" <gi...@apache.org>.
CTTY commented on code in PR #1547:
URL: https://github.com/apache/iceberg-rust/pull/1547#discussion_r2232078220
##########
crates/iceberg/src/writer/file_writer/rolling_writer.rs:
##########
@@ -0,0 +1,331 @@
+// 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.
+
+use arrow_array::RecordBatch;
+use futures::future::try_join_all;
+
+use crate::runtime::{JoinHandle, spawn};
+use crate::spec::DataFileBuilder;
+use crate::writer::CurrentFileStatus;
+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> {
+ inner_builder: B,
+ target_size: usize,
+}
+
+impl<B: FileWriterBuilder> RollingFileWriterBuilder<B> {
+ /// Creates a new `RollingFileWriterBuilder` with the specified inner builder and target size.
+ ///
+ /// # Arguments
+ ///
+ /// * `inner_builder` - The builder for the underlying file writer
+ /// * `target_size` - The target size in bytes before rolling over to a new file
+ pub fn new(inner_builder: B, target_size: usize) -> Self {
+ Self {
+ inner_builder,
+ target_size,
+ }
+ }
+}
+
+impl<B: FileWriterBuilder> FileWriterBuilder for RollingFileWriterBuilder<B> {
+ type R = RollingFileWriter<B>;
+
+ async fn build(self) -> Result<Self::R> {
+ Ok(RollingFileWriter {
+ inner: None,
+ inner_builder: self.inner_builder,
+ target_size: self.target_size,
+ close_handles: vec![],
+ })
+ }
+}
+
+/// 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> {
+ inner: Option<B::R>,
+ inner_builder: B,
+ target_size: usize,
+ close_handles: Vec<JoinHandle<Result<Vec<DataFileBuilder>>>>,
+}
+
+impl<B: FileWriterBuilder> RollingFileWriter<B> {
+ /// Determines if the writer should roll over to a new file.
+ ///
+ /// # Arguments
+ ///
+ /// * `input_size` - The size in bytes of the incoming data
+ ///
+ /// # Returns
+ ///
+ /// `true` if a new file should be started, `false` otherwise
+ pub fn should_roll(&self, input_size: usize) -> bool {
+ self.current_written_size() + input_size > self.target_size
+ }
+}
+
+impl<B: FileWriterBuilder> FileWriter for RollingFileWriter<B> {
+ async fn write(&mut self, input: &RecordBatch) -> Result<()> {
+ let input_size = input.get_array_memory_size();
Review Comment:
Hi @liurenjie1024 , after testing with the suggested changes, I found an interesting issue:
tldr: the existing `ParquetWriter` can only get the correct `current_written_size` when it's closing and flushing data, not when writing data.
This can cause the following case to fail:
```
let writer: RollingWriter = ...
// should create 1 file
// but won't update current_written_size because we won't close the writer in write()
writer.write(batch1).await?
// if this write should rollover, but since the inner.current_written_size is not updated
// it will try to write the data to the same file as the previous batch
writer.write(batch2).await?
```
A more detailed analysis:
- `ParquetWriter` uses `ArrowAsyncWriter` as its inner writer
- `ArrowAsyncWriter` has _async_writer_ (`ArrowRowGroupWriter`) and _sync_writer_ (`TrackWriter` in this case)
- `ArrowAsyncWriter`'s _sync_writer_ will buffer rows based on the config value `max_row_group_size` (default is 1024 x 1024), causing `TrackWriter` won't be able to track the data in the buffer until closing
Basically this issue can happen a lot when the `max_row_group_size` is large and the `target_file_size` is small.
To fix this, I think we'll need to change the `ParquetWriter`'s implementation of `current_file_size()` and use `AsyncArrowWrite`'s `in_progress_size` to take buffered data into account
--
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] feat(core): Implement `RollingFileWriter` to help split data into multiple files [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 merged PR #1547:
URL: https://github.com/apache/iceberg-rust/pull/1547
--
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] feat(core): Implement `RollingFileWriter` to help split data into multiple files [iceberg-rust]
Posted by "CTTY (via GitHub)" <gi...@apache.org>.
CTTY commented on code in PR #1547:
URL: https://github.com/apache/iceberg-rust/pull/1547#discussion_r2232078220
##########
crates/iceberg/src/writer/file_writer/rolling_writer.rs:
##########
@@ -0,0 +1,331 @@
+// 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.
+
+use arrow_array::RecordBatch;
+use futures::future::try_join_all;
+
+use crate::runtime::{JoinHandle, spawn};
+use crate::spec::DataFileBuilder;
+use crate::writer::CurrentFileStatus;
+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> {
+ inner_builder: B,
+ target_size: usize,
+}
+
+impl<B: FileWriterBuilder> RollingFileWriterBuilder<B> {
+ /// Creates a new `RollingFileWriterBuilder` with the specified inner builder and target size.
+ ///
+ /// # Arguments
+ ///
+ /// * `inner_builder` - The builder for the underlying file writer
+ /// * `target_size` - The target size in bytes before rolling over to a new file
+ pub fn new(inner_builder: B, target_size: usize) -> Self {
+ Self {
+ inner_builder,
+ target_size,
+ }
+ }
+}
+
+impl<B: FileWriterBuilder> FileWriterBuilder for RollingFileWriterBuilder<B> {
+ type R = RollingFileWriter<B>;
+
+ async fn build(self) -> Result<Self::R> {
+ Ok(RollingFileWriter {
+ inner: None,
+ inner_builder: self.inner_builder,
+ target_size: self.target_size,
+ close_handles: vec![],
+ })
+ }
+}
+
+/// 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> {
+ inner: Option<B::R>,
+ inner_builder: B,
+ target_size: usize,
+ close_handles: Vec<JoinHandle<Result<Vec<DataFileBuilder>>>>,
+}
+
+impl<B: FileWriterBuilder> RollingFileWriter<B> {
+ /// Determines if the writer should roll over to a new file.
+ ///
+ /// # Arguments
+ ///
+ /// * `input_size` - The size in bytes of the incoming data
+ ///
+ /// # Returns
+ ///
+ /// `true` if a new file should be started, `false` otherwise
+ pub fn should_roll(&self, input_size: usize) -> bool {
+ self.current_written_size() + input_size > self.target_size
+ }
+}
+
+impl<B: FileWriterBuilder> FileWriter for RollingFileWriter<B> {
+ async fn write(&mut self, input: &RecordBatch) -> Result<()> {
+ let input_size = input.get_array_memory_size();
Review Comment:
Hi @liurenjie1024 , after testing with the suggested changes, I found an interesting issue:
tldr: the existing `ParquetWriter` can only get the correct `current_written_size` when it's closing and flushing data, not when writing data.
This can cause the following case to fail:
```
let writer: RollingWriter = ...
// should create 1 file
// but won't update current_written_size because we won't close the writer in write()
writer.write(batch1).await?
// if this write should rollover, but since the inner.current_written_size is not updated
// it will try to write the data to the same file as the previous batch
writer.write(batch2).await?
```
A more detailed analysis:
- `ParquetWriter` uses `ArrowAsyncWriter` as its inner writer
- `ArrowAsyncWriter` has _async_writer_ (`ArrowRowGroupWriter`) and _sync_writer_ (`TrackWriter` in this case)
- `ArrowAsyncWriter`'s _sync_writer_ will buffer rows based on the config value `max_row_group_size` (default is 1024 x 1024), causing `TrackWriter` won't be able to track the data in the buffer until closing
Basically this issue can happen a lot when the `max_row_group_size` is large and the `target_file_size` is small.
To fix this, I think we'll need to change the `ParquetWriter`'s implementation of `current_file_size()` and use `AsyncArrowWrite`'s `in_progress_size` to take buffered data into account. But again, `in_progress_size` is the in-memory size, not the physical size
--
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] feat(core): Implement `RollingFileWriter` to help split data into multiple files [iceberg-rust]
Posted by "CTTY (via GitHub)" <gi...@apache.org>.
CTTY commented on code in PR #1547:
URL: https://github.com/apache/iceberg-rust/pull/1547#discussion_r2229431027
##########
crates/iceberg/src/writer/file_writer/rolling_writer.rs:
##########
@@ -0,0 +1,331 @@
+// 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.
+
+use arrow_array::RecordBatch;
+use futures::future::try_join_all;
+
+use crate::runtime::{JoinHandle, spawn};
+use crate::spec::DataFileBuilder;
+use crate::writer::CurrentFileStatus;
+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> {
+ inner_builder: B,
+ target_size: usize,
+}
+
+impl<B: FileWriterBuilder> RollingFileWriterBuilder<B> {
+ /// Creates a new `RollingFileWriterBuilder` with the specified inner builder and target size.
+ ///
+ /// # Arguments
+ ///
+ /// * `inner_builder` - The builder for the underlying file writer
+ /// * `target_size` - The target size in bytes before rolling over to a new file
+ pub fn new(inner_builder: B, target_size: usize) -> Self {
+ Self {
+ inner_builder,
+ target_size,
+ }
+ }
+}
+
+impl<B: FileWriterBuilder> FileWriterBuilder for RollingFileWriterBuilder<B> {
+ type R = RollingFileWriter<B>;
+
+ async fn build(self) -> Result<Self::R> {
+ Ok(RollingFileWriter {
+ inner: None,
+ inner_builder: self.inner_builder,
+ target_size: self.target_size,
+ close_handles: vec![],
+ })
+ }
+}
+
+/// 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> {
+ inner: Option<B::R>,
+ inner_builder: B,
+ target_size: usize,
+ close_handles: Vec<JoinHandle<Result<Vec<DataFileBuilder>>>>,
+}
+
+impl<B: FileWriterBuilder> RollingFileWriter<B> {
+ /// Determines if the writer should roll over to a new file.
+ ///
+ /// # Arguments
+ ///
+ /// * `input_size` - The size in bytes of the incoming data
+ ///
+ /// # Returns
+ ///
+ /// `true` if a new file should be started, `false` otherwise
+ pub fn should_roll(&self, input_size: usize) -> bool {
+ self.current_written_size() + input_size > self.target_size
+ }
+}
+
+impl<B: FileWriterBuilder> FileWriter for RollingFileWriter<B> {
+ async fn write(&mut self, input: &RecordBatch) -> Result<()> {
+ let input_size = input.get_array_memory_size();
+
+ if self.inner.is_none() {
+ // initialize inner writer
+ self.inner = Some(self.inner_builder.clone().build().await?);
+ }
+
+ if self.should_roll(input_size) {
+ if let Some(inner) = self.inner.take() {
+ // close the current writer, roll to a new file
+ let handle = spawn(async move { inner.close().await });
Review Comment:
I've removed the `close_handles` from this PR and created an issue to track this potential optimization: https://github.com/apache/iceberg-rust/issues/1551
--
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] feat(core): Implement `RollingFileWriter` to help split data into multiple files [iceberg-rust]
Posted by "CTTY (via GitHub)" <gi...@apache.org>.
CTTY commented on code in PR #1547:
URL: https://github.com/apache/iceberg-rust/pull/1547#discussion_r2234203421
##########
crates/iceberg/src/writer/file_writer/rolling_writer.rs:
##########
@@ -0,0 +1,331 @@
+// 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.
+
+use arrow_array::RecordBatch;
+use futures::future::try_join_all;
+
+use crate::runtime::{JoinHandle, spawn};
+use crate::spec::DataFileBuilder;
+use crate::writer::CurrentFileStatus;
+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> {
+ inner_builder: B,
+ target_size: usize,
+}
+
+impl<B: FileWriterBuilder> RollingFileWriterBuilder<B> {
+ /// Creates a new `RollingFileWriterBuilder` with the specified inner builder and target size.
+ ///
+ /// # Arguments
+ ///
+ /// * `inner_builder` - The builder for the underlying file writer
+ /// * `target_size` - The target size in bytes before rolling over to a new file
+ pub fn new(inner_builder: B, target_size: usize) -> Self {
+ Self {
+ inner_builder,
+ target_size,
+ }
+ }
+}
+
+impl<B: FileWriterBuilder> FileWriterBuilder for RollingFileWriterBuilder<B> {
+ type R = RollingFileWriter<B>;
+
+ async fn build(self) -> Result<Self::R> {
+ Ok(RollingFileWriter {
+ inner: None,
+ inner_builder: self.inner_builder,
+ target_size: self.target_size,
+ close_handles: vec![],
+ })
+ }
+}
+
+/// 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> {
+ inner: Option<B::R>,
+ inner_builder: B,
+ target_size: usize,
+ close_handles: Vec<JoinHandle<Result<Vec<DataFileBuilder>>>>,
+}
+
+impl<B: FileWriterBuilder> RollingFileWriter<B> {
+ /// Determines if the writer should roll over to a new file.
+ ///
+ /// # Arguments
+ ///
+ /// * `input_size` - The size in bytes of the incoming data
+ ///
+ /// # Returns
+ ///
+ /// `true` if a new file should be started, `false` otherwise
+ pub fn should_roll(&self, input_size: usize) -> bool {
+ self.current_written_size() + input_size > self.target_size
+ }
+}
+
+impl<B: FileWriterBuilder> FileWriter for RollingFileWriter<B> {
+ async fn write(&mut self, input: &RecordBatch) -> Result<()> {
+ let input_size = input.get_array_memory_size();
Review Comment:
I've created an [issue](https://github.com/apache/iceberg-rust/issues/1554) and will fix the parquet writer behavior in a separate PR
--
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] feat(core): Implement `RollingFileWriter` to help split data into multiple files [iceberg-rust]
Posted by "CTTY (via GitHub)" <gi...@apache.org>.
CTTY commented on PR #1547:
URL: https://github.com/apache/iceberg-rust/pull/1547#issuecomment-3130407854
Hi @stevie9868 , I hope my [reply](https://github.com/apache/iceberg-rust/issues/1382#issuecomment-3130404115) in a different thread can answer your question
--
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] feat(core): Implement `RollingFileWriter` to help split data into multiple files [iceberg-rust]
Posted by "CTTY (via GitHub)" <gi...@apache.org>.
CTTY commented on code in PR #1547:
URL: https://github.com/apache/iceberg-rust/pull/1547#discussion_r2229405980
##########
crates/iceberg/src/writer/file_writer/rolling_writer.rs:
##########
@@ -0,0 +1,331 @@
+// 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.
+
+use arrow_array::RecordBatch;
+use futures::future::try_join_all;
+
+use crate::runtime::{JoinHandle, spawn};
+use crate::spec::DataFileBuilder;
+use crate::writer::CurrentFileStatus;
+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> {
+ inner_builder: B,
+ target_size: usize,
+}
+
+impl<B: FileWriterBuilder> RollingFileWriterBuilder<B> {
+ /// Creates a new `RollingFileWriterBuilder` with the specified inner builder and target size.
+ ///
+ /// # Arguments
+ ///
+ /// * `inner_builder` - The builder for the underlying file writer
+ /// * `target_size` - The target size in bytes before rolling over to a new file
+ pub fn new(inner_builder: B, target_size: usize) -> Self {
+ Self {
+ inner_builder,
+ target_size,
+ }
+ }
+}
+
+impl<B: FileWriterBuilder> FileWriterBuilder for RollingFileWriterBuilder<B> {
+ type R = RollingFileWriter<B>;
+
+ async fn build(self) -> Result<Self::R> {
+ Ok(RollingFileWriter {
+ inner: None,
+ inner_builder: self.inner_builder,
+ target_size: self.target_size,
+ close_handles: vec![],
+ })
+ }
+}
+
+/// 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> {
+ inner: Option<B::R>,
+ inner_builder: B,
+ target_size: usize,
+ close_handles: Vec<JoinHandle<Result<Vec<DataFileBuilder>>>>,
+}
+
+impl<B: FileWriterBuilder> RollingFileWriter<B> {
+ /// Determines if the writer should roll over to a new file.
+ ///
+ /// # Arguments
+ ///
+ /// * `input_size` - The size in bytes of the incoming data
+ ///
+ /// # Returns
+ ///
+ /// `true` if a new file should be started, `false` otherwise
+ pub fn should_roll(&self, input_size: usize) -> bool {
+ self.current_written_size() + input_size > self.target_size
+ }
+}
+
+impl<B: FileWriterBuilder> FileWriter for RollingFileWriter<B> {
+ async fn write(&mut self, input: &RecordBatch) -> Result<()> {
+ let input_size = input.get_array_memory_size();
+
+ if self.inner.is_none() {
+ // initialize inner writer
+ self.inner = Some(self.inner_builder.clone().build().await?);
+ }
+
+ if self.should_roll(input_size) {
+ if let Some(inner) = self.inner.take() {
+ // close the current writer, roll to a new file
+ let handle = spawn(async move { inner.close().await });
Review Comment:
Good point!
I have some rough idea to further improving this: we can use a config to control the maximum parallelism here:
```
struct RollingWriter {
...
buffer: Vec<DataFile>
}
...
while close_handles.len() >= self.max_parallelism() {
// wait until some closers complete, and store the data files in a buffer
self.buffer.push(future::select(self.close_handles))
}
self.close_handles.push(new_handle);
```
I have not thought very clearly on how to prevent `buffer` from eating up the memory as of now.
Either way I agree this can be completed as a follow-up
--
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] feat(core): Implement `RollingFileWriter` to help split data into multiple files [iceberg-rust]
Posted by "CTTY (via GitHub)" <gi...@apache.org>.
CTTY commented on code in PR #1547:
URL: https://github.com/apache/iceberg-rust/pull/1547#discussion_r2229405980
##########
crates/iceberg/src/writer/file_writer/rolling_writer.rs:
##########
@@ -0,0 +1,331 @@
+// 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.
+
+use arrow_array::RecordBatch;
+use futures::future::try_join_all;
+
+use crate::runtime::{JoinHandle, spawn};
+use crate::spec::DataFileBuilder;
+use crate::writer::CurrentFileStatus;
+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> {
+ inner_builder: B,
+ target_size: usize,
+}
+
+impl<B: FileWriterBuilder> RollingFileWriterBuilder<B> {
+ /// Creates a new `RollingFileWriterBuilder` with the specified inner builder and target size.
+ ///
+ /// # Arguments
+ ///
+ /// * `inner_builder` - The builder for the underlying file writer
+ /// * `target_size` - The target size in bytes before rolling over to a new file
+ pub fn new(inner_builder: B, target_size: usize) -> Self {
+ Self {
+ inner_builder,
+ target_size,
+ }
+ }
+}
+
+impl<B: FileWriterBuilder> FileWriterBuilder for RollingFileWriterBuilder<B> {
+ type R = RollingFileWriter<B>;
+
+ async fn build(self) -> Result<Self::R> {
+ Ok(RollingFileWriter {
+ inner: None,
+ inner_builder: self.inner_builder,
+ target_size: self.target_size,
+ close_handles: vec![],
+ })
+ }
+}
+
+/// 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> {
+ inner: Option<B::R>,
+ inner_builder: B,
+ target_size: usize,
+ close_handles: Vec<JoinHandle<Result<Vec<DataFileBuilder>>>>,
+}
+
+impl<B: FileWriterBuilder> RollingFileWriter<B> {
+ /// Determines if the writer should roll over to a new file.
+ ///
+ /// # Arguments
+ ///
+ /// * `input_size` - The size in bytes of the incoming data
+ ///
+ /// # Returns
+ ///
+ /// `true` if a new file should be started, `false` otherwise
+ pub fn should_roll(&self, input_size: usize) -> bool {
+ self.current_written_size() + input_size > self.target_size
+ }
+}
+
+impl<B: FileWriterBuilder> FileWriter for RollingFileWriter<B> {
+ async fn write(&mut self, input: &RecordBatch) -> Result<()> {
+ let input_size = input.get_array_memory_size();
+
+ if self.inner.is_none() {
+ // initialize inner writer
+ self.inner = Some(self.inner_builder.clone().build().await?);
+ }
+
+ if self.should_roll(input_size) {
+ if let Some(inner) = self.inner.take() {
+ // close the current writer, roll to a new file
+ let handle = spawn(async move { inner.close().await });
Review Comment:
Good point!
I have some rough idea to further improve this: we can use a config to control the maximum parallelism here:
```
struct RollingWriter {
...
buffer: Vec<DataFileBuilder>
}
...
while close_handles.len() >= self.max_parallelism() {
// wait until some closers complete, and store the data files in a buffer
self.buffer.extend(future::select(self.close_handles))
}
self.close_handles.push(new_handle);
```
I have not thought very clearly on how to prevent `buffer` from eating up the memory as of now, or do we even need it?
Either way I agree this can be completed as a follow-up
--
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] feat(core): Implement `RollingFileWriter` to help split data into multiple files [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 commented on code in PR #1547:
URL: https://github.com/apache/iceberg-rust/pull/1547#discussion_r2230578083
##########
crates/iceberg/src/writer/file_writer/rolling_writer.rs:
##########
@@ -0,0 +1,331 @@
+// 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.
+
+use arrow_array::RecordBatch;
+use futures::future::try_join_all;
+
+use crate::runtime::{JoinHandle, spawn};
+use crate::spec::DataFileBuilder;
+use crate::writer::CurrentFileStatus;
+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> {
+ inner_builder: B,
+ target_size: usize,
+}
+
+impl<B: FileWriterBuilder> RollingFileWriterBuilder<B> {
+ /// Creates a new `RollingFileWriterBuilder` with the specified inner builder and target size.
+ ///
+ /// # Arguments
+ ///
+ /// * `inner_builder` - The builder for the underlying file writer
+ /// * `target_size` - The target size in bytes before rolling over to a new file
+ pub fn new(inner_builder: B, target_size: usize) -> Self {
+ Self {
+ inner_builder,
+ target_size,
+ }
+ }
+}
+
+impl<B: FileWriterBuilder> FileWriterBuilder for RollingFileWriterBuilder<B> {
+ type R = RollingFileWriter<B>;
+
+ async fn build(self) -> Result<Self::R> {
+ Ok(RollingFileWriter {
+ inner: None,
+ inner_builder: self.inner_builder,
+ target_size: self.target_size,
+ close_handles: vec![],
+ })
+ }
+}
+
+/// 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> {
+ inner: Option<B::R>,
+ inner_builder: B,
+ target_size: usize,
+ close_handles: Vec<JoinHandle<Result<Vec<DataFileBuilder>>>>,
+}
+
+impl<B: FileWriterBuilder> RollingFileWriter<B> {
+ /// Determines if the writer should roll over to a new file.
+ ///
+ /// # Arguments
+ ///
+ /// * `input_size` - The size in bytes of the incoming data
+ ///
+ /// # Returns
+ ///
+ /// `true` if a new file should be started, `false` otherwise
+ pub fn should_roll(&self, input_size: usize) -> bool {
+ self.current_written_size() + input_size > self.target_size
+ }
+}
+
+impl<B: FileWriterBuilder> FileWriter for RollingFileWriter<B> {
+ async fn write(&mut self, input: &RecordBatch) -> Result<()> {
+ let input_size = input.get_array_memory_size();
Review Comment:
No, I mean we should not use this `input_size` to determine if we should roll, but the writer's `current_written_size` only.
--
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] feat(core): Implement `RollingFileWriter` to help split data into multiple files [iceberg-rust]
Posted by "CTTY (via GitHub)" <gi...@apache.org>.
CTTY commented on code in PR #1547:
URL: https://github.com/apache/iceberg-rust/pull/1547#discussion_r2229405980
##########
crates/iceberg/src/writer/file_writer/rolling_writer.rs:
##########
@@ -0,0 +1,331 @@
+// 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.
+
+use arrow_array::RecordBatch;
+use futures::future::try_join_all;
+
+use crate::runtime::{JoinHandle, spawn};
+use crate::spec::DataFileBuilder;
+use crate::writer::CurrentFileStatus;
+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> {
+ inner_builder: B,
+ target_size: usize,
+}
+
+impl<B: FileWriterBuilder> RollingFileWriterBuilder<B> {
+ /// Creates a new `RollingFileWriterBuilder` with the specified inner builder and target size.
+ ///
+ /// # Arguments
+ ///
+ /// * `inner_builder` - The builder for the underlying file writer
+ /// * `target_size` - The target size in bytes before rolling over to a new file
+ pub fn new(inner_builder: B, target_size: usize) -> Self {
+ Self {
+ inner_builder,
+ target_size,
+ }
+ }
+}
+
+impl<B: FileWriterBuilder> FileWriterBuilder for RollingFileWriterBuilder<B> {
+ type R = RollingFileWriter<B>;
+
+ async fn build(self) -> Result<Self::R> {
+ Ok(RollingFileWriter {
+ inner: None,
+ inner_builder: self.inner_builder,
+ target_size: self.target_size,
+ close_handles: vec![],
+ })
+ }
+}
+
+/// 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> {
+ inner: Option<B::R>,
+ inner_builder: B,
+ target_size: usize,
+ close_handles: Vec<JoinHandle<Result<Vec<DataFileBuilder>>>>,
+}
+
+impl<B: FileWriterBuilder> RollingFileWriter<B> {
+ /// Determines if the writer should roll over to a new file.
+ ///
+ /// # Arguments
+ ///
+ /// * `input_size` - The size in bytes of the incoming data
+ ///
+ /// # Returns
+ ///
+ /// `true` if a new file should be started, `false` otherwise
+ pub fn should_roll(&self, input_size: usize) -> bool {
+ self.current_written_size() + input_size > self.target_size
+ }
+}
+
+impl<B: FileWriterBuilder> FileWriter for RollingFileWriter<B> {
+ async fn write(&mut self, input: &RecordBatch) -> Result<()> {
+ let input_size = input.get_array_memory_size();
+
+ if self.inner.is_none() {
+ // initialize inner writer
+ self.inner = Some(self.inner_builder.clone().build().await?);
+ }
+
+ if self.should_roll(input_size) {
+ if let Some(inner) = self.inner.take() {
+ // close the current writer, roll to a new file
+ let handle = spawn(async move { inner.close().await });
Review Comment:
Good point!
I have some rough idea to further improving this: we can use a config to control the maximum parallelism here:
```
struct RollingWriter {
...
buffer: Vec<DataFileBuilder>
}
...
while close_handles.len() >= self.max_parallelism() {
// wait until some closers complete, and store the data files in a buffer
self.buffer.push(future::select(self.close_handles))
}
self.close_handles.push(new_handle);
```
I have not thought very clearly on how to prevent `buffer` from eating up the memory as of now, or do we even need it?
Either way I agree this can be completed as a follow-up
##########
crates/iceberg/src/writer/file_writer/rolling_writer.rs:
##########
@@ -0,0 +1,331 @@
+// 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.
+
+use arrow_array::RecordBatch;
+use futures::future::try_join_all;
+
+use crate::runtime::{JoinHandle, spawn};
+use crate::spec::DataFileBuilder;
+use crate::writer::CurrentFileStatus;
+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> {
+ inner_builder: B,
+ target_size: usize,
+}
+
+impl<B: FileWriterBuilder> RollingFileWriterBuilder<B> {
+ /// Creates a new `RollingFileWriterBuilder` with the specified inner builder and target size.
+ ///
+ /// # Arguments
+ ///
+ /// * `inner_builder` - The builder for the underlying file writer
+ /// * `target_size` - The target size in bytes before rolling over to a new file
+ pub fn new(inner_builder: B, target_size: usize) -> Self {
+ Self {
+ inner_builder,
+ target_size,
+ }
+ }
+}
+
+impl<B: FileWriterBuilder> FileWriterBuilder for RollingFileWriterBuilder<B> {
+ type R = RollingFileWriter<B>;
+
+ async fn build(self) -> Result<Self::R> {
+ Ok(RollingFileWriter {
+ inner: None,
+ inner_builder: self.inner_builder,
+ target_size: self.target_size,
+ close_handles: vec![],
+ })
+ }
+}
+
+/// 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> {
+ inner: Option<B::R>,
+ inner_builder: B,
+ target_size: usize,
+ close_handles: Vec<JoinHandle<Result<Vec<DataFileBuilder>>>>,
+}
+
+impl<B: FileWriterBuilder> RollingFileWriter<B> {
+ /// Determines if the writer should roll over to a new file.
+ ///
+ /// # Arguments
+ ///
+ /// * `input_size` - The size in bytes of the incoming data
+ ///
+ /// # Returns
+ ///
+ /// `true` if a new file should be started, `false` otherwise
+ pub fn should_roll(&self, input_size: usize) -> bool {
+ self.current_written_size() + input_size > self.target_size
+ }
+}
+
+impl<B: FileWriterBuilder> FileWriter for RollingFileWriter<B> {
+ async fn write(&mut self, input: &RecordBatch) -> Result<()> {
+ let input_size = input.get_array_memory_size();
+
+ if self.inner.is_none() {
+ // initialize inner writer
+ self.inner = Some(self.inner_builder.clone().build().await?);
+ }
+
+ if self.should_roll(input_size) {
+ if let Some(inner) = self.inner.take() {
+ // close the current writer, roll to a new file
+ let handle = spawn(async move { inner.close().await });
Review Comment:
Good point!
I have some rough idea to further improving this: we can use a config to control the maximum parallelism here:
```
struct RollingWriter {
...
buffer: Vec<DataFileBuilder>
}
...
while close_handles.len() >= self.max_parallelism() {
// wait until some closers complete, and store the data files in a buffer
self.buffer.extend(future::select(self.close_handles))
}
self.close_handles.push(new_handle);
```
I have not thought very clearly on how to prevent `buffer` from eating up the memory as of now, or do we even need it?
Either way I agree this can be completed as a follow-up
--
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] feat(core): Implement `RollingFileWriter` to help split data into multiple files [iceberg-rust]
Posted by "CTTY (via GitHub)" <gi...@apache.org>.
CTTY commented on PR #1547:
URL: https://github.com/apache/iceberg-rust/pull/1547#issuecomment-3130017652
I've manually tested the rolling writer using some small data with the [ParquetWriter::current_written_size fix](https://github.com/apache/iceberg-rust/issues/1554). Now the generated file size is much closer to the configured `target_file_size`.
The difference between the configured `target_file_size` and the actual written file size can vary depending on the batch size. Generally speaking, rolling file writer rolls more precisely when the target file size is much larger compared to the size of each batch, which is expected.
| target_file_size | batch_rows |batch_size_in_memory (Bytes) | Written File Size (Bytes) |
|---------|--------|-------| --- |
| 10 MB | 500K | 8194584 | 12968410 |
| 1 MB | 50K | 662424 | 1077332 |
| 300 KB | 10K |145816 | 312345 |
| 30 KB | 1K | 16440 | 29595 |
| 30 KB | 500 | 8360| 28800 |
| 30 KB | 100 | 1576| 28218 |
--
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] feat(core): Implement `RollingFileWriter` to help split data into multiple files [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 commented on PR #1547:
URL: https://github.com/apache/iceberg-rust/pull/1547#issuecomment-3131470917
> I've manually tested the rolling writer using some small data with the [ParquetWriter::current_written_size fix](https://github.com/apache/iceberg-rust/issues/1554). Now the generated file size is much closer to the configured `target_file_size`.
>
> The difference between the configured `target_file_size` and the actual written file size can vary depending on the batch size. Generally speaking, rolling file writer rolls more precisely when the target file size is much larger compared to the size of each batch, which is expected.
>
> target_file_size batch_rows batch_size_in_memory (Bytes) Written File Size (Bytes)
> 10 MB 500K 8194584 12968410
> 1 MB 50K 662424 1077332
> 300 KB 10K 145816 312345
> 30 KB 1K 16440 29595
> 30 KB 500 8360 28800
> 30 KB 100 1576 28218
Thanks @CTTY for the tests. The difference are acceptable 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] feat(core): Implement `RollingFileWriter` to help split data into multiple files [iceberg-rust]
Posted by "CTTY (via GitHub)" <gi...@apache.org>.
CTTY commented on code in PR #1547:
URL: https://github.com/apache/iceberg-rust/pull/1547#discussion_r2232078220
##########
crates/iceberg/src/writer/file_writer/rolling_writer.rs:
##########
@@ -0,0 +1,331 @@
+// 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.
+
+use arrow_array::RecordBatch;
+use futures::future::try_join_all;
+
+use crate::runtime::{JoinHandle, spawn};
+use crate::spec::DataFileBuilder;
+use crate::writer::CurrentFileStatus;
+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> {
+ inner_builder: B,
+ target_size: usize,
+}
+
+impl<B: FileWriterBuilder> RollingFileWriterBuilder<B> {
+ /// Creates a new `RollingFileWriterBuilder` with the specified inner builder and target size.
+ ///
+ /// # Arguments
+ ///
+ /// * `inner_builder` - The builder for the underlying file writer
+ /// * `target_size` - The target size in bytes before rolling over to a new file
+ pub fn new(inner_builder: B, target_size: usize) -> Self {
+ Self {
+ inner_builder,
+ target_size,
+ }
+ }
+}
+
+impl<B: FileWriterBuilder> FileWriterBuilder for RollingFileWriterBuilder<B> {
+ type R = RollingFileWriter<B>;
+
+ async fn build(self) -> Result<Self::R> {
+ Ok(RollingFileWriter {
+ inner: None,
+ inner_builder: self.inner_builder,
+ target_size: self.target_size,
+ close_handles: vec![],
+ })
+ }
+}
+
+/// 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> {
+ inner: Option<B::R>,
+ inner_builder: B,
+ target_size: usize,
+ close_handles: Vec<JoinHandle<Result<Vec<DataFileBuilder>>>>,
+}
+
+impl<B: FileWriterBuilder> RollingFileWriter<B> {
+ /// Determines if the writer should roll over to a new file.
+ ///
+ /// # Arguments
+ ///
+ /// * `input_size` - The size in bytes of the incoming data
+ ///
+ /// # Returns
+ ///
+ /// `true` if a new file should be started, `false` otherwise
+ pub fn should_roll(&self, input_size: usize) -> bool {
+ self.current_written_size() + input_size > self.target_size
+ }
+}
+
+impl<B: FileWriterBuilder> FileWriter for RollingFileWriter<B> {
+ async fn write(&mut self, input: &RecordBatch) -> Result<()> {
+ let input_size = input.get_array_memory_size();
Review Comment:
Hi @liurenjie1024 , after testing with the suggested changes, I found an interesting issue:
tldr: the existing `ParquetWriter` can only get the correct `current_written_size` when it's closing and flushing data, not when writing data.
This can cause the following case to fail:
```
let writer: RollingWriter = ...
// should create 1 file
// but won't update current_written_size because we won't close the writer in write()
writer.write(batch1).await?
// if this write should rollover, but since the inner.current_written_size is not updated
// it will try to write the data to the same file as the previous batch
writer.write(batch2).await?
```
A more detailed analysis:
- `ParquetWriter` uses `ArrowAsyncWriter` as its inner writer
- `ArrowAsyncWriter` has _async_writer_ (`ArrowRowGroupWriter`) and _sync_writer_ (`TrackWriter` in this case)
- `ArrowAsyncWriter`'s _sync_writer_ will buffer rows based on the config value `max_row_group_size` (default is 1024 x 1024), causing `TrackWriter` won't be able to track the data in the buffer until closing
Basically this issue can happen a lot when the `max_row_group_size` is large and the `target_file_size` is small
--
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] feat(core): Implement `RollingFileWriter` to help split data into multiple files [iceberg-rust]
Posted by "stevie9868 (via GitHub)" <gi...@apache.org>.
stevie9868 commented on PR #1547:
URL: https://github.com/apache/iceberg-rust/pull/1547#issuecomment-3130297500
Wondering what's your plan to make RollingFileWriter partition aware.
Looks like the parquet writer right now also is not partition aware?
--
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] feat(core): Implement `RollingFileWriter` to help split data into multiple files [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 commented on code in PR #1547:
URL: https://github.com/apache/iceberg-rust/pull/1547#discussion_r2230586125
##########
crates/iceberg/src/writer/file_writer/rolling_writer.rs:
##########
@@ -0,0 +1,331 @@
+// 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.
+
+use arrow_array::RecordBatch;
+use futures::future::try_join_all;
+
+use crate::runtime::{JoinHandle, spawn};
+use crate::spec::DataFileBuilder;
+use crate::writer::CurrentFileStatus;
+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> {
+ inner_builder: B,
+ target_size: usize,
+}
+
+impl<B: FileWriterBuilder> RollingFileWriterBuilder<B> {
+ /// Creates a new `RollingFileWriterBuilder` with the specified inner builder and target size.
+ ///
+ /// # Arguments
+ ///
+ /// * `inner_builder` - The builder for the underlying file writer
+ /// * `target_size` - The target size in bytes before rolling over to a new file
+ pub fn new(inner_builder: B, target_size: usize) -> Self {
+ Self {
+ inner_builder,
+ target_size,
+ }
+ }
+}
+
+impl<B: FileWriterBuilder> FileWriterBuilder for RollingFileWriterBuilder<B> {
+ type R = RollingFileWriter<B>;
+
+ async fn build(self) -> Result<Self::R> {
+ Ok(RollingFileWriter {
+ inner: None,
+ inner_builder: self.inner_builder,
+ target_size: self.target_size,
+ close_handles: vec![],
+ })
+ }
+}
+
+/// 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> {
+ inner: Option<B::R>,
+ inner_builder: B,
+ target_size: usize,
+ close_handles: Vec<JoinHandle<Result<Vec<DataFileBuilder>>>>,
+}
+
+impl<B: FileWriterBuilder> RollingFileWriter<B> {
+ /// Determines if the writer should roll over to a new file.
+ ///
+ /// # Arguments
+ ///
+ /// * `input_size` - The size in bytes of the incoming data
+ ///
+ /// # Returns
+ ///
+ /// `true` if a new file should be started, `false` otherwise
+ pub fn should_roll(&self, input_size: usize) -> bool {
+ self.current_written_size() + input_size > self.target_size
+ }
+}
+
+impl<B: FileWriterBuilder> FileWriter for RollingFileWriter<B> {
+ async fn write(&mut self, input: &RecordBatch) -> Result<()> {
+ let input_size = input.get_array_memory_size();
+
+ if self.inner.is_none() {
+ // initialize inner writer
+ self.inner = Some(self.inner_builder.clone().build().await?);
+ }
+
+ if self.should_roll(input_size) {
+ if let Some(inner) = self.inner.take() {
+ // close the current writer, roll to a new file
+ let handle = spawn(async move { inner.close().await });
Review Comment:
Memory control is a complex topic, and from what I've learned simply add a fix number of in fly doesn't work well when integrated into other systems. I would prefer not to spend too much time on this for now.
--
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] feat(core): Implement `RollingFileWriter` to help split data into multiple files [iceberg-rust]
Posted by "CTTY (via GitHub)" <gi...@apache.org>.
CTTY commented on code in PR #1547:
URL: https://github.com/apache/iceberg-rust/pull/1547#discussion_r2229405980
##########
crates/iceberg/src/writer/file_writer/rolling_writer.rs:
##########
@@ -0,0 +1,331 @@
+// 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.
+
+use arrow_array::RecordBatch;
+use futures::future::try_join_all;
+
+use crate::runtime::{JoinHandle, spawn};
+use crate::spec::DataFileBuilder;
+use crate::writer::CurrentFileStatus;
+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> {
+ inner_builder: B,
+ target_size: usize,
+}
+
+impl<B: FileWriterBuilder> RollingFileWriterBuilder<B> {
+ /// Creates a new `RollingFileWriterBuilder` with the specified inner builder and target size.
+ ///
+ /// # Arguments
+ ///
+ /// * `inner_builder` - The builder for the underlying file writer
+ /// * `target_size` - The target size in bytes before rolling over to a new file
+ pub fn new(inner_builder: B, target_size: usize) -> Self {
+ Self {
+ inner_builder,
+ target_size,
+ }
+ }
+}
+
+impl<B: FileWriterBuilder> FileWriterBuilder for RollingFileWriterBuilder<B> {
+ type R = RollingFileWriter<B>;
+
+ async fn build(self) -> Result<Self::R> {
+ Ok(RollingFileWriter {
+ inner: None,
+ inner_builder: self.inner_builder,
+ target_size: self.target_size,
+ close_handles: vec![],
+ })
+ }
+}
+
+/// 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> {
+ inner: Option<B::R>,
+ inner_builder: B,
+ target_size: usize,
+ close_handles: Vec<JoinHandle<Result<Vec<DataFileBuilder>>>>,
+}
+
+impl<B: FileWriterBuilder> RollingFileWriter<B> {
+ /// Determines if the writer should roll over to a new file.
+ ///
+ /// # Arguments
+ ///
+ /// * `input_size` - The size in bytes of the incoming data
+ ///
+ /// # Returns
+ ///
+ /// `true` if a new file should be started, `false` otherwise
+ pub fn should_roll(&self, input_size: usize) -> bool {
+ self.current_written_size() + input_size > self.target_size
+ }
+}
+
+impl<B: FileWriterBuilder> FileWriter for RollingFileWriter<B> {
+ async fn write(&mut self, input: &RecordBatch) -> Result<()> {
+ let input_size = input.get_array_memory_size();
+
+ if self.inner.is_none() {
+ // initialize inner writer
+ self.inner = Some(self.inner_builder.clone().build().await?);
+ }
+
+ if self.should_roll(input_size) {
+ if let Some(inner) = self.inner.take() {
+ // close the current writer, roll to a new file
+ let handle = spawn(async move { inner.close().await });
Review Comment:
Good point!
I have some rough idea to further improving this: we can use a config to control the maximum parallelism here:
```
struct RollingWriter {
...
buffer: Vec<DataFile>
}
...
while close_handles.len() >= self.max_parallelism() {
// wait until some closers complete, and store the data files in a buffer
self.buffer.push(future::select(self.close_handles))
}
self.close_handles.push(new_handle);
```
I have not thought very clearly on how to prevent `buffer` from eating up the memory as of now, or do we even need it?
Either way I agree this can be completed as a follow-up
--
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