ARROW-13542: [C++][Compute][Dataset] Add dataset::WriteNode for writing rows from an ExecPlan to disk #11017
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
019d847 to
fd2b237
Compare
d6cdf3b to
605913e
Compare
… some new async utilities
…hat all tests are now passing.
… placed arrow::dataset::Initialize method
c651c4a to
4240953
Compare
lidavidm
approved these changes
Oct 5, 2021
Member
lidavidm
left a comment
There was a problem hiding this comment.
LGTM. There are a lot of files touched, but things look straightforward.
cpp/src/arrow/util/future.h
Outdated
| /// | ||
| /// The future will be marked complete if all `futures` complete | ||
| /// successfully. Otherwise, it will be marked failed with the status of | ||
| /// the first failing future. |
Member
There was a problem hiding this comment.
I left a nit about the docstring on the MakeTask PR: #11210 (comment)
Member
Author
There was a problem hiding this comment.
I clarified the comment according to that review and also changed the name to AllFinished.
| // is finished. Use SinkNode if you are transferring the ownership of the data to another | ||
| // system. Use ConsumingSinkNode if the data is being consumed within the exec plan (i.e. | ||
| // the exec plan should not complete until the consumption has completed). | ||
| class ConsumingSinkNode : public ExecNode { |
Member
There was a problem hiding this comment.
Should we add an explicit test of this node in plan_test.cc, as with the other nodes?
Member
Author
There was a problem hiding this comment.
I added a few unit tests to plan_test.cc
Contributor
|
LGTM too me too. I reviewed the code and there are some common patterns to handle signals, input_counter and future |
lidavidm
approved these changes
Oct 6, 2021
Member
lidavidm
left a comment
There was a problem hiding this comment.
Thanks for the updates. LGTM
ViniciusSouzaRoque
pushed a commit
to s1mbi0se/arrow
that referenced
this pull request
Oct 20, 2021
…ng rows from an ExecPlan to disk This PR adds a write node. The write node takes in `FileSystemDatasetWriteOptions` and the projected schema and writes the incoming data to disk. It is a sink node but it is a bit different than the existing sink node. The existing sink node transferred batches via an AsyncGenerator but that puts ownership of the batches outside of the push-based flow of the exec plan. I added a new ConsumingSinkNode which consumes the batches as part of the push-based flow. This makes it possible to block the exec plan from finishing until all data has been written to disk. In addition this PR refines the AsyncTaskGroup a little. `WaitForTasksToFinish` was not a very clearly named method. Once called it actually transitioned the task group from a "top level tasks can be added" state to a "no more top level tasks can be added" state and that was not clear in the name. The new name (`End`) is hopefully more clear. This PR does not solve the backpressure problem. Instead a serial async task group was created. This will run all tasks in order (the default task group allows them to run in parallel) but not necessarily on the calling thread (i.e. unlike SerialTaskGroup we do not block on the AddTask method). This allows tasks to pile up in a queue and, if the write is slow, this will become a pile-up point which will eventually run out of memory (provided there is enough data being written). That problem is solved in a follow-up apache#11286 The `AsyncSerializedTaskGroup` and `AsyncTaskGroup` classes have very similar APIs but I did not create an interface / abstract base class because I don't yet envision any case where they would be interchangeable. The distinction is a "can these tasks run in parallel or not" and is not a performance / resource question. As a consequence of using the ExecPlan dataset writes no longer have reliable ordering. If you pass in three batches and they are all destined for the same file then the batches may be written in any order in the destination file. This is because the ExecPlan creates a thread task for each input batch and so they could arrive at the write node in any order. Closes apache#11017 from westonpace/feature/ARROW-13542-write-node Authored-by: Weston Pace <weston.pace@gmail.com> Signed-off-by: Weston Pace <weston.pace@gmail.com>
Closed
2 tasks
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
This PR adds a write node. The write node takes in
FileSystemDatasetWriteOptionsand the projected schema and writes the incoming data to disk. It is a sink node but it is a bit different than the existing sink node. The existing sink node transferred batches via an AsyncGenerator but that puts ownership of the batches outside of the push-based flow of the exec plan. I added a new ConsumingSinkNode which consumes the batches as part of the push-based flow. This makes it possible to block the exec plan from finishing until all data has been written to disk.In addition this PR refines the AsyncTaskGroup a little.
WaitForTasksToFinishwas not a very clearly named method. Once called it actually transitioned the task group from a "top level tasks can be added" state to a "no more top level tasks can be added" state and that was not clear in the name. The new name (End) is hopefully more clear.This PR does not solve the backpressure problem. Instead a serial async task group was created. This will run all tasks in order (the default task group allows them to run in parallel) but not necessarily on the calling thread (i.e. unlike SerialTaskGroup we do not block on the AddTask method). This allows tasks to pile up in a queue and, if the write is slow, this will become a pile-up point which will eventually run out of memory (provided there is enough data being written). That problem is solved in a follow-up #11286
The
AsyncSerializedTaskGroupandAsyncTaskGroupclasses have very similar APIs but I did not create an interface / abstract base class because I don't yet envision any case where they would be interchangeable. The distinction is a "can these tasks run in parallel or not" and is not a performance / resource question.As a consequence of using the ExecPlan dataset writes no longer have reliable ordering. If you pass in three batches and they are all destined for the same file then the batches may be written in any order in the destination file. This is because the ExecPlan creates a thread task for each input batch and so they could arrive at the write node in any order.