fix(connector): SQL sampling failures → scan_failures (#1140)

connectors/sql_connector.py

@@ -98,6 +98,26 @@
     }
 )
 
+# scan_failures reason for column sampling errors (#1140; taxonomy shared with #828).
+SCAN_FAILURE_REASON_SAMPLING_ERROR = "sampling_error"
+
+
+class ColumnSampleError(RuntimeError):
+    """Column sampling failed; the connector persisted a scan_failures row."""
+
+
+def format_column_sample_failure_details(
+    *,
+    schema: str,
+    table: str,
+    column_name: str,
+    dialect: str,
+    exc: BaseException,
+) -> str:
+    """Stable detail string for scan_failures when SQL column sampling fails."""
+    loc = f"{schema}.{table}.{column_name}" if schema else f"{table}.{column_name}"
+    return f"{loc} dialect={dialect}: {type(exc).__name__}: {exc}"
+
 
 def _get_skip_schemas(dialect: str) -> frozenset[str]:
     """Return schema names to skip when discovering (dialect-specific blacklist)."""
@@ -324,7 +344,10 @@ def _process_one_finding(
         """Sample column, run detection, optionally full-scan for minor; save finding and log."""
         if self._inter_query_delay_s > 0:
             time.sleep(self._inter_query_delay_s)
-        sample = self.sample(schema, table, cname)
+        try:
+            sample = self.sample(schema, table, cname)
+        except ColumnSampleError:
+            return
         res = self.scanner.scan_column(cname, sample, connector_data_type=ctype)
         res = augment_low_id_like_for_persist(res, cname, self.detection_config)
         if (
@@ -337,16 +360,22 @@ def _process_one_finding(
             res.get("pattern_detected") or ""
         ) and self.detection_config.get("minor_full_scan"):
             full_scan_limit = self.detection_config.get("minor_full_scan_limit", 100)
-            full_sample = self.sample(schema, table, cname, limit=full_scan_limit)
-            full_res = self.scanner.scan_column(
-                cname, full_sample, connector_data_type=ctype
-            )
-            if "DOB_POSSIBLE_MINOR" in (full_res.get("pattern_detected") or ""):
-                res = full_res
-                suffix = " (full-scan confirmed)"
-                norm_tag = (
-                    (norm_tag or "").rstrip() + suffix if norm_tag else suffix.lstrip()
+            try:
+                full_sample = self.sample(schema, table, cname, limit=full_scan_limit)
+            except ColumnSampleError:
+                pass
+            else:
+                full_res = self.scanner.scan_column(
+                    cname, full_sample, connector_data_type=ctype
                 )
+                if "DOB_POSSIBLE_MINOR" in (full_res.get("pattern_detected") or ""):
+                    res = full_res
+                    suffix = " (full-scan confirmed)"
+                    norm_tag = (
+                        (norm_tag or "").rstrip() + suffix
+                        if norm_tag
+                        else suffix.lstrip()
+                    )
         self.db_manager.save_finding(
             source_type="database",
             target_name=target_name,
@@ -397,59 +426,71 @@ def sample(
         safe_col = column_name.replace('"', '""')
         safe_table = table.replace('"', '""')
         safe_schema = (schema or "").replace('"', '""')
-        try:
-            tkey = (schema or "", table)
-            if tkey not in self._table_row_cache:
+        tkey = (schema or "", table)
+        if tkey not in self._table_row_cache:
+            try:
                 from connectors.sql_table_row_estimate import estimate_table_rows
 
                 self._table_row_cache[tkey] = estimate_table_rows(
                     self._connection, dialect, schema or "", table
                 )
-            est = self._table_row_cache[tkey]
-            table_meta = TableSamplingMetadata(estimated_row_count=est)
-            to = self._sample_statement_timeout_ms
-            plan = SamplingManager.build_column_sample(
-                dialect,
-                safe_col=safe_col,
-                safe_table=safe_table,
-                safe_schema=safe_schema,
-                schema=schema,
-                limit=use_limit,
-                table_metadata=table_meta,
-                statement_timeout_ms=to,
-            )
-            audit_key = f"{schema or ''}.{table}"
-            if self._sql_sampling_audit_key != audit_key:
-                self._sql_sampling_audit_key = audit_key
-                try:
-                    from utils.logger import get_logger
-
-                    human = plan.human_strategy or plan.strategy_label
-                    msg = "Sampling %s using %s strategy (label=%s dialect=%s)" % (
-                        audit_key,
-                        human,
-                        plan.strategy_label,
-                        dialect,
-                    )
-                    if plan.audit_notes:
-                        msg += " notes=%s" % (plan.audit_notes,)
-                    get_logger().info(msg)
-                except Exception:
-                    pass
+            except Exception:
+                self._table_row_cache[tkey] = None
+        est = self._table_row_cache[tkey]
+        table_meta = TableSamplingMetadata(estimated_row_count=est)
+        to = self._sample_statement_timeout_ms
+        plan = SamplingManager.build_column_sample(
+            dialect,
+            safe_col=safe_col,
+            safe_table=safe_table,
+            safe_schema=safe_schema,
+            schema=schema,
+            limit=use_limit,
+            table_metadata=table_meta,
+            statement_timeout_ms=to,
+        )
+        audit_key = f"{schema or ''}.{table}"
+        if self._sql_sampling_audit_key != audit_key:
+            self._sql_sampling_audit_key = audit_key
+            try:
+                from utils.logger import get_logger
+
+                human = plan.human_strategy or plan.strategy_label
+                msg = "Sampling %s using %s strategy (label=%s dialect=%s)" % (
+                    audit_key,
+                    human,
+                    plan.strategy_label,
+                    dialect,
+                )
+                if plan.audit_notes:
+                    msg += " notes=%s" % (plan.audit_notes,)
+                get_logger().info(msg)
+            except Exception:
+                pass  # best-effort audit logging — never break sampling for a log line
+        try:
             with self._connection.begin():
                 if dialect in ("postgresql", "postgres") and to:
                     self._connection.execute(
                         text("SET LOCAL statement_timeout = :mt").bindparams(mt=int(to))
                     )
                 rows = self._connection.execute(plan.query).fetchall()
-            parts = [str(r[0])[:200] for r in rows if r[0] is not None]
-            return " ".join(parts)
         except Exception as e:
+            target_name = str(self.config.get("name") or "database")
+            details = format_column_sample_failure_details(
+                schema=schema or "",
+                table=table,
+                column_name=column_name,
+                dialect=dialect,
+                exc=e,
+            )
+            self.db_manager.save_failure(
+                target_name, SCAN_FAILURE_REASON_SAMPLING_ERROR, details
+            )
             try:
                 from utils.logger import get_logger
 
                 get_logger().warning(
-                    "sample_skipped schema=%s table=%s col=%s dialect=%s err=%s",
+                    "sample_failed schema=%s table=%s col=%s dialect=%s err=%s",
                     schema or "",
                     table,
                     column_name,
@@ -458,7 +499,9 @@ def sample(
                 )
             except Exception:
                 pass
-            return ""
+            raise ColumnSampleError from e
+        parts = [str(r[0])[:200] for r in rows if r[0] is not None]
+        return " ".join(parts)
 
     def run(self) -> None:
         """Connect, discover, sample each column, detect, save_finding; on error save_failure."""

connectors/sql_sampling.py

@@ -357,9 +357,8 @@ def _plan_mssql_column_sample(
     # SQL Server has **no** query-level execution-time hint equivalent to
     # MySQL's ``/*+ MAX_EXECUTION_TIME(N) */``; ``OPTION (MAX_EXECUTION_TIME = …)``
     # is *not* valid T-SQL and emitting it makes every sample fail with a
-    # syntax error -- which the connector's broad ``except Exception`` then
-    # swallows, returning empty samples and producing **silent zero-finding
-    # scans** on every SQL Server target. Statement-time bounds for MSSQL
+    # syntax error -- the connector records ``scan_failures`` (reason
+    # ``sampling_error``) and skips detection for that column (#1140). Statement-time bounds for MSSQL
     # come from the SQLAlchemy connection-level ``connect_timeout`` (set via
     # ``_connect_args_from_target``); for finer per-statement budgets a
     # dedicated ADR + driver-level ``LOCK_TIMEOUT`` or session option is the

core/database.py

@@ -131,7 +131,7 @@ class ScanFailure(Base):
     target_name = Column(String(100))
     reason = Column(
         String(50)
-    )  # unreachable, auth_failed, permission_denied, timeout, error
+    )  # unreachable, auth_failed, permission_denied, timeout, sampling_error, error
     details = Column(Text, nullable=True)
     created_at = Column(DateTime, default=_utc_now)
 
@@ -163,6 +163,12 @@ def failure_hint(reason: str) -> str:
             "Consider increasing timeout values and re-running during off-peak hours. "
             "You can set timeouts in config (timeouts.connect_seconds, timeouts.read_seconds) or per target; see USAGE.md."
         )
+    if r == "sampling_error":
+        return (
+            "Column sampling failed (SQL syntax, permissions, or connectivity). The column was not "
+            "evaluated — this is not a clean result. Check dialect-specific sampling notes, credentials, "
+            "and the detailed message; fix the target or sampling plan before re-running."
+        )
     return (
         "Unexpected error. Review the detailed message and audit log, verify the target configuration "
         "(host, port, path, credentials) and test connectivity manually before re-running."

tests/test_database.py

@@ -219,6 +219,7 @@ def test_failure_hint_other_reasons():
     )
     assert "auth" in failure_hint("auth_failed").lower()
     assert "permission" in failure_hint("permission_denied").lower()
+    assert "sampling" in failure_hint("sampling_error").lower()
     assert (
         "unexpected" in failure_hint("unknown").lower()
         or "error" in failure_hint("unknown").lower()

tests/test_sql_connector.py

@@ -7,17 +7,22 @@
 import sqlite3
 from unittest.mock import MagicMock
 
+import pytest
 from connectors.sql_connector import (
     DRIVER_MAP,
+    SCAN_FAILURE_REASON_SAMPLING_ERROR,
+    ColumnSampleError,
     SQLConnector,
     _connect_args_from_target,
     _discover_fallback_no_schemas,
     _get_skip_schemas,
     _resolve_sample_statement_timeout_ms,
     _should_skip_schema,
+    format_column_sample_failure_details,
 )
 from sqlalchemy import create_engine, inspect
 from sqlalchemy.engine.url import make_url
+from sqlalchemy.exc import ProgrammingError
 
 
 def test_driver_map_uses_registered_sqlalchemy_dialects():
@@ -254,3 +259,104 @@ def test_sql_connector_sample_sparse_column_prefers_non_null(tmp_path):
         connector.close()
 
     assert "marker_value" in sample
+
+
+def test_format_column_sample_failure_details_includes_location_and_dialect():
+    err = ProgrammingError("stmt", {}, Exception("Incorrect syntax near 'OPTION'"))
+    detail = format_column_sample_failure_details(
+        schema="dbo",
+        table="users",
+        column_name="email",
+        dialect="mssql",
+        exc=err,
+    )
+    assert "dbo.users.email" in detail
+    assert "dialect=mssql" in detail
+    assert "ProgrammingError" in detail
+
+
+def test_sql_connector_sample_syntax_error_records_scan_failure():
+    """Sampling SQL errors must persist scan_failures, not return empty clean samples (#1140)."""
+    target = {"type": "database", "driver": "mssql", "name": "ProdDB"}
+    scanner = MagicMock()
+    db_manager = MagicMock()
+    connector = SQLConnector(target, scanner, db_manager, sample_limit=5)
+    connector.engine = MagicMock()
+    connector.engine.dialect.name = "mssql"
+    connector._connection = MagicMock()
+    ctx = MagicMock()
+    connector._connection.begin.return_value = ctx
+    ctx.__enter__ = MagicMock(return_value=None)
+    ctx.__exit__ = MagicMock(return_value=False)
+    connector._connection.execute.side_effect = ProgrammingError(
+        "SELECT",
+        {},
+        Exception("Incorrect syntax near 'OPTION'"),
+    )
+
+    with pytest.raises(ColumnSampleError):
+        connector.sample("dbo", "users", "email")
+
+    db_manager.save_failure.assert_called_once()
+    args = db_manager.save_failure.call_args[0]
+    assert args[0] == "ProdDB"
+    assert args[1] == SCAN_FAILURE_REASON_SAMPLING_ERROR
+    assert "users" in args[2]
+
+
+def test_process_one_finding_sampling_error_skips_scan_column():
+    """Failed sampling must not run detection on an empty pseudo-clean sample."""
+    target = {"type": "database", "driver": "sqlite", "name": "T"}
+    scanner = MagicMock()
+    db_manager = MagicMock()
+    connector = SQLConnector(target, scanner, db_manager)
+    connector.sample = MagicMock(side_effect=ColumnSampleError())
+
+    connector._process_one_finding("T", "localhost", "sqlite", "", "t1", "c1", "TEXT")
+
+    scanner.scan_column.assert_not_called()
+
+
+def test_minor_full_scan_sample_error_keeps_first_pass_dob_and_records_failure():
+    """Full-scan ColumnSampleError must not discard DOB_POSSIBLE_MINOR from the first pass (#1140)."""
+    target = {"type": "database", "driver": "sqlite", "name": "MinorDB"}
+    scanner = MagicMock()
+    scanner.scan_column.return_value = {
+        "sensitivity_level": "MEDIUM",
+        "pattern_detected": "DOB_POSSIBLE_MINOR",
+        "norm_tag": "",
+        "ml_confidence": 50,
+    }
+    db_manager = MagicMock()
+    connector = SQLConnector(
+        target,
+        scanner,
+        db_manager,
+        detection_config={"minor_full_scan": True, "minor_full_scan_limit": 100},
+    )
+
+    def sample_side_effect(schema, table, cname, limit=None):
+        if limit is not None:
+            db_manager.save_failure(
+                "MinorDB",
+                SCAN_FAILURE_REASON_SAMPLING_ERROR,
+                "dbo.minors.dob dialect=sqlite: ProgrammingError: full scan failed",
+            )
+            raise ColumnSampleError()
+        return "2005-01-01"
+
+    connector.sample = MagicMock(side_effect=sample_side_effect)
+
+    connector._process_one_finding(
+        "MinorDB", "localhost", "sqlite", "", "minors", "dob", "DATE"
+    )
+
+    db_manager.save_failure.assert_called_once_with(
+        "MinorDB",
+        SCAN_FAILURE_REASON_SAMPLING_ERROR,
+        "dbo.minors.dob dialect=sqlite: ProgrammingError: full scan failed",
+    )
+    db_manager.save_finding.assert_called_once()
+    finding_kwargs = db_manager.save_finding.call_args.kwargs
+    assert "DOB_POSSIBLE_MINOR" in finding_kwargs["pattern_detected"]
+    assert "(full-scan confirmed)" not in (finding_kwargs.get("norm_tag") or "")