diff --git a/Cargo.lock b/Cargo.lock index 87608b56..74b9030f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5560,6 +5560,7 @@ dependencies = [ "clap", "csv-core", "duckdb", + "encoding_rs", "flate2", "futures-util", "getrandom 0.3.4", diff --git a/Cargo.toml b/Cargo.toml index 614b9ac8..4e017cdc 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -35,6 +35,7 @@ bytes = "1" 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"] } +encoding_rs = "0.8" clap = { version = "4", features = ["derive", "env"] } csv-core = "0.1" flate2 = "1" diff --git a/crates/sqlload/Cargo.toml b/crates/sqlload/Cargo.toml index 062dcbdf..9f471cab 100644 --- a/crates/sqlload/Cargo.toml +++ b/crates/sqlload/Cargo.toml @@ -55,6 +55,7 @@ chrono-tz.workspace = true clap = { workspace = true, optional = true } csv-core.workspace = true duckdb = { workspace = true, optional = true } +encoding_rs.workspace = true flate2.workspace = true futures-util.workspace = true getrandom.workspace = true 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 ff104c43..5efb72ba 100644 --- a/crates/sqlload/src/adapters/snowflake/_helpers/dialect/text_values.rs +++ b/crates/sqlload/src/adapters/snowflake/_helpers/dialect/text_values.rs @@ -2,12 +2,14 @@ use crate::adapters::contract::types::SqlAdapter; use crate::adapters::snowflake::constants::{ - MAX_JSON_INT, MIN_JSON_INT, MISFIT_MARKER, RFC3339_PATTERN, + LATIN1_ENCODING, MAX_JSON_INT, MIN_JSON_INT, MISFIT_MARKER, RFC3339_PATTERN, + SNOWFLAKE_ENCODINGS, UTF8_ENCODING, }; use crate::adapters::snowflake::models::SnowflakeSql; use crate::column_types::main::combined_type::combined_type; use crate::column_types::models::ColumnType; -use crate::files_source::models::{Compression, CsvOptions}; +use crate::files_source::main::text_charset::text_charset; +use crate::files_source::models::{Charset, Compression, CsvOptions}; /// The `FILE_FORMAT` a direct `COPY` reads JSON and NDJSON with: one record per object, whether the file holds objects one after another or one top-level array of them; a key given twice keeps its last value, as SQLLoad's reader does. pub(crate) fn json_file_format(compression: Compression) -> String { @@ -89,15 +91,34 @@ pub(crate) fn keys_query(from: &str, format: &str, pattern: &str) -> String { /// The `FILE_FORMAT` a direct `COPY` reads CSV with, as SQLLoad's reader does: `options`' delimiter and quote (a quote inside a quoted field written twice, backslashes plain text), the header skipped, every field as text (`csv_field` applies the NULL values, since Snowflake would not turn a quoted empty field into NULL; live, 2026-11-05). pub(crate) fn csv_file_format(options: &CsvOptions, compression: Compression) -> String { let literal = |text: &str| SnowflakeSql.render_loader_value_literal(text); + let encoding = match text_charset(&options.encoding).map(snowflake_encoding) { + Ok(Some(encoding)) if encoding != UTF8_ENCODING => { + format!(" ENCODING = {}", literal(encoding)) + } + Ok(_) | Err(_) => String::new(), + }; format!( - "TYPE = CSV FIELD_DELIMITER = {} FIELD_OPTIONALLY_ENCLOSED_BY = {} ESCAPE = NONE ESCAPE_UNENCLOSED_FIELD = NONE SKIP_HEADER = {} NULL_IF = () EMPTY_FIELD_AS_NULL = FALSE COMPRESSION = {}", + "TYPE = CSV FIELD_DELIMITER = {} FIELD_OPTIONALLY_ENCLOSED_BY = {} ESCAPE = NONE ESCAPE_UNENCLOSED_FIELD = NONE SKIP_HEADER = {} NULL_IF = () EMPTY_FIELD_AS_NULL = FALSE COMPRESSION = {}{encoding}", literal(&options.delimiter.to_string()), literal(&options.quote.to_string()), - u8::from(options.header), + options.skip_rows + u64::from(options.header), compression_name(compression) ) } +/// The `ENCODING` a direct `COPY` reads CSV files in `charset` with, when Snowflake decodes them as SQLLoad's reader does; `None` for a charset it does not map (ASCII, the East Asian ones). +pub(crate) fn snowflake_encoding(charset: Charset) -> Option<&'static str> { + match charset { + Charset::Utf8 => Some(UTF8_ENCODING), + Charset::Latin1 => Some(LATIN1_ENCODING), + Charset::Ascii => None, + Charset::Other(encoding) => SNOWFLAKE_ENCODINGS + .iter() + .find(|(name, _)| *name == encoding.name()) + .map(|(_, snowflake)| *snowflake), + } +} + /// The value of CSV field `field` (`$n`), named `name`, as `column_type` (`sql_type`), as SQLLoad's own reader converts its text; an `inferred` column fails the load on a value sampling would not have inferred it from. pub(crate) fn csv_value( (name, field): (&str, &str), diff --git a/crates/sqlload/src/adapters/snowflake/constants.rs b/crates/sqlload/src/adapters/snowflake/constants.rs index 5c914c1a..19d9cc81 100644 --- a/crates/sqlload/src/adapters/snowflake/constants.rs +++ b/crates/sqlload/src/adapters/snowflake/constants.rs @@ -108,3 +108,28 @@ pub const CONCURRENT_UPLOADS: usize = 4; /// The warehouse's name in messages. pub const SNOWFLAKE: &str = "Snowflake"; + +/// The `ENCODING` a direct `COPY` reads CSV with, for each charset SQLLoad's reader decodes the same way, by its Encoding Standard name; other charsets are refused for the direct route. +pub const SNOWFLAKE_ENCODINGS: [(&str, &str); 15] = [ + ("windows-1250", "WINDOWS1250"), + ("windows-1251", "WINDOWS1251"), + ("windows-1252", "WINDOWS1252"), + ("windows-1253", "WINDOWS1253"), + ("windows-1254", "WINDOWS1254"), + ("windows-1255", "WINDOWS1255"), + ("windows-1256", "WINDOWS1256"), + ("ISO-8859-2", "ISO88592"), + ("ISO-8859-5", "ISO88595"), + ("ISO-8859-6", "ISO88596"), + ("ISO-8859-7", "ISO88597"), + ("ISO-8859-15", "ISO885915"), + ("KOI8-R", "KOI8R"), + ("UTF-16LE", "UTF16LE"), + ("UTF-16BE", "UTF16BE"), +]; + +/// Snowflake's name for UTF-8, the `ENCODING` a direct `COPY` of CSV leaves unsaid. +pub const UTF8_ENCODING: &str = "UTF8"; + +/// Snowflake's name for ISO-8859-1, which `latin-1` reads byte for byte. +pub const LATIN1_ENCODING: &str = "ISO88591"; diff --git a/crates/sqlload/src/adapters/snowflake/main/copy/_direct_check.rs b/crates/sqlload/src/adapters/snowflake/main/copy/_direct_check.rs index d5633201..69cce435 100644 --- a/crates/sqlload/src/adapters/snowflake/main/copy/_direct_check.rs +++ b/crates/sqlload/src/adapters/snowflake/main/copy/_direct_check.rs @@ -1,9 +1,11 @@ //! Whether Snowflake may `COPY` a files table's files itself. +use crate::adapters::snowflake::_helpers::dialect::text_values::snowflake_encoding; use crate::adapters::snowflake::_helpers::write::copy_columns::copy_columns; use crate::adapters::snowflake::main::copy::_direct_problem::direct_problem; use crate::adapters::snowflake::models::{DirectCheck, ExternalStage}; -use crate::files_source::models::{DataFile, FilesRead}; +use crate::files_source::main::text_charset::text_charset; +use crate::files_source::models::{DataFile, FilesRead, SourceFormat}; use crate::loader::models::TableSpec; use crate::project::models::MaskMethod; @@ -14,6 +16,16 @@ pub(in crate::adapters) fn direct_check( stage: Option<&ExternalStage>, ) -> Option { let source = spec.files.as_ref()?; + let label = &source.encoding.csv.encoding; + let encoding = match text_charset(label) { + Ok(charset) => snowflake_encoding(charset), + Err(reason) => return Some(reason), + }; + if source.encoding.format == SourceFormat::Csv && encoding.is_none() { + return Some(format!( + "csv encoding {label:?} has no Snowflake COPY encoding SQLLoad reads the same way; Snowflake reads CSV in utf-8, latin-1, windows-1250 to windows-1256, iso-8859-2, -5, -6, -7 or -15, koi8-r, utf-16le or utf-16be" + )); + } let mut all_refused: Vec<(String, String)> = refused.to_vec(); for column in &source.table.columns { if let Some(arrow_type) = &column.direct_refusal { diff --git a/crates/sqlload/src/adapters/snowflake/tests/helpers.rs b/crates/sqlload/src/adapters/snowflake/tests/helpers.rs index cfe5607d..fd733adb 100644 --- a/crates/sqlload/src/adapters/snowflake/tests/helpers.rs +++ b/crates/sqlload/src/adapters/snowflake/tests/helpers.rs @@ -351,3 +351,23 @@ pub(crate) fn staged_float_lists(rows: Vec>>>) -> Result< .collect::>() .join("|")) } + +/// CSV files in `encoding`, with or without a header, and `skip_rows` lines before it, read whole into the record column. +pub(crate) fn csv_in( + encoding: &str, + header: bool, + skip_rows: u64, +) -> crate::files_source::models::FilesEncoding { + use crate::files_source::models::{CsvOptions, FilesEncoding, SourceFormat}; + FilesEncoding { + format: SourceFormat::Csv, + csv: CsvOptions { + header, + encoding: encoding.to_owned(), + skip_rows, + ..CsvOptions::default() + }, + record: Some(crate::files_source::constants::RECORD_COLUMN.to_owned()), + ..FilesEncoding::default() + } +} diff --git a/crates/sqlload/src/adapters/snowflake/tests/test_direct_copies.rs b/crates/sqlload/src/adapters/snowflake/tests/test_direct_copies.rs index 4c933e02..1f33b4c5 100644 --- a/crates/sqlload/src/adapters/snowflake/tests/test_direct_copies.rs +++ b/crates/sqlload/src/adapters/snowflake/tests/test_direct_copies.rs @@ -1,10 +1,14 @@ +use crate::adapters::snowflake::_helpers::dialect::text_values::snowflake_encoding; use crate::adapters::snowflake::models::CopyValue; use crate::adapters::snowflake::tests::helpers::{ - EXTERNAL, INTERNAL, column, direct_sql, many_files, order_columns, + EXTERNAL, INTERNAL, column, csv_in, direct_sql, many_files, order_columns, +}; +use crate::adapters::snowflake::tests::test_types::{ + DirectCopiesTestCase, SnowflakeEncodingTestCase, }; -use crate::adapters::snowflake::tests::test_types::DirectCopiesTestCase; use crate::column_types::models::ColumnType; use crate::files_source::constants::RECORD_COLUMN; +use crate::files_source::main::text_charset::text_charset; use crate::files_source::models::{CsvOptions, FilesEncoding, SourceFormat, StoredRead}; const WHOLE_FILE: CopyValue = CopyValue::File { @@ -159,6 +163,42 @@ fn given_files_under_the_stage_when_building_direct_copies_then_values_match_the "(SELECT IFF($2 IS NULL OR $3 IS NOT NULL, CAST(TO_VARCHAR(TO_NUMBER('sqlload: a line does not have 2 fields, as the file''s first line does')) AS VARIANT), ARRAY_CONSTRUCT(COALESCE(TO_VARIANT(NULLIF($1, 'NA')), PARSE_JSON('null')), COALESCE(TO_VARIANT(NULLIF($2, 'NA')), PARSE_JSON('null')))) FROM", )), }, + DirectCopiesTestCase { + description: "csv in cp1252 with two lines before the header: Snowflake decodes WINDOWS1252 and skips the two lines and the header", + files: vec![("part-0.csv".to_owned(), "2026-11-04")], + columns: vec![column(RECORD_COLUMN, ColumnType::Json, WHOLE_FILE)], + from: EXTERNAL, + encoding: csv_in("cp1252", true, 2), + header: &["column_1"], + expected_copies: Ok(( + 1, + "SKIP_HEADER = 3 NULL_IF = () EMPTY_FIELD_AS_NULL = FALSE COMPRESSION = AUTO ENCODING = 'WINDOWS1252')", + )), + }, + DirectCopiesTestCase { + description: "csv in latin-1 without a header, one line skipped: ISO88591, and only that line skipped", + files: vec![("part-0.csv".to_owned(), "2026-11-04")], + columns: vec![column(RECORD_COLUMN, ColumnType::Json, WHOLE_FILE)], + from: EXTERNAL, + encoding: csv_in("latin-1", false, 1), + header: &["column_1"], + expected_copies: Ok(( + 1, + "SKIP_HEADER = 1 NULL_IF = () EMPTY_FIELD_AS_NULL = FALSE COMPRESSION = AUTO ENCODING = 'ISO88591')", + )), + }, + DirectCopiesTestCase { + description: "csv in utf-8 under another label leaves ENCODING to its UTF8 default", + files: vec![("part-0.csv".to_owned(), "2026-11-04")], + columns: vec![column(RECORD_COLUMN, ColumnType::Json, WHOLE_FILE)], + from: EXTERNAL, + encoding: csv_in("UTF8", true, 0), + header: &["column_1"], + expected_copies: Ok(( + 1, + "SKIP_HEADER = 1 NULL_IF = () EMPTY_FIELD_AS_NULL = FALSE COMPRESSION = AUTO)", + )), + }, ]; for test_case in test_cases { let copies = direct_sql( @@ -185,3 +225,88 @@ fn given_files_under_the_stage_when_building_direct_copies_then_values_match_the ); } } + +#[test] +fn given_a_csv_encoding_when_choosing_snowflakes_then_only_charsets_decoded_alike_map() { + let test_cases = [ + SnowflakeEncodingTestCase { + description: "utf-8 is Snowflake's default", + label: "utf-8", + expected_encoding: Some("UTF8"), + }, + SnowflakeEncodingTestCase { + description: "cp1252 is windows-1252", + label: "cp1252", + expected_encoding: Some("WINDOWS1252"), + }, + SnowflakeEncodingTestCase { + description: "windows-1250", + label: "windows-1250", + expected_encoding: Some("WINDOWS1250"), + }, + SnowflakeEncodingTestCase { + description: "latin-1 reads byte for byte, as ISO88591 does", + label: "latin-1", + expected_encoding: Some("ISO88591"), + }, + SnowflakeEncodingTestCase { + description: "ISO-8859-1 is latin-1", + label: "ISO-8859-1", + expected_encoding: Some("ISO88591"), + }, + SnowflakeEncodingTestCase { + description: "iso88591, Snowflake's own spelling, is latin-1", + label: "iso88591", + expected_encoding: Some("ISO88591"), + }, + SnowflakeEncodingTestCase { + description: "iso-8859-9 decodes as windows-1254", + label: "iso-8859-9", + expected_encoding: Some("WINDOWS1254"), + }, + SnowflakeEncodingTestCase { + description: "iso-8859-15", + label: "iso-8859-15", + expected_encoding: Some("ISO885915"), + }, + SnowflakeEncodingTestCase { + description: "koi8-r", + label: "koi8-r", + expected_encoding: Some("KOI8R"), + }, + SnowflakeEncodingTestCase { + description: "utf-16le", + label: "utf-16le", + expected_encoding: Some("UTF16LE"), + }, + SnowflakeEncodingTestCase { + description: "Snowflake has no ASCII", + label: "ascii", + expected_encoding: None, + }, + SnowflakeEncodingTestCase { + description: "shift_jis is not mapped", + label: "shift_jis", + expected_encoding: None, + }, + SnowflakeEncodingTestCase { + description: "gb18030 is not mapped", + label: "gb18030", + expected_encoding: None, + }, + SnowflakeEncodingTestCase { + description: "an unknown label", + label: "klingon", + expected_encoding: None, + }, + ]; + for test_case in test_cases { + let encoding = text_charset(test_case.label).map(snowflake_encoding); + assert_eq!( + encoding.unwrap_or_default(), + test_case.expected_encoding, + "{}", + test_case.description + ); + } +} diff --git a/crates/sqlload/src/adapters/snowflake/tests/test_types.rs b/crates/sqlload/src/adapters/snowflake/tests/test_types.rs index 0c2dcc01..5fb8f6ea 100644 --- a/crates/sqlload/src/adapters/snowflake/tests/test_types.rs +++ b/crates/sqlload/src/adapters/snowflake/tests/test_types.rs @@ -97,3 +97,11 @@ pub(crate) struct ListsAsJsonTestCase { /// Each row's JSON text joined by `|`, or the error. pub(crate) expected_text: Result<&'static str, &'static str>, } + +pub(crate) struct SnowflakeEncodingTestCase { + pub(crate) description: &'static str, + /// The `csv.encoding` label. + pub(crate) label: &'static str, + /// Snowflake's `ENCODING` for it, or `None` when the direct route refuses it. + pub(crate) expected_encoding: Option<&'static str>, +} 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 0e5fd9ea..88db63c9 100644 --- a/crates/sqlload/src/files_source/_helpers/text/bytes_reader.rs +++ b/crates/sqlload/src/files_source/_helpers/text/bytes_reader.rs @@ -6,21 +6,70 @@ use object_store::ObjectStoreExt; use tokio::io::{AsyncRead, AsyncReadExt}; use tokio_util::io::StreamReader; +use encoding_rs::{DecoderResult, Encoding, WINDOWS_1252}; + use crate::file_store::models::{OpenFolder, StoredFile}; -use crate::files_source::constants::READ_CHUNK_BYTES; -use crate::files_source::models::Codec; +use crate::files_source::constants::{ + CP1252_UNDEFINED, ENCODING_HINT, READ_CHUNK_BYTES, UTF16_BOMS, +}; +use crate::files_source::models::{Charset, Codec}; const BYTE_ORDER_MARK: &[u8] = b"\xEF\xBB\xBF"; -/// A file's decompressed bytes, read a chunk at a time as they arrive (the whole file is never held). +/// A file's decompressed bytes as UTF-8, read a chunk at a time as they arrive (the whole file is never held). pub(crate) struct BytesReader { inner: Pin>, started: bool, + charset: Charset, + decoder: Option, + /// Bytes of the file read so far, before decoding. + read: u64, + /// Line breaks decoded so far, outside UTF-8, to say where an undecodable byte is. + lines: u64, + /// Lines still to skip at the start of the file. + skip: u64, + /// Whether a skipped line ended in a line feed, and whether skipped text held a carriage return: lines ending in a bare `\r` never end a skipped line. + skipped_breaks: (bool, bool), } impl BytesReader { - /// The next chunk, without a leading UTF-8 byte order mark; empty at the end of the file. + /// 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; + self.skip = skip_rows; + self + } + + /// The next chunk as UTF-8, without a leading byte order mark or the lines to skip; empty at the end of the file. A byte the charset cannot decode is an error naming its line and offset (UTF-8 itself is checked by the CSV reader). pub(crate) async fn next_chunk(&mut self) -> Result, String> { + loop { + let raw = self.raw_chunk().await?; + let end = raw.is_empty(); + let mut text = self.decoded(raw, end)?; + while self.skip > 0 && !text.is_empty() { + match text.iter().position(|byte| *byte == b'\n') { + Some(at) => { + text.drain(..=at); + self.skip -= 1; + self.skipped_breaks.0 = true; + } + None => { + self.skipped_breaks.1 |= text.contains(&b'\r'); + text.clear(); + } + } + } + if end && self.skip > 0 && self.skipped_breaks == (false, true) { + return Err("csv skip_rows found no line feed (\\n) in the file, whose lines end in a bare carriage return (\\r); skip_rows counts lines ending in \\n or \\r\\n, so convert the line ends or set skip_rows = 0".to_owned()); + } + if end || !text.is_empty() { + return Ok(text); + } + } + } + + /// The next chunk of the file's bytes, without a leading UTF-8 byte order mark when the charset is UTF-8. + async fn raw_chunk(&mut self) -> Result, String> { let mut chunk: Vec = Vec::with_capacity(READ_CHUNK_BYTES); let mut limited = (&mut self.inner).take(READ_CHUNK_BYTES as u64); limited @@ -29,12 +78,123 @@ impl BytesReader { .map_err(|error| format!("cannot read the file: {error}"))?; if !self.started && !chunk.is_empty() { self.started = true; - if chunk.starts_with(BYTE_ORDER_MARK) { + if self.charset == Charset::Utf8 && chunk.starts_with(BYTE_ORDER_MARK) { chunk.drain(..BYTE_ORDER_MARK.len()); + self.read += BYTE_ORDER_MARK.len() as u64; } } Ok(chunk) } + + /// `raw` decoded to UTF-8; `end` when it is the last of the file. + fn decoded(&mut self, raw: Vec, end: bool) -> Result, String> { + let text = match self.charset { + Charset::Utf8 => { + self.read += raw.len() as u64; + return Ok(raw); + } + Charset::Latin1 => encoding_rs::mem::decode_latin1(&raw) + .into_owned() + .into_bytes(), + Charset::Ascii => match raw.iter().position(|byte| !byte.is_ascii()) { + Some(at) => return Err(self.undecodable(&raw, (at, 1), "ASCII")), + None => raw.clone(), + }, + Charset::Other(encoding) => self.decode_with(encoding, &raw, end)?, + }; + self.read += raw.len() as u64; + self.lines += text.iter().filter(|byte| **byte == b'\n').count() as u64; + Ok(text) + } + + /// `raw` decoded from `encoding`, a byte order mark of its own dropped; windows-1252's five undefined bytes are errors, as in Python's cp1252, rather than control characters. + fn decode_with( + &mut self, + encoding: &'static Encoding, + raw: &[u8], + end: bool, + ) -> Result, String> { + if encoding == WINDOWS_1252 + && let Some(at) = raw.iter().position(|byte| CP1252_UNDEFINED.contains(byte)) + { + return Err(self.undecodable(raw, (at, 1), encoding.name())); + } + let [little, big] = UTF16_BOMS; + let opposite = match (encoding.name(), raw.get(..2)) { + ("UTF-16LE", Some(start)) if start == big => Some("utf-16be"), + ("UTF-16BE", Some(start)) if start == little => Some("utf-16le"), + _ => None, + }; + if self.decoder.is_none() + && let Some(opposite) = opposite + { + return Err(format!( + "line 1, byte 0: the file starts with a {opposite} byte order mark, but csv.encoding reads it as {}; set csv.encoding = \"{opposite}\"", + encoding.name() + )); + } + let decoder = self + .decoder + .get_or_insert_with(|| encoding.new_decoder_with_bom_removal()); + let room = decoder + .max_utf8_buffer_length_without_replacement(raw.len()) + .ok_or_else(|| "a chunk of the file is too large to decode".to_owned())?; + let mut text = vec![0; room]; + let (result, read, written) = + decoder.decode_to_utf8_without_replacement(raw, &mut text, end); + text.truncate(written); + match result { + DecoderResult::InputEmpty => Ok(text), + DecoderResult::Malformed(bad, after) => { + let lines = text.iter().filter(|byte| **byte == b'\n').count() as u64; + self.lines += lines; + let behind = usize::from(after) + usize::from(bad); + match read.checked_sub(behind) { + Some(at) => { + Err(self.undecodable_at(raw, (at, usize::from(bad)), encoding.name())) + } + None => Err(format!( + "line {}, byte {}: an incomplete {} sequence; {ENCODING_HINT}", + self.lines + 1, + (self.read + read as u64).saturating_sub(behind as u64), + encoding.name() + )), + } + } + DecoderResult::OutputFull => { + Err("cannot decode the file: the decoder ran out of room".to_owned()) + } + } + } + + /// The error for `bad` bytes at `at` in `raw` that `charset` cannot decode, where `raw` is ASCII-compatible, so its line breaks count lines. + fn undecodable(&self, raw: &[u8], bad: (usize, usize), charset: &str) -> String { + let breaks = raw[..bad.0].iter().filter(|byte| **byte == b'\n').count() as u64; + let line = self.lines + breaks + 1; + self.undecodable_text(raw, bad, (line, charset)) + } + + /// The error for `bad` bytes at `at` in `raw` that `charset` cannot decode, after the line breaks already counted. + fn undecodable_at(&self, raw: &[u8], bad: (usize, usize), charset: &str) -> String { + self.undecodable_text(raw, bad, (self.lines + 1, charset)) + } + + fn undecodable_text( + &self, + raw: &[u8], + (at, length): (usize, usize), + (line, charset): (u64, &str), + ) -> String { + let shown: Vec = raw[at..(at + length.max(1)).min(raw.len())] + .iter() + .map(|byte| format!("0x{byte:02X}")) + .collect(); + format!( + "line {line}, byte {}: {} is not valid {charset}; {ENCODING_HINT}", + self.read + at as u64, + shown.join(" ") + ) + } } /// A streaming reader of `file`'s bytes, decompressed with `codec`. @@ -67,6 +227,12 @@ pub(crate) async fn open_bytes( Ok(BytesReader { inner, started: false, + charset: Charset::Utf8, + decoder: None, + read: 0, + lines: 0, + skip: 0, + skipped_breaks: (false, false), }) } diff --git a/crates/sqlload/src/files_source/_helpers/text/csv_reader.rs b/crates/sqlload/src/files_source/_helpers/text/csv_reader.rs index 8e169c95..3025c6c2 100644 --- a/crates/sqlload/src/files_source/_helpers/text/csv_reader.rs +++ b/crates/sqlload/src/files_source/_helpers/text/csv_reader.rs @@ -7,7 +7,8 @@ use regex::Regex; use crate::files_source::_helpers::text::bytes_reader::BytesReader; use crate::files_source::_helpers::text::csv_records::{FirstRecord, QuoteTracker}; -use crate::files_source::constants::MAX_RECORD_BYTES; +use crate::files_source::constants::{ENCODING_HINT, MAX_RECORD_BYTES}; +use crate::files_source::main::text_charset::text_charset; use crate::files_source::models::CsvOptions; /// Reads a CSV file's records in order as text columns, a chunk of bytes at a time. @@ -22,8 +23,10 @@ pub(crate) struct CsvReader { /// Whether the decoder was told the file ended (an empty `decode`), which completes a last record without a line break. end_sent: bool, names: Vec, - /// The line of the next data record: lines count records, the header included. + /// The line of the next data record: lines count records, the header and the skipped lines included. next_line: u64, + /// Lines skipped before the header, which the record parsers' line numbers leave out. + skipped: u64, } impl CsvReader { @@ -45,24 +48,30 @@ impl CsvReader { Self::open_as(bytes, (options, batch_rows), false).await } - /// A reader positioned at the start of `bytes`; with `named`, its columns are the header's names, checked. + /// A reader positioned after the lines `options` skips in `bytes`, decoded from its encoding; with `named`, its columns are the header's names, checked. pub(crate) async fn open_as( - mut bytes: BytesReader, + bytes: BytesReader, (options, batch_rows): (&CsvOptions, usize), named: bool, ) -> Result { let delimiter = ascii(options.delimiter, "delimiter")?; let quote = ascii(options.quote, "quote")?; + let skipped = options.skip_rows; + let mut bytes = bytes.decoding(text_charset(&options.encoding)?, skipped); let mut first_record = FirstRecord::new(delimiter, quote, MAX_RECORD_BYTES); let mut pending: Vec = Vec::new(); let (first, at_end) = loop { let chunk = bytes.next_chunk().await?; pending.extend_from_slice(&chunk); - if let Some(first) = first_record.feed(&chunk)? { + let fed = first_record + .feed(&chunk) + .map_err(|error| shifted(&error, skipped))?; + if let Some(first) = fed { break (first, chunk.is_empty()); } }; - let names = column_names(first, options.header && named)?; + let names = column_names(first, options.header && named) + .map_err(|error| shifted(&error, skipped))?; let decoder = match names.is_empty() { true => None, false => Some(decoder(&names, options, batch_rows)?), @@ -76,7 +85,8 @@ impl CsvReader { at_end, end_sent: false, names, - next_line: 1 + u64::from(options.header), + next_line: 1 + skipped + u64::from(options.header), + skipped, }) } @@ -87,6 +97,8 @@ impl CsvReader { /// The next batch of records and the line its first row is on; `None` at the end of the file. pub(crate) async fn next_batch(&mut self) -> Result, String> { + let skipped = self.skipped; + let parsed = |error: String| shifted(&error, skipped); let Some(decoder) = self.decoder.as_mut() else { return Ok(None); }; @@ -98,19 +110,19 @@ impl CsvReader { } let input = &self.pending[self.offset..]; if !input.is_empty() { - let decoded = decoder.decode(input).map_err(csv_error)?; - self.tracker.push(&input[..decoded])?; + let decoded = decoder.decode(input).map_err(csv_error).map_err(parsed)?; + self.tracker.push(&input[..decoded]).map_err(parsed)?; self.offset += decoded; } else if self.at_end && !self.end_sent && decoder.capacity() > 0 { - decoder.decode(&[]).map_err(csv_error)?; + decoder.decode(&[]).map_err(csv_error).map_err(parsed)?; self.end_sent = true; } let done = self.end_sent && self.offset == self.pending.len(); if done { - self.tracker.finish()?; + self.tracker.finish().map_err(parsed)?; } if decoder.capacity() == 0 || done { - match decoder.flush().map_err(csv_error)? { + match decoder.flush().map_err(csv_error).map_err(parsed)? { Some(batch) => { let line = self.next_line; self.next_line += batch.num_rows() as u64; @@ -183,8 +195,33 @@ fn column_names(first: Vec, header: bool) -> Result, String> Ok(first) } +/// arrow-csv's error without its prefix; invalid UTF-8 says which encoding setting fixes it. fn csv_error(error: arrow_schema::ArrowError) -> String { let text = error.to_string(); - text.strip_prefix("Csv error: ") - .map_or(text.clone(), str::to_owned) + let text = text.strip_prefix("Csv error: ").unwrap_or(&text); + let invalid = text + .strip_prefix("Encountered invalid UTF-8 data for line ") + .and_then(|rest| rest.split_once(" and field ")); + match invalid { + Some((line, field)) => { + format!("line {line}, field {field}: the bytes are not valid UTF-8; {ENCODING_HINT}") + } + None => text.to_owned(), + } +} + +/// `error` with each `line N` it names moved down by the `skipped` lines the parser never saw. +fn shifted(error: &str, skipped: u64) -> String { + if skipped == 0 { + return error.to_owned(); + } + let Ok(lines) = Regex::new(r"\bline (\d+)") else { + return error.to_owned(); + }; + lines + .replace_all(error, |found: ®ex::Captures<'_>| { + let line = found[1].parse::().map_or(0, |line| line + skipped); + format!("line {line}") + }) + .into_owned() } diff --git a/crates/sqlload/src/files_source/_helpers/text/csv_records.rs b/crates/sqlload/src/files_source/_helpers/text/csv_records.rs index 6ceb19a9..0539d922 100644 --- a/crates/sqlload/src/files_source/_helpers/text/csv_records.rs +++ b/crates/sqlload/src/files_source/_helpers/text/csv_records.rs @@ -1,6 +1,7 @@ use csv_core::ReadRecordResult; use crate::files_source::_helpers::text::bytes_reader::too_long; +use crate::files_source::constants::ENCODING_HINT; /// Where a [`QuoteTracker`] is within a record, as csv-core reads it. #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -156,8 +157,9 @@ impl FirstRecord { let mut start = 0; let mut fields: Vec = Vec::new(); for end in &self.ends[..self.ended] { - let field = std::str::from_utf8(&self.output[start..*end]) - .map_err(|_| "line 1: the first record is not UTF-8".to_owned())?; + let field = std::str::from_utf8(&self.output[start..*end]).map_err(|_| { + format!("line 1: the first record is not valid UTF-8; {ENCODING_HINT}") + })?; fields.push(field.to_owned()); start = *end; } 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 692ad706..95707bb9 100644 --- a/crates/sqlload/src/files_source/_helpers/text/text_reader.rs +++ b/crates/sqlload/src/files_source/_helpers/text/text_reader.rs @@ -119,7 +119,7 @@ async fn typed_records( SourceFormat::Csv => { let reader = CsvReader::open(bytes, &request.encoding.csv, request.batch_rows).await?; for name in reader.names() { - checks.check_name("header", name, 1)?; + checks.check_name("header", name, 1 + request.encoding.csv.skip_rows)?; } Records::Csv(Box::new(reader)) } diff --git a/crates/sqlload/src/files_source/constants.rs b/crates/sqlload/src/files_source/constants.rs index ea554c60..86014c66 100644 --- a/crates/sqlload/src/files_source/constants.rs +++ b/crates/sqlload/src/files_source/constants.rs @@ -89,3 +89,35 @@ pub const NESTED_JSON_SLICE_ROWS: usize = 1024; /// The most text one Arrow text array holds: its offsets are 32-bit. pub const MAX_TEXT_ARRAY_BYTES: usize = i32::MAX as usize; + +/// The encoding CSV files are read in unless `csv.encoding` says otherwise. +pub const DEFAULT_CSV_ENCODING: &str = "utf-8"; + +/// `csv.encoding` labels read as ISO-8859-1 itself, byte for byte; the Encoding Standard reads these as windows-1252, which differs in bytes 0x80 to 0x9F. +pub const LATIN1_LABELS: [&str; 12] = [ + "latin-1", + "latin1", + "iso-8859-1", + "iso8859-1", + "iso88591", + "iso_8859-1", + "iso_8859-1:1987", + "iso-ir-100", + "csisolatin1", + "ibm819", + "cp819", + "l1", +]; + +/// The UTF-16 byte order marks, little-endian then big-endian. +pub const UTF16_BOMS: [[u8; 2]; 2] = [[0xFF, 0xFE], [0xFE, 0xFF]]; + +/// `csv.encoding` labels read as US-ASCII itself, failing on any byte above 127; the Encoding Standard reads these as windows-1252. +pub const ASCII_LABELS: [&str; 3] = ["ascii", "us-ascii", "ansi_x3.4-1968"]; + +/// The windows-1252 bytes no character is assigned to, which Python's cp1252 refuses to decode. +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\""; diff --git a/crates/sqlload/src/files_source/main/mod.rs b/crates/sqlload/src/files_source/main/mod.rs index e96d5d60..4e53037b 100644 --- a/crates/sqlload/src/files_source/main/mod.rs +++ b/crates/sqlload/src/files_source/main/mod.rs @@ -12,3 +12,4 @@ pub mod read_files; pub mod read_footers; pub mod select_files; pub mod stored_read_of; +pub mod text_charset; diff --git a/crates/sqlload/src/files_source/main/text_charset.rs b/crates/sqlload/src/files_source/main/text_charset.rs new file mode 100644 index 00000000..3974895e --- /dev/null +++ b/crates/sqlload/src/files_source/main/text_charset.rs @@ -0,0 +1,22 @@ +//! Resolve a CSV source's `encoding` setting. + +use crate::files_source::constants::{ASCII_LABELS, LATIN1_LABELS}; +use crate::files_source::models::Charset; + +/// The charset `label` names, case-insensitively: `latin-1` (and `iso-8859-1`) byte for byte, `ascii` strictly, and otherwise any label of the WHATWG Encoding Standard, such as `utf-8`, `cp1252`, `windows-1250`, `iso-8859-15`, `utf-16le`, `shift_jis` or `gbk`; an unknown label is an error. +pub fn text_charset(label: &str) -> Result { + let normal = label.trim().to_ascii_lowercase(); + if LATIN1_LABELS.contains(&normal.as_str()) { + return Ok(Charset::Latin1); + } + if ASCII_LABELS.contains(&normal.as_str()) { + return Ok(Charset::Ascii); + } + match encoding_rs::Encoding::for_label_no_replacement(normal.as_bytes()) { + Some(found) if found == encoding_rs::UTF_8 => Ok(Charset::Utf8), + Some(found) => Ok(Charset::Other(found)), + None => Err(format!( + "csv encoding {label:?} is not an encoding SQLLoad knows; use a label such as \"utf-8\", \"cp1252\", \"latin-1\" or \"utf-16le\"" + )), + } +} diff --git a/crates/sqlload/src/files_source/models.rs b/crates/sqlload/src/files_source/models.rs index f193e98a..81ad9771 100644 --- a/crates/sqlload/src/files_source/models.rs +++ b/crates/sqlload/src/files_source/models.rs @@ -7,6 +7,7 @@ use serde::Deserialize; use crate::column_types::models::ColumnType; use crate::file_store::models::{OpenFolder, StoredFile}; +use crate::files_source::constants::DEFAULT_CSV_ENCODING; use crate::schema_plan::models::DesiredColumn; /// A data file and its partition values. @@ -228,6 +229,10 @@ pub struct CsvOptions { pub quote: char, /// Field values read as NULL. pub null_values: Vec, + /// The files' character encoding, such as `utf-8` (the default), `cp1252` or `latin-1`; see [`text_charset`](crate::files_source::main::text_charset::text_charset). + pub encoding: String, + /// Lines skipped at the start of each file, before the header (or the first record without one). + pub skip_rows: u64, } impl Default for CsvOptions { @@ -237,10 +242,25 @@ impl Default for CsvOptions { header: true, quote: '"', null_values: vec![String::new()], + encoding: DEFAULT_CSV_ENCODING.to_owned(), + skip_rows: 0, } } } +/// How CSV bytes decode to text. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Charset { + /// UTF-8, read as it is. + Utf8, + /// ISO-8859-1: each byte is the character of the same number. + Latin1, + /// US-ASCII: a byte above 127 is an error. + Ascii, + /// Any other encoding the WHATWG Encoding Standard names, such as windows-1252 or UTF-16LE. + Other(&'static encoding_rs::Encoding), +} + /// How a files source's files are encoded. #[derive(Debug, Clone, Default, PartialEq, Eq)] pub struct FilesEncoding { diff --git a/crates/sqlload/src/files_source/tests/helpers.rs b/crates/sqlload/src/files_source/tests/helpers.rs index 6fc623fb..6cae6499 100644 --- a/crates/sqlload/src/files_source/tests/helpers.rs +++ b/crates/sqlload/src/files_source/tests/helpers.rs @@ -79,13 +79,35 @@ pub(crate) fn record_encoding(format: SourceFormat, csv: CsvOptions) -> FilesEnc /// Turns a test file's text into its bytes. type Encoder = fn(&str) -> Result, String>; -/// How a test file is written, by key suffix: the first suffix that matches wins. -const ENCODERS: &[(&str, Encoder)] = &[(".gz", gzip), ("", plain)]; +/// How a test file is written, by a part of its key: the first that matches wins. +const ENCODERS: &[(&str, Encoder)] = &[ + (".gz", gzip), + (".raw.", raw), + (".utf16.", utf16), + ("", plain), +]; fn plain(text: &str) -> Result, String> { Ok(text.as_bytes().to_vec()) } +/// Each character as the one byte of the same number, so `\u{e9}` writes byte 0xE9. +fn raw(text: &str) -> Result, String> { + text.chars() + .map(|character| { + u8::try_from(u32::from(character)).map_err(|_| format!("{character:?} is not one byte")) + }) + .collect() +} + +/// UTF-16LE with its byte order mark. +fn utf16(text: &str) -> Result, String> { + Ok([0xFF, 0xFE] + .into_iter() + .chain(text.encode_utf16().flat_map(u16::to_le_bytes)) + .collect()) +} + fn gzip(text: &str) -> Result, String> { let mut encoder = GzEncoder::new(Vec::new(), GzipLevel::default()); encoder @@ -94,11 +116,11 @@ fn gzip(text: &str) -> Result, String> { encoder.finish().map_err(|error| error.to_string()) } -/// `text` gzip-compressed when the key ends in `.gz`, as written otherwise. +/// `text` gzip-compressed when the key has `.gz`, byte for byte with `.raw.`, as UTF-16LE with `.utf16.`, as written otherwise. fn file_bytes(key: &str, text: &str) -> Result { let encode = ENCODERS .iter() - .find(|(suffix, _)| key.ends_with(suffix)) + .find(|(part, _)| key.contains(part)) .map_or(plain as Encoder, |(_, encode)| *encode); encode(text).map(Bytes::from) } @@ -522,3 +544,12 @@ pub(crate) fn footer_column_type(written: &DataType) -> Result { .map(ToString::to_string) .map_err(Clone::clone) } + +/// CSV options reading `encoding`, skipping `skip_rows` lines. +pub(crate) fn decoded(encoding: &str, skip_rows: u64) -> CsvOptions { + CsvOptions { + encoding: encoding.to_owned(), + skip_rows, + ..CsvOptions::default() + } +} diff --git a/crates/sqlload/src/files_source/tests/test_read_text_files.rs b/crates/sqlload/src/files_source/tests/test_read_text_files.rs index a9148683..80d45758 100644 --- a/crates/sqlload/src/files_source/tests/test_read_text_files.rs +++ b/crates/sqlload/src/files_source/tests/test_read_text_files.rs @@ -1,7 +1,7 @@ use crate::column_types::models::ColumnType; use crate::files_source::models::{CsvOptions, SourceFormat}; use crate::files_source::tests::helpers::{ - block_on, csv_encoding, encoding, read_rows, record_encoding, + block_on, csv_encoding, decoded, encoding, read_rows, record_encoding, }; use crate::files_source::tests::test_types::ReadTextFilesTestCase; @@ -420,3 +420,183 @@ fn given_json_and_csv_files_when_reading_then_rows_convert_strictly_and_errors_n } Ok(()) } + +#[test] +fn given_csv_files_in_other_encodings_or_with_leading_lines_when_reading_then_text_decodes_or_the_error_names_the_byte() +-> Result<(), String> { + let test_cases = [ + ReadTextFilesTestCase { + description: "cp1252: 0xE9 is \u{e9} and 0x80 the euro sign", + encoding: csv_encoding(decoded("cp1252", 0)), + files: &[("a.raw.csv", "id,name\n1,Caf\u{e9} \u{80}5\n")], + unsampled: &[], + exclude: &[], + pinned: &[], + expected_rows: Ok("1|Caf\u{e9} \u{20ac}5"), + }, + ReadTextFilesTestCase { + description: "latin-1 is ISO-8859-1 byte for byte: 0x80 stays U+0080, unlike cp1252", + encoding: csv_encoding(decoded("Latin-1", 0)), + files: &[("a.raw.csv", "id,name\n1,Caf\u{e9} \u{80}5\n")], + unsampled: &[], + exclude: &[], + pinned: &[], + expected_rows: Ok("1|Caf\u{e9} \u{80}5"), + }, + ReadTextFilesTestCase { + description: "utf-16le with its byte order mark and CRLF line ends", + encoding: csv_encoding(decoded("utf-16le", 0)), + files: &[("a.utf16.csv", "id,name\r\n1,Zo\u{eb}\r\n2,\u{4e2d}\r\n")], + unsampled: &[], + exclude: &[], + pinned: &[], + expected_rows: Ok("1|Zo\u{eb}\n2|\u{4e2d}"), + }, + ReadTextFilesTestCase { + description: "the default UTF-8 refuses a cp1252 byte, naming the encoding setting", + encoding: csv_encoding(CsvOptions::default()), + files: &[("a.raw.csv", "id,name\n1,ok\n2,Caf\u{e9}\n")], + unsampled: &[], + exclude: &[], + pinned: &[], + expected_rows: Err( + "field 2: the bytes are not valid UTF-8; set csv.encoding to the files' encoding", + ), + }, + ReadTextFilesTestCase { + description: "a byte cp1252 leaves undefined fails, with its line and offset, as Python's cp1252 does", + encoding: csv_encoding(decoded("cp1252", 0)), + files: &[("a.raw.csv", "id,name\n1,a\n2,b\u{81}\n")], + unsampled: &[], + exclude: &[], + pinned: &[], + expected_rows: Err( + "a.raw.csv: line 3, byte 15: 0x81 is not valid windows-1252; set csv.encoding", + ), + }, + ReadTextFilesTestCase { + description: "ascii refuses any byte above 127", + encoding: csv_encoding(decoded("ascii", 0)), + files: &[("a.raw.csv", "id\n7\n\u{e9}\n")], + unsampled: &[], + exclude: &[], + pinned: &[], + expected_rows: Err("line 3, byte 5: 0xE9 is not valid ASCII"), + }, + ReadTextFilesTestCase { + description: "shift_jis decodes, and a lead byte without its trail fails", + encoding: csv_encoding(decoded("shift_jis", 0)), + files: &[("a.raw.csv", "id,name\n1,\u{93}\u{fa}\n2,\u{93}\n")], + unsampled: &[], + exclude: &[], + pinned: &[], + expected_rows: Err("line 3, byte 15: 0x93 is not valid Shift_JIS"), + }, + ReadTextFilesTestCase { + description: "iso88591, Snowflake's spelling, is latin-1 too, not cp1252", + encoding: csv_encoding(decoded("iso88591", 0)), + files: &[("a.raw.csv", "id,name\n1,\u{81}\u{80}\n")], + unsampled: &[], + exclude: &[], + pinned: &[], + expected_rows: Ok("1|\u{81}\u{80}"), + }, + ReadTextFilesTestCase { + description: "a big-endian byte order mark under utf-16le fails, naming utf-16be", + encoding: csv_encoding(decoded("utf-16", 0)), + files: &[("a.raw.csv", "\u{fe}\u{ff}\u{0}i\u{0}d\u{0}\n")], + unsampled: &[], + exclude: &[], + pinned: &[], + expected_rows: Err( + "the file starts with a utf-16be byte order mark, but csv.encoding reads it as UTF-16LE; set csv.encoding = \"utf-16be\"", + ), + }, + ReadTextFilesTestCase { + description: "a UTF-16 file cut off inside a character fails at its last byte", + encoding: csv_encoding(decoded("utf-16le", 0)), + files: &[("a.raw.csv", "\u{ff}\u{fe}i\u{0}d\u{0}\n\u{0}1\u{0}\n\u{0}2")], + unsampled: &[], + exclude: &[], + pinned: &[], + expected_rows: Err("line 3, byte 12: an incomplete UTF-16LE sequence"), + }, + ReadTextFilesTestCase { + description: "skip_rows over lines ending in a bare carriage return fails rather than skipping the whole file", + encoding: csv_encoding(decoded("utf-8", 1)), + files: &[("a.csv", "title\rid,name\r1,a\r")], + unsampled: &[], + exclude: &[], + pinned: &[], + expected_rows: Err("found no line feed"), + }, + ReadTextFilesTestCase { + description: "an encoding SQLLoad does not know", + encoding: csv_encoding(decoded("klingon", 0)), + files: &[("a.csv", "id\n1\n")], + unsampled: &[], + exclude: &[], + pinned: &[], + expected_rows: Err("csv encoding \"klingon\" is not an encoding SQLLoad knows"), + }, + ReadTextFilesTestCase { + description: "skip_rows drops lines before the header, even one with an unclosed quote", + encoding: csv_encoding(decoded("utf-8", 2)), + files: &[( + "a.csv", + "Report run 2026-11-12\n\"unclosed, title\nid,name\n1,a\n2,b\n", + )], + unsampled: &[], + exclude: &[], + pinned: &[], + expected_rows: Ok("1|a\n2|b"), + }, + ReadTextFilesTestCase { + description: "skip_rows without a header drops lines before the first record", + encoding: csv_encoding(CsvOptions { + header: false, + ..decoded("utf-8", 1) + }), + files: &[("a.csv", "totals\n1,a\n")], + unsampled: &[], + exclude: &[], + pinned: &[], + expected_rows: Ok("1|a"), + }, + ReadTextFilesTestCase { + description: "errors after skipped lines count the file's own lines", + encoding: csv_encoding(decoded("utf-8", 1)), + files: &[("a.csv", "junk\nid,n\n1,2\n3,4,5\n")], + unsampled: &[], + exclude: &[], + pinned: &[], + expected_rows: Err("line 4"), + }, + ReadTextFilesTestCase { + description: "a skipped cp1252 file still counts skipped lines in where a byte fails", + encoding: csv_encoding(decoded("cp1252", 1)), + files: &[("a.raw.csv", "x\nid\n\u{9d}\n")], + unsampled: &[], + exclude: &[], + pinned: &[], + expected_rows: Err("line 3, byte 5: 0x9D is not valid windows-1252"), + }, + ]; + 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/project/_helpers/references.rs b/crates/sqlload/src/project/_helpers/references.rs index 66e11bb3..8dd3be67 100644 --- a/crates/sqlload/src/project/_helpers/references.rs +++ b/crates/sqlload/src/project/_helpers/references.rs @@ -1,6 +1,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::project::_helpers::targets::{ destination_kind_problems, is_files_target, stage_problems, target_problems, @@ -511,6 +512,9 @@ fn encoding_problems(place: &str, source: &SourceConfig) -> Vec { if csv.delimiter == csv.quote { problems.push(format!("{place} csv delimiter and quote must differ")); } + if let Err(problem) = text_charset(&csv.encoding) { + problems.push(format!("{place} {problem}")); + } } (Some(_), format) => problems.push(format!( "{place} csv is for format = \"csv\", not {format:?}", diff --git a/crates/sqlload/src/project/tests/test_parse_project.rs b/crates/sqlload/src/project/tests/test_parse_project.rs index 800fab2c..1bda4145 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\", \"\"] }, record: None }", + expected_value: "FilesEncoding { format: Csv, compression: Zstd, csv: CsvOptions { delimiter: ';', header: true, quote: '\"', null_values: [\"NA\", \"\"], encoding: \"utf-8\", skip_rows: 0 }, record: None }", }, ParseProjectTestCase { description: "a files source without settings reads Parquet", diff --git a/website/src/content/docs/docs/sources/files.mdx b/website/src/content/docs/docs/sources/files.mdx index 87415d5b..cf9e30f5 100644 --- a/website/src/content/docs/docs/sources/files.mdx +++ b/website/src/content/docs/docs/sources/files.mdx @@ -29,7 +29,7 @@ cursor_column = "dt" | `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. | -| `csv` | `{ delimiter = ",", header = true, quote = '"', null_values = [""] }`, the defaults. Without a header, columns are `column_1`, `column_2`, … | +| `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). | | `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" }`. | @@ -38,6 +38,35 @@ cursor_column = "dt" | `constant_columns` | From 0.26.0. Columns with the same value in every row, such as `{ region = "eu", as_of = 2026-11-08 }`. See [Columns from paths](#columns-from-paths). | | `copy_to_internal_stage` | Snowflake only: upload files no external stage covers to the user stage. See [Load methods](/docs/targets/snowflake/#files-into-snowflake). | +## CSV encodings + +From 0.35.0 (not released yet). `csv.encoding` names the files' character encoding: `utf-8` (the +default), `cp1252` (Windows Western European), `latin-1` (ISO-8859-1, each byte read as the code +point of the same number), `ascii`, `utf-16le`, `utf-16be`, or any other label of the +[Encoding Standard](https://encoding.spec.whatwg.org/#names-and-labels), such as `windows-1250`, +`iso-8859-15` or `shift_jis`. An unknown name fails `plan`. + +A byte the encoding doesn't define, such as `0x81` in cp1252 or any byte over 127 in `ascii`, fails +the table with the file, line and byte, such as +`a.csv: line 3, byte 15: 0x81 is not valid windows-1252`. A UTF-8 file with other bytes fails the +same way, with a hint to set `csv.encoding`. A UTF-16 file whose byte order mark names the other +byte order fails too, naming the encoding to set. + +`csv.skip_rows` skips that many lines at the start of each file, before the header (or the first +record without one), such as a report title. A line ends at `\n` or `\r\n`, so quotes in a skipped +line don't matter, and a file whose lines end in a bare `\r` fails rather than being skipped whole. +Line numbers in errors count the skipped lines. + +```toml +csv = { encoding = "cp1252", skip_rows = 2 } +``` + +Snowflake reading CSV files itself decodes them and skips the lines in its `COPY`, which supports +`utf-8`, `latin-1`, `windows-1250` to `windows-1256`, `iso-8859-2`, `-5`, `-6`, `-7` and `-15`, +`koi8-r`, `utf-16le` and `utf-16be`. Any other encoding fails `plan` for that route. Snowflake loads +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. + ## Which files load | Strategy | Reads | diff --git a/website/src/content/docs/docs/targets/snowflake.mdx b/website/src/content/docs/docs/targets/snowflake.mdx index cb27c540..86fb4772 100644 --- a/website/src/content/docs/docs/targets/snowflake.mdx +++ b/website/src/content/docs/docs/targets/snowflake.mdx @@ -100,5 +100,8 @@ this gives the same values as SQLLoad's own reader. - A `hash`-masked column reaches Snowflake in plain text, in a temporary table that ends with the session, and is hashed there. An `hmac` key must never appear in SQL, so `hmac`-masked files go through SQLLoad instead, which `plan` warns is much slower. +- From 0.35.0 (not released yet), CSV `encoding` and `skip_rows` go into the `COPY` as `ENCODING` + and `SKIP_HEADER`. An encoding Snowflake doesn't support fails `plan`. See + [CSV encodings](/docs/sources/files/#csv-encodings). Files are never read and re-uploaded by SQLLoad otherwise.