feat: add opt-in ASCII varchar string policy

docs/type-mapping.md

@@ -67,6 +67,7 @@ runtime checks.
 | `UInt64` | `UInt64Policy::CheckedBigInt` | `bigint` | runtime check | runtime check | Values greater than `i64::MAX` are rejected. |
 | `Utf8`, `LargeUtf8` | `StringPolicy::NVarCharMax` | `nvarchar(max)` | yes | yes | Default. |
 | `Utf8`, `LargeUtf8` | `StringPolicy::NVarChar(n)` | `nvarchar(n)` | runtime check | runtime check | Runtime rejects values whose UTF-16 length exceeds `n`. |
+| `Utf8`, `LargeUtf8`, `Utf8View` | `StringPolicy::AsciiVarChar(n)` | `varchar(n)` | runtime check | runtime check | `n` must be in `1..=8000`; runtime rejects non-ASCII values and values longer than `n` bytes. |
 | `Utf8`, `LargeUtf8` | `StringPolicy::ObservedNVarChar` | inferred `nvarchar(n)` | schema-only reject | schema-only reject | Requires observed values or statistics; schema-only planning currently rejects it. |
 | `Binary`, `LargeBinary` | `BinaryPolicy::VarBinaryMax` | `varbinary(max)` | yes | yes | Default. |
 | `Binary`, `LargeBinary` | `BinaryPolicy::VarBinary(n)` | `varbinary(n)` | runtime check | runtime check | Runtime rejects values whose byte length exceeds `n`. |

src/conversion/arrow_to_mssql/variable_width.rs

