Simplify Write API: Add InsertExec, port in memory insert to use DataSink
#6347
Conversation
| _state: &SessionState, | ||
| _input: Arc<dyn ExecutionPlan>, | ||
| ) -> Result<Arc<dyn ExecutionPlan>> { | ||
| /// Return a [`DataSink`] suitable for writing to this table |
There was a problem hiding this comment.
Here is the new interface -- I think this probably should also get some sort of "options" parameter too so that we can provide a way to pass in settings like row_group_size from the various inserts).
I prefer to add that parameter at some future point when I have an actual usecase with COPY rather than guessing exactly what is needed.
| let num_partitions = self.batches.len(); | ||
|
|
||
| // buffer up the data round robin stle into num_partitions new buffers | ||
| let mut new_batches = vec![vec![]; num_partitions]; |
There was a problem hiding this comment.
I am quite pleased that by following @tustvold 's guidance and pushing the partitioning choice into the DataSink implementations, that the logic becomes quite a bit simpler and flexible with seemingly no performance penalty
| use crate::physical_plan::Distribution; | ||
| use datafusion_common::DataFusionError; | ||
|
|
||
| /// Execution plan for writing record batches to a [`DataSink`] |
There was a problem hiding this comment.
This is basically a generic version of MemoryWriteExec that calls a dyn DataSink to do the actual writing
|
|
||
| // Test the less-lock mode by inserting a large number of batches into a table. | ||
| #[tokio::test] | ||
| async fn test_one_to_one_mode() -> Result<()> { |
There was a problem hiding this comment.
I am not quite sure how to port these tests yet -- I am thinking about it
There was a problem hiding this comment.
I studied these tests carefully. I believe these cases with different partitioning strategies are already covered by https://github.com/apache/arrow-datafusion/blob/eb918ab217213d5e07e71e53c118a8409d2f71a0/datafusion/core/src/datasource/memory.rs#L455-L476
alamb
force-pushed
the
alamb/simplified_insert
branch
from
f70741b to
925c01b
Compare
|
Given the API change proposal does not have consensus I am closing this PR for now until we get one. I think I can get most of the benefit I was looking for from #6354 |
Which issue does this PR close?
Closes #6339
Rationale for this change
I want to make it easier for DataFusion to be extended with new write sources. I want to use this both for
COPY ... TO ...statements as well as @JanKaul is looking into using it for Delta.rs: #6339 (comment)What changes are included in this PR?
MemTablewrite to be in terms of DataSink (it was really helpful to have a specific implemetation to try this with -- thanks @metesynnada )MemoryWriteExec, theMemTablespecific execution planAre these changes tested?
Yes
Are there any user-facing changes?
The API for insert changes.