feat(files): read Excel, XML and ZIP inputs

Cargo.toml

@@ -32,6 +32,7 @@ async-trait = "0.1.92"
 axum = { version = "0.8", default-features = false, features = ["http1", "tokio"] }
 base64 = "0.22"
 bytes = "1"
+calamine = { version = "0.38", default-features = false, features = ["chrono"] }
 chrono = { version = "0.4", default-features = false, features = ["std"] }
 chrono-tz = { version = "0.10", default-features = false }
 duckdb = { version = "1.20000", features = ["bundled", "appender-arrow", "parquet"] }
@@ -50,13 +51,15 @@ object_store = { version = "0.14", default-features = false, features = ["aws-ba
 jsonwebtoken = { version = "11", default-features = false, features = ["use_pem", "rust_crypto"] }
 parquet = { version = "60", default-features = false, features = ["arrow", "zstd", "async", "object_store", "snap", "lz4", "flate2", "flate2-rust_backend"] }
 prometheus-client = "0.25"
+quick-xml = "0.42"
 pyo3 = { version = "0.29", features = ["abi3-py310"] }
 rdkafka = { version = "0.39", default-features = false, features = ["tokio", "libz", "zstd"] }
 rcgen = { version = "0.14", default-features = false, features = ["crypto", "pem", "ring"] }
 regex = "1"
 reqwest = { version = "0.13", default-features = false, features = ["rustls-no-provider", "http2", "json", "form", "query", "gzip"] }
 rlimit = "0.11"
 rsa = { version = "0.9", features = ["pem"] }
+rust_xlsxwriter = { version = "0.102", default-features = false }
 russh = { version = "0.65", default-features = false, features = ["ring", "rsa"] }
 russh-sftp = { version = "3.0.1", default-features = false }
 rustix = { version = "1.1", default-features = false, features = ["std", "fs"] }
@@ -86,6 +89,7 @@ tracing-subscriber = { version = "0.3", features = ["fmt", "json", "env-filter"]
 uuid = { version = "1", features = ["v4"] }
 webpki-roots = "1"
 zstd = "0.14"
+zip = { version = "9", default-features = false, features = ["deflate"] }
 
 [workspace.lints.rust]
 unsafe_code = "forbid"

crates/sqlload/Cargo.toml

@@ -50,6 +50,7 @@ async-trait.workspace = true
 axum.workspace = true
 base64.workspace = true
 bytes.workspace = true
+calamine.workspace = true
 chrono.workspace = true
 chrono-tz.workspace = true
 clap = { workspace = true, optional = true }
@@ -67,6 +68,7 @@ jsonwebtoken.workspace = true
 object_store.workspace = true
 parquet.workspace = true
 prometheus-client.workspace = true
+quick-xml.workspace = true
 regex.workspace = true
 rdkafka.workspace = true
 reqwest.workspace = true
@@ -90,6 +92,7 @@ tonic.workspace = true
 tonic-prost.workspace = true
 toml.workspace = true
 toml_edit.workspace = true
+zip.workspace = true
 tower-layer.workspace = true
 tower-service.workspace = true
 tracing.workspace = true
@@ -109,6 +112,7 @@ serde_json.workspace = true
 sqlload-testkit.workspace = true
 tokio.workspace = true
 tokio-util.workspace = true
+rust_xlsxwriter.workspace = true
 
 [lints]
 workspace = true

crates/sqlload/src/adapters/files/_helpers/write/csv_files.rs

@@ -52,7 +52,9 @@ fn row_line(columns: &[Vec<Option<String>>], row: usize, csv: &CsvOutput) -> Str
 pub(crate) fn new_csv(compression: OutputCompression) -> CsvOut {
     match compression {
         OutputCompression::None => CsvOut::Plain(Vec::new()),
-        OutputCompression::Gzip => CsvOut::Gzip(GzEncoder::new(Vec::new(), GzipLevel::default())),
+        OutputCompression::Gzip => {
+            CsvOut::Gzip(Box::new(GzEncoder::new(Vec::new(), GzipLevel::default())))
+        }
         OutputCompression::Zstd => CsvOut::Zstd {
             done: Vec::new(),
             pending: Vec::new(),

crates/sqlload/src/adapters/files/models.rs

@@ -197,7 +197,7 @@ pub enum FileBody {
 #[derive(Debug)]
 pub enum CsvOut {
     Plain(Vec<u8>),
-    Gzip(flate2::write::GzEncoder<Vec<u8>>),
+    Gzip(Box<flate2::write::GzEncoder<Vec<u8>>>),
     /// zstd frames compressed so far, and text waiting for the next frame.
     Zstd {
         done: Vec<u8>,

crates/sqlload/src/adapters/files/tests/helpers.rs

@@ -57,6 +57,7 @@ pub(crate) fn stored(key: &str) -> StoredFile {
         size: 1,
         etag: None,
         modified_ms: 0,
+        member: None,
     }
 }
 

crates/sqlload/src/adapters/snowflake/_helpers/dialect/text_values.rs

@@ -213,11 +213,13 @@ fn misfit(
     )
 }
 
+/// The `COMPRESSION` of a `COPY` of files compressed as `compression`; ZIP archives never reach a `COPY`, as SQLLoad reads their members itself.
 fn compression_name(compression: Compression) -> &'static str {
     match compression {
         Compression::Auto => "AUTO",
         Compression::None => "NONE",
         Compression::Gzip => "GZIP",
         Compression::Zstd => "ZSTD",
+        Compression::Zip => "NONE",
     }
 }

crates/sqlload/src/adapters/snowflake/tests/helpers.rs

@@ -153,6 +153,7 @@ pub(crate) fn data_file(key: &str, day: &str) -> DataFile {
             size: 10,
             etag: Some(format!("\"etag-{day}\"")),
             modified_ms: 0,
+            member: None,
         },
         partitions: vec![("day".to_owned(), day.to_owned())],
     }

crates/sqlload/src/batch_plan/tests/helpers.rs

@@ -72,6 +72,7 @@ fn data_file(partition: &PartitionFiles, index: usize) -> DataFile {
             size: partition.file_bytes,
             etag: None,
             modified_ms: 0,
+            member: None,
         },
         partitions: vec![
             ("region".to_owned(), partition.region.to_owned()),

crates/sqlload/src/file_store/main/list_files.rs

@@ -38,6 +38,7 @@ pub async fn list_files(folder: &OpenFolder) -> Result<Vec<StoredFile>, StoreErr
             etag: meta.e_tag,
             modified_ms: meta.last_modified.timestamp_millis(),
             path: meta.location,
+            member: None,
         });
     }
     files.sort_by(|left, right| left.relative_key.cmp(&right.relative_key));

crates/sqlload/src/file_store/models.rs

@@ -22,6 +22,8 @@ pub struct StoredFile {
     /// The store's version tag, used to notice a file rewritten in place.
     pub etag: Option<String>,
     pub modified_ms: i64,
+    /// A member of the ZIP archive at `path`: its name in the archive, whose key is then the archive's key, `/` and this name.
+    pub member: Option<String>,
 }
 
 impl StoredFile {

crates/sqlload/src/files_source/_helpers/archive/mod.rs

@@ -0,0 +1 @@
+pub(crate) mod zip_archive;