diff --git a/Cargo.lock b/Cargo.lock index 8fd65076..0d401d72 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -526,6 +526,16 @@ dependencies = [ "num-traits", ] +[[package]] +name = "atoi_simd" +version = "0.18.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f3cdb3708a128e559a30fb830e8a77a5022ee6902806925c216658652b452a44" +dependencies = [ + "debug_unsafe", + "rustversion", +] + [[package]] name = "atomic-waker" version = "1.1.2" @@ -827,6 +837,24 @@ version = "1.12.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "fc652a48c352aef3ea3aed32080501cf3ef6ed5da78602a020c991775b0aff04" +[[package]] +name = "calamine" +version = "0.38.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f431b67cba6122a1f7d26b7adafc71bbdf07026095fe811f5efc025d8748e1d8" +dependencies = [ + "atoi_simd", + "byteorder", + "chrono", + "codepage", + "encoding_rs", + "fast-float2", + "log", + "quick-xml 0.42.0", + "serde", + "zip", +] + [[package]] name = "calloop" version = "0.13.0" @@ -985,6 +1013,15 @@ version = "0.5.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0c9ea0ac24bc397ab3c98583a3c9ba74fa56b09a4449bbe172b9b1ddb016027a" +[[package]] +name = "codepage" +version = "0.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bdff162541cd8b79de82e2edcc7eff3a8c2a6dc3d75152636028f96d93de3b26" +dependencies = [ + "encoding_rs", +] + [[package]] name = "colorchoice" version = "1.0.5" @@ -1071,6 +1108,15 @@ dependencies = [ "tiny-keccak", ] +[[package]] +name = "constant_time_eq" +version = "0.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9eb6feb175df65d0a3ccd3e73484e940d0ac664e8034f0519fd49a93db3071d9" +dependencies = [ + "rustversion", +] + [[package]] name = "core-foundation" version = "0.9.4" @@ -1386,6 +1432,12 @@ version = "2.11.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4583a4551df46e2792f82ceeac45e850d2e2d5debba0b91f102385cda5b11f06" +[[package]] +name = "debug_unsafe" +version = "0.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7eed2c4702fa172d1ce21078faa7c5203e69f5394d48cc436d25928394a867a2" + [[package]] name = "defmt" version = "1.2.0" @@ -1819,6 +1871,12 @@ version = "0.1.9" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7360491ce676a36bf9bb3c56c1aa791658183a54d2744120f27285738d90465a" +[[package]] +name = "fast-float2" +version = "0.2.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c6e8948ce679d00a02a94739ea185595dca7118ed04feb991127e443bd3d761f" + [[package]] name = "fastrand" version = "2.5.0" @@ -3600,7 +3658,7 @@ dependencies = [ "nix", "parking_lot", "percent-encoding", - "quick-xml", + "quick-xml 0.41.0", "rand 0.10.3", "reqwest", "ring", @@ -4408,6 +4466,16 @@ dependencies = [ "serde", ] +[[package]] +name = "quick-xml" +version = "0.42.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "41b1177fdf999d2321d3fb46ff47159d9c1fb9ad66a4879f8c50a0b504615e9b" +dependencies = [ + "encoding_rs", + "memchr", +] + [[package]] name = "quote" version = "1.0.47" @@ -4880,6 +4948,15 @@ dependencies = [ "wasm-bindgen-futures", ] +[[package]] +name = "rust_xlsxwriter" +version = "0.102.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6ca3d0e879a578c1313c2800ab0c6c029d38eed18098675e50341a30d32372ed" +dependencies = [ + "zip", +] + [[package]] name = "rustc-hash" version = "2.1.3" @@ -5555,6 +5632,7 @@ dependencies = [ "axum", "base64 0.22.1", "bytes", + "calamine", "chrono", "chrono-tz", "clap", @@ -5574,6 +5652,7 @@ dependencies = [ "pgwire-replication", "prometheus-client", "prost", + "quick-xml 0.42.0", "rdkafka", "regex", "reqwest", @@ -5581,6 +5660,7 @@ dependencies = [ "rsa 0.9.10", "russh", "russh-sftp", + "rust_xlsxwriter", "rustix 1.1.6", "rustls", "serde", @@ -5603,6 +5683,7 @@ dependencies = [ "tracing-subscriber", "uuid", "webpki-roots", + "zip", "zstd", ] @@ -6341,6 +6422,12 @@ version = "2.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "86a801b3cea342a06d468c8710662aa29e5e05e4f5c0d62f00bbb7f2ad7941c2" +[[package]] +name = "typed-path" +version = "0.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9ce898823d8cd0f028662d2f349694b1af0a5cc9be83f1d213a00d3147445e1e" + [[package]] name = "typenum" version = "1.20.1" @@ -7158,6 +7245,26 @@ dependencies = [ "syn 3.0.9", ] +[[package]] +name = "zip" +version = "9.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e3476fd497ac8af5b1ec7b51acab7bef380f7e8dd8e0487595bafd839417a302" +dependencies = [ + "constant_time_eq", + "crc32fast", + "flate2", + "getrandom 0.4.3", + "indexmap 2.14.2", + "libc", + "memchr", + "nix", + "typed-path", + "windows-sys 0.61.2", + "zeroize", + "zopfli", +] + [[package]] name = "zlib-rs" version = "0.6.8" @@ -7170,6 +7277,18 @@ version = "1.0.23" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "29666d0abbfad1e3dc4dcf6144730dd3a3ab225bbbdac83319345b1b44ccfc1b" +[[package]] +name = "zopfli" +version = "0.8.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "aaf7fc5d30c28483d93805c4a5e12b05bbb52407fa67c5f8bd552374cd01fb11" +dependencies = [ + "bumpalo", + "crc32fast", + "log", + "simd-adler32", +] + [[package]] name = "zstd" version = "0.14.3" diff --git a/Cargo.toml b/Cargo.toml index 187a736e..d3e92f51 100644 --- a/Cargo.toml +++ b/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,6 +51,7 @@ 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"] } @@ -57,6 +59,7 @@ 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" diff --git a/crates/sqlload/Cargo.toml b/crates/sqlload/Cargo.toml index 9f471cab..5701a017 100644 --- a/crates/sqlload/Cargo.toml +++ b/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 diff --git a/crates/sqlload/src/adapters/files/_helpers/write/csv_files.rs b/crates/sqlload/src/adapters/files/_helpers/write/csv_files.rs index fad5cf8d..d885da61 100644 --- a/crates/sqlload/src/adapters/files/_helpers/write/csv_files.rs +++ b/crates/sqlload/src/adapters/files/_helpers/write/csv_files.rs @@ -52,7 +52,9 @@ fn row_line(columns: &[Vec>], 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(), diff --git a/crates/sqlload/src/adapters/files/models.rs b/crates/sqlload/src/adapters/files/models.rs index b04d4c0d..3337d7d0 100644 --- a/crates/sqlload/src/adapters/files/models.rs +++ b/crates/sqlload/src/adapters/files/models.rs @@ -197,7 +197,7 @@ pub enum FileBody { #[derive(Debug)] pub enum CsvOut { Plain(Vec), - Gzip(flate2::write::GzEncoder>), + Gzip(Box>>), /// zstd frames compressed so far, and text waiting for the next frame. Zstd { done: Vec, diff --git a/crates/sqlload/src/adapters/files/tests/helpers.rs b/crates/sqlload/src/adapters/files/tests/helpers.rs index 6541236a..7becb4f0 100644 --- a/crates/sqlload/src/adapters/files/tests/helpers.rs +++ b/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, } } diff --git a/crates/sqlload/src/adapters/snowflake/_helpers/dialect/text_values.rs b/crates/sqlload/src/adapters/snowflake/_helpers/dialect/text_values.rs index 5efb72ba..adac95ea 100644 --- a/crates/sqlload/src/adapters/snowflake/_helpers/dialect/text_values.rs +++ b/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", } } diff --git a/crates/sqlload/src/adapters/snowflake/tests/helpers.rs b/crates/sqlload/src/adapters/snowflake/tests/helpers.rs index fd733adb..d91c915d 100644 --- a/crates/sqlload/src/adapters/snowflake/tests/helpers.rs +++ b/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())], } diff --git a/crates/sqlload/src/batch_plan/tests/helpers.rs b/crates/sqlload/src/batch_plan/tests/helpers.rs index e1abfa43..5d721672 100644 --- a/crates/sqlload/src/batch_plan/tests/helpers.rs +++ b/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()), diff --git a/crates/sqlload/src/file_store/main/list_files.rs b/crates/sqlload/src/file_store/main/list_files.rs index 5e54419b..bcd3764e 100644 --- a/crates/sqlload/src/file_store/main/list_files.rs +++ b/crates/sqlload/src/file_store/main/list_files.rs @@ -38,6 +38,7 @@ pub async fn list_files(folder: &OpenFolder) -> Result, 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)); diff --git a/crates/sqlload/src/file_store/models.rs b/crates/sqlload/src/file_store/models.rs index 68d95ccc..b23cc1c5 100644 --- a/crates/sqlload/src/file_store/models.rs +++ b/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, 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, } impl StoredFile { diff --git a/crates/sqlload/src/files_source/_helpers/archive/mod.rs b/crates/sqlload/src/files_source/_helpers/archive/mod.rs new file mode 100644 index 00000000..dc4ba5e6 --- /dev/null +++ b/crates/sqlload/src/files_source/_helpers/archive/mod.rs @@ -0,0 +1 @@ +pub(crate) mod zip_archive; diff --git a/crates/sqlload/src/files_source/_helpers/archive/zip_archive.rs b/crates/sqlload/src/files_source/_helpers/archive/zip_archive.rs new file mode 100644 index 00000000..3bf766fa --- /dev/null +++ b/crates/sqlload/src/files_source/_helpers/archive/zip_archive.rs @@ -0,0 +1,171 @@ +//! Read the members of ZIP archives with range requests: the index at an archive's end, then one member, never the whole archive. + +use std::io::{Read, Seek, SeekFrom}; +use std::sync::Arc; + +use bytes::Bytes; +use object_store::path::Path; +use object_store::{ObjectStore, ObjectStoreExt}; +use tokio::runtime::Handle; +use zip::ZipArchive; + +use crate::file_store::models::{OpenFolder, StoredFile}; +use crate::files_source::_helpers::plan::encoding::member_fits; +use crate::files_source::constants::{MAX_WHOLE_FILE_BYTES, ZIP_READ_BLOCK_BYTES}; +use crate::files_source::models::FilesEncoding; + +/// A stored file read as it is sought and read, a block per range request, from a blocking thread. +struct RangedFile { + store: Arc, + path: Path, + size: u64, + position: u64, + /// The last block read and where it starts. + block: (u64, Bytes), + handle: Handle, +} + +impl RangedFile { + fn new(folder: &OpenFolder, path: &Path, size: u64) -> Self { + Self { + store: folder.store.clone(), + path: path.clone(), + size, + position: 0, + block: (0, Bytes::new()), + handle: Handle::current(), + } + } +} + +impl Read for RangedFile { + fn read(&mut self, out: &mut [u8]) -> std::io::Result { + let (start, block) = &self.block; + let held = self.position >= *start && self.position < start + block.len() as u64; + if !held && self.position < self.size { + let end = self.size.min(self.position + ZIP_READ_BLOCK_BYTES); + let bytes = self + .handle + .block_on(self.store.get_range(&self.path, self.position..end)) + .map_err(std::io::Error::other)?; + self.block = (self.position, bytes); + } + let (start, block) = &self.block; + let offset = usize::try_from(self.position.saturating_sub(*start)) + .unwrap_or(usize::MAX) + .min(block.len()); + let count = out.len().min(block.len() - offset); + out[..count].copy_from_slice(&block[offset..offset + count]); + self.position += count as u64; + Ok(count) + } +} + +impl Seek for RangedFile { + fn seek(&mut self, to: SeekFrom) -> std::io::Result { + let position = match to { + SeekFrom::Start(offset) => Some(offset), + SeekFrom::End(offset) => self.size.checked_add_signed(offset), + SeekFrom::Current(offset) => self.position.checked_add_signed(offset), + }; + self.position = position.ok_or_else(|| { + std::io::Error::new( + std::io::ErrorKind::InvalidInput, + "seek before the start of the file", + ) + })?; + Ok(self.position) + } +} + +/// The members of `archive`, a listed ZIP archive, that are `encoding`'s files, each a file of its own keyed by the archive's key, `/` and its name, in the archive's order; only the archive's index is read. One of them declaring more than SQLLoad reads is an error. +pub(crate) async fn archive_members( + folder: &OpenFolder, + archive: &StoredFile, + encoding: &FilesEncoding, +) -> Result, String> { + let file = RangedFile::new(folder, &archive.path, archive.size); + let entries = tokio::task::spawn_blocking(move || member_sizes(file)) + .await + .map_err(|error| error.to_string())??; + let mut members: Vec = Vec::new(); + for (name, size) in entries { + if member_fits(&name, encoding)? { + if size > MAX_WHOLE_FILE_BYTES { + return Err(too_large(&name, size)); + } + members.push(StoredFile { + path: archive.path.clone(), + relative_key: format!("{}/{name}", archive.relative_key), + size, + etag: archive.etag.clone(), + modified_ms: archive.modified_ms, + member: Some(name), + }); + } + } + Ok(members) +} + +/// Member `name` of the ZIP archive `file` is in, decompressed; only the archive's index and that member are read. +pub(crate) async fn member_bytes( + folder: &OpenFolder, + file: &StoredFile, + name: &str, +) -> Result, String> { + let size = folder + .store + .head(&file.path) + .await + .map_err(|error| format!("cannot read the file: {error}"))? + .size; + let ranged = RangedFile::new(folder, &file.path, size); + let name = name.to_owned(); + tokio::task::spawn_blocking(move || { + let mut archive = open_archive(ranged)?; + let mut member = archive + .by_name(&name) + .map_err(|error| format!("member {name} of the ZIP archive: {error}"))?; + if member.size() > MAX_WHOLE_FILE_BYTES { + return Err(too_large(&name, member.size())); + } + let mut data: Vec = Vec::new(); + member + .read_to_end(&mut data) + .map_err(|error| format!("member {name} of the ZIP archive: {error}"))?; + Ok(data) + }) + .await + .map_err(|error| error.to_string())? +} + +/// Each file member's name and decompressed size, folders left out; an archive naming a member twice is an error. +fn member_sizes(file: RangedFile) -> Result, String> { + let archive = open_archive(file)?; + if archive.has_duplicate_names() { + return Err("the ZIP archive holds two members with the same name".to_owned()); + } + let mut entries: Vec<(String, u64)> = Vec::new(); + for index in 0..archive.len() { + let member = archive + .by_index_data(index) + .map_err(|error| format!("cannot read the ZIP archive: {error}"))?; + if member.is_file() { + let name = member + .name() + .map_err(|error| format!("cannot read the ZIP archive: {error}"))?; + entries.push((name.into_owned(), member.size())); + } + } + Ok(entries) +} + +fn too_large(name: &str, size: u64) -> String { + format!( + "member {name} of the ZIP archive is {size} bytes decompressed; SQLLoad reads members whole, up to {MAX_WHOLE_FILE_BYTES} bytes" + ) +} + +fn open_archive(file: RangedFile) -> Result, String> { + ZipArchive::new(file).map_err(|error| format!("cannot read the ZIP archive: {error}")) +} diff --git a/crates/sqlload/src/files_source/_helpers/cells/cell_reader.rs b/crates/sqlload/src/files_source/_helpers/cells/cell_reader.rs new file mode 100644 index 00000000..41ea8a11 --- /dev/null +++ b/crates/sqlload/src/files_source/_helpers/cells/cell_reader.rs @@ -0,0 +1,102 @@ +//! Read Excel worksheets and XML documents whole, as records of typed cells. + +use std::collections::HashMap; +use std::sync::Arc; + +use arrow_array::{ArrayRef, RecordBatch, RecordBatchOptions, StringArray}; +use arrow_schema::{DataType, Field, Schema}; + +use crate::column_types::main::combined_type::combined_type; +use crate::column_types::models::ColumnType; +use crate::files_source::_helpers::cells::xlsx_records::xlsx_records; +use crate::files_source::_helpers::cells::xml_records::xml_records; +use crate::files_source::_helpers::text::bytes_reader::BytesReader; +use crate::files_source::_helpers::text::checks::{ColumnChecks, misfit, value_fits}; +use crate::files_source::models::{CellRecord, FilesEncoding, SourceFormat}; + +/// A worksheet's or XML document's records, read whole, handed out a batch at a time. +pub(crate) struct CellReader { + /// A worksheet's column names, in sheet order; empty for XML, whose records name their own. + names: Vec, + records: std::vec::IntoIter, +} + +impl CellReader { + /// The records of the file `bytes` reads, an xlsx workbook or an XML document as `encoding` says. + pub(crate) async fn open( + mut bytes: BytesReader, + encoding: &FilesEncoding, + ) -> Result { + let whole = bytes.whole().await?; + let encoding = encoding.clone(); + let (names, records) = tokio::task::spawn_blocking(move || match encoding.format { + SourceFormat::Xlsx => xlsx_records(whole, &encoding.xlsx), + _ => xml_records(&whole, &encoding.xml).map(|records| (Vec::new(), records)), + }) + .await + .map_err(|error| error.to_string())??; + Ok(Self { + names, + records: records.into_iter(), + }) + } + + pub(crate) fn names(&self) -> &[String] { + &self.names + } + + /// Up to `count` more records; empty at the end. + pub(crate) fn next_records(&mut self, count: usize) -> Vec { + self.records.by_ref().take(count).collect() + } +} + +/// The type of a column holding cells of types `seen` and `value`: a date-formatted cell at midnight reads as a `date`, so a date and a timestamp make a `timestamp_ntz`; otherwise as for any text format. +pub(crate) fn cell_type(seen: ColumnType, value: ColumnType) -> ColumnType { + match (seen, value) { + (ColumnType::Date, ColumnType::TimestampNtz) + | (ColumnType::TimestampNtz, ColumnType::Date) => ColumnType::TimestampNtz, + (seen, value) => combined_type(seen, value), + } +} + +/// `records` as text columns `columns`, with the line each starts on; a cell no column takes, or a value an inferred column would not have been inferred from, is an error. +pub(crate) fn cells_batch( + records: &[CellRecord], + columns: &[(String, ColumnType)], + checks: &ColumnChecks, +) -> Result<(RecordBatch, Vec), String> { + let index: HashMap<&str, usize> = columns + .iter() + .enumerate() + .map(|(position, (name, _))| (name.as_str(), position)) + .collect(); + let mut values: Vec>> = vec![vec![None; records.len()]; columns.len()]; + for (row, record) in records.iter().enumerate() { + for (name, cell) in &record.cells { + let Some(position) = index.get(name.as_str()) else { + checks.check_name("column", name, record.line)?; + continue; + }; + if let Some(column_type) = checks.inferred_type(name) + && !value_fits(column_type, Some(cell.kind.clone())) + && cell_type(column_type.clone(), cell.kind.clone()) != *column_type + { + return Err(misfit(record.line, name, &cell.text, column_type)); + } + values[*position][row] = Some(cell.text.clone()); + } + } + let fields: Vec = columns + .iter() + .map(|(name, _)| Field::new(name, DataType::Utf8, true)) + .collect(); + let arrays: Vec = values + .into_iter() + .map(|column| Arc::new(StringArray::from(column)) as ArrayRef) + .collect(); + let options = RecordBatchOptions::new().with_row_count(Some(records.len())); + let batch = RecordBatch::try_new_with_options(Arc::new(Schema::new(fields)), arrays, &options) + .map_err(|error| error.to_string())?; + Ok((batch, records.iter().map(|record| record.line).collect())) +} diff --git a/crates/sqlload/src/files_source/_helpers/cells/mod.rs b/crates/sqlload/src/files_source/_helpers/cells/mod.rs new file mode 100644 index 00000000..4491e66d --- /dev/null +++ b/crates/sqlload/src/files_source/_helpers/cells/mod.rs @@ -0,0 +1,3 @@ +pub(crate) mod cell_reader; +pub(crate) mod xlsx_records; +pub(crate) mod xml_records; diff --git a/crates/sqlload/src/files_source/_helpers/cells/xlsx_records.rs b/crates/sqlload/src/files_source/_helpers/cells/xlsx_records.rs new file mode 100644 index 00000000..71d97bf6 --- /dev/null +++ b/crates/sqlload/src/files_source/_helpers/cells/xlsx_records.rs @@ -0,0 +1,157 @@ +//! Read an Excel worksheet's rows as typed cells. + +use std::collections::BTreeMap; +use std::io::Cursor; + +use calamine::{Data, ExcelDateTime, Reader, Xlsx, XlsxError}; +use chrono::NaiveTime; + +use crate::column_types::models::ColumnType; +use crate::files_source::constants::{ + EXCEL_DAY, MAX_EXACT_FLOAT_INT, MAX_SHEET_CELLS, NORMAL_DATE, NORMAL_TIME, NORMAL_TIMESTAMP, +}; +use crate::files_source::models::{Cell, CellRecord, XlsxOptions}; + +/// The column names and rows of `options`' worksheet in `workbook`, each row numbered as in the sheet; empty rows are left out, and so is a column with an empty header and no values. Cells are read one at a time, so a sheet's stated size is never allocated; a sheet over `MAX_SHEET_CELLS` values is an error. +pub(crate) fn xlsx_records( + workbook: Vec, + options: &XlsxOptions, +) -> Result<(Vec, Vec), String> { + let mut book = Xlsx::new(Cursor::new(workbook)) + .map_err(|error| format!("cannot read the workbook: {error}"))?; + let sheets = book.sheet_names(); + let sheet = match &options.sheet { + Some(sheet) if sheets.contains(sheet) => sheet.clone(), + Some(sheet) => { + return Err(format!( + "the workbook has no sheet {sheet:?}; its sheets are {sheets:?}" + )); + } + None => sheets + .first() + .cloned() + .ok_or_else(|| "the workbook has no sheets".to_owned())?, + }; + let sheet_error = |error: XlsxError| format!("sheet {sheet:?}: {error}"); + let mut reader = book.worksheet_cells_reader(&sheet).map_err(sheet_error)?; + let mut rows: BTreeMap> = BTreeMap::new(); + let mut count = 0_usize; + while let Some(found) = reader.next_cell().map_err(sheet_error)? { + let (row, column) = found.get_position(); + let row = u64::from(row) + 1; + let Some(cell) = cell_of(&Data::from(found.get_value().clone())) else { + continue; + }; + if row <= u64::from(options.skip_rows) { + continue; + } + count += 1; + if count > MAX_SHEET_CELLS { + return Err(format!( + "sheet {sheet:?} holds over {MAX_SHEET_CELLS} values; SQLLoad reads a worksheet whole, up to that many" + )); + } + rows.entry(row).or_default().push((column, cell)); + } + let mut rows = rows.into_iter(); + let mut headers: BTreeMap = BTreeMap::new(); + if options.header + && let Some((row, cells)) = rows.next() + { + for (column, cell) in cells { + if headers.values().any(|name| *name == cell.text) { + return Err(format!( + "row {row}: column {} appears twice in the header", + cell.text + )); + } + headers.insert(column, cell.text); + } + } + let mut names = headers; + let mut records: Vec = Vec::new(); + for (row, cells) in rows { + let mut record = CellRecord { + line: row, + cells: Vec::with_capacity(cells.len()), + }; + for (column, cell) in cells { + let name = match names.get(&column) { + Some(name) => name.clone(), + None => { + let name = placeholder(&names, column, row)?; + names.insert(column, name.clone()); + name + } + }; + record.cells.push((name, cell)); + } + records.push(record); + } + Ok((names.into_values().collect(), records)) +} + +/// The name of headerless column `column`, `column_` and its place in the sheet; a header in `names` already using that name is an error naming `row`. +fn placeholder(names: &BTreeMap, column: u32, row: u64) -> Result { + let name = format!("column_{}", u64::from(column) + 1); + if names.values().any(|taken| *taken == name) { + return Err(format!( + "row {row}: the column without a header in place {} would be named {name}, which the header already uses", + u64::from(column) + 1 + )); + } + Ok(name) +} + +/// A cell's text and the type it suggests; `None` for an empty cell or an error value such as `#N/A`, which load as NULL. +fn cell_of(data: &Data) -> Option { + let cell = |text: String, kind: ColumnType| Some(Cell { text, kind }); + match data { + Data::Empty | Data::Error(_) => None, + Data::Int(value) => cell(value.to_string(), ColumnType::Int), + Data::Float(value) => Some(number_cell(*value)), + Data::Bool(value) => cell(value.to_string(), ColumnType::Bool), + Data::String(text) | Data::DateTimeIso(text) | Data::DurationIso(text) => { + cell(text.clone(), ColumnType::Text) + } + Data::DateTime(value) => Some(date_cell(value)), + } +} + +/// A number: Excel stores every number as a float, so a whole one within a float's exact range is an `int`. +fn number_cell(value: f64) -> Cell { + match value.trunc() == value && value.abs() < MAX_EXACT_FLOAT_INT { + true => Cell { + text: format!("{value:.0}"), + kind: ColumnType::Int, + }, + false => Cell { + text: value.to_string(), + kind: ColumnType::Float, + }, + } +} + +/// A date-formatted number: a `time` under one day, a `date` at midnight, otherwise a `timestamp_ntz`; a duration is text, as ISO 8601. +fn date_cell(value: &ExcelDateTime) -> Cell { + let moment = value.as_datetime().filter(|_| value.is_datetime()); + match (moment, value.as_duration()) { + (Some(moment), _) if value.as_f64() < EXCEL_DAY => Cell { + text: moment.time().format(NORMAL_TIME).to_string(), + kind: ColumnType::Time, + }, + (Some(moment), _) if moment.time() == NaiveTime::MIN => Cell { + text: moment.date().format(NORMAL_DATE).to_string(), + kind: ColumnType::Date, + }, + (Some(moment), _) => Cell { + text: moment.format(NORMAL_TIMESTAMP).to_string(), + kind: ColumnType::TimestampNtz, + }, + (None, Some(duration)) => Cell { + text: duration.to_string(), + kind: ColumnType::Text, + }, + (None, None) => number_cell(value.as_f64()), + } +} diff --git a/crates/sqlload/src/files_source/_helpers/cells/xml_records.rs b/crates/sqlload/src/files_source/_helpers/cells/xml_records.rs new file mode 100644 index 00000000..1c074fdf --- /dev/null +++ b/crates/sqlload/src/files_source/_helpers/cells/xml_records.rs @@ -0,0 +1,280 @@ +//! Read the elements at a path of an XML document as records. + +use quick_xml::NsReader; +use quick_xml::escape::resolve_predefined_entity; +use quick_xml::events::{BytesStart, Event}; +use quick_xml::name::{NamespaceResolver, ResolveResult}; + +use crate::column_types::models::ColumnType; +use crate::files_source::_helpers::text::inference::csv_value_type; +use crate::files_source::constants::{ + UTF16_BOMS, XML_NIL_VALUES, XML_TEXT_KEY, XSI_NAMESPACE, XSI_NIL, +}; +use crate::files_source::models::{Cell, CellRecord, XmlOptions}; + +/// An element being read: its attributes, child elements and text, in document order. +struct Element { + name: String, + attributes: Vec<(String, String)>, + children: Vec, + text: String, + /// Whether `xsi:nil` marks it as NULL. + nil: bool, +} + +/// A record's field: text, or JSON text for repeated or structured elements. +enum Field { + Text(String), + Json(String), + Null, +} + +/// Each element at `options.record_path` in `document` as a record, numbered by the line its tag starts on: attributes and child elements are its columns, by local name. A document that ends inside an element, is UTF-16, or has no element where the path's parent leads is an error. +pub(crate) fn xml_records( + document: &[u8], + options: &XmlOptions, +) -> Result, String> { + if UTF16_BOMS.iter().any(|bom| document.starts_with(bom)) { + return Err("the document is UTF-16; SQLLoad reads XML documents in UTF-8".to_owned()); + } + let path: Vec<&str> = options + .record_path + .trim_matches('/') + .split('/') + .map(local_part) + .collect(); + let mut reader = NsReader::from_reader(document); + let mut found = false; + let mut buffer: Vec = Vec::new(); + let mut outside: Vec = Vec::new(); + let mut open: Vec = Vec::new(); + let mut records: Vec = Vec::new(); + let (mut counted, mut line, mut record_line) = (0_usize, 1_u64, 1_u64); + loop { + let position = usize::try_from(reader.buffer_position()).unwrap_or(document.len()); + let end = position.min(document.len()); + line += document[counted.min(end)..end] + .iter() + .filter(|byte| **byte == b'\n') + .count() as u64; + counted = counted.max(end); + let event = reader + .read_event_into(&mut buffer) + .map_err(|error| format!("line {line}: {error}"))?; + let (started, ended) = match &event { + Event::Start(start) => (Some(element(start, reader.resolver(), line)?), false), + Event::Empty(start) => (Some(element(start, reader.resolver(), line)?), true), + Event::End(_) => (None, true), + Event::Text(text) => { + push_text(&mut open, &text.xml10_content()); + (None, false) + } + Event::CData(data) => { + push_text(&mut open, &data.xml10_content()); + (None, false) + } + Event::GeneralRef(entity) => { + let resolved = match entity.resolve_char_ref() { + Ok(Some(character)) => character.to_string(), + Ok(None) => resolve_predefined_entity(entity) + .ok_or_else(|| format!("line {line}: unknown entity &{};", &**entity))? + .to_owned(), + Err(error) => return Err(format!("line {line}: {error}")), + }; + push_text(&mut open, &resolved); + (None, false) + } + Event::Eof => break, + _ => (None, false), + }; + if let Some(started) = started { + match open.is_empty() { + true => { + outside.push(started.name.clone()); + found |= + outside.len() + 1 == path.len() && path[..outside.len()] == outside[..]; + if outside == path { + found = true; + record_line = line; + open.push(started); + } + } + false => open.push(started), + } + } + if ended { + match open.pop() { + Some(done) if open.is_empty() => { + outside.pop(); + records.push(record(done, record_line, options)?); + } + Some(done) => { + if let Some(parent) = open.last_mut() { + parent.children.push(done); + } + } + None => { + outside.pop(); + } + } + } + buffer.clear(); + } + if let Some(inside) = open.last().map(|element| &element.name).or(outside.last()) { + return Err(format!("line {line}: the document ends inside <{inside}>")); + } + match found { + true => Ok(records), + false => Err(format!( + "no element at record_path {:?}; its names match elements without their namespace prefix", + options.record_path + )), + } +} + +/// A name without its namespace prefix. +fn local_part(name: &str) -> &str { + name.rsplit(':').next().unwrap_or(name) +} + +/// The element `start` opens, on `line`: its attributes by local name, namespace declarations and attributes in the XML Schema instance namespace left out; `xsi:nil` marks it NULL. Two attributes with one local name are an error. +fn element( + start: &BytesStart<'_>, + resolver: &NamespaceResolver, + line: u64, +) -> Result { + let name = start.local_name().into_inner().to_owned(); + let mut attributes: Vec<(String, String)> = Vec::new(); + let mut nil = false; + for attribute in start.attributes() { + let attribute = attribute.map_err(|error| format!("line {line}: {error}"))?; + if attribute.key.as_namespace_binding().is_some() { + continue; + } + let value = attribute + .normalized_value(quick_xml::XmlVersion::Implicit1_0) + .map_err(|error| format!("line {line}: {error}"))?; + let local = attribute.key.local_name().into_inner().to_owned(); + if let (ResolveResult::Bound(namespace), _) = resolver.resolve_attribute(attribute.key) + && namespace.0 == XSI_NAMESPACE + { + nil |= local == XSI_NIL && XML_NIL_VALUES.contains(&value.trim()); + continue; + } + if attributes.iter().any(|(taken, _)| *taken == local) { + return Err(format!( + "line {line}: <{name}> has two attributes named {local}" + )); + } + attributes.push((local, value.into_owned())); + } + Ok(Element { + name, + attributes, + children: Vec::new(), + text: String::new(), + nil, + }) +} + +fn push_text(open: &mut [Element], text: &str) { + if let Some(element) = open.last_mut() { + element.text.push_str(text); + } +} + +/// The record element `done` as cells: attributes, then child elements by first appearance, then its own text under `#text`. +fn record(done: Element, line: u64, options: &XmlOptions) -> Result { + let mut cells: Vec<(String, Cell)> = Vec::new(); + for (name, field) in fields(&done, options).map_err(|error| format!("line {line}: {error}"))? { + let cell = match field { + Field::Null => continue, + Field::Text(text) => Cell { + kind: csv_value_type(&text), + text, + }, + Field::Json(text) => Cell { + text, + kind: ColumnType::Json, + }, + }; + cells.push((name, cell)); + } + Ok(CellRecord { line, cells }) +} + +/// An element's fields: each attribute as text; each child element by name, as its value, or a JSON array of values where it repeats or `options.lists` names it; and its own text, trimmed, where it has attributes or children. +fn fields(element: &Element, options: &XmlOptions) -> Result, String> { + let mut fields: Vec<(String, Field)> = element + .attributes + .iter() + .map(|(name, value)| (name.clone(), Field::Text(value.clone()))) + .collect(); + let mut names: Vec<&str> = Vec::new(); + for child in &element.children { + if !names.contains(&child.name.as_str()) { + names.push(&child.name); + } + } + for name in names { + if element + .attributes + .iter() + .any(|(attribute, _)| attribute == name) + { + return Err(format!( + "<{}> has an attribute and a child element both named {name}", + element.name + )); + } + let values: Vec<&Element> = element + .children + .iter() + .filter(|child| child.name == name) + .collect(); + let field = match values.as_slice() { + [one] if !options.lists.iter().any(|list| list == name) => value(one, options)?, + many => Field::Json(format!( + "[{}]", + many.iter() + .map(|child| value(child, options).map(json)) + .collect::, String>>()? + .join(",") + )), + }; + fields.push((name.to_owned(), field)); + } + let text = element.text.trim(); + if !text.is_empty() { + fields.push((XML_TEXT_KEY.to_owned(), Field::Text(text.to_owned()))); + } + Ok(fields) +} + +/// A child element's value: NULL where `xsi:nil` says so, its trimmed text (NULL when empty), or a JSON object of its fields. +fn value(element: &Element, options: &XmlOptions) -> Result { + if element.nil { + return Ok(Field::Null); + } + if element.attributes.is_empty() && element.children.is_empty() { + let text = element.text.trim(); + return Ok(match text.is_empty() { + true => Field::Null, + false => Field::Text(text.to_owned()), + }); + } + let members: Vec = fields(element, options)? + .into_iter() + .map(|(name, field)| format!("{}:{}", serde_json::Value::String(name), json(field))) + .collect(); + Ok(Field::Json(format!("{{{}}}", members.join(",")))) +} + +/// A field as JSON text: text as a JSON string. +fn json(field: Field) -> String { + match field { + Field::Text(text) => serde_json::Value::String(text).to_string(), + Field::Json(text) => text, + Field::Null => "null".to_owned(), + } +} diff --git a/crates/sqlload/src/files_source/_helpers/mod.rs b/crates/sqlload/src/files_source/_helpers/mod.rs index 1e933b69..5a7c5069 100644 --- a/crates/sqlload/src/files_source/_helpers/mod.rs +++ b/crates/sqlload/src/files_source/_helpers/mod.rs @@ -1,3 +1,5 @@ +pub(crate) mod archive; +pub(crate) mod cells; pub(crate) mod parquet; pub(crate) mod plan; pub(crate) mod text; diff --git a/crates/sqlload/src/files_source/_helpers/plan/encoding.rs b/crates/sqlload/src/files_source/_helpers/plan/encoding.rs index c96984c5..78c97187 100644 --- a/crates/sqlload/src/files_source/_helpers/plan/encoding.rs +++ b/crates/sqlload/src/files_source/_helpers/plan/encoding.rs @@ -1,5 +1,8 @@ +use globset::Glob; + use crate::files_source::constants::{ - CSV_SUFFIX, GZIP_SUFFIX, JSON_SUFFIXES, PARQUET_SUFFIX, ZSTD_SUFFIX, + CSV_SUFFIX, GZIP_SUFFIX, JSON_SUFFIXES, PARQUET_SUFFIX, XLSX_SUFFIX, XML_SUFFIX, ZIP_SUFFIX, + ZSTD_SUFFIX, }; use crate::files_source::models::{Codec, Compression, FilesEncoding, SourceFormat}; @@ -9,10 +12,12 @@ fn format_suffixes(format: SourceFormat) -> &'static [&'static str] { SourceFormat::Parquet => std::slice::from_ref(&PARQUET_SUFFIX), SourceFormat::Ndjson | SourceFormat::Json => JSON_SUFFIXES, SourceFormat::Csv => std::slice::from_ref(&CSV_SUFFIX), + SourceFormat::Xlsx => std::slice::from_ref(&XLSX_SUFFIX), + SourceFormat::Xml => std::slice::from_ref(&XML_SUFFIX), } } -/// The decompression `key` is read with, or `None` when its name does not belong to `encoding` (case-insensitive). +/// The decompression `key` is read with, or `None` when its name does not belong to `encoding` (case-insensitive); with `compression = "zip"`, `Plain` for a ZIP archive, whose members [`member_fits`] picks. pub(crate) fn file_codec(key: &str, encoding: &FilesEncoding) -> Option { let lower = key.to_ascii_lowercase(); let (stem, suffix_codec) = match ( @@ -24,6 +29,7 @@ pub(crate) fn file_codec(key: &str, encoding: &FilesEncoding) -> Option { (None, None) => (lower.as_str(), None), }; let codec = match (encoding.format, encoding.compression, suffix_codec) { + (_, Compression::Zip, _) => return lower.ends_with(ZIP_SUFFIX).then_some(Codec::Plain), (SourceFormat::Parquet, _, Some(_)) => return None, (SourceFormat::Parquet, _, None) => Codec::Plain, (_, Compression::Auto, suffix) => suffix.unwrap_or(Codec::Plain), @@ -38,9 +44,31 @@ pub(crate) fn file_codec(key: &str, encoding: &FilesEncoding) -> Option { .then_some(codec) } +/// Whether a ZIP archive's member `name` is one of `encoding`'s files: its name ends with the format's suffix (case-insensitive) and matches `members` when set. +pub(crate) fn member_fits(name: &str, encoding: &FilesEncoding) -> Result { + let lower = name.to_ascii_lowercase(); + let suffix = format_suffixes(encoding.format) + .iter() + .any(|suffix| lower.ends_with(suffix)); + let pattern = match &encoding.members { + Some(pattern) => Glob::new(pattern) + .map_err(|error| format!("members {pattern:?}: {error}"))? + .compile_matcher() + .is_match(name), + None => true, + }; + Ok(!name.ends_with('/') && suffix && pattern) +} + /// The file names `encoding` reads, for messages: `.csv`, `.csv.gz`, ... pub(crate) fn file_names(encoding: &FilesEncoding) -> String { let compressed: &[&str] = match (encoding.format, encoding.compression) { + (format, Compression::Zip) => { + return format!( + "{} members of {ZIP_SUFFIX}", + format_suffixes(format).join(", ") + ); + } (SourceFormat::Parquet, _) | (_, Compression::None) => &[""], (_, Compression::Auto) => &["", GZIP_SUFFIX, ZSTD_SUFFIX], (_, Compression::Gzip) => &["", GZIP_SUFFIX], diff --git a/crates/sqlload/src/files_source/_helpers/text/bytes_reader.rs b/crates/sqlload/src/files_source/_helpers/text/bytes_reader.rs index 88db63c9..fdcff21d 100644 --- a/crates/sqlload/src/files_source/_helpers/text/bytes_reader.rs +++ b/crates/sqlload/src/files_source/_helpers/text/bytes_reader.rs @@ -9,8 +9,9 @@ use tokio_util::io::StreamReader; use encoding_rs::{DecoderResult, Encoding, WINDOWS_1252}; use crate::file_store::models::{OpenFolder, StoredFile}; +use crate::files_source::_helpers::archive::zip_archive::member_bytes; use crate::files_source::constants::{ - CP1252_UNDEFINED, ENCODING_HINT, READ_CHUNK_BYTES, UTF16_BOMS, + CP1252_UNDEFINED, ENCODING_HINT, MAX_WHOLE_FILE_BYTES, READ_CHUNK_BYTES, UTF16_BOMS, }; use crate::files_source::models::{Charset, Codec}; @@ -33,6 +34,37 @@ pub(crate) struct BytesReader { } impl BytesReader { + /// A reader of `inner`'s bytes as UTF-8, as they are. + fn new(inner: Pin>) -> Self { + Self { + inner, + started: false, + charset: Charset::Utf8, + decoder: None, + read: 0, + lines: 0, + skip: 0, + skipped_breaks: (false, false), + } + } + + /// The rest of the file, whole, decoded. + pub(crate) async fn whole(&mut self) -> Result, String> { + let mut all: Vec = Vec::new(); + loop { + let chunk = self.next_chunk().await?; + if chunk.is_empty() { + return Ok(all); + } + all.extend(chunk); + if all.len() as u64 > MAX_WHOLE_FILE_BYTES { + return Err(format!( + "the file is over {MAX_WHOLE_FILE_BYTES} bytes; SQLLoad reads workbooks and XML documents whole, up to that size" + )); + } + } + } + /// This reader, decoding the file from `charset` and dropping its first `skip_rows` lines. pub(crate) fn decoding(mut self, charset: Charset, skip_rows: u64) -> Self { self.charset = charset; @@ -203,6 +235,10 @@ pub(crate) async fn open_bytes( file: &StoredFile, codec: Codec, ) -> Result { + if let Some(name) = &file.member { + let member = member_bytes(folder, file, name).await?; + return Ok(BytesReader::new(Box::pin(std::io::Cursor::new(member)))); + } let stream = folder .store .get(&file.path) @@ -224,16 +260,7 @@ pub(crate) async fn open_bytes( Box::pin(decoder) } }; - Ok(BytesReader { - inner, - started: false, - charset: Charset::Utf8, - decoder: None, - read: 0, - lines: 0, - skip: 0, - skipped_breaks: (false, false), - }) + Ok(BytesReader::new(inner)) } /// The error for a record longer than `max_record_bytes`, starting on `line`. diff --git a/crates/sqlload/src/files_source/_helpers/text/sample.rs b/crates/sqlload/src/files_source/_helpers/text/sample.rs index 1d88c1f3..dcc150dd 100644 --- a/crates/sqlload/src/files_source/_helpers/text/sample.rs +++ b/crates/sqlload/src/files_source/_helpers/text/sample.rs @@ -5,6 +5,7 @@ use arrow_array::{Array, RecordBatch, StringArray}; use crate::column_types::main::combined_type::combined_type; use crate::column_types::models::ColumnType; use crate::file_store::models::{OpenFolder, StoredFile}; +use crate::files_source::_helpers::cells::cell_reader::{CellReader, cell_type}; use crate::files_source::_helpers::plan::columns::SampledColumn; use crate::files_source::_helpers::text::bytes_reader::open_bytes; use crate::files_source::_helpers::text::csv_reader::CsvReader; @@ -12,7 +13,7 @@ use crate::files_source::_helpers::text::inference::{csv_value_type, json_value_ use crate::files_source::_helpers::text::json_reader::JsonReader; use crate::files_source::_helpers::text::json_records::JsonRecord; use crate::files_source::constants::SAMPLE_ROWS; -use crate::files_source::models::{Codec, FilesEncoding, SourceFormat, StoredRead}; +use crate::files_source::models::{CellRecord, Codec, FilesEncoding, SourceFormat, StoredRead}; /// The columns in the first records of a JSON or CSV `file`, with the types their values suggest. pub(crate) async fn sample_file( @@ -23,6 +24,11 @@ pub(crate) async fn sample_file( ) -> Result, String> { let bytes = open_bytes(folder, file, codec).await?; match encoding.format { + SourceFormat::Xlsx | SourceFormat::Xml => { + let mut reader = CellReader::open(bytes, encoding).await?; + let records = reader.next_records(SAMPLE_ROWS); + Ok(cell_columns(reader.names(), &records)) + } SourceFormat::Csv => { let mut reader = CsvReader::open(bytes, &encoding.csv, SAMPLE_ROWS).await?; let batch = reader.next_batch().await?.map(|(batch, _)| batch); @@ -57,6 +63,38 @@ pub(crate) fn json_columns(records: &[JsonRecord]) -> Vec { columns } +/// A worksheet's columns `names`, then any other key in `records` in first-seen order, each with the type its cells suggest. +pub(crate) fn cell_columns(names: &[String], records: &[CellRecord]) -> Vec { + let mut columns: Vec = names.iter().map(|name| text_column(name)).collect(); + for record in records { + for (name, cell) in &record.cells { + let position = match columns.iter().position(|column| column.name == *name) { + Some(position) => position, + None => { + columns.push(text_column(name)); + columns.len() - 1 + } + }; + let column = &mut columns[position]; + column.column_type = match column.column_type.take() { + Some(Ok(seen)) => Some(Ok(cell_type(seen, cell.kind.clone()))), + seen => observe(seen, Some(cell.kind.clone())), + }; + } + } + columns +} + +/// A sampled column `name` with no value seen yet. +fn text_column(name: &str) -> SampledColumn { + SampledColumn { + name: name.to_owned(), + column_type: None, + direct_refusal: None, + stored_read: StoredRead::Plain, + } +} + /// CSV columns `names`, with the type the values in `batch` (text columns) suggest. pub(crate) fn csv_columns(names: &[String], batch: Option<&RecordBatch>) -> Vec { let mut columns: Vec = Vec::new(); diff --git a/crates/sqlload/src/files_source/_helpers/text/text_reader.rs b/crates/sqlload/src/files_source/_helpers/text/text_reader.rs index 95707bb9..ffe777fa 100644 --- a/crates/sqlload/src/files_source/_helpers/text/text_reader.rs +++ b/crates/sqlload/src/files_source/_helpers/text/text_reader.rs @@ -2,6 +2,7 @@ use arrow_array::{Array, RecordBatch, StringArray}; use crate::column_types::models::ColumnType; use crate::file_store::models::{OpenFolder, StoredFile}; +use crate::files_source::_helpers::cells::cell_reader::{CellReader, cells_batch}; use crate::files_source::_helpers::text::bytes_reader::{BytesReader, open_bytes}; use crate::files_source::_helpers::text::checks::{ColumnChecks, misfit, value_fits}; use crate::files_source::_helpers::text::csv_reader::CsvReader; @@ -30,6 +31,12 @@ enum Records { reader: Box, column: String, }, + /// An xlsx worksheet's rows or an XML document's records, picking out `columns`. + Cells { + reader: CellReader, + columns: Vec<(String, ColumnType)>, + batch_rows: usize, + }, } /// Reads a JSON or CSV file as batches of text columns, which the shape step converts to the planned types. @@ -49,6 +56,15 @@ impl TextReader { let checks = ColumnChecks::new(request)?; let bytes = open_bytes(folder, file, codec).await?; let records = match (request.encoding.format, &request.encoding.record) { + (SourceFormat::Xlsx | SourceFormat::Xml, _) => Records::Cells { + reader: CellReader::open(bytes, &request.encoding).await?, + columns: request + .columns + .iter() + .map(|column| (column.name.clone(), column.column_type.clone())) + .collect(), + batch_rows: request.batch_rows, + }, (SourceFormat::Csv, Some(column)) => Records::CsvWhole { reader: Box::new( CsvReader::open_lines(bytes, &request.encoding.csv, request.batch_rows).await?, @@ -79,6 +95,17 @@ impl TextReader { false => records_batch(&records, columns, &self.checks).map(Some), } } + Records::Cells { + reader, + columns, + batch_rows, + } => { + let records = reader.next_records(*batch_rows); + match records.is_empty() { + true => Ok(None), + false => cells_batch(&records, columns, &self.checks).map(Some), + } + } Records::JsonWhole { reader, column, diff --git a/crates/sqlload/src/files_source/constants.rs b/crates/sqlload/src/files_source/constants.rs index 86014c66..3ad03a2d 100644 --- a/crates/sqlload/src/files_source/constants.rs +++ b/crates/sqlload/src/files_source/constants.rs @@ -24,6 +24,36 @@ pub const JSON_SUFFIXES: &[&str] = &[".ndjson", ".jsonl", ".json"]; /// The file name suffix of CSV data files. pub const CSV_SUFFIX: &str = ".csv"; +/// The suffix of Excel workbooks. +pub const XLSX_SUFFIX: &str = ".xlsx"; + +/// The suffix of XML documents. +pub const XML_SUFFIX: &str = ".xml"; + +/// The suffix of ZIP archives. +pub const ZIP_SUFFIX: &str = ".zip"; + +/// The largest xlsx workbook, XML document or ZIP archive member read: each is read whole into memory. +pub const MAX_WHOLE_FILE_BYTES: u64 = 1024 * 1024 * 1024; + +/// The bytes of a ZIP archive each range request reads: its index, at its end, then one member. +pub const ZIP_READ_BLOCK_BYTES: u64 = 4 * 1024 * 1024; + +/// The most non-empty cells read from one worksheet, each held in memory. +pub const MAX_SHEET_CELLS: usize = 20_000_000; + +/// The key an XML element's own text loads under when it also has attributes or child elements. +pub const XML_TEXT_KEY: &str = "#text"; + +/// The XML Schema instance namespace, whose attributes such as `xsi:type` describe an element rather than hold data. +pub const XSI_NAMESPACE: &str = "http://www.w3.org/2001/XMLSchema-instance"; + +/// The `xsi:` attribute that marks an element NULL. +pub const XSI_NIL: &str = "nil"; + +/// The values of `xsi:nil` that make an element NULL. +pub const XML_NIL_VALUES: [&str; 2] = ["true", "1"]; + /// The file name suffix of gzip-compressed files. pub const GZIP_SUFFIX: &str = ".gz"; @@ -121,3 +151,9 @@ pub const CP1252_UNDEFINED: [u8; 5] = [0x81, 0x8D, 0x8F, 0x90, 0x9D]; /// The fix an error about bytes that do not decode ends with. pub const ENCODING_HINT: &str = "set csv.encoding to the files' encoding, such as \"cp1252\", \"latin-1\" or \"utf-16le\""; + +/// Floats below this hold every whole number exactly: 2^53. +pub const MAX_EXACT_FLOAT_INT: f64 = 9_007_199_254_740_992.0; + +/// One day in Excel's date numbers; a date-formatted value below it is a time of day. +pub const EXCEL_DAY: f64 = 1.0; diff --git a/crates/sqlload/src/files_source/main/discover_files.rs b/crates/sqlload/src/files_source/main/discover_files.rs index 86bfb66d..7abba732 100644 --- a/crates/sqlload/src/files_source/main/discover_files.rs +++ b/crates/sqlload/src/files_source/main/discover_files.rs @@ -3,6 +3,7 @@ use crate::column_types::models::ColumnType; use crate::file_store::main::list_files::list_files; use crate::file_store::models::OpenFolder; +use crate::files_source::_helpers::archive::zip_archive::archive_members; use crate::files_source::_helpers::plan::columns::{ ColumnMerge, SampledColumn, case_problems, drop_loader_columns, file_columns, merge_columns, }; @@ -14,7 +15,7 @@ use crate::files_source::constants::{ }; use crate::files_source::errors::FilesError; use crate::files_source::models::{ - FilesEncoding, FilesTable, ListedFiles, PathValues, SourceFormat, StoredRead, + Compression, FilesEncoding, FilesTable, ListedFiles, PathValues, SourceFormat, StoredRead, }; /// The `encoding` files under `folder` with their path `values`, and the schema of the newest ones without the loader's own columns, which this load fills; when a file's path lacks a value, only the listing and its problems, before any file is read. @@ -110,11 +111,24 @@ async fn listed_files( listed.skipped += 1; continue; }; - let (data, problem) = data_file(file, values); - listed.problems.extend(problem); - match data { - Some(data) => listed.files.push((data, codec)), - None => listed.unmatched += 1, + let files = match encoding.compression { + Compression::Zip => { + archive_members(folder, &file, encoding) + .await + .map_err(|reason| FilesError::File { + key: file.relative_key.clone(), + reason, + })? + } + _ => vec![file], + }; + for file in files { + let (data, problem) = data_file(file, values); + listed.problems.extend(problem); + match data { + Some(data) => listed.files.push((data, codec)), + None => listed.unmatched += 1, + } } } let unmatched = match listed.unmatched { diff --git a/crates/sqlload/src/files_source/models.rs b/crates/sqlload/src/files_source/models.rs index 81ad9771..364a6948 100644 --- a/crates/sqlload/src/files_source/models.rs +++ b/crates/sqlload/src/files_source/models.rs @@ -184,6 +184,10 @@ pub enum SourceFormat { /// JSON objects one after another (over any number of lines), or one top-level array of objects. Json, Csv, + /// Excel workbooks: one worksheet's rows, each cell typed as the workbook stores it. + Xlsx, + /// XML documents: each element at `xml.record_path` one row. + Xml, } impl std::fmt::Display for SourceFormat { @@ -193,6 +197,8 @@ impl std::fmt::Display for SourceFormat { Self::Ndjson => "ndjson", Self::Json => "json", Self::Csv => "csv", + Self::Xlsx => "xlsx", + Self::Xml => "xml", }) } } @@ -207,6 +213,8 @@ pub enum Compression { None, Gzip, Zstd, + /// ZIP archives (`.zip`): each member whose name fits the format is a file of its own. + Zip, } /// The decompression one file is read with. @@ -261,12 +269,48 @@ pub enum Charset { Other(&'static encoding_rs::Encoding), } +/// Which rows of an Excel workbook load. +#[derive(Debug, Clone, PartialEq, Eq, Deserialize)] +#[serde(default, deny_unknown_fields)] +pub struct XlsxOptions { + /// The worksheet, by name; the first one when omitted. + pub sheet: Option, + /// Whether the first row read holds the column names; without one, columns are `column_1`, `column_2`, ... by their place in the sheet. + pub header: bool, + /// Rows skipped at the top of the sheet, before the header (or the first row without one). + pub skip_rows: u32, +} + +impl Default for XlsxOptions { + fn default() -> Self { + Self { + sheet: None, + header: true, + skip_rows: 0, + } + } +} + +/// Which elements of an XML document are rows. +#[derive(Debug, Clone, Default, PartialEq, Eq, Deserialize)] +#[serde(default, deny_unknown_fields)] +pub struct XmlOptions { + /// The element names from the root to each row's element, separated by `/`, such as `orders/order`. + pub record_path: String, + /// Child elements loaded as a JSON array even where one appears only once. + pub lists: Vec, +} + /// How a files source's files are encoded. #[derive(Debug, Clone, Default, PartialEq, Eq)] pub struct FilesEncoding { pub format: SourceFormat, pub compression: Compression, pub csv: CsvOptions, + pub xlsx: XlsxOptions, + pub xml: XmlOptions, + /// With `compression = "zip"`, a pattern the members' names must match, beyond the format's suffix. + pub members: Option, /// With `shape = "record"`, the one column each JSON object or CSV line loads into whole; `None` for typed columns. pub record: Option, } @@ -288,6 +332,7 @@ impl std::fmt::Display for FilesEncoding { (format, Compression::None) => write!(formatter, "{format}, uncompressed"), (format, Compression::Gzip) => write!(formatter, "{format}, gzip"), (format, Compression::Zstd) => write!(formatter, "{format}, zstd"), + (format, Compression::Zip) => write!(formatter, "{format} in ZIP archives"), }?; match &self.record { Some(column) => write!(formatter, ", each record whole in column {column}"), @@ -307,6 +352,20 @@ pub(crate) struct ListedFiles { pub(crate) problems: Vec, } +/// One value of a spreadsheet row or XML record: its text, which the shape step converts to the column's type, and the type it suggests. +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct Cell { + pub(crate) text: String, + pub(crate) kind: ColumnType, +} + +/// One spreadsheet row or XML record: its cells by column name, in order, a NULL cell left out; and the row or line it starts on. +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct CellRecord { + pub(crate) line: u64, + pub(crate) cells: Vec<(String, Cell)>, +} + /// One file of a read, and what its batches are shaped into. #[derive(Debug)] pub(crate) struct FileRead<'a> { diff --git a/crates/sqlload/src/files_source/tests.rs b/crates/sqlload/src/files_source/tests.rs index 2389d627..056cd090 100644 --- a/crates/sqlload/src/files_source/tests.rs +++ b/crates/sqlload/src/files_source/tests.rs @@ -18,6 +18,8 @@ mod merge_columns; mod nested_parquet; #[path = "tests/test_path_columns.rs"] mod path_columns; +#[path = "tests/test_read_containers.rs"] +mod read_containers; #[path = "tests/test_read_text_files.rs"] mod read_text_files; #[path = "tests/test_select_files.rs"] diff --git a/crates/sqlload/src/files_source/tests/helpers.rs b/crates/sqlload/src/files_source/tests/helpers.rs index 6cae6499..dfa61c2f 100644 --- a/crates/sqlload/src/files_source/tests/helpers.rs +++ b/crates/sqlload/src/files_source/tests/helpers.rs @@ -27,7 +27,7 @@ use crate::files_source::main::discover_files::discover_files; use crate::files_source::main::read_files::read_files; use crate::files_source::models::{ BatchFooters, Compression, CsvOptions, FilesEncoding, FilesRead, FilesTable, PathValues, - SourceFormat, TextChecks, + SourceFormat, TextChecks, XlsxOptions, XmlOptions, }; use crate::files_source::tests::test_types::{ ReadTextFilesTestCase, SplitJsonTestCase, TrackCsvTestCase, @@ -53,6 +53,7 @@ pub(crate) fn encoding(format: SourceFormat) -> FilesEncoding { compression: Compression::Auto, csv: CsvOptions::default(), record: None, + ..FilesEncoding::default() } } @@ -63,6 +64,7 @@ pub(crate) fn csv_encoding(csv: CsvOptions) -> FilesEncoding { compression: Compression::Auto, csv, record: None, + ..FilesEncoding::default() } } @@ -73,6 +75,7 @@ pub(crate) fn record_encoding(format: SourceFormat, csv: CsvOptions) -> FilesEnc compression: Compression::Auto, csv, record: Some(RECORD_COLUMN.to_owned()), + ..FilesEncoding::default() } } @@ -81,6 +84,8 @@ type Encoder = fn(&str) -> Result, String>; /// How a test file is written, by a part of its key: the first that matches wins. const ENCODERS: &[(&str, Encoder)] = &[ + (".zip", zip_archive), + (".xlsx", workbook), (".gz", gzip), (".raw.", raw), (".utf16.", utf16), @@ -108,6 +113,106 @@ fn utf16(text: &str) -> Result, String> { .collect()) } +/// A ZIP archive of the members `text` lists, separated by `\u{1e}`, each `name\u{1f}content`, with each member written as its own name says (so a member may be a workbook); a name ending in `/` is a folder. +fn zip_archive(text: &str) -> Result, String> { + let mut archive = zip::ZipWriter::new(std::io::Cursor::new(Vec::new())); + let options = zip::write::SimpleFileOptions::default(); + let members: Vec<(&str, &str)> = text + .split('\u{1e}') + .map(|member| member.split_once('\u{1f}').unwrap_or((member, ""))) + .collect(); + for (name, _) in members.iter().filter(|(name, _)| name.ends_with('/')) { + archive + .add_directory(*name, options) + .map_err(|error| error.to_string())?; + } + for (name, content) in members.iter().filter(|(name, _)| !name.ends_with('/')) { + archive + .start_file(*name, options) + .map_err(|error| error.to_string())?; + archive + .write_all(&file_bytes(name, content)?) + .map_err(|error| error.to_string())?; + } + archive + .finish() + .map(std::io::Cursor::into_inner) + .map_err(|error| error.to_string()) +} + +/// Writes one workbook cell's text, its type prefix removed. +type CellWriter = + fn(&mut rust_xlsxwriter::Worksheet, (u32, u16), &str) -> Result<(), rust_xlsxwriter::XlsxError>; + +/// How a workbook cell is written, by its prefix: the first that matches wins, and a cell without one is a string. +const CELL_WRITERS: &[(&str, CellWriter)] = &[ + ("n:", |sheet, (row, column), text| { + sheet + .write_number(row, column, text.parse::().unwrap_or(f64::NAN)) + .map(|_| ()) + }), + ("b:", |sheet, (row, column), text| { + sheet.write_boolean(row, column, text == "true").map(|_| ()) + }), + ("dt:", |sheet, place, text| { + moment(sheet, place, (text, "yyyy-mm-dd hh:mm:ss")) + }), + ("d:", |sheet, place, text| { + moment(sheet, place, (text, "yyyy-mm-dd")) + }), + ("tm:", |sheet, place, text| { + moment(sheet, place, (text, "hh:mm:ss")) + }), + ("s:", |sheet, (row, column), text| { + sheet.write_string(row, column, text).map(|_| ()) + }), + ("", |sheet, (row, column), text| { + sheet.write_string(row, column, text).map(|_| ()) + }), +]; + +/// An xlsx workbook of the sheets in `text`: each starts with a `[Name]` line, then one line per row (from row 1) with cells separated by `|`. A cell is a string, empty, or typed by a prefix: `n:` number, `b:` boolean, `d:` date, `dt:` date and time, `tm:` time of day, `s:` string as written. +fn workbook(text: &str) -> Result, String> { + let failed = |error: rust_xlsxwriter::XlsxError| error.to_string(); + let mut book = rust_xlsxwriter::Workbook::new(); + for sheet in text.split('[').filter(|sheet| !sheet.is_empty()) { + let (name, rows) = sheet.split_once("]\n").unwrap_or((sheet, "")); + let worksheet = book.add_worksheet(); + worksheet.set_name(name).map_err(failed)?; + for (row, line) in rows.lines().enumerate() { + let cells = line + .split('|') + .enumerate() + .filter(|(_, cell)| !cell.is_empty()); + for (column, cell) in cells { + let place = ( + u32::try_from(row).map_err(|error| error.to_string())?, + u16::try_from(column).map_err(|error| error.to_string())?, + ); + let (prefix, write) = CELL_WRITERS + .iter() + .find(|(prefix, _)| cell.starts_with(prefix)) + .ok_or_else(|| format!("no writer for cell {cell:?}"))?; + write(worksheet, place, &cell[prefix.len()..]).map_err(failed)?; + } + } + } + book.save_to_buffer().map_err(failed) +} + +/// A date, time or both, written with number `format`. +fn moment( + sheet: &mut rust_xlsxwriter::Worksheet, + (row, column): (u32, u16), + (text, format): (&str, &str), +) -> Result<(), rust_xlsxwriter::XlsxError> { + let value = rust_xlsxwriter::ExcelDateTime::parse_from_str(text)?; + let format = rust_xlsxwriter::Format::new().set_num_format(format); + sheet + .write_datetime_with_format(row, column, value, &format) + .map(|_| ()) +} + fn gzip(text: &str) -> Result, String> { let mut encoder = GzEncoder::new(Vec::new(), GzipLevel::default()); encoder @@ -553,3 +658,29 @@ pub(crate) fn decoded(encoding: &str, skip_rows: u64) -> CsvOptions { ..CsvOptions::default() } } + +/// `format` files, with `xlsx` and `xml` options, in ZIP archives when `members` is set (`Some("")` for any member). +pub(crate) fn container_encoding( + format: SourceFormat, + (xlsx, xml): (XlsxOptions, XmlOptions), + members: Option<&str>, +) -> FilesEncoding { + FilesEncoding { + format, + compression: members.map_or(Compression::Auto, |_| Compression::Zip), + xlsx, + xml, + members: members + .filter(|pattern| !pattern.is_empty()) + .map(str::to_owned), + ..FilesEncoding::default() + } +} + +/// XML options with rows at `record_path` and `lists` loaded as arrays. +pub(crate) fn xml_at(record_path: &str, lists: &[&str]) -> XmlOptions { + XmlOptions { + record_path: record_path.to_owned(), + lists: lists.iter().map(|name| (*name).to_owned()).collect(), + } +} diff --git a/crates/sqlload/src/files_source/tests/test_path_columns.rs b/crates/sqlload/src/files_source/tests/test_path_columns.rs index f2fe9cbd..5224bffa 100644 --- a/crates/sqlload/src/files_source/tests/test_path_columns.rs +++ b/crates/sqlload/src/files_source/tests/test_path_columns.rs @@ -156,6 +156,7 @@ fn given_a_path_template_when_listing_keys_then_captures_become_typed_values() - size: 1, etag: None, modified_ms: 0, + member: None, }; let (data, problem) = data_file(file, &values); let found: Option = data.map(|data| { diff --git a/crates/sqlload/src/files_source/tests/test_read_containers.rs b/crates/sqlload/src/files_source/tests/test_read_containers.rs new file mode 100644 index 00000000..3a054a5d --- /dev/null +++ b/crates/sqlload/src/files_source/tests/test_read_containers.rs @@ -0,0 +1,301 @@ +use crate::files_source::models::{SourceFormat, XlsxOptions, XmlOptions}; +use crate::files_source::tests::helpers::{block_on, container_encoding, read_rows, xml_at}; +use crate::files_source::tests::test_types::ReadTextFilesTestCase; + +const ORDERS_XML: &str = "\n\n \n A & B\n x\n y\n Oslo\n \n ]]>z\n not a row\n\n"; + +#[test] +fn given_xlsx_xml_and_zip_files_when_reading_then_cells_load_typed_and_errors_name_the_place() +-> Result<(), String> { + let sheet = |sheet: Option<&str>, header: bool, skip_rows: u32| XlsxOptions { + sheet: sheet.map(str::to_owned), + header, + skip_rows, + }; + let xlsx = |options: XlsxOptions, members: Option<&'static str>| { + container_encoding( + SourceFormat::Xlsx, + (options, XmlOptions::default()), + members, + ) + }; + let xml = |options: XmlOptions, members: Option<&'static str>| { + container_encoding( + SourceFormat::Xml, + (XlsxOptions::default(), options), + members, + ) + }; + let csv_zip = |members: Option<&'static str>| { + container_encoding(SourceFormat::Csv, Default::default(), members) + }; + let noise: String = (0..700_000_u64) + .map(|index| format!("{:016x}", index.wrapping_mul(0x9E37_79B9_7F4A_7C15))) + .collect(); + let large_archive: &'static str = format!( + "a.csv\u{1f}id,name\n1,a\n\u{1e}noise.txt\u{1f}{noise}\u{1e}b.csv\u{1f}id,name\n2,b\n" + ) + .leak(); + let large_files: &'static [(&'static str, &'static str)] = + vec![("a.zip", large_archive)].leak(); + let test_cases = [ + ReadTextFilesTestCase { + description: "zip: members of an archive larger than one range request are read from its start and end, its other members skipped", + encoding: csv_zip(Some("")), + files: large_files, + unsampled: &[], + exclude: &[], + pinned: &[], + expected_rows: Ok("1|a\n2|b"), + }, + ReadTextFilesTestCase { + description: "xlsx: numbers, text, booleans, dates, timestamps and times keep their cell types; an empty cell is NULL; a whole number is an int", + encoding: xlsx(sheet(None, true, 0), None), + files: &[( + "a.xlsx", + "[Orders]\nid|name|amount|paid|day|at|t\nn:1|Caf\u{e9}|n:2.5|b:true|d:2026-11-12|dt:2026-11-12T10:30:00|tm:10:30:00\nn:2||n:3|b:false|d:2026-11-13|dt:2026-11-13T00:00:00|tm:23:59:59\n", + )], + unsampled: &[], + exclude: &[], + pinned: &[], + expected_rows: Ok( + "1|Caf\u{e9}|2.5|true|2026-11-12|2026-11-12T10:30:00|10:30:00\n2|NULL|3.0|false|2026-11-13|2026-11-13T00:00:00|23:59:59", + ), + }, + ReadTextFilesTestCase { + description: "xlsx: the named sheet, its title rows skipped, the header after them; empty rows are left out", + encoding: xlsx(sheet(Some("Data"), true, 2), None), + files: &[( + "a.xlsx", + "[Cover]\nn:99\n[Data]\nMonthly report|\n\nid|v\nn:7|x\n\nn:8|y\n", + )], + unsampled: &[], + exclude: &[], + pinned: &[], + expected_rows: Ok("7|x\n8|y"), + }, + ReadTextFilesTestCase { + description: "xlsx without a header: columns are numbered by their place in the sheet", + encoding: xlsx(sheet(None, false, 0), None), + files: &[("a.xlsx", "[S]\nn:1|a\nn:2|b\n")], + unsampled: &[], + exclude: &[], + pinned: &[], + expected_rows: Ok("1|a\n2|b"), + }, + ReadTextFilesTestCase { + description: "xlsx: a sheet the workbook lacks fails, naming its sheets", + encoding: xlsx(sheet(Some("Nope"), true, 0), None), + files: &[("a.xlsx", "[Orders]\nid\nn:1\n")], + unsampled: &[], + exclude: &[], + pinned: &[], + expected_rows: Err("the workbook has no sheet \"Nope\"; its sheets are [\"Orders\"]"), + }, + ReadTextFilesTestCase { + description: "xlsx: a header naming a column twice fails with its row", + encoding: xlsx(sheet(None, true, 0), None), + files: &[("a.xlsx", "[S]\nid|id\nn:1|n:2\n")], + unsampled: &[], + exclude: &[], + pinned: &[], + expected_rows: Err("row 1: column id appears twice in the header"), + }, + ReadTextFilesTestCase { + description: "xlsx: a value later rows hold that the sampled type does not fit fails with its row", + encoding: xlsx(sheet(None, true, 0), None), + files: &[("a.xlsx", "[S]\nid\nn:1\n")], + unsampled: &[("b.xlsx", "[S]\nid\nn:2\nlate text\n")], + exclude: &[], + pinned: &[], + expected_rows: Err("b.xlsx: line 3: id: late text does not fit int"), + }, + ReadTextFilesTestCase { + description: "xml: attributes and child elements by name, a repeated or listed child as a JSON array, a nested element as a JSON object, entities and CDATA decoded, elements off the path ignored", + encoding: xml(xml_at("orders/order", &["tag"]), None), + files: &[("a.xml", ORDERS_XML)], + unsampled: &[], + exclude: &[], + pinned: &[], + expected_rows: Ok("1|A & B|[\"x\",\"y\"]|{\"city\":\"Oslo\"}\n2||[\"z\"]|NULL"), + }, + ReadTextFilesTestCase { + description: "xml: names match without their namespace prefix", + encoding: xml(xml_at("/orders/order/", &[]), None), + files: &[( + "a.xml", + "2", + )], + unsampled: &[], + exclude: &[], + pinned: &[], + expected_rows: Ok("5|2"), + }, + ReadTextFilesTestCase { + description: "xml: a malformed document fails with its line", + encoding: xml(xml_at("orders/order", &[]), None), + files: &[("a.xml", "\n\n1\n")], + unsampled: &[], + exclude: &[], + pinned: &[], + expected_rows: Err("a.xml: line 3: "), + }, + ReadTextFilesTestCase { + description: "xlsx: a header may use a numbered column's name when no headerless column takes it", + encoding: xlsx(sheet(None, true, 0), None), + files: &[("a.xlsx", "[S]\ncolumn_2|column_1\nn:1|n:2\n")], + unsampled: &[], + exclude: &[], + pinned: &[], + expected_rows: Ok("1|2"), + }, + ReadTextFilesTestCase { + description: "xlsx: a headerless column whose numbered name the header already uses fails with its row", + encoding: xlsx(sheet(None, true, 0), None), + files: &[("a.xlsx", "[S]\ncolumn_2|\nn:1|n:2\n")], + unsampled: &[], + exclude: &[], + pinned: &[], + expected_rows: Err( + "row 2: the column without a header in place 2 would be named column_2, which the header already uses", + ), + }, + ReadTextFilesTestCase { + description: "xml: a document that ends inside an element fails rather than loading the records before the cut", + encoding: xml(xml_at("orders/order", &[]), None), + files: &[( + "a.xml", + "\n\n5\n", + )], + unsampled: &[], + exclude: &[], + pinned: &[], + expected_rows: Err("a.xml: line 4: the document ends inside "), + }, + ReadTextFilesTestCase { + description: "xml: a UTF-16 document fails, naming its encoding", + encoding: xml(xml_at("orders/order", &[]), None), + files: &[("a.utf16.xml", "")], + unsampled: &[], + exclude: &[], + pinned: &[], + expected_rows: Err("a.utf16.xml: the document is UTF-16"), + }, + ReadTextFilesTestCase { + description: "xml: namespace declarations and xsi attributes are not columns; xsi:nil is NULL; record_path may name prefixes", + encoding: xml(xml_at("p:orders/p:order", &[]), None), + files: &[( + "a.xml", + "34.51", + )], + unsampled: &[], + exclude: &[], + pinned: &[], + expected_rows: Ok("1|3|NULL\n2|1|4.5"), + }, + ReadTextFilesTestCase { + description: "xml: two attributes with one local name fail rather than one overwriting the other", + encoding: xml(xml_at("orders/order", &[]), None), + files: &[( + "a.xml", + "", + )], + unsampled: &[], + exclude: &[], + pinned: &[], + expected_rows: Err("a.xml: line 1: has two attributes named id"), + }, + ReadTextFilesTestCase { + description: "xml: a record_path whose parent element is there but holds no such element loads no rows", + encoding: xml(xml_at("orders/ordr", &[]), None), + files: &[("a.xml", ORDERS_XML)], + unsampled: &[], + exclude: &[], + pinned: &[], + expected_rows: Ok(""), + }, + ReadTextFilesTestCase { + description: "xml: a record_path matching no element fails", + encoding: xml(xml_at("order/item", &[]), None), + files: &[("a.xml", ORDERS_XML)], + unsampled: &[], + exclude: &[], + pinned: &[], + expected_rows: Err("a.xml: no element at record_path \"order/item\""), + }, + ReadTextFilesTestCase { + description: "zip: each CSV member is a file of its own, other members and folders left out", + encoding: csv_zip(Some("")), + files: &[ + ( + "a.zip", + "orders/\u{1e}orders/a.csv\u{1f}id,name\n1,a\n\u{1e}readme.txt\u{1f}hello\n\u{1e}b.csv\u{1f}id,name\n2,b\n", + ), + ("c.csv", "id,name\n3,loose\n"), + ], + unsampled: &[], + exclude: &[], + pinned: &[], + expected_rows: Ok("1|a\n2|b"), + }, + ReadTextFilesTestCase { + description: "zip: members picks the members by pattern", + encoding: csv_zip(Some("orders/*.csv")), + files: &[( + "a.zip", + "orders/a.csv\u{1f}id,name\n1,a\n\u{1e}b.csv\u{1f}id,name\n2,b\n", + )], + unsampled: &[], + exclude: &[], + pinned: &[], + expected_rows: Ok("1|a"), + }, + ReadTextFilesTestCase { + description: "zip: an xlsx workbook in a ZIP archive", + encoding: xlsx(sheet(None, true, 0), Some("")), + files: &[("a.zip", "book.xlsx\u{1f}[S]\nid|name\nn:5|e\n")], + unsampled: &[], + exclude: &[], + pinned: &[], + expected_rows: Ok("5|e"), + }, + ReadTextFilesTestCase { + description: "zip: XML documents in a ZIP archive", + encoding: xml(xml_at("orders/order", &["tag"]), Some("*.xml")), + files: &[( + "a.zip", + "one.xml\u{1f}\u{1e}two.xml\u{1f}", + )], + unsampled: &[], + exclude: &[], + pinned: &[], + expected_rows: Ok("1\n2"), + }, + ReadTextFilesTestCase { + description: "zip: a ZIP archive without a member of the format fails as an empty source", + encoding: csv_zip(Some("")), + files: &[("a.zip", "readme.txt\u{1f}hello\n")], + unsampled: &[], + exclude: &[], + pinned: &[], + expected_rows: Err("no .csv members of .zip files found"), + }, + ]; + for test_case in test_cases { + let rows = block_on(async { Ok(read_rows(&test_case).await) })?; + assert_eq!( + rows.as_deref().ok(), + test_case.expected_rows.ok(), + "{}: {rows:?}", + test_case.description + ); + assert!( + rows.as_ref().err().is_none_or( + |error| error.contains(test_case.expected_rows.err().unwrap_or_default()) + ), + "{}: {rows:?}", + test_case.description + ); + } + Ok(()) +} diff --git a/crates/sqlload/src/files_source/tests/test_select_files.rs b/crates/sqlload/src/files_source/tests/test_select_files.rs index d4ea091b..2c3790d4 100644 --- a/crates/sqlload/src/files_source/tests/test_select_files.rs +++ b/crates/sqlload/src/files_source/tests/test_select_files.rs @@ -72,6 +72,7 @@ fn given_listed_files_when_selecting_then_loaded_files_and_those_outside_the_win size: 1, etag: None, modified_ms: 0, + member: None, }, partitions: vec![("day".to_owned(), (*day).to_owned())], }) diff --git a/crates/sqlload/src/loader/_helpers/publish.rs b/crates/sqlload/src/loader/_helpers/publish.rs index 90a81933..c373574c 100644 --- a/crates/sqlload/src/loader/_helpers/publish.rs +++ b/crates/sqlload/src/loader/_helpers/publish.rs @@ -763,6 +763,7 @@ impl Publisher { }, ) } + SourceFormat::Xlsx | SourceFormat::Xml => return Ok(BatchFooters::default()), SourceFormat::Json | SourceFormat::Ndjson => { let (Some(read), true) = (&self.table.files, is_direct(self.table.route.as_ref())) else { diff --git a/crates/sqlload/src/loader/_helpers/table_plan.rs b/crates/sqlload/src/loader/_helpers/table_plan.rs index f37865ad..0543fa89 100644 --- a/crates/sqlload/src/loader/_helpers/table_plan.rs +++ b/crates/sqlload/src/loader/_helpers/table_plan.rs @@ -32,7 +32,7 @@ use crate::column_types::models::ColumnType; use crate::column_types::models::Warehouse; use crate::files_source::constants::SOURCE_ROW_COLUMN; use crate::files_source::models::FilesRead; -use crate::files_source::models::SourceFormat; +use crate::files_source::models::{Compression, SourceFormat}; use crate::loader::_helpers::python::with_live_columns; use crate::loader::_helpers::sources::column_fix; use crate::loader::_helpers::sources::files_read_columns; @@ -763,6 +763,8 @@ fn reader_offer( SourceFormat::Parquet => FileFormat::Parquet, SourceFormat::Ndjson | SourceFormat::Json => FileFormat::Json, SourceFormat::Csv => FileFormat::Csv, + SourceFormat::Xlsx => FileFormat::Xlsx, + SourceFormat::Xml => FileFormat::Xml, }, }); let direct_blocked = read.and_then(|read| direct_blocked(spec, read, stage)); @@ -782,11 +784,24 @@ fn reader_offer( (true, false, Some(_)) => { Some("the table has checks, which run on the rows SQLLoad stages".to_owned()) } - (true, false, None) => spec - .files - .as_ref() - .filter(|files| matches!(files.location.store, StoreKind::Sftp { .. })) - .map(|_| "no destination reads files over SFTP itself".to_owned()), + (true, false, None) => spec.files.as_ref().and_then(|files| { + match ( + &files.location.store, + files.encoding.format, + files.encoding.compression, + ) { + (StoreKind::Sftp { .. }, _, _) => { + Some("no destination reads files over SFTP itself".to_owned()) + } + (_, _, Compression::Zip) => { + Some("no destination reads ZIP archives' members itself".to_owned()) + } + (_, format @ (SourceFormat::Xlsx | SourceFormat::Xml), _) => { + Some(format!("no destination reads {format} files itself")) + } + _ => None, + } + }), }; let unload_to: Vec = match &spec.warehouse_source { Some(WarehouseSource { diff --git a/crates/sqlload/src/project/_helpers/references.rs b/crates/sqlload/src/project/_helpers/references.rs index 8dd3be67..c25a4df8 100644 --- a/crates/sqlload/src/project/_helpers/references.rs +++ b/crates/sqlload/src/project/_helpers/references.rs @@ -2,7 +2,7 @@ use crate::files_source::constants::LOADER_COLUMNS; use crate::files_source::main::constant_columns::constant_columns; use crate::files_source::main::path_template::path_template; use crate::files_source::main::text_charset::text_charset; -use crate::files_source::models::{RecordShape, SourceFormat}; +use crate::files_source::models::{Compression, RecordShape, SourceFormat}; use crate::project::_helpers::targets::{ destination_kind_problems, is_files_target, stage_problems, target_problems, }; @@ -360,11 +360,14 @@ fn files_only_problems(place: &str, source: &SourceConfig) -> Vec { if source.format != SourceFormat::Parquet || source.compression.is_some() || source.csv.is_some() + || source.xlsx.is_some() + || source.xml.is_some() + || source.members.is_some() || source.shape != RecordShape::Columns || source.record_column.is_some() { problems.push(format!( - "{place} format, compression, csv, shape and record_column are for kind = \"files\"" + "{place} format, compression, csv, xlsx, xml, members, shape and record_column are for kind = \"files\"" )); } problems @@ -454,6 +457,41 @@ fn path_value_problems(place: &str, source: &SourceConfig) -> Vec { problems } +/// Settings of the xlsx and xml formats and of ZIP archives that do not fit the source. +fn container_problems(place: &str, source: &SourceConfig) -> Vec { + let mut problems: Vec = Vec::new(); + let zip = source.compression == Some(Compression::Zip); + if source.xlsx.is_some() && source.format != SourceFormat::Xlsx { + problems.push(format!("{place} xlsx is for format = \"xlsx\"")); + } + match (&source.xml, source.format) { + (Some(_), format) if format != SourceFormat::Xml => { + problems.push(format!("{place} xml is for format = \"xml\"")); + } + (Some(xml), _) if !xml.record_path.trim_matches('/').is_empty() => {} + (_, SourceFormat::Xml) => problems.push(format!( + "{place} format = \"xml\" needs xml.record_path, the element names from the root to each row's element, such as \"orders/order\"" + )), + _ => {} + } + match (&source.members, zip) { + (Some(_), false) => problems.push(format!("{place} members is for compression = \"zip\"")), + (Some(pattern), true) => { + if let Err(error) = globset::Glob::new(pattern) { + problems.push(format!("{place} members {pattern:?}: {error}")); + } + } + (None, _) => {} + } + let loader_only = zip || matches!(source.format, SourceFormat::Xlsx | SourceFormat::Xml); + if source.copy_to_internal_stage && loader_only { + problems.push(format!( + "{place} copy_to_internal_stage is for files Snowflake reads itself; SQLLoad reads xlsx and xml files and ZIP archives' members" + )); + } + problems +} + /// Settings that do not fit a files source's format. fn encoding_problems(place: &str, source: &SourceConfig) -> Vec { let mut problems: Vec = Vec::new(); @@ -461,6 +499,11 @@ fn encoding_problems(place: &str, source: &SourceConfig) -> Vec { (RecordShape::Record, SourceFormat::Parquet, _) => problems.push(format!( "{place} shape = \"record\" is for ndjson, json and csv files; Parquet files already have typed columns" )), + (RecordShape::Record, format @ (SourceFormat::Xlsx | SourceFormat::Xml), _) => { + problems.push(format!( + "{place} shape = \"record\" is for ndjson, json and csv files, not {format}" + )); + } (RecordShape::Columns, _, Some(_)) => problems.push(format!( "{place} record_column is for shape = \"record\", which is not set" )), @@ -494,9 +537,10 @@ fn encoding_problems(place: &str, source: &SourceConfig) -> Vec { } } } + problems.extend(container_problems(place, source)); if source.format == SourceFormat::Parquet && source.compression.is_some() { problems.push(format!( - "{place} compression is for ndjson, json and csv files; Parquet files compress inside the file" + "{place} compression is for ndjson, json, csv, xlsx and xml files; Parquet files compress inside the file, and SQLLoad does not read them from ZIP archives" )); } match (&source.csv, source.format) { diff --git a/crates/sqlload/src/project/constants.rs b/crates/sqlload/src/project/constants.rs index 6bf18c6b..92d81705 100644 --- a/crates/sqlload/src/project/constants.rs +++ b/crates/sqlload/src/project/constants.rs @@ -40,6 +40,9 @@ pub const SOURCE_ONLY_KEYS: &[&str] = &[ "format", "compression", "csv", + "xlsx", + "xml", + "members", "shape", "record_column", "destination_path", diff --git a/crates/sqlload/src/project/models.rs b/crates/sqlload/src/project/models.rs index 5aed2744..586d2c3a 100644 --- a/crates/sqlload/src/project/models.rs +++ b/crates/sqlload/src/project/models.rs @@ -7,7 +7,8 @@ use serde::Deserialize; use crate::config::models::AuthConfig; use crate::files_source::constants::RECORD_COLUMN; use crate::files_source::models::{ - Compression, CsvOptions, FilesEncoding, RecordShape, SourceFormat, UnmatchedPaths, + Compression, CsvOptions, FilesEncoding, RecordShape, SourceFormat, UnmatchedPaths, XlsxOptions, + XmlOptions, }; use crate::project::_helpers::defaults; use crate::project::_helpers::serde_values::{ @@ -383,6 +384,15 @@ pub struct SourceConfig { /// How CSV files are written (files sources with `format = "csv"`). #[serde(default)] pub csv: Option, + /// Which worksheet rows load (files sources with `format = "xlsx"`). + #[serde(default)] + pub xlsx: Option, + /// Which elements are rows (files sources with `format = "xml"`). + #[serde(default)] + pub xml: Option, + /// With `compression = "zip"`, a pattern the members' names must match (files sources). + #[serde(default)] + pub members: Option, /// Typed columns (default), or each JSON object or CSV line whole in one `json` column (files sources with a text format). #[serde(default)] pub shape: RecordShape, @@ -519,6 +529,9 @@ impl SourceConfig { format: self.format, compression: self.compression.unwrap_or_default(), csv: self.csv.clone().unwrap_or_default(), + xlsx: self.xlsx.clone().unwrap_or_default(), + xml: self.xml.clone().unwrap_or_default(), + members: self.members.clone(), record: (self.shape == RecordShape::Record).then(|| { self.record_column .clone() diff --git a/crates/sqlload/src/project/tests/test_parse_project.rs b/crates/sqlload/src/project/tests/test_parse_project.rs index 1bda4145..9afa31e3 100644 --- a/crates/sqlload/src/project/tests/test_parse_project.rs +++ b/crates/sqlload/src/project/tests/test_parse_project.rs @@ -149,7 +149,7 @@ fn given_a_files_target_when_parsing_then_paths_and_defaults_are_read() -> Resul ParseProjectTestCase { description: "a files source reads its format, compression and CSV options; unset options keep their defaults", read: |project| format!("{:?}", project.sources["events"].files_encoding()), - expected_value: "FilesEncoding { format: Csv, compression: Zstd, csv: CsvOptions { delimiter: ';', header: true, quote: '\"', null_values: [\"NA\", \"\"], encoding: \"utf-8\", skip_rows: 0 }, record: None }", + expected_value: "FilesEncoding { format: Csv, compression: Zstd, csv: CsvOptions { delimiter: ';', header: true, quote: '\"', null_values: [\"NA\", \"\"], encoding: \"utf-8\", skip_rows: 0 }, xlsx: XlsxOptions { sheet: None, header: true, skip_rows: 0 }, xml: XmlOptions { record_path: \"\", lists: [] }, members: None, record: None }", }, ParseProjectTestCase { description: "a files source without settings reads Parquet", @@ -296,13 +296,43 @@ fn given_an_invalid_project_when_parsing_then_every_problem_is_named() { ParseProjectErrorTestCase { description: "compression is for text formats", text: "[connections.lake]\nkind = \"filesystem\"\nroot = \"/data\"\n[sources.raw]\nkind = \"files\"\nconnection = \"lake\"\npath = \"raw/\"\ncompression = \"gzip\"\n", - expected_message: "[sources.raw] compression is for ndjson, json and csv files; Parquet files compress inside the file", + expected_message: "[sources.raw] compression is for ndjson, json, csv, xlsx and xml files; Parquet files compress inside the file", }, ParseProjectErrorTestCase { description: "csv options need format = csv", text: "[connections.lake]\nkind = \"filesystem\"\nroot = \"/data\"\n[sources.raw]\nkind = \"files\"\nconnection = \"lake\"\npath = \"raw/\"\nformat = \"ndjson\"\ncsv = { header = false }\n", expected_message: "[sources.raw] csv is for format = \"csv\", not \"ndjson\"", }, + ParseProjectErrorTestCase { + description: "xml files need a record path", + text: "[connections.lake]\nkind = \"filesystem\"\nroot = \"/data\"\n[sources.raw]\nkind = \"files\"\nconnection = \"lake\"\npath = \"raw/\"\nformat = \"xml\"\n", + expected_message: "[sources.raw] format = \"xml\" needs xml.record_path", + }, + ParseProjectErrorTestCase { + description: "xlsx options need format = xlsx", + text: "[connections.lake]\nkind = \"filesystem\"\nroot = \"/data\"\n[sources.raw]\nkind = \"files\"\nconnection = \"lake\"\npath = \"raw/\"\nformat = \"csv\"\nxlsx = { sheet = \"Orders\" }\n", + expected_message: "[sources.raw] xlsx is for format = \"xlsx\"", + }, + ParseProjectErrorTestCase { + description: "members need compression = zip", + text: "[connections.lake]\nkind = \"filesystem\"\nroot = \"/data\"\n[sources.raw]\nkind = \"files\"\nconnection = \"lake\"\npath = \"raw/\"\nformat = \"csv\"\nmembers = \"*.csv\"\n", + expected_message: "[sources.raw] members is for compression = \"zip\"", + }, + ParseProjectErrorTestCase { + description: "whole records are not for spreadsheets", + text: "[connections.lake]\nkind = \"filesystem\"\nroot = \"/data\"\n[sources.raw]\nkind = \"files\"\nconnection = \"lake\"\npath = \"raw/\"\nformat = \"xlsx\"\nshape = \"record\"\n", + expected_message: "[sources.raw] shape = \"record\" is for ndjson, json and csv files, not xlsx", + }, + ParseProjectErrorTestCase { + description: "ZIP archives' members never go to a warehouse stage", + text: "[connections.lake]\nkind = \"filesystem\"\nroot = \"/data\"\n[sources.raw]\nkind = \"files\"\nconnection = \"lake\"\npath = \"raw/\"\nformat = \"csv\"\ncompression = \"zip\"\ncopy_to_internal_stage = true\n", + expected_message: "[sources.raw] copy_to_internal_stage is for files Snowflake reads itself", + }, + ParseProjectErrorTestCase { + description: "an unknown xlsx option is rejected", + text: "[connections.lake]\nkind = \"filesystem\"\nroot = \"/data\"\n[sources.raw]\nkind = \"files\"\nconnection = \"lake\"\npath = \"raw/\"\nformat = \"xlsx\"\nxlsx = { tab = \"Orders\" }\n", + expected_message: "unknown field `tab`", + }, ParseProjectErrorTestCase { description: "a csv delimiter is one ASCII character", text: "[connections.lake]\nkind = \"filesystem\"\nroot = \"/data\"\n[sources.raw]\nkind = \"files\"\nconnection = \"lake\"\npath = \"raw/\"\nformat = \"csv\"\ncsv = { delimiter = \"\u{e9}\" }\n", @@ -376,7 +406,7 @@ fn given_an_invalid_project_when_parsing_then_every_problem_is_named() { ParseProjectErrorTestCase { description: "format settings are for files sources", text: "[sources.app]\nkind = \"postgres\"\nconnection = \"pg\"\ninclude = [\"*\"]\nformat = \"csv\"\n", - expected_message: "[sources.app] format, compression, csv, shape and record_column are for kind = \"files\"", + expected_message: "[sources.app] format, compression, csv, xlsx, xml, members, shape and record_column are for kind = \"files\"", }, ParseProjectErrorTestCase { description: "batch_interval is for postgres_cdc sources, never silently ignored", @@ -436,7 +466,7 @@ fn given_an_invalid_project_when_parsing_then_every_problem_is_named() { ParseProjectErrorTestCase { description: "shape is for files sources", text: "[sources.app]\nkind = \"postgres\"\nconnection = \"pg\"\ninclude = [\"*\"]\nshape = \"record\"\n", - expected_message: "[sources.app] format, compression, csv, shape and record_column are for kind = \"files\"", + expected_message: "[sources.app] format, compression, csv, xlsx, xml, members, shape and record_column are for kind = \"files\"", }, ParseProjectErrorTestCase { description: "unmatched_paths without path_columns is never silently ignored", diff --git a/crates/sqlload/src/routing/models.rs b/crates/sqlload/src/routing/models.rs index eaeea426..de5b766f 100644 --- a/crates/sqlload/src/routing/models.rs +++ b/crates/sqlload/src/routing/models.rs @@ -5,6 +5,10 @@ pub enum FileFormat { /// Newline-delimited JSON, optionally gzip-compressed. Json, Csv, + /// Excel workbooks, which only SQLLoad reads. + Xlsx, + /// XML documents, which only SQLLoad reads. + Xml, } impl std::fmt::Display for FileFormat { @@ -13,6 +17,8 @@ impl std::fmt::Display for FileFormat { Self::Parquet => "parquet", Self::Json => "json", Self::Csv => "csv", + Self::Xlsx => "xlsx", + Self::Xml => "xml", }) } } diff --git a/website/src/content/docs/docs/sources/files.mdx b/website/src/content/docs/docs/sources/files.mdx index cf9e30f5..e6521a4e 100644 --- a/website/src/content/docs/docs/sources/files.mdx +++ b/website/src/content/docs/docs/sources/files.mdx @@ -27,9 +27,12 @@ cursor_column = "dt" |---------|---------| | `connection` | An `s3` or `filesystem` connection, or from 0.32.0 an `sftp` connection. | | `path` | The folder holding the files, relative to the connection. | -| `format` | `parquet` (default), `ndjson` (one object per line), `json` (objects one after another, or one top-level array) or `csv`. | -| `compression` | For text formats: `auto` (default, by file name: `.gz` is gzip, `.zst` is zstd), `none`, `gzip` or `zstd`. Parquet compresses inside the file. | +| `format` | `parquet` (default), `ndjson` (one object per line), `json` (objects one after another, or one top-level array) or `csv`; from 0.35.0 (not released yet), also `xlsx` and `xml`. See [Excel, XML and ZIP files](#excel-xml-and-zip-files). | +| `compression` | For text formats: `auto` (default, by file name: `.gz` is gzip, `.zst` is zstd), `none`, `gzip` or `zstd`; from 0.35.0 (not released yet), also `zip`, which reads the `format`'s files inside `.zip` archives. Parquet compresses inside the file. | | `csv` | `{ delimiter = ",", header = true, quote = '"', null_values = [""] }`, the defaults. Without a header, columns are `column_1`, `column_2`, … From 0.35.0 (not released yet), also `encoding = "utf-8"` and `skip_rows = 0`; see [CSV encodings](#csv-encodings). | +| `xlsx` | From 0.35.0 (not released yet). `{ sheet = "...", header = true, skip_rows = 0 }`: the worksheet (the first when omitted), whether its first row read names the columns, and rows skipped above it. | +| `xml` | From 0.35.0 (not released yet). `{ record_path = "orders/order", lists = [] }`: the elements from the root to each row's element (required), and child elements always loaded as a JSON array. | +| `members` | From 0.35.0 (not released yet). With `compression = "zip"`, a pattern the members' names must match, such as `"exports/*.csv"`. | | `shape` | From 0.27.0. `columns` (default) loads typed columns; `record` loads each JSON object or CSV line whole into one `json` column. See [Whole records](#whole-records). | | `record_column` | From 0.27.0. The column `shape = "record"` loads into; `record` when omitted. | | `partition_columns` | Hive `key=value` folders loaded as typed columns, such as `{ dt = "date" }`. | @@ -67,6 +70,73 @@ Snowflake reading CSV files itself decodes them and skips the lines in its `COPY a byte the encoding leaves undefined, such as `0x81` in cp1252, as a control character rather than failing; `plan` finds one only in the rows it samples. +## Excel, XML and ZIP files + +From 0.35.0 (not released yet). SQLLoad reads these itself on every destination, as no warehouse +does; `copy_to_internal_stage` and `shape = "record"` don't apply. + +**Excel (`format = "xlsx"`).** One worksheet loads per source: `xlsx.sheet`, or the first sheet. +`xlsx.skip_rows` skips rows at the top, such as a title; the next row names the columns unless +`header = false`, when columns are `column_1`, `column_2`, … by their place in the sheet. Empty rows +are left out. Cells keep the type the workbook stores: a whole number is `int`, other numbers +`float`, `TRUE`/`FALSE` `bool`, and a date-formatted cell `date`, `timestamp_ntz` or `time` (a +column holding dates and timestamps is `timestamp_ntz`). An empty cell or an error such as `#N/A` is +NULL. Columns are typed from a sample, as for CSV, and a later value that doesn't fit fails with its +row; pin a type in `columns` to convert instead. Formulas load their cached values. Older `.xls` +workbooks aren't read. A headerless column whose `column_N` name the header already uses fails, and +so does a sheet holding over 20 million values. + +**XML (`format = "xml"`).** Each element at `xml.record_path` is a row. Its attributes and child +elements are its columns, by name without a namespace prefix. A child that appears more than once +in a row, or one `xml.lists` names, loads as a JSON array; a child with attributes or children of +its own loads as a JSON object, keys in document order. Text is trimmed of surrounding whitespace and +typed as CSV fields are; an element with `xsi:nil="true"` is NULL. Namespace declarations and other +attributes in the XML Schema instance namespace, such as `xsi:type`, aren't columns. `record_path` may name prefixes, which are ignored. These fail, +with the line where it applies: malformed XML, a document that ends inside an element, an element +with two attributes of one name or an attribute and a child of one name, a UTF-16 document, and a +`record_path` whose parent elements the document doesn't have. + +```toml +[sources.customers] +kind = "files" +connection = "lake" +path = "drop/" +format = "xml" +xml = { record_path = "customers/customer", lists = ["phone"] } +description = "Customers from the nightly XML export." +``` + +**ZIP archives (`compression = "zip"`).** Each `.zip` file under `path` is opened, and each member +whose name ends with the format's suffix (`.csv`, `.xlsx`, `.xml`, …) and matches `members` loads +as a file of its own, recorded in `_sqlload_source_file` as `.zip/`. Other members +are left out. For an archive holding several formats, define one source per format over the same +path: + +```toml +[sources.orders] +kind = "files" +connection = "lake" +path = "drop/" +format = "csv" +compression = "zip" +members = "*.csv" +description = "Orders from the nightly ZIP export." + +[sources.lines] +kind = "files" +connection = "lake" +path = "drop/" +format = "xlsx" +compression = "zip" +xlsx = { sheet = "Lines" } +description = "Order lines from the workbook in the nightly ZIP export." +``` + +A member's version is its archive's, so with `append` an archive rewritten in place fails the table, +as any rewritten file does. Only an archive's index and the members read are fetched, with range +requests. Workbooks, XML documents and ZIP members are read whole, up to 1 GiB each; a larger member, +or an archive naming two members alike, fails. + ## Which files load | Strategy | Reads |