feat(files): read CSV in other encodings and skip leading lines

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"

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

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),

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";

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<String> {
     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 {

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

@@ -351,3 +351,23 @@ pub(crate) fn staged_float_lists(rows: Vec<Option<Vec<Option<f64>>>>) -> Result<
         .collect::<Vec<_>>()
         .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()
+    }
+}

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
+        );
+    }
+}

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>,
+}