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/07/07 05:22:35 UTC
[PR] feat(parquet): Implement AsyncFileWriter for obejct_store::buffered::BufWriter [arrow-rs]
Xuanwo opened a new pull request, #6013:
URL: https://github.com/apache/arrow-rs/pull/6013
# 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 https://github.com/apache/arrow-rs/issues/5766
# Rationale for this change
Implement `AsyncFileWriter` for `object_store::buffered::BufWriter`
# What changes are included in this PR?
Add a new impl for `AsyncFileWriter`
# Are there any user-facing changes?
New impl added.
--
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(parquet): Implement AsyncFileWriter for obejct_store::buffered::BufWriter [arrow-rs]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on PR #6013:
URL: https://github.com/apache/arrow-rs/pull/6013#issuecomment-2264496738
Hi, @alamb, Sorry for the long waiting, this PR is ready to review 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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(parquet): Implement AsyncFileWriter for `object_store::buffered::BufWriter` [arrow-rs]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on PR #6013:
URL: https://github.com/apache/arrow-rs/pull/6013#issuecomment-2266356682
Hi, @alamb, I have addressed all comments. 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(parquet): Implement AsyncFileWriter for obejct_store::buffered::BufWriter [arrow-rs]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on PR #6013:
URL: https://github.com/apache/arrow-rs/pull/6013#issuecomment-2213344194
Hi, @alamb @tustvold, is it possible to run a patch release for `object_store`?
--
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(parquet): Implement AsyncFileWriter for `object_store::buffered::BufWriter` [arrow-rs]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #6013:
URL: https://github.com/apache/arrow-rs/pull/6013#discussion_r1702443992
##########
parquet/Cargo.toml:
##########
@@ -82,7 +82,7 @@ serde_json = { version = "1.0", features = ["std"], default-features = false }
arrow = { workspace = true, features = ["ipc", "test_utils", "prettyprint", "json"] }
tokio = { version = "1.0", default-features = false, features = ["macros", "rt", "io-util", "fs"] }
rand = { version = "0.8", default-features = false, features = ["std", "std_rng"] }
-object_store = { version = "0.10.0", default-features = false, features = ["azure"] }
Review Comment:
Addressed.
--
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(parquet): Implement AsyncFileWriter for obejct_store::buffered::BufWriter [arrow-rs]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #6013:
URL: https://github.com/apache/arrow-rs/pull/6013#issuecomment-2214848822
Thanks for the reminder @Xuanwo
I filed this ticket to track the next release https://github.com/apache/arrow-rs/issues/6022
I looked at the commits that went into object_store: https://github.com/apache/arrow-rs/commits/master/object_store
As of today there do not appear to be any breaking API changes to object store and thus I think we can release a minor 0.10.1 version
--
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(parquet): Implement AsyncFileWriter for `object_store::buffered::BufWriter` [arrow-rs]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #6013:
URL: https://github.com/apache/arrow-rs/pull/6013#issuecomment-2270943822
Thanks again @Xuanwo
--
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(parquet): Implement AsyncFileWriter for `object_store::buffered::BufWriter` [arrow-rs]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb merged PR #6013:
URL: https://github.com/apache/arrow-rs/pull/6013
--
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(parquet): Implement AsyncFileWriter for `object_store::buffered::BufWriter` [arrow-rs]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on code in PR #6013:
URL: https://github.com/apache/arrow-rs/pull/6013#discussion_r1702121666
##########
parquet/src/arrow/async_writer/store.rs:
##########
@@ -0,0 +1,111 @@
+// 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 bytes::Bytes;
+use futures::future::BoxFuture;
+use std::sync::Arc;
+
+use crate::arrow::async_writer::AsyncFileWriter;
+use crate::errors::{ParquetError, Result};
+use object_store::buffered::BufWriter;
+use object_store::path::Path;
+use object_store::ObjectStore;
+use tokio::io::AsyncWriteExt;
+
Review Comment:
```suggestion
//! [`ParquetObjectWriter`] for writing to parquet to [`ObjectStore`]
```
##########
parquet/Cargo.toml:
##########
@@ -82,7 +82,7 @@ serde_json = { version = "1.0", features = ["std"], default-features = false }
arrow = { workspace = true, features = ["ipc", "test_utils", "prettyprint", "json"] }
tokio = { version = "1.0", default-features = false, features = ["macros", "rt", "io-util", "fs"] }
rand = { version = "0.8", default-features = false, features = ["std", "std_rng"] }
-object_store = { version = "0.10.0", default-features = false, features = ["azure"] }
Review Comment:
I think this update to `0.10.2` should also be done in the `dependencies` section above too (this change is the `dev-dependencies`)
```toml
# Intentionally not a path dependency as object_store is released separately
object_store = { version = "0.10.0", default-features = false, optional = true }
```
##########
parquet/src/arrow/async_writer/store.rs:
##########
@@ -0,0 +1,111 @@
+// 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 bytes::Bytes;
+use futures::future::BoxFuture;
+use std::sync::Arc;
+
+use crate::arrow::async_writer::AsyncFileWriter;
+use crate::errors::{ParquetError, Result};
+use object_store::buffered::BufWriter;
+use object_store::path::Path;
+use object_store::ObjectStore;
+use tokio::io::AsyncWriteExt;
+
+#[derive(Debug)]
+pub struct ParquetObjectWriter {
+ w: BufWriter,
+}
+
+impl ParquetObjectWriter {
+ /// Create a new [`ParquetObjectWriter`] that writes to the specified path in the given store.
+ ///
+ /// To configure the writer behavior, please build [`BufWriter`] and then use [`Self::from_raw`]
+ pub fn new(store: Arc<dyn ObjectStore>, path: Path) -> Self {
+ Self::from_raw(BufWriter::new(store, path))
+ }
+
+ /// Construct a new ParquetObjectWriter via a existing BufWriter.
+ pub fn from_raw(w: BufWriter) -> Self {
Review Comment:
```suggestion
pub fn from_buf_writer(w: BufWriter) -> Self {
```
##########
parquet/src/arrow/async_writer/store.rs:
##########
@@ -0,0 +1,111 @@
+// 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 bytes::Bytes;
+use futures::future::BoxFuture;
+use std::sync::Arc;
+
+use crate::arrow::async_writer::AsyncFileWriter;
+use crate::errors::{ParquetError, Result};
+use object_store::buffered::BufWriter;
+use object_store::path::Path;
+use object_store::ObjectStore;
+use tokio::io::AsyncWriteExt;
+
+#[derive(Debug)]
+pub struct ParquetObjectWriter {
+ w: BufWriter,
+}
+
+impl ParquetObjectWriter {
+ /// Create a new [`ParquetObjectWriter`] that writes to the specified path in the given store.
+ ///
+ /// To configure the writer behavior, please build [`BufWriter`] and then use [`Self::from_raw`]
+ pub fn new(store: Arc<dyn ObjectStore>, path: Path) -> Self {
+ Self::from_raw(BufWriter::new(store, path))
+ }
+
+ /// Construct a new ParquetObjectWriter via a existing BufWriter.
+ pub fn from_raw(w: BufWriter) -> Self {
+ Self { w }
+ }
+
+ /// Consume the writer and return the underlying BufWriter.
+ pub fn into_raw(self) -> BufWriter {
Review Comment:
I think `into_inner` is a common name for this kind of API, for example in std::io::BufWriter: https://doc.rust-lang.org/std/io/struct.BufWriter.html#method.into_inner
```suggestion
pub fn into_inner(self) -> BufWriter {
```
##########
parquet/src/arrow/async_writer/store.rs:
##########
@@ -0,0 +1,111 @@
+// 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 bytes::Bytes;
+use futures::future::BoxFuture;
+use std::sync::Arc;
+
+use crate::arrow::async_writer::AsyncFileWriter;
+use crate::errors::{ParquetError, Result};
+use object_store::buffered::BufWriter;
+use object_store::path::Path;
+use object_store::ObjectStore;
+use tokio::io::AsyncWriteExt;
+
+#[derive(Debug)]
+pub struct ParquetObjectWriter {
+ w: BufWriter,
+}
+
+impl ParquetObjectWriter {
+ /// Create a new [`ParquetObjectWriter`] that writes to the specified path in the given store.
+ ///
+ /// To configure the writer behavior, please build [`BufWriter`] and then use [`Self::from_raw`]
+ pub fn new(store: Arc<dyn ObjectStore>, path: Path) -> Self {
+ Self::from_raw(BufWriter::new(store, path))
+ }
+
+ /// Construct a new ParquetObjectWriter via a existing BufWriter.
+ pub fn from_raw(w: BufWriter) -> Self {
+ Self { w }
+ }
+
+ /// Consume the writer and return the underlying BufWriter.
+ pub fn into_raw(self) -> BufWriter {
+ self.w
+ }
+}
+
+impl AsyncFileWriter for ParquetObjectWriter {
+ fn write(&mut self, bs: Bytes) -> BoxFuture<'_, Result<()>> {
+ Box::pin(async {
+ self.w
+ .put(bs)
+ .await
+ .map_err(|err| ParquetError::External(Box::new(err)))
+ })
+ }
+
+ fn complete(&mut self) -> BoxFuture<'_, Result<()>> {
+ Box::pin(async {
+ self.w
+ .shutdown()
+ .await
+ .map_err(|err| ParquetError::External(Box::new(err)))
+ })
+ }
+}
+
Review Comment:
It might also be nice to provide a `From` impl too for convenience
```suggestion
impl From<BufWriter> for ParquetObjectWriter {
fn from(w: BufWriter) -> Self {
Self::from_raw(w)
}
}
```
--
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(parquet): Implement AsyncFileWriter for `object_store::buffered::BufWriter` [arrow-rs]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on code in PR #6013:
URL: https://github.com/apache/arrow-rs/pull/6013#discussion_r1702710661
##########
parquet/src/arrow/async_writer/store.rs:
##########
@@ -0,0 +1,157 @@
+// 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 bytes::Bytes;
+use futures::future::BoxFuture;
+use std::sync::Arc;
+
+use crate::arrow::async_writer::AsyncFileWriter;
+use crate::errors::{ParquetError, Result};
+use object_store::buffered::BufWriter;
+use object_store::path::Path;
+use object_store::ObjectStore;
+use tokio::io::AsyncWriteExt;
+
+/// [`ParquetObjectWriter`] for writing to parquet to [`ObjectStore`]
+///
+/// ```
+/// # use arrow_array::{ArrayRef, Int64Array, RecordBatch};
Review Comment:
❤️ -- this is great -- 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(parquet): Implement AsyncFileWriter for obejct_store::buffered::BufWriter [arrow-rs]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on PR #6013:
URL: https://github.com/apache/arrow-rs/pull/6013#issuecomment-2212329521
This PR is blocked by `BufWriter::put` API is not released yet.
--
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