datafusion icon indicating copy to clipboard operation
datafusion copied to clipboard
verified publisher icon apache

[EPIC] Streaming partitioned writes

Open alamb opened this issue 2 years ago • 11 comments

Is your feature request related to a problem or challenge?

This is a tracking epic for a collection of features related to writing data.

The basic idea is better / full support for writing data:

  1. to multiple (possibly Partitoned by value) files
  2. To different file types (parquet, csv, json, avro, arrow)
  3. In a streaming fashion (input doesn't need to be entirely buffered)
  4. From SQL (via INSERT, INSERT INTO, COPY, Etc)
  5. Stream to a target object_store (aka multi-part S3 upload)

This is partially supported today programmatically (see SessionContext::write_csv, etc)

Subtasks:

  • [x] https://github.com/apache/arrow-datafusion/issues/5130
  • [x] https://github.com/apache/arrow-datafusion/pull/6526
  • [x] https://github.com/apache/arrow-datafusion/issues/5654
  • [x] End to end SQL tests for writing data to a file (e.g for https://github.com/apache/arrow-datafusion/pull/6526)
  • [ ] Tests for streaming writes (that they actually stream while writing rather than buffering)
  • [ ] Tests for abort behavior (making sure all writers are canceled and the correct error is returned)
  • [x] Implement the write API for JSON and parquet files (https://github.com/apache/arrow-datafusion/pull/7141 and https://github.com/apache/arrow-datafusion/pull/7244)
  • [ ] Update the user guide Documentation to say that DataFusion supports streaming multi-part writes
  • [x] https://github.com/apache/arrow-datafusion/issues/5076
  • [x] https://github.com/apache/arrow-datafusion/issues/7079
  • [x] https://github.com/apache/arrow-datafusion/issues/7036
  • [x] https://github.com/apache/arrow-datafusion/issues/7298
  • [ ] https://github.com/apache/arrow-datafusion/issues/7317
  • [x] https://github.com/apache/arrow-datafusion/issues/7319
  • [x] https://github.com/apache/arrow-datafusion/issues/7322
  • [x] https://github.com/apache/arrow-datafusion/issues/7354
  • [x] https://github.com/apache/arrow-datafusion/issues/7442
  • [x] https://github.com/apache/arrow-datafusion/issues/7499
  • [x] https://github.com/apache/arrow-datafusion/issues/7536
  • [x] https://github.com/apache/arrow-datafusion/issues/7590
  • [x] https://github.com/apache/arrow-datafusion/issues/7589
  • [x] https://github.com/apache/arrow-datafusion/issues/7591
  • [ ] https://github.com/apache/arrow-datafusion/issues/7679
  • [x] https://github.com/apache/arrow-datafusion/issues/7744
  • [x] https://github.com/apache/arrow-datafusion/issues/6983
  • [x] https://github.com/apache/arrow-datafusion/issues/7767
  • [x] https://github.com/apache/arrow-datafusion/issues/5383
  • [x] https://github.com/apache/arrow-datafusion/issues/7891
  • [x] https://github.com/apache/arrow-datafusion/issues/7892
  • [x] https://github.com/apache/arrow-datafusion/issues/8493
  • [x] https://github.com/apache/arrow-datafusion/issues/8502
  • [x] https://github.com/apache/arrow-datafusion/issues/8503
  • [x] https://github.com/apache/arrow-datafusion/issues/8504
  • [x] https://github.com/apache/arrow-datafusion/issues/8547
  • [x] https://github.com/apache/arrow-datafusion/issues/8621
  • [ ] https://github.com/apache/arrow-datafusion/issues/8635
  • [x] https://github.com/apache/arrow-datafusion/issues/8657
  • [x] https://github.com/apache/arrow-datafusion/issues/9237

alamb avatar Jun 06 '23 19:06 alamb

FYI @devinjdangelo I updated this ticket with various issues related to the write code you are working on

alamb avatar Aug 14 '23 12:08 alamb

@alamb I opened #7298 to track improving statement level options/overrides relevant to this epic.

devinjdangelo avatar Aug 16 '23 11:08 devinjdangelo

@alamb I opened https://github.com/apache/arrow-datafusion/issues/7298 to track improving statement level options/overrides relevant to this epic.

Thanks @devinjdangelo -- I added it to the list

alamb avatar Aug 16 '23 16:08 alamb

An additional issue we should cut and add to this epic is allowing inserts to a sorted ListingTable. In the case of appending new files to a directory, I think it is as simple as having FileSinkExec require its input be sorted.

It can't really be supported efficiently for Append to existing file since it would require reading the existing file, sorting with the new data and rewriting the whole file. For this case, you could use insert overwrite instead if you really want to do this (which is another thing which we could cut a ticket to add support for).

Alternatively, we could have a check to see if 1) the table is sorted and 2) the input to FileSinkExec is sorted. If 1) is true but 2) is not, we would need to update the metadata about the table to indicate for subsequent queries it is no longer guaranteed to be sorted.

devinjdangelo avatar Aug 18 '23 11:08 devinjdangelo

Filed https://github.com/apache/arrow-datafusion/issues/7354 to track

alamb avatar Aug 21 '23 18:08 alamb

@alamb I opened #7442 to track adding support for parquet column level settings via SQL options.

devinjdangelo avatar Aug 29 '23 13:08 devinjdangelo

@alamb I opened https://github.com/apache/arrow-datafusion/issues/7442 to track adding support for parquet column level settings via SQL options.

Thanks @devinjdangelo -- I added it to the list on this ticket

alamb avatar Sep 05 '23 17:09 alamb

I added #7679 to track adding avro support for writes @alamb

devinjdangelo avatar Sep 28 '23 01:09 devinjdangelo

@alamb I made some progress on inserts to sorted tables https://github.com/apache/arrow-datafusion/issues/7354

This also got me thinking about inserts to partitioned tables, so I opened issue to track: https://github.com/apache/arrow-datafusion/issues/7744

Lastly, I've been thinking we may want to deprecate and eventually remove the SessionContext write methods or alternatively hook them into FileSinkExec similarly to how the data frame write methods are set up. https://github.com/apache/arrow-datafusion/blob/0408c2b1596417ba55a636fa3c8a601ffbdb0e60/datafusion/core/src/execution/context.rs#L1271-L1278

devinjdangelo avatar Oct 05 '23 02:10 devinjdangelo

This also got me thinking about inserts to partitioned tables, so I opened issue to track: https://github.com/apache/arrow-datafusion/issues/7744

Thank you -- I added https://github.com/apache/arrow-datafusion/issues/7744 to the list on this ticket

Lastly, I've been thinking we may want to deprecate and eventually remove the SessionContext write methods or alternatively hook them into FileSinkExec similarly to how the data frame write methods are set up.

That sounds like a reasonable idea to me. One challenge might be that the SessionContext::write_csv take an ExecutionPlan where the apis on DataFrame require a LogicalPlan.

Hooking them into FileSinkExec sounds like a great idea to consolidate the code. Shall I file a ticket for it?

alamb avatar Oct 05 '23 18:10 alamb

@alamb I wrote up an issue describing what we discussed on #7743 regarding empty files being written out and some potential solutions.

https://github.com/apache/arrow-datafusion/issues/7767

devinjdangelo avatar Oct 07 '23 12:10 devinjdangelo