Land the sqlserver connector: session and full-refresh loads

[victorchutw]

Implements ADR-0064 (full refresh), the connection facts of ADR-0060, and ADR-0068 (parallelism declaration); first live consumer of the #134 mapping and #135 Accept Family modules; test posture per ADR-0066. Part of the SQL Server destination cycle (map #106, sliced from #113).

Blocked by #131, #132, #135

Context

The first live mode. The connector bridges the sync Destination/DestinationWriter seam (src/connector.rs:290, :356) to async tiberius with a connector-contained current-thread tokio runtime — one runtime and one client per load, nothing async escaping the connector (the #107 research posture). Full refresh is the only mode that bootstraps a table; production tables are owned by their environment, so the write never replaces the table object.

The contract

• Session: SQL auth; encryption: required → tiberius default encryption, optional → EncryptionLevel::Off (login encrypted where possible, data traffic per server policy — the production SSMS Encrypt=Optional parity posture); trust_server_certificate honored; a hardcoded 15 s connect timeout so a hang becomes a clean failure; application_name pinned to data-spark. The password is read from the password_env variable at connect time (its presence was validated offline in the Definition phase).
• begin(full_refresh): table absent → CREATE TABLE in the created shape (Generate SQL Server DDL and bulk rows from the dataset schema #134); table present → introspect and validate against the Accept Family (Validate existing SQL Server tables against the Accept Family #135, full-refresh rules; violations fail incompatible_destination_table before any write). Then BEGIN TRAN and DELETE FROM [schema].[table] — object-preserving, never drop/rename (a staged sp_rename swap is rejected by ADR-0064: it strips permissions, indexes, triggers, FKs).
• write_chunk: one bulk_insert per chunk inside the transaction (bulk load inside a user-defined transaction is documented TDS behavior and rolls back with it), rows in table column order via Generate SQL Server DDL and bulk rows from the dataset schema #134.
• commit: terminal COMMIT. Write facts: atomicity: "atomic", strategy: "transactional_delete_insert" (new string, additive under the report contract's version-1 clause). A zero-survivor full refresh commits an empty table — ADR-0056's mirror semantics.
• Failure: explicit best-effort ROLLBACK; failures surface as destination_write_failed (or incompatible_destination_table from validation). Every raising site classifies Transience::Terminal in this slice — the transience slice flips the decided sites later (ADR-0067's terminal-first-then-flip path).
• Modes: supported_load_modes = full refresh only; append and merge keep declining via unsupported_load_mode_for_destination until their slices widen the list.
• Parallelism: parallelism_limit declared 1 for all three modes — declared, not defaulted (ADR-0068); the report keeps stating the effective value beside the connector limit.
• Tests (ADR-0066): pure logic ungated; live tests carry #[ignore = "needs SQL Server"], connect with fixed defaults matching the CI container of Run CI and release tests against a live SQL Server with --include-ignored #131, overridable via the DATA_SPARK_TEST_MSSQL_* env family (host, port, user, password), and never start containers themselves. CI proves them via --include-ignored.

Acceptance criteria

1. Gated live tests:
   • Bootstrap: full refresh onto an absent table creates the created shape (introspection read-back asserts exact types — BIGINT, FLOAT, BIT, NVARCHAR(MAX), DATETIME2(6), DECIMAL(p,s) — and nullability) and lands the rows; values round-trip exactly (instants UTC-normalized, microsecond precision, i64::MAX, negative decimals, empty string ≠ NULL).
   • Object preservation: an index created out-of-band on the table survives a second full refresh; rows are replaced.
   • Zero survivors: an all-rejected source under a permissive threshold commits an empty table with atomic/transactional_delete_insert facts.
   • Incompatibility: an existing table with a VARCHAR column fails incompatible_destination_table at begin; the target's rows are untouched.
   • Representability sliver: a >32,767-character utf8 value fails the load as a write failure (never a Rejected Record), the transaction rolls back, and the destination is unchanged.
2. Ungated units: config→client posture mapping (encryption/trust matrix), error-shape mapping.
3. The load report carries the destination_summary echo, the atomic facts, and connector_parallelism_limit: 1; exit codes follow the existing contract.
4. On a serverless machine cargo test --locked passes with the live tests visibly ignored; CI (with the Run CI and release tests against a live SQL Server with --include-ignored #131 container) runs them for real.
5. CHANGELOG.md Unreleased entry: the sqlserver destination lands with full-refresh support, naming incompatible_destination_table in prose.

[victorchutw]

Hand-off from #134 (PR #158 → b361d6d), facts the live connector should build on:

• Consume the module and drop the attribute. src/sqlserver.rs provides create_table_ddl(dataset, schema, table), BulkRowPlan::new(dataset, table_columns), BulkRowPlan::rows(&RecordBatch), and quote_identifier. lib.rs registers the module under #[cfg_attr(not(test), allow(dead_code))]; remove that with the first real consumer.
• Row lifetimes. rows() yields Result<TokenRow<'a>, LoadFailure> borrowing the chunk, and tiberius's BulkLoadRequest::send takes TokenRow<'a> with the request's own 'a, so the chunk must outlive the bulk request (or the rows must be converted to owned Cows). A row that fails to encode ends the chunk's rows: treat it as the chunk's write failure.
• Column order. For a table this slice creates, table_columns is the dataset field order (create_table_ddl emits fields in that order). For an existing table, pass the introspected ordinal order from Validate existing SQL Server tables against the Accept Family #135 — same-typed columns in the wrong order land transposed with no error (Spike: validate Tiberius against a containerized SQL Server #115).
• Representability slivers arrive two ways. A timestamp outside 0001-01-01..=9999-12-31 comes back from the encoder as destination_write_failed (ADR-0062's dated "Range check activated" note explains why the encoder, not the server, reports it). An NVARCHAR value over 32,767 characters reaches tiberius and returns Err(BulkInput(...)) at send(). Both are chunk write failures with rollback, never Rejected Records.
• Accept Family before bulk_insert, always. The fork's bulk Numeric arm todo!()-panics on a row/column scale mismatch (details on Validate existing SQL Server tables against the Accept Family #135), so an existing table must pass ADR-0065 validation before the first row is sent.
• DATETIME2 encoding. Values are built directly at scale 6, so a DATETIME2(6) column takes them with no rescale; for an accepted DATETIME2(7) column tiberius rescales ×10 through f64, which is exact at these magnitudes.