You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@arrow.apache.org by "Xuanwo (via GitHub)" <gi...@apache.org> on 2024/05/11 11:23:54 UTC
[PR] feat: Make AsyncArrowWriter accepts AsyncFileWriter [arrow-rs]
Xuanwo opened a new pull request, #5753:
URL: https://github.com/apache/arrow-rs/pull/5753
# Which issue does this PR close?
Closes https://github.com/apache/arrow-rs/issues/5738
# Rationale for this change
This makes `AsyncArrowWriter` accepts `AsyncFileWriter` so users can save an extra copy if they have a writer that can accept `Bytes` directly.
# What changes are included in this PR?
A bit change from the proposal but mainly the same.
# Are there any user-facing changes?
`AsyncArrowWriter` now accepts `AsyncFileWriter` instead of `AsyncWrite`. But `AsyncFileWriter` has been implemented for all `AsyncWrite`, so existing code should be working as before.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Make AsyncArrowWriter accepts AsyncFileWriter [arrow-rs]
Posted by "tustvold (via GitHub)" <gi...@apache.org>.
tustvold commented on code in PR #5753:
URL: https://github.com/apache/arrow-rs/pull/5753#discussion_r1597461850
##########
parquet/src/arrow/async_writer/mod.rs:
##########
@@ -188,12 +234,7 @@ impl<W: AsyncWrite + Unpin + Send> AsyncArrowWriter<W> {
let buffer = self.sync_writer.inner_mut();
self.async_writer
- .write_all(buffer.as_slice())
- .await
- .map_err(|e| ParquetError::External(Box::new(e)))?;
-
- self.async_writer
- .flush()
+ .write(Bytes::copy_from_slice(buffer))
Review Comment:
Could we avoid this copy, perhaps by taking the buffer and replacing it with a freshly allocated one?
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Make AsyncArrowWriter accepts AsyncFileWriter [arrow-rs]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #5753:
URL: https://github.com/apache/arrow-rs/pull/5753#discussion_r1597467300
##########
parquet/src/arrow/async_writer/mod.rs:
##########
@@ -188,12 +234,7 @@ impl<W: AsyncWrite + Unpin + Send> AsyncArrowWriter<W> {
let buffer = self.sync_writer.inner_mut();
self.async_writer
- .write_all(buffer.as_slice())
- .await
- .map_err(|e| ParquetError::External(Box::new(e)))?;
-
- self.async_writer
- .flush()
+ .write(Bytes::copy_from_slice(buffer))
Review Comment:
I'm somewhat concerned about the changes in behavior: previously, if a write operation encountered an error, users could safely retry it because the buffer remained unchanged. However, if we use the buffer directly now, users might discover that their data already written is missing.
This is also the reason I raise https://github.com/apache/arrow-rs/pull/5753#discussion_r1597429609
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Make AsyncArrowWriter accepts AsyncFileWriter [arrow-rs]
Posted by "tustvold (via GitHub)" <gi...@apache.org>.
tustvold commented on code in PR #5753:
URL: https://github.com/apache/arrow-rs/pull/5753#discussion_r1597482302
##########
parquet/src/arrow/async_writer/mod.rs:
##########
@@ -188,12 +234,7 @@ impl<W: AsyncWrite + Unpin + Send> AsyncArrowWriter<W> {
let buffer = self.sync_writer.inner_mut();
self.async_writer
- .write_all(buffer.as_slice())
- .await
- .map_err(|e| ParquetError::External(Box::new(e)))?;
-
- self.async_writer
- .flush()
+ .write(Bytes::copy_from_slice(buffer))
Review Comment:
BytesMut would be the same, freeze is consuming - https://docs.rs/bytes/latest/bytes/struct.BytesMut.html#method.freeze.
Preserving retries I think could be done by simply stashing the produced Bytes on AsyncWriter
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Make AsyncArrowWriter accepts AsyncFileWriter [arrow-rs]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #5753:
URL: https://github.com/apache/arrow-rs/pull/5753#discussion_r1597467300
##########
parquet/src/arrow/async_writer/mod.rs:
##########
@@ -188,12 +234,7 @@ impl<W: AsyncWrite + Unpin + Send> AsyncArrowWriter<W> {
let buffer = self.sync_writer.inner_mut();
self.async_writer
- .write_all(buffer.as_slice())
- .await
- .map_err(|e| ParquetError::External(Box::new(e)))?;
-
- self.async_writer
- .flush()
+ .write(Bytes::copy_from_slice(buffer))
Review Comment:
I'm somewhat concerned about the changes in behavior: previously, if a write operation encountered an error, users could safely retry it because the buffer remained unchanged. However, if we use the buffer directly now, users might discover that their data already written is missing.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Make AsyncArrowWriter accepts AsyncFileWriter [arrow-rs]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #5753:
URL: https://github.com/apache/arrow-rs/pull/5753#discussion_r1599328506
##########
parquet/src/arrow/async_writer/mod.rs:
##########
@@ -59,13 +59,59 @@ use crate::{
};
use arrow_array::RecordBatch;
use arrow_schema::SchemaRef;
+use bytes::Bytes;
+use futures::future::BoxFuture;
+use futures::FutureExt;
use tokio::io::{AsyncWrite, AsyncWriteExt};
-/// Encodes [`RecordBatch`] to parquet, outputting to an [`AsyncWrite`]
+/// The asynchronous interface used by [`AsyncArrowWriter`] to write parquet files
+pub trait AsyncFileWriter: Send {
+ /// Write the provided bytes to the underlying writer
+ ///
+ /// The underlying writer CAN decide to buffer the data or write it immediately.
+ /// This design allows the writer implementer to control the buffering and I/O scheduling.
+ fn write(&mut self, bs: Bytes) -> BoxFuture<'_, Result<()>>;
Review Comment:
I find current ArrowWriter's design is also not safe to retry. I will update the behavior notes of `write` to make it more clear and just take the `Vec<u8>` here to make this PR more simple.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Make AsyncArrowWriter accepts AsyncFileWriter [arrow-rs]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on PR #5753:
URL: https://github.com/apache/arrow-rs/pull/5753#issuecomment-2109216732
Hi @tustvold, PTAL. Thanks!
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Make AsyncArrowWriter accepts AsyncFileWriter [arrow-rs]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #5753:
URL: https://github.com/apache/arrow-rs/pull/5753#discussion_r1599332346
##########
parquet/src/arrow/async_writer/mod.rs:
##########
@@ -188,12 +234,7 @@ impl<W: AsyncWrite + Unpin + Send> AsyncArrowWriter<W> {
let buffer = self.sync_writer.inner_mut();
self.async_writer
- .write_all(buffer.as_slice())
- .await
- .map_err(|e| ParquetError::External(Box::new(e)))?;
-
- self.async_writer
- .flush()
+ .write(Bytes::copy_from_slice(buffer))
Review Comment:
See https://github.com/apache/arrow-rs/pull/5753#discussion_r1599328506
It's not a good idea to make this function able to retry, I revert my idea.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Make AsyncArrowWriter accepts AsyncFileWriter [arrow-rs]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #5753:
URL: https://github.com/apache/arrow-rs/pull/5753#discussion_r1597429609
##########
parquet/src/arrow/async_writer/mod.rs:
##########
@@ -188,12 +232,7 @@ impl<W: AsyncWrite + Unpin + Send> AsyncArrowWriter<W> {
let buffer = self.sync_writer.inner_mut();
self.async_writer
- .write_all(buffer.as_slice())
- .await
- .map_err(|e| ParquetError::External(Box::new(e)))?;
-
- self.async_writer
- .flush()
+ .write(Bytes::from(buffer))
Review Comment:
Maybe we can use `BytesMut` here so we can reclaim the memory for all reference to `Bytes` has been dropped?
---
I'm willing to submit a PR for 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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Make AsyncArrowWriter accepts AsyncFileWriter [arrow-rs]
Posted by "tustvold (via GitHub)" <gi...@apache.org>.
tustvold commented on PR #5753:
URL: https://github.com/apache/arrow-rs/pull/5753#issuecomment-2109633274
Thank you
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Make AsyncArrowWriter accepts AsyncFileWriter [arrow-rs]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #5753:
URL: https://github.com/apache/arrow-rs/pull/5753#discussion_r1597429609
##########
parquet/src/arrow/async_writer/mod.rs:
##########
@@ -188,12 +232,7 @@ impl<W: AsyncWrite + Unpin + Send> AsyncArrowWriter<W> {
let buffer = self.sync_writer.inner_mut();
self.async_writer
- .write_all(buffer.as_slice())
- .await
- .map_err(|e| ParquetError::External(Box::new(e)))?;
-
- self.async_writer
- .flush()
+ .write(Bytes::from(buffer))
Review Comment:
Maybe we can use `BytesMut` here so we can reclaim the memory after all reference to `Bytes` has been dropped?
---
I'm willing to submit a PR for 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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Make AsyncArrowWriter accepts AsyncFileWriter [arrow-rs]
Posted by "tustvold (via GitHub)" <gi...@apache.org>.
tustvold merged PR #5753:
URL: https://github.com/apache/arrow-rs/pull/5753
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Make AsyncArrowWriter accepts AsyncFileWriter [arrow-rs]
Posted by "tustvold (via GitHub)" <gi...@apache.org>.
tustvold commented on code in PR #5753:
URL: https://github.com/apache/arrow-rs/pull/5753#discussion_r1597482498
##########
parquet/src/arrow/async_writer/mod.rs:
##########
@@ -59,13 +59,59 @@ use crate::{
};
use arrow_array::RecordBatch;
use arrow_schema::SchemaRef;
+use bytes::Bytes;
+use futures::future::BoxFuture;
+use futures::FutureExt;
use tokio::io::{AsyncWrite, AsyncWriteExt};
-/// Encodes [`RecordBatch`] to parquet, outputting to an [`AsyncWrite`]
+/// The asynchronous interface used by [`AsyncArrowWriter`] to write parquet files
+pub trait AsyncFileWriter: Send {
+ /// Write the provided bytes to the underlying writer
+ ///
+ /// The underlying writer CAN decide to buffer the data or write it immediately.
+ /// This design allows the writer implementer to control the buffering and I/O scheduling.
+ fn write(&mut self, bs: Bytes) -> BoxFuture<'_, Result<()>>;
Review Comment:
We should document atomicity requirements for this method
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Make AsyncArrowWriter accepts AsyncFileWriter [arrow-rs]
Posted by "tustvold (via GitHub)" <gi...@apache.org>.
tustvold commented on code in PR #5753:
URL: https://github.com/apache/arrow-rs/pull/5753#discussion_r1597482302
##########
parquet/src/arrow/async_writer/mod.rs:
##########
@@ -188,12 +234,7 @@ impl<W: AsyncWrite + Unpin + Send> AsyncArrowWriter<W> {
let buffer = self.sync_writer.inner_mut();
self.async_writer
- .write_all(buffer.as_slice())
- .await
- .map_err(|e| ParquetError::External(Box::new(e)))?;
-
- self.async_writer
- .flush()
+ .write(Bytes::copy_from_slice(buffer))
Review Comment:
BytesMut would be the same, freeze is consuming - https://docs.rs/bytes/latest/bytes/struct.BytesMut.html#method.freeze.
Preserving retries I think could be done by simply stashing a copy of the produced Bytes on AsyncWriter. Although this would be assuming that the underlying write API is atomic
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org