From 69e85bb1890eeb630a62c8695d055d0955628ed3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=9C=B1=E4=BD=99=E6=99=9F?= Date: Thu, 3 Sep 2026 08:28:17 +0800 Subject: [PATCH 1/3] Generate SQL Server DDL and bulk rows from the dataset schema Add the offline SQL Server type-mapping module of ADR-0062: the created shape as CREATE TABLE DDL with exact-fit column types, mirrored nullability, bracket-quoted identifiers, and no COLLATE clause; and a BulkRowPlan that encodes each chunk's records into tiberius TokenRows in destination table column order, preserving NULLs, keeping empty strings present, carrying microsecond DATETIME2(6) values built directly at scale 6, and emitting decimals as Numeric at the declared scale (the decimal(38,38) scaled -1 regression from #156 is pinned). The encoder refuses a timestamp outside DATETIME2's 0001-9999 range as a write failure rather than reaching tiberius, whose Date::new asserts on a pre-0001 day count instead of returning an error; a panic is not the write failure ADR-0062 assigns that sliver. Over-long NVARCHAR values are not preflighted: tiberius reports them at send(). Define Created Shape in the glossary, a term ADR-0062/0064/0065 already use. No CHANGELOG entry: nothing is user-observable until a mode goes live (#136). Co-Authored-By: Claude Fable 5.1 --- CONTEXT.md | 4 + src/lib.rs | 6 + src/sqlserver.rs | 1132 ++++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 1142 insertions(+) create mode 100644 src/sqlserver.rs diff --git a/CONTEXT.md b/CONTEXT.md index ec5f3a2..b8c7907 100644 --- a/CONTEXT.md +++ b/CONTEXT.md @@ -222,6 +222,10 @@ _Avoid_: Good row, valid row, kept record A temporary data object prepared so a destination can load records through its native batch loading path. _Avoid_: Temp file, intermediate file, upload +**Created Shape**: +The destination table a load creates for a dataset schema: one column per dataset field, in field order, each with the exact-fit destination column type and the field's nullability. +_Avoid_: Generated DDL, table template, default table + **Accept Family**: The set of existing destination column shapes a dataset field may write into without silent value degradation, plus explicitly opted-in exceptions. _Avoid_: Compatible types, type whitelist, column tolerance diff --git a/src/lib.rs b/src/lib.rs index 6c97a6b..7984838 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -15,6 +15,12 @@ mod dispatch; mod rejection; mod retry; mod schema; +// The SQL Server mapping module has no live consumer until the first +// sqlserver write session (#136); until then its items are reached only by +// tests and would trip dead_code in non-test builds. Drop the attribute +// with that consumer. +#[cfg_attr(not(test), allow(dead_code))] +mod sqlserver; use connector::{ destination_connector, resolved_source_format, source_connector, DestinationWriteFacts, diff --git a/src/sqlserver.rs b/src/sqlserver.rs new file mode 100644 index 0000000..a45d251 --- /dev/null +++ b/src/sqlserver.rs @@ -0,0 +1,1132 @@ +//! The SQL Server type mapping (ADR-0062): the created shape a dataset +//! schema takes as a `CREATE TABLE` statement ([`create_table_ddl`]), and +//! the encoding of one chunk's records into the tiberius bulk rows every +//! live load mode consumes ([`BulkRowPlan`], ADR-0064). Pure and offline — +//! nothing here opens a connection, and nothing here preflights the +//! driver's representability caps: an over-long `NVARCHAR` value reaches +//! the wire and fails there as a chunk write failure. The one value the +//! encoder refuses itself is a timestamp outside the `DATETIME2` year +//! range 0001–9999, because tiberius asserts on a pre-0001 day count +//! instead of reporting it, and a panic is not the write failure ADR-0062 +//! assigns that sliver. + +use crate::LoadFailure; +use arrow_array::{ + Array, ArrayRef, BooleanArray, Decimal128Array, Float64Array, Int64Array, RecordBatch, + StringArray, TimestampMicrosecondArray, +}; +use arrow_schema::{DataType, Field, Schema, TimeUnit}; +use std::borrow::Cow; +use tiberius::numeric::Numeric; +use tiberius::time::{Date, DateTime2, Time}; +use tiberius::{ColumnData, TokenRow}; + +/// Renders the created shape of a dataset schema (ADR-0062) as one +/// `CREATE TABLE [schema].[table] (...)` statement. +pub(crate) fn create_table_ddl( + dataset: &Schema, + schema_name: &str, + table_name: &str, +) -> Result { + if dataset.fields().is_empty() { + return Err(LoadFailure { + code: "destination_write_failed", + message: format!( + "cannot create SQL Server table {}.{}: the dataset schema has no fields", + quote_identifier(schema_name), + quote_identifier(table_name) + ), + }); + } + let columns = dataset + .fields() + .iter() + .map(|field| column_ddl(field)) + .collect::, _>>()?; + Ok(format!( + "CREATE TABLE {}.{} ({})", + quote_identifier(schema_name), + quote_identifier(table_name), + columns.join(", ") + )) +} + +/// One column of the created shape: the bracket-quoted field name, its +/// exact-fit column type, and the nullability the field declares. +fn column_ddl(field: &Field) -> Result { + let column_type = ColumnType::from_arrow(field)?; + let nullability = if field.is_nullable() { + "NULL" + } else { + "NOT NULL" + }; + Ok(format!( + "{} {} {nullability}", + quote_identifier(field.name()), + column_type.ddl() + )) +} + +/// Bracket-quotes a T-SQL identifier, doubling any closing bracket the name +/// itself contains — the only character bracket quoting has to escape. +fn quote_identifier(name: &str) -> String { + format!("[{}]", name.replace(']', "]]")) +} + +/// How one chunk of a dataset becomes tiberius bulk rows for one destination +/// table: every table column, in table column order, reading exactly one +/// dataset field. The order is load-bearing — the TDS bulk path carries no +/// column list, so rows ride in the order the server's metadata states, and +/// same-typed columns in the wrong order would land transposed without an +/// error (#115). A plan is built once per write session and encodes every +/// chunk of the session through [`BulkRowPlan::rows`]. +/// +/// This slice plans full-schema rows: the table columns and the dataset +/// fields must match one-to-one by exact name. Extra destination columns — +/// the nullable, `IDENTITY`, or defaulted columns ADR-0065 admits — need +/// the introspected column type to choose their placeholder, so they join +/// the plan with the Accept Family slices. +#[derive(Debug)] +pub(crate) struct BulkRowPlan { + dataset: Schema, + columns: Vec, +} + +/// One table column of a [`BulkRowPlan`]: which chunk column it reads and +/// the column type that fixes the wire encoding. +#[derive(Debug)] +struct PlannedColumn { + name: String, + field_index: usize, + column_type: ColumnType, +} + +impl BulkRowPlan { + /// Plans the emission order for `table_columns` — the destination + /// table's column names in table order — against the dataset schema. + /// Every table column must name a dataset field exactly once, and every + /// dataset field must have a table column; anything else is a write + /// failure naming the offending column, never a silently partial row. + pub(crate) fn new(dataset: &Schema, table_columns: &[String]) -> Result { + let mismatch = |message: String| LoadFailure { + code: "destination_write_failed", + message, + }; + let mut planned = vec![false; dataset.fields().len()]; + let mut columns = Vec::with_capacity(table_columns.len()); + for name in table_columns { + let field_index = dataset.index_of(name).map_err(|_| { + mismatch(format!( + "SQL Server table column {name} has no dataset field of that name" + )) + })?; + if std::mem::replace(&mut planned[field_index], true) { + return Err(mismatch(format!( + "SQL Server table column {name} is listed more than once" + ))); + } + columns.push(PlannedColumn { + name: name.clone(), + field_index, + column_type: ColumnType::from_arrow(dataset.field(field_index))?, + }); + } + if let Some(index) = planned.iter().position(|planned| !planned) { + return Err(mismatch(format!( + "dataset field {} has no SQL Server table column", + dataset.field(index).name() + ))); + } + Ok(BulkRowPlan { + dataset: dataset.clone(), + columns, + }) + } + + /// Encodes one chunk as bulk rows: one [`TokenRow`] per record, in + /// record order, each holding the planned columns in table order. Values + /// borrow from the chunk, so the rows live no longer than it. The chunk + /// must carry the planned dataset schema; a differing chunk fails before + /// any row is produced. + pub(crate) fn rows<'a>(&self, batch: &'a RecordBatch) -> Result, LoadFailure> { + if batch.schema().fields() != self.dataset.fields() { + return Err(LoadFailure { + code: "destination_write_failed", + message: "chunk schema differs from the dataset schema the SQL Server bulk rows \ + were planned for" + .to_string(), + }); + } + let readers = self + .columns + .iter() + .map(|column| ColumnReader::new(column, batch.column(column.field_index))) + .collect::, _>>()?; + Ok(BulkRows { + readers, + next: 0, + len: batch.num_rows(), + }) + } +} + +/// The bulk rows of one chunk (see [`BulkRowPlan::rows`]). A record whose +/// value the wire cannot carry — a timestamp outside the `DATETIME2` range +/// — yields the write failure in its place and ends the rows, so a chunk +/// is either fully encodable or fails at the first offending record. +pub(crate) struct BulkRows<'a> { + readers: Vec>, + next: usize, + len: usize, +} + +impl<'a> Iterator for BulkRows<'a> { + type Item = Result, LoadFailure>; + + fn next(&mut self) -> Option { + if self.next >= self.len { + return None; + } + let record = self.next; + self.next += 1; + let mut row = TokenRow::with_capacity(self.readers.len()); + for reader in &self.readers { + match reader.value(record) { + Ok(value) => row.push(value), + Err(failure) => { + self.next = self.len; + return Some(Err(failure)); + } + } + } + Some(Ok(row)) + } + + fn size_hint(&self) -> (usize, Option) { + let remaining = self.len - self.next; + (remaining, Some(remaining)) + } +} + +/// One planned column bound to its chunk column: the typed Arrow array the +/// values are read from, resolved once per chunk so per-record encoding +/// never downcasts. +enum ColumnReader<'a> { + BigInt(&'a Int64Array), + Float53(&'a Float64Array), + Bit(&'a BooleanArray), + NvarcharMax(&'a StringArray), + DateTime2Micros { + name: String, + array: &'a TimestampMicrosecondArray, + }, + Numeric { + scale: u8, + array: &'a Decimal128Array, + }, +} + +impl<'a> ColumnReader<'a> { + fn new(column: &PlannedColumn, array: &'a ArrayRef) -> Result { + fn typed<'a, T: 'static>( + column: &PlannedColumn, + array: &'a ArrayRef, + ) -> Result<&'a T, LoadFailure> { + array + .as_any() + .downcast_ref::() + .ok_or_else(|| LoadFailure { + code: "destination_write_failed", + message: format!( + "chunk column {} is not the {} array its dataset field materializes as", + column.name, + column.column_type.ddl() + ), + }) + } + Ok(match column.column_type { + ColumnType::BigInt => ColumnReader::BigInt(typed(column, array)?), + ColumnType::Float53 => ColumnReader::Float53(typed(column, array)?), + ColumnType::Bit => ColumnReader::Bit(typed(column, array)?), + ColumnType::NvarcharMax => ColumnReader::NvarcharMax(typed(column, array)?), + ColumnType::DateTime2Micros => ColumnReader::DateTime2Micros { + name: column.name.clone(), + array: typed(column, array)?, + }, + ColumnType::Decimal { scale, .. } => ColumnReader::Numeric { + scale, + array: typed(column, array)?, + }, + }) + } + + /// The wire value of one record: `None` inside the variant for a null + /// slot — an empty string is a present value, never a null — and the + /// exact value otherwise. + fn value(&self, record: usize) -> Result, LoadFailure> { + Ok(match self { + &ColumnReader::BigInt(array) => { + ColumnData::I64(array.is_valid(record).then(|| array.value(record))) + } + &ColumnReader::Float53(array) => { + ColumnData::F64(array.is_valid(record).then(|| array.value(record))) + } + &ColumnReader::Bit(array) => { + ColumnData::Bit(array.is_valid(record).then(|| array.value(record))) + } + &ColumnReader::NvarcharMax(array) => ColumnData::String( + array + .is_valid(record) + .then(|| Cow::Borrowed(array.value(record))), + ), + ColumnReader::DateTime2Micros { name, array } => { + ColumnData::DateTime2(if array.is_valid(record) { + let micros = array.value(record); + Some( + datetime2_from_micros(micros) + .ok_or_else(|| datetime2_out_of_range(name, record, micros))?, + ) + } else { + None + }) + } + // The scale is the dataset field's declared scale, which the + // created shape carries verbatim and the Accept Family + // (ADR-0065) requires of an existing column — tiberius encodes + // a value only at the column's own scale. + &ColumnReader::Numeric { scale, array } => ColumnData::Numeric( + array + .is_valid(record) + .then(|| Numeric::new_with_scale(array.value(record), scale)), + ), + }) + } +} + +const MICROS_PER_DAY: i64 = 86_400_000_000; +/// Days from 0001-01-01, the `DATETIME2` day origin, to 1970-01-01, the +/// origin of Arrow's microsecond timestamps. +const DAYS_FROM_DATETIME2_ORIGIN_TO_UNIX_EPOCH: i64 = 719_162; +/// 9999-12-31 counted from 0001-01-01: the last day `DATETIME2` holds. +const LAST_DATETIME2_DAY: i64 = 3_652_058; + +/// Converts an Arrow microsecond timestamp — a wall-clock reading, or an +/// instant already normalized to UTC (ADR-0043) — to the `DATETIME2(6)` +/// wire value carrying the same microsecond exactly: the day count from +/// 0001-01-01 plus the microseconds into that day at scale 6, so tiberius +/// writes the value as-is with no rescaling. `None` when the day lies +/// outside 0001-01-01..=9999-12-31, the `DATETIME2` range: SQL Server +/// cannot store such a value, and tiberius asserts on a negative day +/// count rather than reporting it, so the range is checked here. +fn datetime2_from_micros(micros_since_epoch: i64) -> Option { + let day = + micros_since_epoch.div_euclid(MICROS_PER_DAY) + DAYS_FROM_DATETIME2_ORIGIN_TO_UNIX_EPOCH; + if !(0..=LAST_DATETIME2_DAY).contains(&day) { + return None; + } + let micros_of_day = micros_since_epoch.rem_euclid(MICROS_PER_DAY); + Some(DateTime2::new( + Date::new(day as u32), + Time::new(micros_of_day as u64, 6), + )) +} + +/// The write failure for a timestamp outside the `DATETIME2` range: the +/// second representability sliver of ADR-0062, a write failure rather than +/// a Rejected Record because the record satisfies the dataset schema. +fn datetime2_out_of_range(column: &str, record: usize, micros_since_epoch: i64) -> LoadFailure { + let rendered = chrono::DateTime::from_timestamp_micros(micros_since_epoch) + .map(|instant| format!("{:?}", instant.naive_utc())) + .unwrap_or_else(|| format!("{micros_since_epoch} microseconds since 1970-01-01")); + LoadFailure { + code: "destination_write_failed", + message: format!( + "record {record} of the chunk holds timestamp {rendered} in dataset field {column}, \ + outside the SQL Server DATETIME2 range 0001-01-01 to 9999-12-31" + ), + } +} + +/// The exact-fit SQL Server column type of one dataset field type +/// (ADR-0062): the narrowest type that holds every value of the field +/// exactly. Both timestamp types land as `DATETIME2(6)` — the instant +/// type's UTC normalization already happened at parse (ADR-0043), so the +/// column holds the UTC wall-clock reading — and a declared decimal keeps +/// its precision and scale verbatim, the declaration cap of 38 (ADR-0044) +/// being SQL Server's own. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum ColumnType { + BigInt, + Float53, + Bit, + NvarcharMax, + DateTime2Micros, + Decimal { precision: u8, scale: u8 }, +} + +impl ColumnType { + /// Maps one materialized dataset field to its column type. The dataset + /// schema only ever materializes the seven field types of + /// `schema::FieldType`, so an Arrow type outside that vocabulary is a + /// contract breach surfaced as a write failure naming the field rather + /// than a silent approximate mapping. + fn from_arrow(field: &Field) -> Result { + let unmapped = || LoadFailure { + code: "destination_write_failed", + message: format!( + "dataset field {} has no SQL Server column type mapping: {}", + field.name(), + field.data_type() + ), + }; + match field.data_type() { + DataType::Int64 => Ok(ColumnType::BigInt), + DataType::Float64 => Ok(ColumnType::Float53), + DataType::Boolean => Ok(ColumnType::Bit), + DataType::Utf8 => Ok(ColumnType::NvarcharMax), + DataType::Timestamp(TimeUnit::Microsecond, None) => Ok(ColumnType::DateTime2Micros), + DataType::Timestamp(TimeUnit::Microsecond, Some(zone)) if zone.as_ref() == "UTC" => { + Ok(ColumnType::DateTime2Micros) + } + DataType::Decimal128(precision, scale) => { + let precision = *precision; + let scale = u8::try_from(*scale).map_err(|_| unmapped())?; + if (1..=38).contains(&precision) && scale <= precision { + Ok(ColumnType::Decimal { precision, scale }) + } else { + Err(unmapped()) + } + } + _ => Err(unmapped()), + } + } + + fn ddl(self) -> String { + match self { + ColumnType::BigInt => "BIGINT".to_string(), + ColumnType::Float53 => "FLOAT(53)".to_string(), + ColumnType::Bit => "BIT".to_string(), + ColumnType::NvarcharMax => "NVARCHAR(MAX)".to_string(), + ColumnType::DateTime2Micros => "DATETIME2(6)".to_string(), + ColumnType::Decimal { precision, scale } => format!("DECIMAL({precision},{scale})"), + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use arrow_array::{ + BooleanArray, Decimal128Array, Float64Array, Int64Array, RecordBatch, StringArray, + TimestampMicrosecondArray, + }; + use arrow_schema::{DataType, Field, TimeUnit}; + use std::sync::Arc; + + // ---- Created-shape DDL ---- + + #[test] + fn ddl_renders_one_required_bigint_column_in_a_bracket_quoted_table() { + let dataset = Schema::new(vec![Field::new("id", DataType::Int64, false)]); + + let ddl = create_table_ddl(&dataset, "dbo", "customers").expect("mapped schema"); + + assert_eq!(ddl, "CREATE TABLE [dbo].[customers] ([id] BIGINT NOT NULL)"); + } + + /// Every dataset field type beside the exact-fit column type ADR-0062 + /// assigns it; the decimal cases span the declaration range's edges. + fn type_mapping() -> Vec<(DataType, &'static str)> { + vec![ + (DataType::Int64, "BIGINT"), + (DataType::Float64, "FLOAT(53)"), + (DataType::Boolean, "BIT"), + (DataType::Utf8, "NVARCHAR(MAX)"), + ( + DataType::Timestamp(TimeUnit::Microsecond, None), + "DATETIME2(6)", + ), + ( + DataType::Timestamp(TimeUnit::Microsecond, Some("UTC".into())), + "DATETIME2(6)", + ), + (DataType::Decimal128(18, 4), "DECIMAL(18,4)"), + (DataType::Decimal128(1, 0), "DECIMAL(1,0)"), + (DataType::Decimal128(38, 38), "DECIMAL(38,38)"), + ] + } + + #[test] + fn ddl_renders_every_type_and_nullability_combination_exactly() { + for (data_type, column_type) in type_mapping() { + for (nullable, nullability) in [(true, "NULL"), (false, "NOT NULL")] { + let dataset = Schema::new(vec![Field::new("v", data_type.clone(), nullable)]); + + let ddl = create_table_ddl(&dataset, "dbo", "t").expect("mapped schema"); + + assert_eq!( + ddl, + format!("CREATE TABLE [dbo].[t] ([v] {column_type} {nullability})"), + "{data_type} nullable={nullable}" + ); + } + } + } + + #[test] + fn ddl_keeps_the_dataset_field_order() { + let dataset = Schema::new(vec![ + Field::new("amount", DataType::Decimal128(10, 2), true), + Field::new("id", DataType::Int64, false), + Field::new("name", DataType::Utf8, true), + ]); + + let ddl = create_table_ddl(&dataset, "sales", "orders").expect("mapped schema"); + + assert_eq!( + ddl, + "CREATE TABLE [sales].[orders] ([amount] DECIMAL(10,2) NULL, [id] BIGINT NOT NULL, \ + [name] NVARCHAR(MAX) NULL)" + ); + } + + #[test] + fn ddl_doubles_closing_brackets_inside_every_identifier() { + let dataset = Schema::new(vec![Field::new("odd]name", DataType::Utf8, true)]); + + let ddl = create_table_ddl(&dataset, "sch]ema", "tab]le").expect("mapped schema"); + + assert_eq!( + ddl, + "CREATE TABLE [sch]]ema].[tab]]le] ([odd]]name] NVARCHAR(MAX) NULL)" + ); + } + + #[test] + fn ddl_never_carries_a_collate_clause() { + let fields = type_mapping() + .into_iter() + .enumerate() + .map(|(index, (data_type, _))| Field::new(format!("c{index}"), data_type, true)) + .collect::>(); + + let ddl = create_table_ddl(&Schema::new(fields), "dbo", "t").expect("mapped schema"); + + assert!( + !ddl.to_ascii_uppercase().contains("COLLATE"), + "columns inherit the database collation: {ddl}" + ); + } + + #[test] + fn ddl_fails_for_a_schema_without_fields() { + let error = create_table_ddl(&Schema::empty(), "dbo", "t").expect_err("no columns"); + + assert_eq!(error.code, "destination_write_failed"); + assert_eq!( + error.message, + "cannot create SQL Server table [dbo].[t]: the dataset schema has no fields" + ); + } + + #[test] + fn ddl_fails_for_an_arrow_type_outside_the_dataset_vocabulary() { + let unmapped = [ + DataType::Int32, + DataType::Timestamp(TimeUnit::Nanosecond, None), + DataType::Timestamp(TimeUnit::Microsecond, Some("+08:00".into())), + DataType::Decimal128(39, 0), + DataType::Decimal128(10, -1), + DataType::Decimal128(0, 0), + DataType::Decimal128(5, 6), + DataType::Decimal256(10, 2), + DataType::LargeUtf8, + ]; + for data_type in unmapped { + let dataset = Schema::new(vec![ + Field::new("id", DataType::Int64, false), + Field::new("v", data_type.clone(), true), + ]); + + let error = create_table_ddl(&dataset, "dbo", "t").expect_err("unmapped type"); + + assert_eq!(error.code, "destination_write_failed", "{data_type}"); + assert_eq!( + error.message, + format!("dataset field v has no SQL Server column type mapping: {data_type}"), + ); + } + } + + // ---- Bulk rows ---- + + fn names(names: &[&str]) -> Vec { + names.iter().map(|name| name.to_string()).collect() + } + + /// Materializes every row of a chunk under a plan, each row as its + /// column values in emission order. + fn rows_of(plan: &BulkRowPlan, batch: &RecordBatch) -> Vec>> { + plan.rows(batch) + .expect("chunk matches the plan") + .map(|row| { + row.expect("encodable record") + .into_iter() + .map(owned) + .collect() + }) + .collect() + } + + /// Detaches a row value from the chunk it borrows so rows can outlive + /// the batch inside assertions. + fn owned(value: ColumnData<'_>) -> ColumnData<'static> { + match value { + ColumnData::String(text) => { + ColumnData::String(text.map(|text| text.into_owned().into())) + } + ColumnData::I64(v) => ColumnData::I64(v), + ColumnData::F64(v) => ColumnData::F64(v), + ColumnData::Bit(v) => ColumnData::Bit(v), + ColumnData::Numeric(v) => ColumnData::Numeric(v), + ColumnData::DateTime2(v) => ColumnData::DateTime2(v), + other => panic!("the mapping never emits {other:?}"), + } + } + + /// The dataset schema of the row tests: one field per column type, in + /// an order the table orders differently. + fn every_type_dataset() -> Schema { + Schema::new(vec![ + Field::new("id", DataType::Int64, false), + Field::new("name", DataType::Utf8, true), + Field::new("amount", DataType::Decimal128(18, 4), true), + Field::new("active", DataType::Boolean, true), + Field::new("ratio", DataType::Float64, true), + Field::new( + "seen_at", + DataType::Timestamp(TimeUnit::Microsecond, Some("UTC".into())), + true, + ), + Field::new( + "created_at", + DataType::Timestamp(TimeUnit::Microsecond, None), + true, + ), + ]) + } + + /// Epoch microseconds of a UTC wall-clock reading, the value both + /// timestamp columns store (ADR-0043). + fn micros(year: i32, month: u32, day: u32, h: u32, m: u32, s: u32, micro: u32) -> i64 { + chrono::NaiveDate::from_ymd_opt(year, month, day) + .expect("calendar date") + .and_hms_micro_opt(h, m, s, micro) + .expect("wall-clock time") + .and_utc() + .timestamp_micros() + } + + /// The `DATETIME2(6)` wire value of a day count from 0001-01-01 and the + /// microseconds into that day — the literal shape the tests expect. + fn datetime2(days: u32, micros_of_day: u64) -> ColumnData<'static> { + ColumnData::DateTime2(Some(DateTime2::new( + Date::new(days), + Time::new(micros_of_day, 6), + ))) + } + + fn numeric(value: i128, scale: u8) -> ColumnData<'static> { + ColumnData::Numeric(Some(Numeric::new_with_scale(value, scale))) + } + + fn text(value: &str) -> ColumnData<'static> { + ColumnData::String(Some(Cow::Owned(value.to_string()))) + } + + #[test] + fn rows_carry_int64_values_and_nulls_one_token_row_per_record() { + let dataset = Schema::new(vec![Field::new("id", DataType::Int64, true)]); + let batch = RecordBatch::try_new( + Arc::new(dataset.clone()), + vec![Arc::new(Int64Array::from(vec![ + Some(1), + None, + Some(i64::MAX), + ]))], + ) + .expect("batch"); + let plan = BulkRowPlan::new(&dataset, &names(&["id"])).expect("plan"); + + assert_eq!( + rows_of(&plan, &batch), + vec![ + vec![ColumnData::I64(Some(1))], + vec![ColumnData::I64(None)], + vec![ColumnData::I64(Some(i64::MAX))], + ] + ); + } + + #[test] + fn rows_emit_every_type_in_table_column_order_not_dataset_order() { + let dataset = every_type_dataset(); + let batch = RecordBatch::try_new( + Arc::new(dataset.clone()), + vec![ + Arc::new(Int64Array::from(vec![Some(7), Some(8), Some(9)])), + Arc::new(StringArray::from(vec![Some("Ada"), None, Some("")])), + Arc::new( + Decimal128Array::from(vec![Some(-1), None, Some(0)]) + .with_precision_and_scale(18, 4) + .expect("decimal(18,4)"), + ), + Arc::new(BooleanArray::from(vec![Some(true), None, Some(false)])), + Arc::new(Float64Array::from(vec![Some(0.5), None, Some(0.0)])), + Arc::new( + TimestampMicrosecondArray::from(vec![ + Some(micros(2024, 2, 29, 12, 34, 56, 1)), + None, + Some(micros(1970, 1, 1, 0, 0, 0, 0)), + ]) + .with_timezone("UTC"), + ), + Arc::new(TimestampMicrosecondArray::from(vec![ + Some(micros(2001, 9, 9, 1, 46, 40, 999_999)), + None, + Some(micros(1969, 12, 31, 23, 59, 59, 999_999)), + ])), + ], + ) + .expect("batch"); + let table_order = names(&[ + "created_at", + "active", + "id", + "amount", + "name", + "seen_at", + "ratio", + ]); + let plan = BulkRowPlan::new(&dataset, &table_order).expect("plan"); + + // 2024-02-29 is day 738,944 from 0001-01-01 (719,162 to the Unix + // epoch plus 19,782 days); 2001-09-09 is day 730,736. + assert_eq!( + rows_of(&plan, &batch), + vec![ + vec![ + datetime2(730_736, 6_400_999_999), + ColumnData::Bit(Some(true)), + ColumnData::I64(Some(7)), + numeric(-1, 4), + text("Ada"), + datetime2(738_944, 45_296_000_001), + ColumnData::F64(Some(0.5)), + ], + vec![ + ColumnData::DateTime2(None), + ColumnData::Bit(None), + ColumnData::I64(Some(8)), + ColumnData::Numeric(None), + ColumnData::String(None), + ColumnData::DateTime2(None), + ColumnData::F64(None), + ], + vec![ + datetime2(719_161, 86_399_999_999), + ColumnData::Bit(Some(false)), + ColumnData::I64(Some(9)), + numeric(0, 4), + text(""), + datetime2(719_162, 0), + ColumnData::F64(Some(0.0)), + ], + ] + ); + } + + #[test] + fn rows_keep_an_empty_string_distinct_from_null() { + let dataset = Schema::new(vec![Field::new("name", DataType::Utf8, true)]); + let batch = RecordBatch::try_new( + Arc::new(dataset.clone()), + vec![Arc::new(StringArray::from(vec![Some(""), None]))], + ) + .expect("batch"); + let plan = BulkRowPlan::new(&dataset, &names(&["name"])).expect("plan"); + + let rows = rows_of(&plan, &batch); + + assert_eq!(rows[0], vec![text("")]); + assert_eq!(rows[1], vec![ColumnData::String(None)]); + assert_ne!(rows[0], rows[1]); + } + + #[test] + fn rows_carry_the_numeric_boundaries_exactly() { + let dataset = Schema::new(vec![ + Field::new("i", DataType::Int64, false), + Field::new("f", DataType::Float64, false), + Field::new("d", DataType::Decimal128(38, 0), false), + ]); + let max_decimal = 10i128.pow(38) - 1; + let batch = RecordBatch::try_new( + Arc::new(dataset.clone()), + vec![ + Arc::new(Int64Array::from(vec![i64::MAX, i64::MIN, 0])), + Arc::new(Float64Array::from(vec![ + f64::MAX, + f64::MIN, + f64::MIN_POSITIVE, + ])), + Arc::new( + Decimal128Array::from(vec![max_decimal, -max_decimal, -1]) + .with_precision_and_scale(38, 0) + .expect("decimal(38,0)"), + ), + ], + ) + .expect("batch"); + let plan = BulkRowPlan::new(&dataset, &names(&["i", "f", "d"])).expect("plan"); + + assert_eq!( + rows_of(&plan, &batch), + vec![ + vec![ + ColumnData::I64(Some(i64::MAX)), + ColumnData::F64(Some(f64::MAX)), + numeric(max_decimal, 0), + ], + vec![ + ColumnData::I64(Some(i64::MIN)), + ColumnData::F64(Some(f64::MIN)), + numeric(-max_decimal, 0), + ], + vec![ + ColumnData::I64(Some(0)), + ColumnData::F64(Some(f64::MIN_POSITIVE)), + numeric(-1, 0), + ], + ] + ); + } + + #[test] + fn rows_carry_negative_decimals_with_the_declared_scale() { + let dataset = Schema::new(vec![Field::new("d", DataType::Decimal128(10, 3), false)]); + let batch = RecordBatch::try_new( + Arc::new(dataset.clone()), + vec![Arc::new( + Decimal128Array::from(vec![-1, -1_000, -9_999_999_999]) + .with_precision_and_scale(10, 3) + .expect("decimal(10,3)"), + )], + ) + .expect("batch"); + let plan = BulkRowPlan::new(&dataset, &names(&["d"])).expect("plan"); + + let rows = rows_of(&plan, &batch); + + // -0.001, -1.000, -9999999.999: value and scale both preserved. + assert_eq!( + rows, + vec![ + vec![numeric(-1, 3)], + vec![numeric(-1_000, 3)], + vec![numeric(-9_999_999_999, 3)], + ] + ); + for row in rows { + match &row[0] { + ColumnData::Numeric(Some(value)) => assert_eq!(value.scale(), 3), + other => panic!("expected a numeric, got {other:?}"), + } + } + } + + #[test] + fn rows_carry_the_datetime2_range_edges_and_microseconds_exactly() { + let dataset = Schema::new(vec![Field::new( + "t", + DataType::Timestamp(TimeUnit::Microsecond, None), + false, + )]); + let batch = RecordBatch::try_new( + Arc::new(dataset.clone()), + vec![Arc::new(TimestampMicrosecondArray::from(vec![ + micros(1, 1, 1, 0, 0, 0, 0), + micros(9999, 12, 31, 23, 59, 59, 999_999), + micros(2024, 6, 15, 8, 30, 0, 1), + micros(2024, 6, 15, 8, 30, 0, 999_999), + ]))], + ) + .expect("batch"); + let plan = BulkRowPlan::new(&dataset, &names(&["t"])).expect("plan"); + + // 2024-06-15 is day 739,051 from 0001-01-01; 08:30:00 is + // 30,600 seconds into the day. + assert_eq!( + rows_of(&plan, &batch), + vec![ + vec![datetime2(0, 0)], + vec![datetime2(3_652_058, 86_399_999_999)], + vec![datetime2(739_051, 30_600_000_001)], + vec![datetime2(739_051, 30_600_999_999)], + ] + ); + } + + /// Cross-checks the direct `DATETIME2` construction against tiberius's + /// own chrono conversion — the path the spike round-tripped through a + /// real server — across the range edges, leap days, and both signs of + /// the Unix epoch: the same day count, and increments scaled 7 → 6 by + /// exactly ten. + #[test] + fn rows_agree_with_the_tiberius_chrono_conversion() { + use tiberius::IntoSql; + + let samples = [ + (1, 1, 1, 0, 0, 0, 0), + (1, 1, 1, 0, 0, 0, 1), + (1600, 2, 29, 23, 59, 59, 999_999), + (1900, 3, 1, 0, 0, 0, 0), + (1969, 12, 31, 23, 59, 59, 999_999), + (1970, 1, 1, 0, 0, 0, 0), + (2000, 2, 29, 12, 0, 0, 500_000), + (2038, 1, 19, 3, 14, 8, 0), + (9999, 12, 31, 0, 0, 0, 0), + (9999, 12, 31, 23, 59, 59, 999_999), + ]; + for (year, month, day, h, m, s, micro) in samples { + let naive = chrono::NaiveDate::from_ymd_opt(year, month, day) + .expect("calendar date") + .and_hms_micro_opt(h, m, s, micro) + .expect("wall-clock time"); + let expected = match naive.into_sql() { + ColumnData::DateTime2(Some(value)) => value, + other => panic!("chrono maps to datetime2, got {other:?}"), + }; + + let actual = datetime2_from_micros(naive.and_utc().timestamp_micros()) + .unwrap_or_else(|| panic!("{naive:?} is inside the DATETIME2 range")); + + assert_eq!( + actual.date().days(), + expected.date().days(), + "{naive:?} day" + ); + assert_eq!(actual.time().scale(), 6, "{naive:?} scale"); + assert_eq!(expected.time().scale(), 7, "tiberius chrono scale"); + assert_eq!( + actual.time().increments() * 10, + expected.time().increments(), + "{naive:?} increments" + ); + } + } + + /// The #134 regression seed (ADR-0069 fork rung, #156): the scaled + /// value `-1` of a `decimal(38,38)` field — `-0.000…01` with 38 + /// fractional digits — becomes a Numeric of value `-1` at scale 38 that + /// tiberius describes as `numeric(38,38)`, deterministically and + /// without rounding. + #[test] + fn rows_convert_decimal_38_38_scaled_minus_one_to_a_scale_38_numeric() { + let dataset = Schema::new(vec![Field::new("d", DataType::Decimal128(38, 38), false)]); + let batch = RecordBatch::try_new( + Arc::new(dataset.clone()), + vec![Arc::new( + Decimal128Array::from(vec![-1]) + .with_precision_and_scale(38, 38) + .expect("decimal(38,38)"), + )], + ) + .expect("batch"); + let plan = BulkRowPlan::new(&dataset, &names(&["d"])).expect("plan"); + + let first = rows_of(&plan, &batch); + let second = rows_of(&plan, &batch); + + assert_eq!(first, second, "encoding is deterministic"); + assert_eq!(first, vec![vec![numeric(-1, 38)]]); + match &first[0][0] { + ColumnData::Numeric(Some(value)) => { + assert_eq!(value.value(), -1); + assert_eq!(value.scale(), 38); + assert_eq!( + value.precision(), + 38, + "numeric(38,38), never numeric(39,38)" + ); + } + other => panic!("expected a numeric, got {other:?}"), + } + } + + #[test] + fn rows_fail_instead_of_panicking_for_timestamps_outside_the_datetime2_range() { + let dataset = Schema::new(vec![ + Field::new("id", DataType::Int64, false), + Field::new( + "t", + DataType::Timestamp(TimeUnit::Microsecond, Some("UTC".into())), + true, + ), + ]); + // Year 0000 is reachable: `0001-01-01T00:00:00+01:00` normalizes + // to 0000-12-31T23:00:00Z (ADR-0043); the far ends guard the + // arithmetic itself. + let cases = [ + ( + micros(0, 12, 31, 23, 0, 0, 0), + "record 1 of the chunk holds timestamp 0000-12-31T23:00:00 in dataset field t, \ + outside the SQL Server DATETIME2 range 0001-01-01 to 9999-12-31", + ), + ( + micros(10_000, 1, 1, 0, 0, 0, 0), + "record 1 of the chunk holds timestamp +10000-01-01T00:00:00 in dataset field t, \ + outside the SQL Server DATETIME2 range 0001-01-01 to 9999-12-31", + ), + ( + i64::MIN, + "record 1 of the chunk holds timestamp -9223372036854775808 microseconds since \ + 1970-01-01 in dataset field t, outside the SQL Server DATETIME2 range \ + 0001-01-01 to 9999-12-31", + ), + ( + i64::MAX, + "record 1 of the chunk holds timestamp 9223372036854775807 microseconds since \ + 1970-01-01 in dataset field t, outside the SQL Server DATETIME2 range \ + 0001-01-01 to 9999-12-31", + ), + ]; + for (value, message) in cases { + let batch = RecordBatch::try_new( + Arc::new(dataset.clone()), + vec![ + Arc::new(Int64Array::from(vec![1, 2, 3])), + Arc::new( + TimestampMicrosecondArray::from(vec![Some(0), Some(value), Some(0)]) + .with_timezone("UTC"), + ), + ], + ) + .expect("batch"); + let plan = BulkRowPlan::new(&dataset, &names(&["id", "t"])).expect("plan"); + + let mut rows = plan.rows(&batch).expect("chunk matches the plan"); + + let first = rows.next().expect("record 0").expect("record 0 encodes"); + assert_eq!(first.len(), 2); + let error = rows + .next() + .expect("record 1") + .expect_err("record 1 is out of range"); + assert_eq!(error.code, "destination_write_failed"); + assert_eq!(error.message, message); + assert!( + rows.next().is_none(), + "the failing record ends the chunk's rows" + ); + } + } + + #[test] + fn plan_requires_table_columns_and_dataset_fields_to_match_one_to_one() { + let dataset = Schema::new(vec![ + Field::new("id", DataType::Int64, false), + Field::new("name", DataType::Utf8, true), + ]); + let cases: [(&[&str], &str); 4] = [ + ( + &["id", "Name"], + "SQL Server table column Name has no dataset field of that name", + ), + ( + &["id", "name", "extra"], + "SQL Server table column extra has no dataset field of that name", + ), + ( + &["id", "name", "id"], + "SQL Server table column id is listed more than once", + ), + (&["name"], "dataset field id has no SQL Server table column"), + ]; + for (table_columns, message) in cases { + let error = BulkRowPlan::new(&dataset, &names(table_columns)) + .err() + .unwrap_or_else(|| panic!("{table_columns:?} must not plan")); + + assert_eq!(error.code, "destination_write_failed"); + assert_eq!(error.message, message); + } + } + + #[test] + fn plan_fails_for_an_arrow_type_outside_the_dataset_vocabulary() { + let dataset = Schema::new(vec![Field::new("v", DataType::Int32, true)]); + + let error = BulkRowPlan::new(&dataset, &names(&["v"])).expect_err("unmapped type"); + + assert_eq!(error.code, "destination_write_failed"); + assert_eq!( + error.message, + "dataset field v has no SQL Server column type mapping: Int32" + ); + } + + #[test] + fn rows_fail_for_a_chunk_carrying_a_different_schema() { + let dataset = Schema::new(vec![Field::new("id", DataType::Int64, false)]); + let other = Schema::new(vec![Field::new("id", DataType::Int64, true)]); + let batch = RecordBatch::try_new( + Arc::new(other), + vec![Arc::new(Int64Array::from(vec![Some(1)]))], + ) + .expect("batch"); + let plan = BulkRowPlan::new(&dataset, &names(&["id"])).expect("plan"); + + let error = plan.rows(&batch).err().expect("schema mismatch"); + + assert_eq!(error.code, "destination_write_failed"); + assert_eq!( + error.message, + "chunk schema differs from the dataset schema the SQL Server bulk rows were \ + planned for" + ); + } + + #[test] + fn rows_of_an_empty_chunk_are_empty_and_sized_exactly() { + let dataset = every_type_dataset(); + let batch = RecordBatch::new_empty(Arc::new(dataset.clone())); + let order = dataset + .fields() + .iter() + .map(|field| field.name().clone()) + .collect::>(); + let plan = BulkRowPlan::new(&dataset, &order).expect("plan"); + + let rows = plan.rows(&batch).expect("chunk matches the plan"); + + assert_eq!(rows.size_hint(), (0, Some(0))); + assert_eq!(rows.count(), 0); + } + + #[test] + fn datetime2_day_constants_match_the_proleptic_gregorian_calendar() { + let origin = chrono::NaiveDate::from_ymd_opt(1, 1, 1).expect("0001-01-01"); + let unix_epoch = chrono::NaiveDate::from_ymd_opt(1970, 1, 1).expect("1970-01-01"); + let last_day = chrono::NaiveDate::from_ymd_opt(9999, 12, 31).expect("9999-12-31"); + + assert_eq!( + unix_epoch.signed_duration_since(origin).num_days(), + DAYS_FROM_DATETIME2_ORIGIN_TO_UNIX_EPOCH + ); + assert_eq!( + last_day.signed_duration_since(origin).num_days(), + LAST_DATETIME2_DAY + ); + } +} From 05ca6e515c39b576c7ca2d09d352c31192be2b17 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=9C=B1=E4=BD=99=E6=99=9F?= Date: Thu, 3 Sep 2026 08:35:24 +0800 Subject: [PATCH 2/3] Settle the two-axis review of the SQL Server mapping module Promote the write-failure constructor to one module-level helper, spell out-of-range timestamps in the dataset's own notation instead of chrono's Debug output, state that the DATETIME2 upper bound is the lower bound's rule applied symmetrically (only the pre-0001 side is forced by tiberius's assert), note that plan mismatches are invariant breaches behind the Accept Family rather than table validation, and drop "Generated DDL" from the Created Shape avoid list because ADR-0062 uses that phrase itself. Co-Authored-By: Claude Fable 5.1 --- CONTEXT.md | 2 +- src/sqlserver.rs | 129 +++++++++++++++++++++++++++-------------------- 2 files changed, 76 insertions(+), 55 deletions(-) diff --git a/CONTEXT.md b/CONTEXT.md index b8c7907..597cb53 100644 --- a/CONTEXT.md +++ b/CONTEXT.md @@ -224,7 +224,7 @@ _Avoid_: Temp file, intermediate file, upload **Created Shape**: The destination table a load creates for a dataset schema: one column per dataset field, in field order, each with the exact-fit destination column type and the field's nullability. -_Avoid_: Generated DDL, table template, default table +_Avoid_: Table template, default table, auto schema **Accept Family**: The set of existing destination column shapes a dataset field may write into without silent value degradation, plus explicitly opted-in exceptions. diff --git a/src/sqlserver.rs b/src/sqlserver.rs index a45d251..719f68f 100644 --- a/src/sqlserver.rs +++ b/src/sqlserver.rs @@ -6,9 +6,12 @@ //! driver's representability caps: an over-long `NVARCHAR` value reaches //! the wire and fails there as a chunk write failure. The one value the //! encoder refuses itself is a timestamp outside the `DATETIME2` year -//! range 0001–9999, because tiberius asserts on a pre-0001 day count -//! instead of reporting it, and a panic is not the write failure ADR-0062 -//! assigns that sliver. +//! range 0001–9999. The lower bound is forced — tiberius asserts on a +//! pre-0001 day count instead of reporting it, and a panic is not the +//! write failure ADR-0062 assigns that sliver; the upper bound is the +//! same rule applied symmetrically, so the refused range is exactly the +//! sliver ADR-0062 enumerates rather than the driver's own limit, and +//! every out-of-range value fails with one message. use crate::LoadFailure; use arrow_array::{ @@ -21,6 +24,16 @@ use tiberius::numeric::Numeric; use tiberius::time::{Date, DateTime2, Time}; use tiberius::{ColumnData, TokenRow}; +/// Every failure this module raises is a write-phase failure: the dataset +/// schema and the load rules were honored, and the destination side could +/// not take the shape or the value. +fn write_failure(message: String) -> LoadFailure { + LoadFailure { + code: "destination_write_failed", + message, + } +} + /// Renders the created shape of a dataset schema (ADR-0062) as one /// `CREATE TABLE [schema].[table] (...)` statement. pub(crate) fn create_table_ddl( @@ -29,14 +42,11 @@ pub(crate) fn create_table_ddl( table_name: &str, ) -> Result { if dataset.fields().is_empty() { - return Err(LoadFailure { - code: "destination_write_failed", - message: format!( - "cannot create SQL Server table {}.{}: the dataset schema has no fields", - quote_identifier(schema_name), - quote_identifier(table_name) - ), - }); + return Err(write_failure(format!( + "cannot create SQL Server table {}.{}: the dataset schema has no fields", + quote_identifier(schema_name), + quote_identifier(table_name) + ))); } let columns = dataset .fields() @@ -85,7 +95,11 @@ fn quote_identifier(name: &str) -> String { /// fields must match one-to-one by exact name. Extra destination columns — /// the nullable, `IDENTITY`, or defaulted columns ADR-0065 admits — need /// the introspected column type to choose their placeholder, so they join -/// the plan with the Accept Family slices. +/// the plan with the Accept Family slices. The Accept Family validates an +/// existing table's shape before a plan is built, listing every violation +/// as `incompatible_destination_table`; a mismatch caught here is +/// therefore an invariant breach between two already-validated shapes, +/// and it fails loudly rather than being reported as table validation. #[derive(Debug)] pub(crate) struct BulkRowPlan { dataset: Schema, @@ -108,20 +122,16 @@ impl BulkRowPlan { /// dataset field must have a table column; anything else is a write /// failure naming the offending column, never a silently partial row. pub(crate) fn new(dataset: &Schema, table_columns: &[String]) -> Result { - let mismatch = |message: String| LoadFailure { - code: "destination_write_failed", - message, - }; let mut planned = vec![false; dataset.fields().len()]; let mut columns = Vec::with_capacity(table_columns.len()); for name in table_columns { let field_index = dataset.index_of(name).map_err(|_| { - mismatch(format!( + write_failure(format!( "SQL Server table column {name} has no dataset field of that name" )) })?; if std::mem::replace(&mut planned[field_index], true) { - return Err(mismatch(format!( + return Err(write_failure(format!( "SQL Server table column {name} is listed more than once" ))); } @@ -132,7 +142,7 @@ impl BulkRowPlan { }); } if let Some(index) = planned.iter().position(|planned| !planned) { - return Err(mismatch(format!( + return Err(write_failure(format!( "dataset field {} has no SQL Server table column", dataset.field(index).name() ))); @@ -150,12 +160,11 @@ impl BulkRowPlan { /// any row is produced. pub(crate) fn rows<'a>(&self, batch: &'a RecordBatch) -> Result, LoadFailure> { if batch.schema().fields() != self.dataset.fields() { - return Err(LoadFailure { - code: "destination_write_failed", - message: "chunk schema differs from the dataset schema the SQL Server bulk rows \ - were planned for" + return Err(write_failure( + "chunk schema differs from the dataset schema the SQL Server bulk rows were \ + planned for" .to_string(), - }); + )); } let readers = self .columns @@ -232,17 +241,13 @@ impl<'a> ColumnReader<'a> { column: &PlannedColumn, array: &'a ArrayRef, ) -> Result<&'a T, LoadFailure> { - array - .as_any() - .downcast_ref::() - .ok_or_else(|| LoadFailure { - code: "destination_write_failed", - message: format!( - "chunk column {} is not the {} array its dataset field materializes as", - column.name, - column.column_type.ddl() - ), - }) + array.as_any().downcast_ref::().ok_or_else(|| { + write_failure(format!( + "chunk column {} is not the {} array its dataset field materializes as", + column.name, + column.column_type.ddl() + )) + }) } Ok(match column.column_type { ColumnType::BigInt => ColumnReader::BigInt(typed(column, array)?), @@ -317,7 +322,10 @@ const LAST_DATETIME2_DAY: i64 = 3_652_058; /// writes the value as-is with no rescaling. `None` when the day lies /// outside 0001-01-01..=9999-12-31, the `DATETIME2` range: SQL Server /// cannot store such a value, and tiberius asserts on a negative day -/// count rather than reporting it, so the range is checked here. +/// count rather than reporting it, so the whole range is checked here — +/// the upper bound by the same rule, although the driver itself would +/// only assert from day 2^24 (year 45,941) on and let the server reject +/// the years between. fn datetime2_from_micros(micros_since_epoch: i64) -> Option { let day = micros_since_epoch.div_euclid(MICROS_PER_DAY) + DAYS_FROM_DATETIME2_ORIGIN_TO_UNIX_EPOCH; @@ -333,18 +341,32 @@ fn datetime2_from_micros(micros_since_epoch: i64) -> Option { /// The write failure for a timestamp outside the `DATETIME2` range: the /// second representability sliver of ADR-0062, a write failure rather than -/// a Rejected Record because the record satisfies the dataset schema. +/// a Rejected Record because the record satisfies the dataset schema. The +/// value is spelled in the dataset's own timestamp notation with all six +/// fractional digits; a value beyond what a calendar can even name falls +/// back to its raw microsecond count. fn datetime2_out_of_range(column: &str, record: usize, micros_since_epoch: i64) -> LoadFailure { + use chrono::{Datelike, Timelike}; + let rendered = chrono::DateTime::from_timestamp_micros(micros_since_epoch) - .map(|instant| format!("{:?}", instant.naive_utc())) + .map(|instant| { + let value = instant.naive_utc(); + format!( + "{:04}-{:02}-{:02}T{:02}:{:02}:{:02}.{:06}", + value.year(), + value.month(), + value.day(), + value.hour(), + value.minute(), + value.second(), + value.nanosecond() / 1_000 + ) + }) .unwrap_or_else(|| format!("{micros_since_epoch} microseconds since 1970-01-01")); - LoadFailure { - code: "destination_write_failed", - message: format!( - "record {record} of the chunk holds timestamp {rendered} in dataset field {column}, \ - outside the SQL Server DATETIME2 range 0001-01-01 to 9999-12-31" - ), - } + write_failure(format!( + "record {record} of the chunk holds timestamp {rendered} in dataset field {column}, \ + outside the SQL Server DATETIME2 range 0001-01-01 to 9999-12-31" + )) } /// The exact-fit SQL Server column type of one dataset field type @@ -371,13 +393,12 @@ impl ColumnType { /// contract breach surfaced as a write failure naming the field rather /// than a silent approximate mapping. fn from_arrow(field: &Field) -> Result { - let unmapped = || LoadFailure { - code: "destination_write_failed", - message: format!( + let unmapped = || { + write_failure(format!( "dataset field {} has no SQL Server column type mapping: {}", field.name(), field.data_type() - ), + )) }; match field.data_type() { DataType::Int64 => Ok(ColumnType::BigInt), @@ -980,13 +1001,13 @@ mod tests { let cases = [ ( micros(0, 12, 31, 23, 0, 0, 0), - "record 1 of the chunk holds timestamp 0000-12-31T23:00:00 in dataset field t, \ - outside the SQL Server DATETIME2 range 0001-01-01 to 9999-12-31", + "record 1 of the chunk holds timestamp 0000-12-31T23:00:00.000000 in dataset \ + field t, outside the SQL Server DATETIME2 range 0001-01-01 to 9999-12-31", ), ( - micros(10_000, 1, 1, 0, 0, 0, 0), - "record 1 of the chunk holds timestamp +10000-01-01T00:00:00 in dataset field t, \ - outside the SQL Server DATETIME2 range 0001-01-01 to 9999-12-31", + micros(10_000, 1, 1, 0, 0, 0, 1), + "record 1 of the chunk holds timestamp 10000-01-01T00:00:00.000001 in dataset \ + field t, outside the SQL Server DATETIME2 range 0001-01-01 to 9999-12-31", ), ( i64::MIN, From 59bdee037596d644576a0ad41537c5b7a9656d27 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=9C=B1=E4=BD=99=E6=99=9F?= Date: Thu, 3 Sep 2026 08:47:55 +0800 Subject: [PATCH 3/3] Record the DATETIME2 range check as a dated note on ADR-0062 The maintainer ruled on PR #158: keep the encoder's symmetric 0001-9999 range check, and record it on the ADR that owns the sliver rather than leaving the decision in code alone. The note states what forced the lower bound (tiberius asserts on a pre-0001 day count), why the upper bound follows the same rule, and why this is a write failure rather than a preflight or a Rejected Record. Co-Authored-By: Claude Fable 5.1 --- ...2-map-dataset-fields-to-exact-fit-sql-server-column-types.md | 2 ++ 1 file changed, 2 insertions(+) diff --git a/docs/adr/0062-map-dataset-fields-to-exact-fit-sql-server-column-types.md b/docs/adr/0062-map-dataset-fields-to-exact-fit-sql-server-column-types.md index bc1637b..dd1e20f 100644 --- a/docs/adr/0062-map-dataset-fields-to-exact-fit-sql-server-column-types.md +++ b/docs/adr/0062-map-dataset-fields-to-exact-fit-sql-server-column-types.md @@ -9,3 +9,5 @@ Every dataset field type maps to the narrowest SQL Server column type that holds Two mappings are judgment calls rather than forced moves. `utf8` becomes `NVARCHAR(MAX)` because the dataset type is unbounded text and the mapping must stay true under chunked streaming: the writer sees one chunk at a time and can never know a global maximum length, and an append or merge table created from early data must not become invalid when later loads carry longer values. The cost — LOB storage and no plain index on the column — is schema-design territory that belongs to the table's consumers, not the loader. Instant timestamps land as UTC-normalized `DATETIME2(6)` rather than `DATETIMEOFFSET(6)` because the maintainer's production environment stores datetimes exclusively in the `DATETIME` family, so `DATETIME2` is what every downstream consumer already reads; the price, stated plainly, is that wall-clock and instant columns are indistinguishable by destination type — the offset discipline of ADR-0043 lives only in the load definition and the report. Nothing of the instant is lost beyond that marker: ADR-0043 already normalizes instants to UTC at parse without retaining the original offset, so a `DATETIMEOFFSET` column would have carried a constant `+00:00`. Precision is 6, not SQL Server's default 7, because the dataset contract is microseconds (ADR-0043): the column type states the real contract instead of implying 100-nanosecond precision that never arrives. Because every mapping is exact, a value that satisfies the dataset schema can fail destination typing only in two enumerated slivers: `NVARCHAR` values longer than 32,767 characters — the Tiberius bulk-path client guard measured by the spike (encoded UCS-2 length ≤ 65,535 bytes), a driver-path limit rather than a server one — and timestamps whose value lies outside SQL Server's year range of 0001–9999, reachable because the strict four-digit-year parse menu admits year 0000 and instant UTC normalization can carry a boundary value across the range edge (DuckDB's ±290k-year range never sees either case). These land as write failures, never Rejected Records: a Rejected Record is a record that violates the chosen schema or load rules, and these records violate neither — they exceed one destination path's representable range. Freezing a driver cap into record-level semantics would let the same record load on DuckDB, reject on SQL Server, and load again if the write path ever changes — semantics drifting with implementation. Failure codes, retry classification, and any client-side preflight are write-semantics and error-classification decisions (#109, #112); what an existing table's columns are allowed to look like on append or merge — including production tables whose datetime columns are legacy `DATETIME` — is likewise a named #109 question, not part of this mapping. Non-finite `float64` values are absent from the sliver list because ADR-0063 removes them at parse. `decimal(38,38)` is likewise not a third sliver: the released tiberius 0.12.3 panicked on scale 38 and would have described a scale-38 fraction as the invalid `numeric(39,38)`, but that was a dependency defect, not a representability limit — the patched revision ADR-0069 pins (#156) accepts scale 38 and declares such a value as `numeric(38,38)`, so the exact-fit `DECIMAL(p,s)` mapping holds across the complete declaration range. Alternatives rejected: `NVARCHAR(4000)` (an arbitrary bound that trades the measured 32,767 cap for a lower, server-side failure surface); length sized from observed data (unknowable under streaming, invalidates evolving append targets); pinning a collation (asserts an opinion the database already answers); `DATETIMEOFFSET(6)` for instants (a self-describing marker, but it buys two bytes per row of type distinction the report already records, against the grain of the environment this tool actually loads into); `DATETIME2(7)` (implied false precision); legacy `DATETIME` as the created type (~3.33 ms rounding against the never-round ethos of ADR-0044, and a 1753 floor below the menu's range); and tightening the parse menu to 0001–9999 (welds one destination's range into the shared contract every destination pays for). + +Range check activated (2026-09-03, #134). The bulk-row encoder refuses a timestamp outside 0001-01-01..=9999-12-31 itself, as `destination_write_failed` naming the record, the dataset field, and the value, instead of letting the value reach the driver. This is forced, not chosen: the pinned tiberius revision's `Date::new` asserts on a day count with bits above 24 — which is what a pre-0001 day becomes once cast to `u32` — rather than returning an error, and a year-0000 value is reachable both through the four-digit-year menu and through instant normalization (`0001-01-01T00:00:00+01:00`), so without the check this sliver would surface as a panic, which is not the write failure this ADR assigns it. The upper bound is the same rule applied symmetrically: the driver would let years 10000–45941 travel to the server and only assert again from day 2^24, and one range with one message states this ADR's sliver rather than the driver's accident, while naming the offending record where a server-side rejection of the bulk batch would not. The check is the conversion reporting what it cannot produce — not a preflight pass, and not record semantics: the record remains a Surviving Record of the dataset schema, and the failure is a chunk write failure like any other. The `NVARCHAR` sliver stays unpreflighted, because tiberius reports an over-long value as an error at `send()`. No exit condition is needed: should tiberius ever return an error instead of asserting, the check becomes redundant and stays correct.