@@ -13,6 +13,8 @@ use crate::{
 pub(crate) enum VariableWidthArrowToMssql {
     /// Arrow string family to SQL Server `nvarchar(n|max)`.
     StringToNVarChar { length: MssqlTypeLength },
+    /// Arrow string family to SQL Server `varchar(n)` with ASCII-only values.
+    StringToAsciiVarChar { length: MssqlTypeLength },
     /// Arrow binary family to SQL Server `varbinary(n|max)`.
     BytesToVarBinary { length: MssqlTypeLength },
 }
@@ -24,6 +26,9 @@ impl VariableWidthArrowToMssql {
             (data_type, MssqlType::NVarChar(length)) if is_arrow_string_family(data_type) => {
                 Self::StringToNVarChar { length: *length }
             }
+            (data_type, MssqlType::VarChar(length)) if is_arrow_string_family(data_type) => {
+                Self::StringToAsciiVarChar { length: *length }
+            }
             (data_type, MssqlType::VarBinary(length)) if is_arrow_binary_family(data_type) => {
                 Self::BytesToVarBinary { length: *length }
             }
@@ -45,10 +50,13 @@ impl VariableWidthArrowToMssql {
     }
 }
 
-/// Returns true when a planned mapping writes Arrow string-family values to `nvarchar`.
-pub(crate) fn is_string_family_to_nvarchar(mapping: &SchemaMapping) -> bool {
+/// Returns true when a planned mapping writes Arrow string-family values to a SQL text type.
+pub(crate) fn is_string_family_to_sql_text(mapping: &SchemaMapping) -> bool {
     is_arrow_string_family(mapping.arrow().data_type())
-        && matches!(mapping.mssql().ty(), MssqlType::NVarChar(_))
+        && matches!(
+            mapping.mssql().ty(),
+            MssqlType::NVarChar(_) | MssqlType::VarChar(_)
+        )
 }
 
 /// Returns true when a planned mapping writes Arrow binary-family values to `varbinary`.
@@ -63,7 +71,7 @@ pub(crate) fn arrow_type_compatible_with_mapping(
     mapping: &SchemaMapping,
 ) -> bool {
     runtime == mapping.arrow().data_type()
-        || (is_arrow_string_family(runtime) && is_string_family_to_nvarchar(mapping))
+        || (is_arrow_string_family(runtime) && is_string_family_to_sql_text(mapping))
         || (is_arrow_binary_family(runtime) && is_binary_family_to_varbinary(mapping))
 }
 
@@ -115,6 +123,13 @@ mod tests {
                     length: MssqlTypeLength::Bounded(32),
                 },
             ),
+            (
+                DataType::Utf8,
+                MssqlType::VarChar(MssqlTypeLength::Bounded(32)),
+                VariableWidthArrowToMssql::StringToAsciiVarChar {
+                    length: MssqlTypeLength::Bounded(32),
+                },
+            ),
             (
                 DataType::LargeUtf8,
                 MssqlType::NVarChar(MssqlTypeLength::Max),
@@ -202,17 +217,17 @@ mod tests {
     }
 
     #[test]
-    fn accepts_string_family_runtime_types_for_nvarchar_mappings() {
-        for planned in [DataType::Utf8, DataType::LargeUtf8, DataType::Utf8View] {
-            let mapping = mapping(
-                0,
-                "text",
-                planned,
-                MssqlType::NVarChar(MssqlTypeLength::Max),
-            );
+    fn accepts_string_family_runtime_types_for_sql_text_mappings() {
+        for target in [
+            MssqlType::NVarChar(MssqlTypeLength::Max),
+            MssqlType::VarChar(MssqlTypeLength::Bounded(32)),
+        ] {
+            for planned in [DataType::Utf8, DataType::LargeUtf8, DataType::Utf8View] {
+                let mapping = mapping(0, "text", planned, target.clone());
 
-            for runtime in [DataType::Utf8, DataType::LargeUtf8, DataType::Utf8View] {
-                assert!(arrow_type_compatible_with_mapping(&runtime, &mapping));
+                for runtime in [DataType::Utf8, DataType::LargeUtf8, DataType::Utf8View] {
+                    assert!(arrow_type_compatible_with_mapping(&runtime, &mapping));
+                }
             }
         }
     }

src/mssql/cell.rs

@@ -34,6 +34,8 @@ pub(crate) enum MssqlCell<'a> {
     Float(Option<f64>),
     /// SQL Server `nvarchar` cell.
     NVarChar(Option<&'a str>),
+    /// SQL Server `varchar` cell containing only ASCII characters.
+    VarChar(Option<&'a str>),
     /// SQL Server `varbinary` cell.
     VarBinary(Option<&'a [u8]>),
 }

src/mssql/cell/from_arrow.rs

@@ -21,7 +21,7 @@ use temporal::{
     mssql_time_value, null_datetime_cell, null_datetime2_cell, null_datetimeoffset_cell,
     null_time_cell,
 };
-use variable_width::{binary_cell, nvar_char_cell, var_binary_cell};
+use variable_width::{ascii_var_char_cell, binary_cell, nvar_char_cell, var_binary_cell};
 
 /// Direction-specific runtime context for Arrow-to-MSSQL value conversion.
 #[derive(Debug, Clone, Copy)]
@@ -121,6 +121,7 @@ pub(crate) fn mssql_cell_from_arrow_cell<'a>(
             mssql_datetimeoffset_value(runtime_mapping, row_index, cell)?,
         ))),
         MssqlType::NVarChar(length) => nvar_char_cell(mapping, row_index, *length, cell),
+        MssqlType::VarChar(length) => ascii_var_char_cell(mapping, row_index, *length, cell),
         MssqlType::VarBinary(length) => var_binary_cell(mapping, row_index, *length, cell),
         MssqlType::Binary(length) => binary_cell(mapping, row_index, *length, cell),
     }
@@ -144,6 +145,7 @@ fn null_mssql_cell<'a>(mapping: &SchemaMapping, row_index: usize) -> Result<Mssq
         MssqlType::Real => Ok(MssqlCell::Real(None)),
         MssqlType::Float { .. } => Ok(MssqlCell::Float(None)),
         MssqlType::NVarChar(_) => Ok(MssqlCell::NVarChar(None)),
+        MssqlType::VarChar(_) => Ok(MssqlCell::VarChar(None)),
         MssqlType::VarBinary(_) => Ok(MssqlCell::VarBinary(None)),
         MssqlType::Binary(_) => Ok(MssqlCell::VarBinary(None)),
         ty => Err(unsupported_value_conversion(

src/mssql/cell/from_arrow/variable_width.rs

@@ -30,7 +30,7 @@ pub(super) fn nvar_char_cell<'a>(
     };
     debug_assert_eq!(length, classified);
 
-    let value = mssql_nvarchar_value(mapping, row_index, cell)?;
+    let value = mssql_string_value(mapping, row_index, cell)?;
     let code_units = value.encode_utf16().count();
 
     if exceeds_length(length, code_units) {
@@ -47,6 +47,55 @@ pub(super) fn nvar_char_cell<'a>(
     Ok(MssqlCell::NVarChar(Some(value)))
 }
 
+pub(super) fn ascii_var_char_cell<'a>(
+    mapping: &SchemaMapping,
+    row_index: usize,
+    length: MssqlTypeLength,
+    cell: ArrowCell<'a>,
+) -> Result<MssqlCell<'a>> {
+    let classified = match VariableWidthArrowToMssql::classify(mapping, row_index)? {
+        VariableWidthArrowToMssql::StringToAsciiVarChar { length } => length,
+        other => {
+            return Err(value_conversion_error(row_mapping_diagnostic(
+                mapping,
+                row_index,
+                DiagnosticCode::ValueConversionUnsupported,
+                format!(
+                    "variable-width mapping {other:?} is not supported by ASCII varchar conversion"
+                ),
+            )));
+        }
+    };
+    debug_assert_eq!(length, classified);
+
+    let value = mssql_string_value(mapping, row_index, cell)?;
+    if !value.is_ascii() {
+        return Err(value_conversion_error(row_mapping_diagnostic(
+            mapping,
+            row_index,
+            DiagnosticCode::ValueConversionUnsupported,
+            format!(
+                "string value contains non-ASCII characters and cannot be written as planned {}",
+                mapping.mssql().ty().to_sql()
+            ),
+        )));
+    }
+
+    if exceeds_length(length, value.len()) {
+        return Err(value_too_long_error(
+            mapping,
+            row_index,
+            format!(
+                "ASCII string value has {} byte(s), exceeding planned {}",
+                value.len(),
+                mapping.mssql().ty().to_sql()
+            ),
+        ));
+    }
+
+    Ok(MssqlCell::VarChar(Some(value)))
+}
+
 pub(super) fn var_binary_cell<'a>(
     mapping: &SchemaMapping,
     row_index: usize,
@@ -114,7 +163,7 @@ pub(super) fn binary_cell<'a>(
     Ok(MssqlCell::VarBinary(Some(value)))
 }
 
-fn mssql_nvarchar_value<'a>(
+fn mssql_string_value<'a>(
     mapping: &SchemaMapping,
     row_index: usize,
     cell: ArrowCell<'a>,
@@ -269,6 +318,30 @@ mod tests {
         );
     }
 
