fix: keep SQLite loader and source timestamps as text and add a waffle shop E2E

src/sqlbuild/.agents/skills/sqlbuild/references/docs/concepts/adapters/sqlite.md

@@ -75,7 +75,9 @@ SQLite stores values by type affinity, so SQLBuild keeps column types explicitly
   its declared type; a computed column gets the type SQLBuild infers at compile time.
 - Timestamps and dates are ISO-8601 text. `CAST(x AS TIMESTAMP)` and `CAST(x AS DATE)` do not
   produce timestamps in SQLite (they keep only the leading number), so write timestamps as text
-  literals or use SQLite date functions.
+  literals or use SQLite date functions. SQLBuild never applies such a cast itself: source,
+  contract and type-change casts to a date or time type keep the stored text, and source loaders
+  write Python `datetime` values as UTC text in the form `YYYY-MM-DD HH:MM:SS[.ffffff]`.
 - Cursor windows, watermarks and window deletes compare timestamp cursors as UTC text in the
   form `YYYY-MM-DD HH:MM:SS[.ffffff]`, so values written with a `T` separator, fractional seconds
   or a UTC offset fall in the right window. A value with an offset belongs to the window of its UTC

src/sqlbuild/adapters/sqlite/_helpers/tables.py

@@ -15,6 +15,7 @@
     r"(\"(?:[^\"]|\"\")+\"|[^\s(]+)\s+AS\s+(.+)$",
     re.IGNORECASE | re.DOTALL,
 )
+_UNTYPED_VIEW_COLUMN_TYPE: str = "BLOB"
 _DECIMAL_TYPE: re.Pattern[str] = re.compile(EXACT_DECIMAL_TYPE_PATTERN, re.IGNORECASE)
 
 
