add support for xz file compression and compression feature (#3993)

* add support for xz file compression * fix Cargo.toml formatting
apache · Nov 25, 2022 · e34c6c3 · e34c6c3
1 parent 010aded
commit e34c6c3
Show file tree

Hide file tree

Showing 10 changed files with 186 additions and 78 deletions.
diff --git a/datafusion-cli/Cargo.lock b/datafusion-cli/Cargo.lock
diff --git a/datafusion/core/Cargo.toml b/datafusion/core/Cargo.toml
@@ -40,8 +40,9 @@ path = "src/lib.rs"
 [features]
 # Used to enable the avro format
 avro = ["apache-avro", "num-traits", "datafusion-common/avro"]
+compression = ["xz2", "bzip2", "flate2", "async-compression"]
 crypto_expressions = ["datafusion-physical-expr/crypto_expressions"]
-default = ["crypto_expressions", "regex_expressions", "unicode_expressions"]
+default = ["crypto_expressions", "regex_expressions", "unicode_expressions", "compression"]
 # Used for testing ONLY: causes all values to hash to the same value (test for collisions)
 force_hash_collisions = []
 # Used to enable JIT code generation
@@ -57,10 +58,10 @@ unicode_expressions = ["datafusion-physical-expr/regex_expressions", "datafusion
 ahash = { version = "0.8", default-features = false, features = ["runtime-rng"] }
 apache-avro = { version = "0.14", optional = true }
 arrow = { version = "27.0.0", features = ["prettyprint"] }
-async-compression = { version = "0.3.14", features = ["bzip2", "gzip", "futures-io", "tokio"] }
+async-compression = { version = "0.3.14", features = ["bzip2", "gzip", "xz", "futures-io", "tokio"], optional = true }
 async-trait = "0.1.41"
 bytes = "1.1"
-bzip2 = "0.4.3"
+bzip2 = { version = "0.4.3", optional = true }
 chrono = { version = "0.4.23", default-features = false }
 dashmap = "5.4.0"
 datafusion-common = { path = "../common", version = "14.0.0", features = ["parquet", "object_store"] }
@@ -70,7 +71,7 @@ datafusion-optimizer = { path = "../optimizer", version = "14.0.0" }
 datafusion-physical-expr = { path = "../physical-expr", version = "14.0.0" }
 datafusion-row = { path = "../row", version = "14.0.0" }
 datafusion-sql = { path = "../sql", version = "14.0.0" }
-flate2 = "1.0.24"
+flate2 = { version = "1.0.24", optional = true }
 futures = "0.3"
 glob = "0.3.0"
 hashbrown = { version = "0.13", features = ["raw"] }
@@ -96,6 +97,7 @@ tokio-stream = "0.1"
 tokio-util = { version = "0.7.4", features = ["io"] }
 url = "2.2"
 uuid = { version = "1.0", features = ["v4"] }
+xz2 = { version = "0.1", optional = true }
 
 [dev-dependencies]
 arrow = { version = "27.0.0", features = ["prettyprint", "dyn_cmp_dict"] }

diff --git a/datafusion/core/src/datasource/file_format/csv.rs b/datafusion/core/src/datasource/file_format/csv.rs
@@ -127,7 +127,7 @@ impl FileFormat for CsvFormat {
                 .await
                 .map_err(|e| DataFusionError::External(Box::new(e)))?;
 
-            let decoder = self.file_compression_type.convert_read(data.reader());
+            let decoder = self.file_compression_type.convert_read(data.reader())?;
             let (schema, records_read) = arrow::csv::reader::infer_reader_schema(
                 decoder,
                 self.delimiter,