Use object_store:BufWriter to replace put_multipart by yyy1000 · Pull Request #9648 · apache/datafusion

yyy1000 · 2024-03-17T00:53:45Z

Which issue does this PR close?

Closes #9614 .

Rationale for this change

What changes are included in this PR?

Are these changes tested?

Are there any user-facing changes?

yyy1000 · 2024-03-17T01:00:44Z

It seems that AbortableWrite needs MultiPart information. 🤔
https://github.com/apache/arrow-datafusion/blob/37253e57beb25f0f1a4412b75421a489c2cb3c6a/datafusion/core/src/datasource/file_format/write/mod.rs#L155-L167
I'm reading the code to see whether put_multipart here can be replaced.

yyy1000 · 2024-03-17T02:23:33Z

Close and reopen to rerun CI

devinjdangelo · 2024-03-17T14:06:27Z

It seems that AbortableWrite needs MultiPart information. 🤔

https://github.com/apache/arrow-datafusion/blob/37253e57beb25f0f1a4412b75421a489c2cb3c6a/datafusion/core/src/datasource/file_format/write/mod.rs#L155-L167

I'm reading the code to see whether put_multipart here can be replaced.

We could drop the AbortableWrite construct entirely and no longer make a manual effort to abort on failure. The manual cleanup will not work in all cases anyway, and users can configure automatic cleanup with their cloud provider, e.g.: https://aws.amazon.com/blogs/aws-cloud-financial-management/discovering-and-deleting-incomplete-multipart-uploads-to-lower-amazon-s3-costs/.

I believe this is in line with what @tustvold is thinking with ongoing improvements of ObjectStore apache/arrow-rs#5500 (comment).

devinjdangelo · 2024-03-17T14:07:28Z

Close and reopen to rerun CI

Another option is to push an empty commit.

git commit --allow-empty -m 'retry ci'

yyy1000 · 2024-03-17T16:10:34Z

Thanks for your review! @devinjdangelo
I will look into the code to see how to drop AbortableWrite. :)

tustvold · 2024-03-18T02:13:04Z

I will look into the code to see how to drop AbortableWrite. :)

FWIW the multipart ID is exposed but it might take some more type plumbing. IMO removing AbortableWrite is probably a simpler path forward

yyy1000

Thanks for your review! @devinjdangelo @tustvold
I updated the PR to address your comment, let me know whether it needs further improvement :)

yyy1000 · 2024-03-18T02:43:27Z

datafusion/core/src/datasource/file_format/write/mod.rs

-    ))
+) -> Result<Box<dyn AsyncWrite + Send + Unpin>> {
+    let buf_writer = BufWriter::new(object_store, location.clone());
+    Ok(file_compression_type.convert_async_writer(buf_writer)?)


I think create_writer could still be saved cause it create the writer with compression?

yyy1000 · 2024-03-18T02:44:01Z

datafusion/core/src/datasource/file_format/write/orchestration.rs

-                writer.shutdown()
+        writer.shutdown()
                    .await
                    .map_err(|_| internal_datafusion_err!("Error encountered while finalizing writes! Partial results may have been written to ObjectStore!"))?;


Don't know whether it's proper to let all just shutdown here. But I think according to #9648 (comment), it's OK? 👀

yyy1000 · 2024-03-18T02:44:41Z

datafusion/core/src/datasource/file_format/write/mod.rs

-        Pin::new(&mut self.get_mut().writer).poll_shutdown(cx)
-    }
-}
-


I also removed struct MultiPart since they're also not used anymore.

yyy1000 · 2024-03-18T02:47:16Z

datafusion/core/src/datasource/file_format/file_compression_type.rs

    pub fn convert_async_writer(
        &self,
-        w: Box<dyn AsyncWrite + Send + Unpin>,
+        w: BufWriter,


I'm thinking whether it's OK to change the param type here cause it's public, but keeping Box<dyn AsyncWrite + Send + Unpin> makes the type incompatible. 🤔

I think it depends on if this code is always used with object_store (aka if DataFusion code always writes output using the object_store API). If this is the case, then switching to BufWriter here makes sense to me

BTW I think we need to update the comments on this function to match the new implementation

alamb

Thank you very much @yyy1000 -- this looks like a good improvement to me

I would like a few more opinions about the change in abort behavior, but as I understand the implications it seems a reasonable change to me.

With a few more updates to doc comments I think this PR will be ready to go in my view

cc @wiedld @tustvold

alamb · 2024-03-19T14:39:57Z

datafusion/core/src/datasource/file_format/file_compression_type.rs

    pub fn convert_async_writer(
        &self,
-        w: Box<dyn AsyncWrite + Send + Unpin>,
+        w: BufWriter,


I think it depends on if this code is always used with object_store (aka if DataFusion code always writes output using the object_store API). If this is the case, then switching to BufWriter here makes sense to me

BTW I think we need to update the comments on this function to match the new implementation

alamb · 2024-03-19T14:47:36Z

datafusion/core/src/datasource/file_format/write/mod.rs

    }
 }

-/// Stores data needed during abortion of MultiPart writers


Is it correct to say that the implications of removing AbortableWrite is that if certain (larger) writes to object store fail / abort for some reason, "garbage" (unreferenced partial uploads) may be left around indefinitely on the provider?

