A statement the warehouse refuses is classified, names its object, and reports what it spent; profiling a non-PII string column no longer kills the statement on Redshift

CHANGELOG.md

@@ -11,6 +11,44 @@ tag releases both in lockstep, so entries below are keyed by the engine version.
 
 ### Fixed
 
+- **Profiling a non-PII string column no longer kills the statement on
+  Redshift** ([#310]). The declared-type-vs-content probe (#204) guarded every
+  `CAST` behind a length-bounded shape predicate inside a `CASE`, on the
+  premise that a dialect without `TRY_CAST` would then never evaluate the cast
+  for a row the `WHEN` excluded. Postgres honors that premise; Redshift does
+  not, and evaluates the branch for rows it never selects, so one ordinary
+  varchar status or category column was enough to fail the whole profiling
+  statement with `Invalid digit, Value 'p', Pos 0, Type: Long`. The probe
+  shipped in 1.6.2, so on Redshift that took down `explore profile`,
+  `explore relationships`, and `explore map`, plus the commands that profile
+  the object they name before running (`explore query`, `explore cluster`).
+
+  Every `CAST` the probe builds is now total: the argument is a `CASE` that
+  yields a digit-only string on every row of the column, so no evaluation
+  order can reach a cast with something it cannot parse, and the sentinel it
+  falls back to is rejected by every predicate built on the cast. The
+  measurements are unchanged, denominators included; the "shaped versus not
+  shaped" distinction the fractions need now comes from a separate uncast
+  expression instead of from the cast result being NULL. The fix is
+  dialect-agnostic and lands in the shared expression builder, so the standing
+  assumption about lazy `CASE` evaluation is gone from all six adapters, and an
+  offline invariant test asserts the shape on every one of them.
+
+- **A statement the warehouse refuses is classified, names its object, and
+  reports what it spent** ([#310]). A server-side SQL error escaped the
+  adapters untranslated. Not being a `DexError`, it fell through every reason
+  override and arrived as `reason: internal` ("not a deliberate dex refusal")
+  with `data: {}`: nothing to branch on, no object named, and no spend, on a
+  connector that had already billed the seconds the statement ran before it
+  died. Every adapter now raises a typed `WarehouseQueryError` (exported from
+  the package root, `reason: execution_failure`) carrying the server's own
+  message and error code, trimmed to one line and capped; profiling names the
+  object the refused statement was reading; and an error envelope from a
+  metered connector reports the spend the ledger recorded, as the
+  budget-exhaustion path already did. The live suites now assert with a helper
+  that prints `errors` rather than an envelope repr pytest truncates, which is
+  what kept the Redshift message out of sixteen CI logs.
+
 - **`maintain check` carries each axis's findings in the command envelope**
   ([#279]). The top-level `data.findings` ranking could report drift while the
   adjacent per-axis result did not carry those findings, making an axis look

packages/dex-core/src/exmergo_dex_core/__init__.py

@@ -109,6 +109,7 @@
     "StoreContext": "storage",
     "StoreFactory": "storage",
     "StoreRequiredError": "errors",
+    "WarehouseQueryError": "errors",
     "generate_demo_warehouse": "demo",
     "render_er_mermaid": "explore.diagram",
     "to_envelope": "results",
@@ -143,6 +144,7 @@
         RepoRootRequiredError,
         RequestError,
         StoreRequiredError,
+        WarehouseQueryError,
     )
     from .explore.cluster import ClusterDependencyError, ClusterError
     from .explore.commands import CacheRequiredError
@@ -249,6 +251,7 @@
     "StoreContext",
     "StoreFactory",
     "StoreRequiredError",
+    "WarehouseQueryError",
     "__version__",
     "generate_demo_warehouse",
     "render_er_mermaid",

packages/dex-core/src/exmergo_dex_core/adapters/base.py

@@ -27,6 +27,7 @@
 from typing import Protocol, runtime_checkable
 
 from ..envelope import Paradigm
+from ..errors import WarehouseQueryError
 
 
 @dataclass(frozen=True)
@@ -242,6 +243,30 @@ def blame(origin: str, error: type[Exception]):
         raise error(f"{exc} [from {origin}]") from exc
 
 
+# How much of a driver's error text survives into the envelope. Generous
+# enough for any real server message, short enough that a driver which appends
+# the whole statement (or a stack of context lines) cannot turn one refusal
+# into a wall of stdout.
+_SERVER_DETAIL_CAP = 400
+
+
+def warehouse_refusal(message: str, *, code: str | None = None) -> WarehouseQueryError:
+    """The typed error for one server-side statement failure.
+
+    Every adapter funnels through here so the envelope reads the same whichever
+    warehouse said no, and so the server's words get the same trim: first line
+    only (drivers append the statement, a caret diagram, or their whole error
+    payload after it) and capped. ``code`` is the connector's own error code
+    where it has one, which is what a caller looking the failure up needs.
+    """
+
+    first = next((ln.strip() for ln in message.splitlines() if ln.strip()), "")
+    detail = first or "the server gave no message"
+    if len(detail) > _SERVER_DETAIL_CAP:
+        detail = detail[:_SERVER_DETAIL_CAP].rstrip() + "..."
+    return WarehouseQueryError(f"{detail} [{code}]" if code else detail)
+
+
 def json_safe(value: object | None) -> object | None:
     """Coerce a connector scalar to a JSON-serializable primitive for the envelope."""
 
@@ -451,6 +476,15 @@ def temporal_units_for(data_type: str) -> tuple[str, ...]:
 # anything else is reported by its bare length instead of guessing further.
 _HEX_LENGTH_NAMES = {32: "md5", 40: "sha1", 64: "sha256"}
 
+# What a shape-gated CAST reads on every row the shape predicate rejects, so
+# the cast's argument is digit-only for the whole column and the cast is total
+# (see `type_contradiction_expressions`). It has to parse as an integer on
+# every dialect and be *rejected* by every predicate built on top of the cast:
+# the epoch ranges start in the year 2000 and the slash-component test asks for
+# > 12, so zero is evidence of nothing and a row that reaches the cast only
+# because the cast is unconditional can never be counted.
+_CAST_SENTINEL = "'0'"
+
 
 def type_contradiction_expressions(
     qcol: str,
@@ -463,16 +497,37 @@ def type_contradiction_expressions(
 ) -> list[str]:
     """Declared-type-vs-content aggregate expressions for one column.
 
-    Every CAST is guarded behind a length-bounded shape predicate inside a
-    CASE -- never combined with AND, which the SQL standard does not
-    guarantee to evaluate left-to-right -- so a non-numeric string never
-    reaches a CAST on a dialect with no ``TRY_CAST`` (Postgres, Redshift), and
-    the gating patterns are always length-bounded so the CAST itself can never
-    overflow BIGINT/INT64. Only fractions and translated-to-date integers
-    (read back and converted to a calendar date by the caller) ever leave the
-    engine through this path. ``qcol`` must already be quoted/escaped by the
-    calling adapter. Returns ``[]`` for a column that is neither string- nor
-    integer-typed (nothing to check).
+    Every CAST here is **total**: its argument is a CASE that yields a
+    digit-only string on every row of the column (the value where a
+    length-bounded shape predicate matches, ``_CAST_SENTINEL`` where it does
+    not), so the cast cannot raise whatever rows the dialect decides to
+    evaluate it for, and the length bound means it can never overflow
+    BIGINT/INT64 either.
+
+    The obvious shape is the opposite one, and it is a bug (#310). Guarding
+    the cast *inside* a CASE branch (``CASE WHEN <shape> THEN CAST(col AS
+    BIGINT) END``) reads as safe and is safe on Postgres, but Redshift
+    evaluates a branch's cast for rows the WHEN never selects, so a single
+    non-numeric string killed the whole profiling statement server-side with
+    ``Invalid digit, Value 'p', Pos 0, Type: Long``. The SQL standard does not
+    promise the lazy evaluation that guard needs, no offline test can catch a
+    dialect that disagrees, and so nothing here depends on it: correctness
+    comes from the cast's argument being castable for every row, which is a
+    property of the expression rather than of the engine's evaluation order.
+
+    What the sentinel cannot express is the difference between "not shaped"
+    and "shaped but out of range", and the fractions' denominators are exactly
+    that distinction (``ts_ep_s_{i}`` is the in-range share *of the
+    epoch-shaped rows*, not of the column). That comes from a second,
+    uncast expression whose NULL-ness is the denominator test, which is where
+    the CASE-returns-NULL trick belongs: it carries a string, so it can raise
+    nothing.
+
+    Only fractions and translated-to-date integers (read back and converted to
+    a calendar date by the caller) ever leave the engine through this path.
+    ``qcol`` must already be quoted/escaped by the calling adapter. Returns
+    ``[]`` for a column that is neither string- nor integer-typed (nothing to
+    check).
     """
 
     def fraction(value_expr: str, condition: str, alias: str) -> str:
@@ -484,6 +539,21 @@ def fraction(value_expr: str, condition: str, alias: str) -> str:
     def plain_fraction(pattern: str, alias: str) -> str:
         return fraction(qcol, regexp_predicate(qcol, pattern), alias)
 
+    def shaped(predicate: str) -> str:
+        """The column itself where the shape matches, NULL everywhere else:
+        the denominator test, uncast and so incapable of raising."""
+
+        return f"CASE WHEN {predicate} THEN {qcol} END"
+
+    def total_cast(predicate: str, inner: str) -> str:
+        """``inner`` as an integer, with the sentinel standing in wherever the
+        shape predicate does not match, so every row casts digits."""
+
+        return (
+            f"CAST(CASE WHEN {predicate} THEN {inner} "
+            f"ELSE {_CAST_SENTINEL} END AS {bigint_type})"
+        )
+
     exprs: list[str] = []
     if is_string:
         exprs += [
@@ -499,34 +569,35 @@ def plain_fraction(pattern: str, alias: str) -> str:
             fraction(qcol, slash_datetime_pred, f"ts_sl_dt_{i}"),
         ]
         slash_either = f"({slash_date_pred} OR {slash_datetime_pred})"
-        first_val = (
-            f"CASE WHEN {slash_either} THEN "
-            f"CAST(SUBSTR({qcol}, 1, 2) AS {bigint_type}) END"
-        )
-        second_val = (
-            f"CASE WHEN {slash_either} THEN "
-            f"CAST(SUBSTR({qcol}, 4, 2) AS {bigint_type}) END"
-        )
+        slash_shaped = shaped(slash_either)
+        first_val = total_cast(slash_either, f"SUBSTR({qcol}, 1, 2)")
+        second_val = total_cast(slash_either, f"SUBSTR({qcol}, 4, 2)")
         exprs += [
-            fraction(first_val, f"{first_val} > 12", f"ts_sl1_{i}"),
-            fraction(second_val, f"{second_val} > 12", f"ts_sl2_{i}"),
+            fraction(slash_shaped, f"{first_val} > 12", f"ts_sl1_{i}"),
+            fraction(slash_shaped, f"{second_val} > 12", f"ts_sl2_{i}"),
         ]
         seconds_shape = regexp_predicate(qcol, EPOCH_SECONDS_SHAPE_PATTERN)
         millis_shape = regexp_predicate(qcol, EPOCH_MILLIS_SHAPE_PATTERN)
-        seconds_val = (
-            f"CASE WHEN {seconds_shape} THEN CAST({qcol} AS {bigint_type}) END"
-        )
-        millis_val = f"CASE WHEN {millis_shape} THEN CAST({qcol} AS {bigint_type}) END"
+        seconds_shaped = shaped(seconds_shape)
+        millis_shaped = shaped(millis_shape)
+        seconds_val = total_cast(seconds_shape, qcol)
+        millis_val = total_cast(millis_shape, qcol)
     elif is_integer:
-        seconds_val = millis_val = qcol  # already numeric: no CAST, no overflow surface
+        # Already numeric: no CAST to make total, no overflow surface, and
+        # every non-null row is in the denominator.
+        seconds_shaped = millis_shaped = seconds_val = millis_val = qcol
     else:
         return []
 
     seconds_cond = f"{seconds_val} BETWEEN {EPOCH_SECONDS_LOW} AND {EPOCH_SECONDS_HIGH}"
     millis_cond = f"{millis_val} BETWEEN {EPOCH_MILLIS_LOW} AND {EPOCH_MILLIS_HIGH}"
     exprs += [
-        fraction(seconds_val, seconds_cond, f"ts_ep_s_{i}"),
-        fraction(millis_val, millis_cond, f"ts_ep_ms_{i}"),
+        fraction(seconds_shaped, seconds_cond, f"ts_ep_s_{i}"),
+        fraction(millis_shaped, millis_cond, f"ts_ep_ms_{i}"),
+        # The MIN/MAX branch value is the total cast, not the shaped string:
+        # the range condition already excludes the sentinel (zero is not a
+        # plausible epoch), so a dialect that evaluates the branch for every
+        # row computes a cast that cannot raise and reports only in-range rows.
         f"MIN(CASE WHEN {seconds_cond} THEN {seconds_val} END) AS ts_ep_s_mn_{i}",
         f"MAX(CASE WHEN {seconds_cond} THEN {seconds_val} END) AS ts_ep_s_mx_{i}",
         f"MIN(CASE WHEN {millis_cond} THEN {millis_val} END) AS ts_ep_ms_mn_{i}",
@@ -579,10 +650,11 @@ def key_shape_expressions(
     is valid input to a hex-charset pattern too), which is what keeps
     ``numeric_string_fraction`` directly reusable here unchanged and keeps
     the two buckets from double-counting the same value. Plain boolean
-    predicates ANDed together carry none of the CAST-ordering risk
-    ``type_contradiction_expressions`` has to guard against; nothing here
-    can raise. ``qcol`` must already be quoted/escaped by the calling
-    adapter.
+    predicates ANDed together cast nothing, so the total-CAST discipline
+    ``type_contradiction_expressions`` has to keep does not apply here and
+    nothing in these expressions can raise on any dialect, whatever it
+    chooses to evaluate. ``qcol`` must already be quoted/escaped by the
+    calling adapter.
     """
 
     numeric_pred = regexp_predicate(qcol, NUMERIC_PATTERN)

packages/dex-core/src/exmergo_dex_core/adapters/bigquery.py

@@ -49,6 +49,7 @@
     temporal_units_for,
     type_contradiction_aggregate_kwargs,
     type_contradiction_expressions,
+    warehouse_refusal,
 )
 
 PARADIGM = "bytes_scanned"
@@ -1091,7 +1092,12 @@ def _run(
                     "(server-side maximum_bytes_billed); raise --budget or "
                     "narrow the query"
                 ) from exc
-            raise
+            # Every other BadRequest is BigQuery refusing the statement itself
+            # (an invalid query, a type it will not coerce). Typed, so the
+            # envelope carries `execution_failure` and BigQuery's own words
+            # rather than the `internal` an untyped API exception falls
+            # through to.
+            raise warehouse_refusal(str(exc)) from exc
         except TimeoutError as exc:
             # concurrent.futures.TimeoutError is the builtin on Python 3.11+.
             self._cancel(job)

packages/dex-core/src/exmergo_dex_core/adapters/databricks.py

@@ -60,6 +60,7 @@
     temporal_units_for,
     type_contradiction_aggregate_kwargs,
     type_contradiction_expressions,
+    warehouse_refusal,
 )
 
 PARADIGM = "compute_time"
@@ -145,6 +146,22 @@ def _date_diff_expr(unit: str, later: str, earlier: str) -> str:
     return f"TIMESTAMPDIFF({unit.upper()}, {earlier}, {later})"
 
 
+def _is_server_error(exc: Exception) -> bool:
+    """Whether the driver labelled this "the warehouse answered with an error".
+
+    Matched on the class name along the MRO rather than with ``isinstance``:
+    the SQL driver is injected into this adapter rather than imported by it
+    (which is what lets the offline suite build the adapter without the
+    library installed), so there is no class object here to compare against.
+    ``ServerOperationError`` is databricks-sql-connector's class for a
+    server-side statement failure; its transport failures (``RequestError``,
+    ``OperationalError``) are deliberately excluded, because a connection that
+    died is not a statement the server refused.
+    """
+
+    return any(cls.__name__ == "ServerOperationError" for cls in type(exc).__mro__)
+
+
 def warehouse_http_path(value: str) -> str:
     """The SQL driver's HTTP path for a pinned warehouse, which the config may
     name by ID or by full path."""
@@ -1293,6 +1310,13 @@ def _translate(self, exc: Exception, timeout_seconds: float | None) -> Exception
                 "remaining budget (STATEMENT_TIMEOUT); raise --budget or "
                 "narrow the work"
             )
+        if _is_server_error(exc):
+            # The warehouse answered with an error. Typed, so the envelope
+            # carries `execution_failure` and Databricks' own words rather
+            # than the `internal` an untyped driver exception falls through
+            # to. Transport failures are left alone deliberately: a connection
+            # that died is not a statement the server refused.
+            return warehouse_refusal(str(exc))
         return exc
 
     # --- helpers ------------------------------------------------------------------

packages/dex-core/src/exmergo_dex_core/adapters/duckdb.py

@@ -13,7 +13,7 @@
 from pathlib import Path
 
 from ..envelope import Paradigm
-from ..errors import ConnectorError
+from ..errors import ConnectorError, WarehouseQueryError
 from ..guards.sql_guard import assert_select_only
 from .base import (
     ColumnAggregate,
@@ -35,6 +35,7 @@
     temporal_units_for,
     type_contradiction_aggregate_kwargs,
     type_contradiction_expressions,
+    warehouse_refusal,
 )
 
 
@@ -528,7 +529,10 @@ def _interrupt() -> None:
                     f"query exceeded {timeout_seconds:g}s and was interrupted; "
                     "narrow it (tighter filter, fewer columns) and retry"
                 ) from exc
-            raise
+            refusal = _refusal(exc)
+            if refusal is None:
+                raise
+            raise refusal from exc
         finally:
             watchdog.cancel()
 
@@ -545,14 +549,37 @@ def _run_select(self, sql: str, params: list | None = None):
         # Single read-only door for every query: parsed and refused if it is not a
         # SELECT, on top of the read-only connection.
         assert_select_only(sql, dialect=self.dialect)
-        return self._conn.execute(sql, params or []).fetchall()
+        try:
+            return self._conn.execute(sql, params or []).fetchall()
+        except Exception as exc:
+            refusal = _refusal(exc)
+            if refusal is None:
+                raise
+            raise refusal from exc
 
     def close(self) -> None:
         if self._conn is not None:
             self._conn.close()
             self._conn = None
 
 
+def _refusal(exc: Exception) -> WarehouseQueryError | None:
+    """The typed refusal for a statement DuckDB itself rejected, or ``None``
+    when the failure did not come from the database at all.
+
+    The local engine is still an engine that answers with errors, and it gets
+    the same treatment as the cloud connectors: a refused statement reads as
+    ``execution_failure`` carrying DuckDB's own words rather than falling
+    through to ``internal``. Anything that is not a ``duckdb.Error`` (an
+    interrupt, an out-of-memory kill) is left alone, so the caller re-raises it
+    as itself.
+    """
+
+    import duckdb
+
+    return warehouse_refusal(str(exc)) if isinstance(exc, duckdb.Error) else None
+
+
 def _quote_ident(name: str) -> str:
     """Quote one identifier component for DuckDB, doubling embedded quotes."""