Keep bind processing for non-plain import cells

Exact integer cells pass preflight but SQLite cannot bind integers beyond
int64, so only rows proven float/None/str bypass SQLAlchemy. Strings that
cannot be UTF-8 encoded fall back to the row-major checks.

Author: TonisOrmisson

src/openstatspec/sql/profiles.py

@@ -128,8 +128,12 @@ def preflight(
     *,
     rows: Iterable[Mapping[str, Any]] | None = None,
     require_canonical_mapping: bool = True,
-) -> None:
-    """Validate strict target capabilities before any source dataset is created."""
+) -> bool:
+    """Validate strict target capabilities before any source dataset is created.
+
+    Returns True only when every cell was proven to be float/None (numeric) or
+    str, so callers may bypass per-value bind processing.
+    """
     variables = None if isinstance(variables_or_count, int) else list(variables_or_count)
     variable_count = variables_or_count if isinstance(variables_or_count, int) else len(variables)
     if variable_count > profile.max_source_variables:
@@ -143,7 +147,7 @@ def preflight(
             max_physical=profile.max_source_variables + 1,
         )
     if variables is None:
-        return
+        return False
 
     used = {"__case_ordinal"}
     source_names: set[str] = set()
@@ -216,7 +220,7 @@ def preflight(
         )
     rows = list(rows or ())
     if not rows or _rows_within_limits(profile, variables, rows):
-        return
+        return True
     # Slow path only reports the first failing row in row-major order.
     for row_ordinal, row in enumerate(rows, start=1):
         row_bytes = 8
@@ -317,6 +321,7 @@ def preflight(
                 row_ordinal=row_ordinal, statement_bytes=statement_bytes,
                 maximum=profile.max_statement_bytes,
             )
+    return False
 
 
 def _rows_within_limits(
@@ -346,7 +351,10 @@ def _rows_within_limits(
             continue
         if set(map(type, column)) != {str}:
             return False
-        longest = max(map(len, map(str.encode, column)))
+        try:
+            longest = max(map(len, map(str.encode, column)))
+        except UnicodeEncodeError:
+            return False
         if longest > profile.max_text_value_bytes:
             return False
         row_bytes += 20 if profile.name in MYSQL_WIRE_PROFILES else longest

src/openstatspec/sql/wide.py

@@ -871,7 +871,7 @@ def create_wide_dataset(
     fidelity_events = tuple(fidelity_events)
     source_rows = _canonicalize_database_numeric_rows(rows, variables)
     try:
-        preflight(profile, variables, rows=source_rows)
+        plain_rows = preflight(profile, variables, rows=source_rows)
         validate_spss_catalog(
             variables,
             case_weight_variable=case_weight_variable,
@@ -982,14 +982,17 @@ def create_wide_dataset(
                 ),
             )
             if source_rows:
-                # Preflight proved cells are binary64/None or str, so the only
-                # bind processor (SQLite to_float) is a no-op; skip the per-row
-                # SQLAlchemy parameter pass and hand rows to the driver.
+                # Plain rows hold only float/None and str, so the only bind
+                # processor (SQLite to_float) is a no-op: hand them straight to
+                # the driver. Anything else (e.g. exact ints) keeps SQLAlchemy.
                 compiled = insert(data_table).compile(dialect=connection.dialect)
                 keys = compiled.positiontup
                 for batch in _bounded_batches(
                     source_rows, variables, profile.max_statement_bytes,
                 ):
+                    if not plain_rows:
+                        connection.execute(insert(data_table), batch)
+                        continue
                     connection.exec_driver_sql(compiled.string, (
                         [tuple(map(row.__getitem__, keys)) for row in batch]
                         if keys else batch

tests/test_sql_profiles.py

@@ -771,3 +771,27 @@ def test_column_preflight_fast_path_agrees_with_row_checks():
             preflight(MYSQL, variables, rows=[*plain, bad])
     too_long = [{"q": 1.0, "s": "x" * (MYSQL.max_text_value_bytes + 1)}]
     assert not _rows_within_limits(MYSQL, variables, too_long)
+    assert not _rows_within_limits(MYSQL, variables, [{"q": 1.0, "s": "\ud800"}])
+    # Exact ints are valid but not plain, so callers keep bind processing.
+    assert preflight(MYSQL, variables, rows=[*plain, {"q": 2**63, "s": ""}]) is False
+    with pytest.raises(UnsupportedOperationError, match="binary64"):
+        preflight(MYSQL, variables, rows=[*plain, {"q": 2**53 + 1, "s": ""}])
+
+
+def test_import_stores_exact_integer_cells_as_binary64(tmp_path):
+    from openstatspec.sql.wide import create_wide_dataset
+    import sqlite3
+
+    database = f"sqlite:///{tmp_path / 'ints.sqlite'}"
+    variables = [{
+        "source_name": "q", "physical_name": "q", "ordinal": 1,
+        "storage_kind": "numeric", "format": "F20.0",
+    }]
+    create_wide_dataset(
+        database_url=database, dataset_id="ints", source_name="ints.sav",
+        source_format="SAV", rows=[{"q": 2**63}, {"q": None}], variables=variables,
+    )
+    stored = sqlite3.connect(tmp_path / "ints.sqlite").execute(
+        "select q from data_ints order by __case_ordinal"
+    ).fetchall()
+    assert stored == [(float(2**63),), (None,)]