when inferring the schema of compressed CSV, decompress before newline-delimited chunking #5860

jiangzhx · 2023-04-04T06:52:27Z

Which issue does this PR close?

base on @kmitchener work when inferring the schema of compressed CSV, decompress before newline-delimited chunking #5109
rebase with main branch
add example
add testcase
update testing data

Closes #5657.

Rationale for this change

#5041
#1736

What changes are included in this PR?

Are these changes tested?

Are there any user-facing changes?

jiangzhx · 2023-04-06T02:43:09Z

@jackwener @Jefffrey I wonder if you have time to review this PR.

Jefffrey

Thanks for this, left some comments

Jefffrey · 2023-04-06T23:53:27Z

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

@@ -124,8 +154,13 @@ impl FileFormat for CsvFormat {
        let mut records_to_read = self.schema_infer_max_rec.unwrap_or(usize::MAX);

        for object in objects {
-            // stream to only read as many rows as needed into memory
-            let stream = read_to_delimited_chunks(store, object).await;
+            let stream = store.get(&object.location).await.unwrap();


safe unwrap?

thanks. fixed

Jefffrey · 2023-04-06T23:56:17Z

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

+                    .unwrap()
+                    .downcast::<DataFusionError>()
+                    .unwrap())


safe unwraps?

I think it's fine.
Because err_converter only accepts std::io::Error type as input, and DataFusionError has also implemented it.

https://github.com/apache/arrow-datafusion/blob/fe46a1ed9833f2f9ea4c4ccd4d77718e5c371ab1/datafusion-common/src/error.rs#L78-L82

Jefffrey · 2023-04-06T23:57:38Z

datafusion-examples/examples/csv_sql.rs

@@ -48,5 +49,22 @@ async fn main() -> Result<()> {
    // print the results
    df.show().await?;

+    // query with


expand on this comment a bit? seems it was cutoff, so like query compressed CSV with specific options etc.

jiangzhx · 2023-04-07T04:21:42Z

Thanks for this, left some comments

@Jefffrey Thanks for the review.

Jefffrey · 2023-04-07T07:14:22Z

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

+        // #[cfg(feature = "compression")]
+        let err_converter = |e: std::io::Error| match e
+            .get_ref()
+            .and_then(|e| e.downcast_ref::<DataFusionError>())
+        {
+            Some(_) => {
+                *(e.into_inner()
+                    .unwrap()
+                    .downcast::<DataFusionError>()
+                    .unwrap())
+            }
+            None => Into::<DataFusionError>::into(e),
+        };


I find this syntax to be a confusing and not ideal, however didn't find any alternative when I experimented myself (without introducing a clone)

This unstable feature seems to solve the usecase, if ever get around to refactoring this in future if/when it becomes available: rust-lang/rust#99262

Some other points:

Is that first commented line about cfg supposed to be there or not?

None => Into::<DataFusionError>::into(e) -> None => DataFusionError::from(e) looks cleaner I think

Is that first commented line about cfg supposed to be there or not?

None => Into::<DataFusionError>::into(e) -> None => DataFusionError::from(e) looks cleaner I think

fixed.

jiangzhx · 2023-04-07T10:12:10Z

Thank you again for your review @Jefffrey

alamb

I think this code cold be simplified, and I left some suggestions. However, I think it works as written so we could also merge it in as is and simpify afterwards. Thank you @jiangzhx @izveigor and @Jefffrey

alamb · 2023-04-07T15:54:19Z

testing

I verified this is a commit on master of testing: 👍 apache/arrow-testing@47f7b56

alamb · 2023-04-07T16:04:38Z

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

+            GZIP => Box::new(
+                ReaderStream::new(AsyncGzEncoder::new(StreamReader::new(s)))
+                    .map_err(err_converter),


I am not sure why we need the err_converer -- this worked for me locally:

Suggested change

GZIP => Box::new(

ReaderStream::new(AsyncGzEncoder::new(StreamReader::new(s)))

.map_err(err_converter),

GZIP => Box::new(

ReaderStream::new(AsyncGzEncoder::new(StreamReader::new(s)))

.map_err(DataFusionError::from),

fixed. thanks

alamb · 2023-04-07T16:11:33Z

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

+    /// Return a newline delimited stream from the specified file on
+    /// Stream, decompressing if necessary
+    /// Each returned `Bytes` has a whole number of newline delimited rows
+    async fn read_to_delimited_chunks(


I feel like there must be a simpler way to express this code, but this does appear to work.

I wonder if we could use BoxStream rather than impl Stream.... 🤔

fixed,thanks a lot.

izveigor · 2023-04-07T17:06:53Z

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

+            ),
+            #[cfg(feature = "compression")]
+            ZSTD => Box::new(
+                ReaderStream::new(AsyncZstdEncoer::new(StreamReader::new(s)))


I think it should be ´AsyncZstdEncoder´

thanks, fixed.

…e-delimited chunking

alamb

Looks good -- thank you @jiangzhx

…e-delimited chunking (apache#5860)

github-actions bot added the core Core DataFusion crate label Apr 4, 2023

jiangzhx mentioned this pull request Apr 4, 2023

Request for documentation for compressed CSV/JSON support #5657

Closed

izveigor approved these changes Apr 4, 2023

View reviewed changes

jiangzhx marked this pull request as draft April 4, 2023 07:32

jiangzhx force-pushed the issues/5657 branch 3 times, most recently from 2e62a39 to 3b8f4e9 Compare April 4, 2023 12:04

jiangzhx marked this pull request as ready for review April 4, 2023 12:09

Jefffrey reviewed Apr 6, 2023

View reviewed changes

jiangzhx force-pushed the issues/5657 branch from 3b8f4e9 to 0bf639d Compare April 7, 2023 04:22

jiangzhx requested a review from Jefffrey April 7, 2023 04:27

Jefffrey approved these changes Apr 7, 2023

View reviewed changes

jiangzhx force-pushed the issues/5657 branch from 801aa67 to 38ca504 Compare April 7, 2023 09:36

alamb approved these changes Apr 7, 2023

View reviewed changes

izveigor reviewed Apr 7, 2023

View reviewed changes

jiangzhx force-pushed the issues/5657 branch 2 times, most recently from 3101231 to 9667c21 Compare April 10, 2023 07:37

when inferring the schema of compressed CSV, decompress before newlin…

aea685d

…e-delimited chunking

jiangzhx force-pushed the issues/5657 branch from 9667c21 to aea685d Compare April 10, 2023 07:48

jiangzhx requested a review from alamb April 11, 2023 10:53

alamb approved these changes Apr 11, 2023

View reviewed changes

alamb merged commit 155d0a9 into apache:main Apr 11, 2023

korowa pushed a commit to korowa/arrow-datafusion that referenced this pull request Apr 13, 2023

when inferring the schema of compressed CSV, decompress before newlin…

04bcf7c

…e-delimited chunking (apache#5860)

jiangzhx deleted the issues/5657 branch April 24, 2023 03:55

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

when inferring the schema of compressed CSV, decompress before newline-delimited chunking #5860

when inferring the schema of compressed CSV, decompress before newline-delimited chunking #5860

jiangzhx commented Apr 4, 2023 •

edited

Loading

jiangzhx commented Apr 6, 2023

Jefffrey left a comment

Jefffrey Apr 6, 2023

jiangzhx Apr 7, 2023 •

edited

Loading

Jefffrey Apr 6, 2023

jiangzhx Apr 7, 2023 •

edited

Loading

Jefffrey Apr 6, 2023

jiangzhx Apr 7, 2023

jiangzhx commented Apr 7, 2023

Jefffrey Apr 7, 2023 •

edited

Loading

jiangzhx Apr 7, 2023

jiangzhx commented Apr 7, 2023

alamb left a comment •

edited

Loading

alamb Apr 7, 2023

alamb Apr 7, 2023

jiangzhx Apr 10, 2023

alamb Apr 7, 2023

jiangzhx Apr 10, 2023

izveigor Apr 7, 2023

jiangzhx Apr 10, 2023

alamb left a comment

when inferring the schema of compressed CSV, decompress before newline-delimited chunking #5860

when inferring the schema of compressed CSV, decompress before newline-delimited chunking #5860

Conversation

jiangzhx commented Apr 4, 2023 • edited Loading

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?

jiangzhx commented Apr 6, 2023

Jefffrey left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jiangzhx Apr 7, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jiangzhx Apr 7, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jiangzhx commented Apr 7, 2023

Jefffrey Apr 7, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jiangzhx commented Apr 7, 2023

alamb left a comment • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

alamb left a comment

Choose a reason for hiding this comment

jiangzhx commented Apr 4, 2023 •

edited

Loading

jiangzhx Apr 7, 2023 •

edited

Loading

jiangzhx Apr 7, 2023 •

edited

Loading

Jefffrey Apr 7, 2023 •

edited

Loading

alamb left a comment •

edited

Loading