+    #[test]
+    fn accepts_bounded_ascii_varchar_and_rejects_invalid_values() {
+        let mappings = mappings_for_schema_with_options(
+            Schema::new(vec![Field::new("text", DataType::Utf8, true)]),
+            PlanOptions {
+                string_policy: StringPolicy::AsciiVarChar(2),
+                ..PlanOptions::default()
+            },
+        );
+
+        assert_eq!(
+            convert_cell(&mappings[0], ArrowCell::Utf8("ab"), 0).unwrap(),
+            MssqlCell::VarChar(Some("ab"))
+        );
+
+        for (row_index, value, code) in [
+            (1, "abc", DiagnosticCode::ValueTooLong),
+            (2, "e\u{301}", DiagnosticCode::ValueConversionUnsupported),
+        ] {
+            let err = convert_cell(&mappings[0], ArrowCell::Utf8(value), row_index).unwrap_err();
+            assert_single_diagnostic(err, code, Some(row_index), Some((0, "text")));
+        }
+    }
+
     #[test]
     fn rejects_bounded_varbinary_by_byte_count() {
         let mappings = mappings_for_schema_with_options(

src/mssql/ty.rs

@@ -70,6 +70,8 @@ pub enum MssqlType {
     },
     /// SQL Server `nvarchar(n|max)`.
     NVarChar(MssqlTypeLength),
+    /// SQL Server `varchar(n|max)` with ASCII-only Arrow string conversion.
+    VarChar(MssqlTypeLength),
     /// SQL Server `varbinary(n|max)`.
     VarBinary(MssqlTypeLength),
     /// SQL Server `binary(n)`.
@@ -111,6 +113,7 @@ impl MssqlType {
             Self::Real => "real".to_owned(),
             Self::Float { precision } => format!("float({precision})"),
             Self::NVarChar(length) => format!("nvarchar({})", length.render()),
+            Self::VarChar(length) => format!("varchar({})", length.render()),
             Self::VarBinary(length) => format!("varbinary({})", length.render()),
             Self::Binary(length) => format!("binary({length})"),
             Self::Decimal { precision, scale } => format!("decimal({precision},{scale})"),
@@ -148,6 +151,10 @@ mod tests {
             MssqlType::NVarChar(MssqlTypeLength::Bounded(128)).to_sql(),
             "nvarchar(128)"
         );
+        assert_eq!(
+            MssqlType::VarChar(MssqlTypeLength::Bounded(128)).to_sql(),
+            "varchar(128)"
+        );
         assert_eq!(
             MssqlType::VarBinary(MssqlTypeLength::Max).to_sql(),
             "varbinary(max)"

src/observability/schema.rs

@@ -221,6 +221,7 @@ fn mssql_type_family(ty: &MssqlType) -> &'static str {
         MssqlType::Real => "real",
         MssqlType::Float { .. } => "float",
         MssqlType::NVarChar(_) => "nvarchar",
+        MssqlType::VarChar(_) => "varchar",
         MssqlType::VarBinary(_) => "varbinary",
         MssqlType::Binary(_) => "binary",
         MssqlType::Decimal { .. } => "decimal",

src/schema/type_conversion.rs

@@ -96,6 +96,18 @@ fn plan_arrow_string_as_mssql_type(
     match policy {
         StringPolicy::NVarCharMax => Ok(MssqlType::NVarChar(MssqlTypeLength::Max)),
         StringPolicy::NVarChar(length) => Ok(MssqlType::NVarChar(MssqlTypeLength::Bounded(length))),
+        StringPolicy::AsciiVarChar(length)
+            if (1..=SQL_SERVER_MAX_VARCHAR_LEN).contains(&length) =>
+        {
+            Ok(MssqlType::VarChar(MssqlTypeLength::Bounded(length)))
+        }
+        StringPolicy::AsciiVarChar(length) => Err(unsupported_arrow_mapping_for_arrow_to_mssql(
+            index,
+            field,
+            format!(
+                "ASCII varchar length {length} is outside SQL Server varchar(n) range 1..={SQL_SERVER_MAX_VARCHAR_LEN}"
+            ),
+        )),
         StringPolicy::ObservedNVarChar => Err(observed_data_required_for_arrow_to_mssql(
             index,
             field,
@@ -128,15 +140,15 @@ fn plan_arrow_fixed_size_binary_as_mssql_type(
     field: &Field,
 ) -> std::result::Result<MssqlType, Diagnostic> {
     let Ok(length) = usize::try_from(length) else {
-        return Err(fixed_size_binary_out_of_range_for_arrow_to_mssql(
+        return Err(unsupported_arrow_mapping_for_arrow_to_mssql(
             index,
             field,
             "fixed-size binary length must be non-negative",
         ));
     };
 
     if !(1..=SQL_SERVER_MAX_BINARY_LEN).contains(&length) {
-        return Err(fixed_size_binary_out_of_range_for_arrow_to_mssql(
+        return Err(unsupported_arrow_mapping_for_arrow_to_mssql(
             index,
             field,
             format!(
@@ -330,7 +342,7 @@ fn decimal_out_of_range_for_arrow_to_mssql(
         .with_field(FieldRef::new(index, field.name()))
 }
 
-fn fixed_size_binary_out_of_range_for_arrow_to_mssql(
+fn unsupported_arrow_mapping_for_arrow_to_mssql(
     index: usize,
     field: &Field,
     message: impl Into<String>,
@@ -375,6 +387,7 @@ fn unsupported_arrow_type_family(data_type: &DataType) -> &'static str {
 
 const SQL_SERVER_MAX_DECIMAL_PRECISION: u8 = 38;
 const SQL_SERVER_MAX_BINARY_LEN: usize = 8000;
+const SQL_SERVER_MAX_VARCHAR_LEN: usize = 8000;
 
 #[cfg(test)]
 mod tests {
@@ -456,6 +469,17 @@ mod tests {
             .unwrap(),
             MssqlType::NVarChar(MssqlTypeLength::Bounded(128))
         );
+        assert_eq!(
+            plan_type(
+                DataType::Utf8,
+                PlanOptions {
+                    string_policy: StringPolicy::AsciiVarChar(128),
+                    ..PlanOptions::default()
+                },
+            )
+            .unwrap(),
+            MssqlType::VarChar(MssqlTypeLength::Bounded(128))
+        );
         assert_eq!(
             plan_type(
                 DataType::Binary,
@@ -480,6 +504,23 @@ mod tests {
         );
     }
 
+    #[test]
+    fn rejects_ascii_varchar_lengths_outside_sql_server_range() {
+        for length in [0, 8001] {
+            let diagnostic = plan_type(
+                DataType::Utf8,
+                PlanOptions {
+                    string_policy: StringPolicy::AsciiVarChar(length),
+                    ..PlanOptions::default()
+                },
+            )
+            .unwrap_err();
+
+            assert_eq!(diagnostic.code(), DiagnosticCode::UnsupportedArrowType);
+            assert!(diagnostic.message().contains("range 1..=8000"));
+        }
+    }
+
     #[test]
     fn maps_uint64_when_explicit_policy_is_selected() {
         assert_eq!(