While I understand that some object stores (maybe all) can be configured to automatically clean up such parts, I think reverting the "try to cleanup on failure" behavior is worth reconsidering.

I think I could be convinced with an argument like "the software can't ensure clean up anyways (for example, if it is SIGKILLed) for some reason, and thus we don't even try to clean up in paths we could", but if we go that route I think we should explicitly document the behavior and rationale in comments somewhere

I think @metesynnada or @mustafasrepo originally added this code (though I may be wrong) so perhaps they have some perspective to share

Yeah, I think according to #9648 (comment), 'garbage' cleanup will be only on cloud provider if removing AbortableWrite 🤔, also @devinjdangelo , is it right?

Is it correct to say that the implications of removing AbortableWrite is that if certain (larger) writes to object store fail / abort for some reason, "garbage" (unreferenced partial uploads) may be left around indefinitely on the provider?

Yes. I have mixed feelings about removing any attempt to clean up on failure.

I think I could be convinced with an argument like "the software can't ensure clean up anyways (for example, if it is SIGKILLed) for some reason, and thus we don't even try to clean up in paths we could", but if we go that route I think we should explicitly document the behavior and rationale in comments somewhere

This argument is valid. A hardware/network fault will prevent any cleanup code we write from working, so to truly protect against partial writes would require logic outside of DataFusion's process (e.g. on the cloud service provider side).

On the other hand, this change may be annoying for simple failures when writing to a local file system. Encountering any execution error will leave dangling files when before they often could be cleaned up.

I think this is a case where one will draw different conclusions depending on if they are imagining an individual user of something like datafusion-cli vs. a production database system implemented on top DataFusion. The latter user will have little use for our attempts at clean up (they will need much better anyway), but the former may appreciate it.

Local file system automatically cleans up on drop, or at least makes a best effort to do so. FWIW this same mechanism is used for ALL uploads, even the non-multipart ones so as to provide atomicity. Given nobody has complained about this, I suspect it is doing a fairly good job

I am not aware of a cloud provider that provides multipart uploads without some automated way to reap aborted uploads after a given time

Ok, so I vote we leave the code as is (no attempt to explicitly abort in write) and add a note in the documentation. If it turns out this is an important behavior, we can add it back in

@yyy1000 can you handle adding the note in the documentation?

Sure, I added it around create_writer function, would it be enough?

Looks reasonable to me -- thank you

alamb · 2024-03-19T14:49:42Z

datafusion/core/src/datasource/file_format/write/mod.rs

    location: &Path,
    object_store: Arc<dyn ObjectStore>,
-) -> Result<AbortableWrite<Box<dyn AsyncWrite + Send + Unpin>>> {
-    let (multipart_id, writer) = object_store


For anyone else following along, the BufWriter internally does a multi-part put when appropriate

metesynnada · 2024-03-19T19:01:53Z

I can check this PR as well tomorrow.

* feat: use BufWriter to replace put_multipart * feat: remove AbortableWrite * fix clippy * fix: add doc comment

feat: use BufWriter to replace put_multipart

6e58483

github-actions bot added the core Core DataFusion crate label Mar 17, 2024

yyy1000 closed this Mar 17, 2024

yyy1000 reopened this Mar 17, 2024

feat: remove AbortableWrite

16de811

yyy1000 commented Mar 18, 2024

View reviewed changes

fix clippy

6b89661

alamb mentioned this pull request Mar 18, 2024

DataFusion weekly project plan (Andrew Lamb) - March 18, 2024 #9675

Closed

7 tasks

alamb approved these changes Mar 19, 2024

View reviewed changes

alamb mentioned this pull request Mar 19, 2024

Replace AsyncWrite with Upload trait and rename MultiPartStore to MultipartStore (#5458) apache/arrow-rs#5500

Merged

yyy1000 added 2 commits March 20, 2024 14:11

fix: add doc comment

d970287

Merge branch 'main' into BufWriter

7128a62

tustvold merged commit dbfb153 into apache:main Mar 20, 2024

yyy1000 deleted the BufWriter branch March 20, 2024 22:13

tillrohrmann pushed a commit to restatedev/datafusion that referenced this pull request Jul 25, 2024

Use object_store:BufWriter to replace put_multipart (apache#9648)

9d292b0

* feat: use BufWriter to replace put_multipart * feat: remove AbortableWrite * fix clippy * fix: add doc comment

Comments

Conversation

yyy1000 commented Mar 17, 2024

Which issue does this PR close?

Rationale for this change

What changes are included in this PR?

Are these changes tested?

Are there any user-facing changes?

Uh oh!

yyy1000 commented Mar 17, 2024

Uh oh!

yyy1000 commented Mar 17, 2024

Uh oh!

devinjdangelo commented Mar 17, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

devinjdangelo commented Mar 17, 2024

Uh oh!

yyy1000 commented Mar 17, 2024

Uh oh!

tustvold commented Mar 18, 2024

Uh oh!

yyy1000 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

alamb left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

devinjdangelo Mar 19, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

tustvold Mar 19, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

alamb Mar 20, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

metesynnada commented Mar 19, 2024

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

devinjdangelo commented Mar 17, 2024 •

edited

Loading

devinjdangelo Mar 19, 2024 •

edited

Loading

tustvold Mar 19, 2024 •

edited

Loading

alamb Mar 20, 2024 •

edited

Loading