@@ -49,7 +50,7 @@ def create_table_typed(
         column_sql: str = ", ".join(
             _column_definition(
                 name=str(row[1]),
-                declared=str(row[2] or "") or inferred_types.get(str(row[1]).lower(), ""),
+                declared=_probed_type(row[2]) or inferred_types.get(str(row[1]).lower(), ""),
                 column_types=column_types,
             )
             for row in rows
@@ -83,6 +84,13 @@ def decimal_scale(target_type: str) -> int | None:
     return int(match.group(2) or 0)
 
 
+def _probed_type(declared: object) -> str:
+    """A view reports an untyped source column as BLOB; keep such columns untyped."""
+
+    text: str = str(declared or "")
+    return "" if text.upper() == _UNTYPED_VIEW_COLUMN_TYPE else text
+
+
 def _column_definition(*, name: str, declared: str, column_types: Mapping[str, str]) -> str:
     override: str | None = column_types.get(name.lower())
     column_type: str = storage_type(override) if override is not None else declared

src/sqlbuild/adapters/sqlite/classes/sqlite_adapter.py

@@ -143,6 +143,7 @@
     MISSING_RELATION_ERROR_FRAGMENT,
     SCHEMA_REGISTRY_TABLE,
     SQLITE_TABLE_TYPE,
+    TEMPORAL_TYPE_PATTERN,
     UNSUPPORTED_FEATURE_CODE,
     VERSION_TOO_OLD_CODE,
 )
@@ -156,6 +157,7 @@
 _INFERRED_COLUMN_TYPES: ContextVar[Mapping[str, str]] = ContextVar(
     "sqlite_inferred_column_types", default=MappingProxyType({})
 )
+_TEMPORAL_TYPE: re.Pattern[str] = re.compile(TEMPORAL_TYPE_PATTERN, re.IGNORECASE)
 _TRUE_SEED_VALUES: frozenset[str] = frozenset({"true", "t", "yes", "y", "1"})
 _FALSE_SEED_VALUES: frozenset[str] = frozenset({"false", "f", "no", "n", "0"})
 
@@ -2237,7 +2239,9 @@ def render_loader_value_literal(
             return "1" if value else "0"
         if isinstance(value, int | float | Decimal):
             return str(value)
-        if isinstance(value, datetime | date):
+        if isinstance(value, datetime):
+            return _quote_sql_string(canonical_timestamp_text(value.isoformat()))
+        if isinstance(value, date):
             return _quote_sql_string(value.isoformat())
         return _quote_sql_string(str(value))
 
@@ -2249,13 +2253,12 @@ def render_loader_rows_select(
         column_sql_types: dict[str, str],
         inferred_types: dict[str, LoaderLogicalType],
     ) -> str:
-        """Render rows as one multi-row VALUES; SQLite names VALUES columns column1, column2."""
+        """Render uncast rows as one VALUES; staging declares the types and affinity converts."""
 
+        del column_sql_types
         if not rows:
             projections: str = ", ".join(
-                f"CAST(NULL AS {column_sql_types.get(column_name, 'TEXT')}) AS "
-                f"{self.render_identifier(column_name)}"
-                for column_name in column_names
+                f"NULL AS {self.render_identifier(column_name)}" for column_name in column_names
             )
             return f"SELECT {projections} WHERE 1 = 0"
         value_rows: list[str] = []
@@ -2270,21 +2273,18 @@ def render_loader_rows_select(
             value_rows.append(f"({literals})")
         values_sql: str = ", ".join(value_rows)
         select_sql: str = ", ".join(
-            (
-                f"column{index} AS {self.render_identifier(column_name)}"
-                if column_name not in column_sql_types
-                else f"CAST(column{index} AS {column_sql_types[column_name]}) "
-                f"AS {self.render_identifier(column_name)}"
-            )
+            f"column{index} AS {self.render_identifier(column_name)}"
             for index, column_name in enumerate(column_names, start=1)
         )
         return f"SELECT {select_sql} FROM (VALUES {values_sql}) AS __loader_rows"
 
     def render_source_expression_cast(
         self, *, expression: str, target_type: str, alias: str
     ) -> str:
-        """Cast, rendering exact decimals as canonical fixed-scale text."""
+        """Cast, keeping date and time text and rendering exact decimals as fixed-scale text."""
 
+        if _TEMPORAL_TYPE.match(target_type) is not None:
+            return f"{expression} AS {alias}"
         scale: int | None = decimal_scale(target_type)
         if scale is None:
             return f"CAST({expression} AS {target_type}) AS {alias}"

src/sqlbuild/adapters/sqlite/constants.py

@@ -30,6 +30,7 @@
 NUMBER_DECIMAL_TYPE_PATTERN: str = r"^\s*(?:DECIMAL|NUMERIC)\b"
 NUMBER_DECIMAL_REPORTED_TYPE: str = "NUMERIC"
 BOOLEAN_TYPE_MARKER: str = "BOOL"
+TEMPORAL_TYPE_PATTERN: str = r"^\s*(?:DATE|TIME)"
 INSERT_KEYWORD: str = "INSERT"
 SQLITE_VIEW_TYPE: str = "view"
 SQLITE_TABLE_TYPE: str = "table"

src/sqlbuild/executor/load/_helpers/rows.py

@@ -136,29 +136,45 @@ def build_rows_sql(
     )
     if not resolved_column_names:
         raise ExecutorInputError("Source loader returned no rows and source declares no columns")
-    declared_types: dict[str, str] = _declared_types(columns=columns)
     resolved_inferred_types: dict[str, LoaderLogicalType] = inferred_types or _infer_column_types(
         rows=rows
     )
-    column_sql_types: dict[str, str] = {
+    return adapter.render_loader_rows_select(
+        rows=rows,
+        column_names=resolved_column_names,
+        column_sql_types=loader_column_sql_types(
+            adapter=adapter,
+            columns=columns,
+            column_names=resolved_column_names,
+            inferred_types=resolved_inferred_types,
+        ),
+        inferred_types=resolved_inferred_types,
+    )
+
+
+def loader_column_sql_types(
+    *,
+    adapter: BaseAdapter,
+    columns: tuple[SourceColumnEntry, ...],
+    column_names: tuple[str, ...],
+    inferred_types: dict[str, LoaderLogicalType],
+) -> dict[str, str]:
+    """Return each loader column's declared or inferred warehouse type, when known."""
+
+    declared_types: dict[str, str] = _declared_types(columns=columns)
+    return {
         column_name: sql_type
-        for column_name in resolved_column_names
+        for column_name in column_names
         if (
             sql_type := _column_sql_type(
                 adapter=adapter,
                 column_name=column_name,
                 declared_types=declared_types,
-                inferred_types=resolved_inferred_types,
+                inferred_types=inferred_types,
             )
         )
         is not None
     }
-    return adapter.render_loader_rows_select(
-        rows=rows,
-        column_names=resolved_column_names,
-        column_sql_types=column_sql_types,
-        inferred_types=resolved_inferred_types,
-    )
 
 
 def _declared_types(*, columns: tuple[SourceColumnEntry, ...]) -> dict[str, str]:

src/sqlbuild/executor/load/_helpers/staging.py

@@ -14,6 +14,7 @@
 from sqlbuild.executor.load._helpers.rows import (
     build_rows_sql,
     iter_loader_row_batches,
+    loader_column_sql_types,
     update_loader_rows_schema,
 )
 from sqlbuild.executor.load.models import LoaderRowsSchema
@@ -83,6 +84,12 @@ def write_loader_rows_to_staging(
                 connection=connection,
                 destination=staging,
                 sql=sql,
+                config=_staging_config(
+                    adapter=adapter,
+                    source_entry=source_entry,
+                    column_names=column_names,
+                    inferred_types=inferred_types,
+                ),
                 statement_recorder=statement_recorder,
             )
             staging_created = True
@@ -99,11 +106,36 @@ def write_loader_rows_to_staging(
             connection=connection,
             destination=staging,
             sql=sql,
+            config=_staging_config(
+                adapter=adapter,
+                source_entry=source_entry,
+                column_names=column_names,
+                inferred_types=inferred_types,
+            ),
             statement_recorder=statement_recorder,
         )
     return rows_loaded
 
 
+def _staging_config(
+    *,
+    adapter: BaseAdapter,
+    source_entry: SourceEntry,
+    column_names: tuple[str, ...],
+    inferred_types: dict[str, LoaderLogicalType],
+) -> dict[str, Any]:
+    """Hand loader column types to adapters whose CTAS cannot keep them."""
+
+    return {
+        "inferred_column_types": loader_column_sql_types(
+            adapter=adapter,
+            columns=source_entry.columns,
+            column_names=column_names,
+            inferred_types=inferred_types,
+        )
+    }
+
+
 def resolve_loader_relations(
     *, adapter: BaseAdapter, source_entry: SourceEntry, destination_name: str
 ) -> tuple[str, str]:

tests/e2e/src/sqlbuild/cli/commands/main/sqlite/_test_types.py

@@ -85,3 +85,21 @@ class SqliteCursorWindowE2ETestCase:
     events_csv: str
     commands: tuple[tuple[str, ...], ...]
     expected_order_ids: tuple[tuple[object, ...], ...]
+
+
+@dataclass(frozen=True)
+class SqliteWaffleShopE2ETestCase:
+    description: str
+    commands: tuple[tuple[str, ...], ...]
+    expected_fragments: tuple[str, ...]
+    expected_row_counts: dict[str, int]
+    expected_return_code: int = 0
+
+
+@dataclass(frozen=True)
+class SqliteLoaderE2ETestCase:
+    description: str
+    builds: int
+    relations: tuple[str, ...]
+    expected_rows: tuple[tuple[object, ...], ...]
+    expected_declared_types: tuple[tuple[object, ...], ...]