TabularIntrospector + on_tabular_uploaded trigger + 31 tests

PROGRESS.md

@@ -2,8 +2,8 @@
 
 Persistent tracker mirroring the 42-item ownership table in `REPO_CONTEXT.md` "Team — division of work". Update as PRs land. Future Claude Code sessions read this to know what's already done.
 
-**Last updated**: 2026-05-07 (PR3-DB — SQL compiler + DB executor shipped)
-**Current open PR**: PR3-DB (DB owner — SqlCompiler + DbExecutor + golden IR→SQL tests)
+**Last updated**: 2026-05-07 (PR1-tab — tabular introspector + on_tabular_uploaded trigger + 31 tests)
+**Current open PRs**: PR3-DB (DB owner — SqlCompiler + DbExecutor + golden IR→SQL tests) · PR1-tab (TAB owner — tabular introspector)
 
 ---
 
@@ -21,7 +21,7 @@ Persistent tracker mirroring the 42-item ownership table in `REPO_CONTEXT.md` "T
 | PR | Status | Owner(s) | Scope |
 |---|---|---|---|
 | PR1 | `[x]` merged | DB | Contract locks + catalog plumbing + DB introspector + IR validator + tests |
-| PR1-tab | `[ ]` | TAB | Tabular introspector + golden IR examples for tabular |
+| PR1-tab | `[~]` open | TAB | Tabular introspector + on_tabular_uploaded trigger + 31 unit tests |
 | PR2a | `[x]` merged | DB | CatalogEnricher + StructuredPipeline + on_db_registered trigger + FK extension on Table |
 | PR2b | `[ ]` | B | IntentRouter + planner prompt (pair) + planner LLM service |
 | PR3-DB | `[~]` open | DB | SqlCompiler (Postgres) + DbExecutor (sqlglot guard, RO + statement_timeout, asyncio.to_thread) + 36 golden IR→SQL tests |
@@ -53,7 +53,7 @@ Persistent tracker mirroring the 42-item ownership table in `REPO_CONTEXT.md` "T
 | # | Item | Owner | Status | Notes |
 |---|---|---|---|---|
 | 5 | DB introspector (`catalog/introspect/database.py`) | DB | `[x]` | PR1 — reuses Phase 1 `database_client_service`, `db_credential_encryption`, `db_pipeline_service.engine_scope`, `extractor.get_schema/profile_column/get_row_count`. PR2a wired FK extraction (was discarded before). |
-| 6 | Tabular introspector (`catalog/introspect/tabular.py`) | TAB | `[ ]` | XLSX sheet → one Table; `Source.source_id = document_id` |
+| 6 | Tabular introspector (`catalog/introspect/tabular.py`) | TAB | `[~]` | PR1-tab — downloads original blob (CSV/XLSX/Parquet), one Table per sheet (XLSX) or one Table (CSV/Parquet). `source_id = document_id`. `fetch_doc`/`fetch_blob` injectable for unit tests (no Settings). |
 | 7 | `BaseIntrospector` ABC (`catalog/introspect/base.py`) | B | `[x]` | Pre-existing; signature locked |
 
 ### Ingestion — shared catalog plumbing
@@ -71,7 +71,7 @@ Persistent tracker mirroring the 42-item ownership table in `REPO_CONTEXT.md` "T
 | # | Item | Owner | Status | Notes |
 |---|---|---|---|---|
 | 13 | Structured pipeline (`pipeline/structured_pipeline.py`) | B | `[x]` | PR2a (DB owner) — `introspect → enrich → merge with existing → validate → upsert`. Source-type-agnostic: caller supplies the introspector. `default_structured_pipeline()` factory wires production deps lazily so tests can inject mocks without `Settings()` construction. |
-| 14 | Triggers (`pipeline/triggers.py`) | B | `[~]` | PR2a — `on_db_registered` implemented (DB owner). `on_tabular_uploaded`, `on_document_uploaded`, `on_catalog_rebuild_requested` still stubs. |
+| 14 | Triggers (`pipeline/triggers.py`) | B | `[~]` | PR2a — `on_db_registered` implemented (DB owner). PR1-tab — `on_tabular_uploaded` implemented (TAB owner). `on_document_uploaded`, `on_catalog_rebuild_requested` still stubs. |
 | 15 | Ingestion orchestrator (`pipeline/orchestrator.py`) | B | `[ ]` | Likely redundant — StructuredPipeline already takes the introspector at run() time. Revisit if a higher-level routing layer is needed. |
 | 16 | Document pipeline (`pipeline/document_pipeline.py`) | TAB | `[ ]` | Tabular-adjacent (file uploads). Phase 1 implementation exists at `pipeline/document_pipeline/document_pipeline.py` — reuse or rewrite TBD |
 
@@ -132,7 +132,7 @@ Persistent tracker mirroring the 42-item ownership table in `REPO_CONTEXT.md` "T
 | — | Catalog validator tests (`tests/catalog/test_validator.py`) | B | `[x]` | PR1 — 5 tests |
 | — | Catalog store integration test (`tests/catalog/test_store.py`) | DB | `[x]` | PR1 — module-level skip without `RUN_INTEGRATION_TESTS=1` |
 | — | DB introspector test | DB | `[ ]` | Deferred to PR2 — needs Postgres testcontainer or fixture infra |
-| — | Tabular introspector test | TAB | `[ ]` | TAB to add when introspector lands |
+| — | Tabular introspector test | TAB | `[~]` | PR1-tab — 31 unit tests (CSV/XLSX/Parquet, stats, PII, error paths). No DB/blob I/O — mocks injected via constructor. |
 | 41 | Planner eval (`tests/query/planner/`) | B | `[ ]` | PR6 — golden question → IR examples; each side contributes |
 | 42 | E2E smoke tests (`tests/e2e/`) | B | `[ ]` | PR4 — pair |
 | — | Golden IR fixtures (`tests/fixtures/golden_irs.json`) | B | `[~]` | PR1 seeded with 5 DB-targeting examples; TAB extends in PR1-tab |
@@ -140,6 +140,19 @@ Persistent tracker mirroring the 42-item ownership table in `REPO_CONTEXT.md` "T
 
 ---
 
+## What just shipped (PR1-tab — TAB owner)
+
+**Files implemented**:
+- `src/catalog/introspect/tabular.py` — `TabularIntrospector` reads original blob (CSV/XLSX/Parquet), profiles each column (dtype, stats, sample values), runs PIIDetector. For XLSX: one `Table` per sheet (`Table.name = sheet_name`); for CSV/Parquet: one `Table` (`Table.name = filename stem`). `fetch_doc`/`fetch_blob` are constructor-injectable for unit tests — no `Settings` or DB required at import time.
+- `src/pipeline/triggers.py` — `on_tabular_uploaded` wired (mirrors `on_db_registered` pattern).
+
+**Tests added** (31 new):
+- `tests/unit/catalog/test_introspect_tabular.py` — CSV / XLSX / Parquet shapes, per-column stats, nullable detection, PII name + value matching, sample capping, all error paths. Pure Python, no network I/O.
+
+**Executor contract note**: introspector downloads the *original* blob for schema reading. The tabular executor (PR3-TAB) downloads *Parquet* blobs for query execution. For CSV/Parquet sources (single table), the executor must call `parquet_blob_name(uid, did, sheet_name=None)`; for XLSX (multi-table), `parquet_blob_name(uid, did, table.name)`.
+
+---
+
 ## What just shipped (PR3-DB — DB owner)
 
 **Files implemented**:

src/catalog/introspect/tabular.py

@@ -1,15 +1,231 @@
 """Tabular file schema introspection (Parquet / CSV / XLSX).
 
 Reads file headers + samples ~100 rows. For XLSX, each sheet becomes a Table.
-Files are expected to live in Azure Blob (location_ref like az_blob://...).
+Files are expected to live in Azure Blob (location_ref like az_blob://{user_id}/{document_id}).
+
+Table.name convention (executor contract)
+-----------------------------------------
+  CSV / Parquet  → Table.name = filename stem (e.g. "sales_data").
+                   Parquet blob was uploaded without a sheet suffix, so the
+                   executor must call parquet_blob_name(uid, did, sheet_name=None).
+  XLSX           → Table.name = sheet_name (e.g. "Sheet1").
+                   Executor calls parquet_blob_name(uid, did, table.name).
 """
 
-from ..models import Source
+import asyncio
+import hashlib
+from collections.abc import Callable, Coroutine
+from datetime import UTC, datetime
+from io import BytesIO
+from pathlib import Path
+from typing import Any
+
+import pandas as pd
+
+from src.middlewares.logging import get_logger
+
+from ..models import Column, ColumnStats, DataType, Source, Table
+from ..pii_detector import PIIDetector
 from .base import BaseIntrospector
 
+logger = get_logger("tabular_introspector")
+
+_AZ_BLOB_PREFIX = "az_blob://"
+
+
+def _stable_id(prefix: str, *parts: str) -> str:
+    h = hashlib.sha1(
+        "/".join(parts).encode("utf-8"), usedforsecurity=False
+    ).hexdigest()[:12]
+    return f"{prefix}{h}"
+
+
+def _map_pandas_type(dtype: Any) -> DataType:
+    s = str(dtype).lower()
+    if "int" in s:
+        return "int"
+    if "float" in s or "decimal" in s:
+        return "decimal"
+    if "bool" in s:
+        return "bool"
+    if "datetime" in s:
+        return "datetime"
+    if "date" in s:
+        return "date"
+    return "string"
+
+
+def _normalize(v: Any) -> Any:
+    """Coerce non-JSON-native scalars to types that survive the jsonb round-trip."""
+    if v is None:
+        return None
+    try:
+        import numpy as np
+
+        if isinstance(v, np.generic):
+            return v.item()
+    except ImportError:
+        pass
+    if isinstance(v, datetime):
+        return v.isoformat()
+    return v
+
 
 class TabularIntrospector(BaseIntrospector):
-    """Read column names, dtypes, and sample values from Parquet/CSV/XLSX."""
+    """Read column names, dtypes, and sample values from Parquet/CSV/XLSX.
+
+    Heavy I/O dependencies (`fetch_doc`, `fetch_blob`) are injectable so unit
+    tests can pass mocks without triggering Settings or DB construction — same
+    pattern as CatalogEnricher's `structured_chain` parameter.
+    """
+
+    def __init__(
+        self,
+        fetch_doc: Callable[[str], Coroutine[Any, Any, Any]] | None = None,
+        fetch_blob: Callable[[str], Coroutine[Any, Any, bytes]] | None = None,
+    ) -> None:
+        self._pii = PIIDetector()
+        self._fetch_doc = fetch_doc or self._default_fetch_doc
+        self._fetch_blob = fetch_blob or self._default_fetch_blob
+
+    @staticmethod
+    async def _default_fetch_doc(document_id: str) -> Any:
+        from sqlalchemy import select
+
+        from src.db.postgres.connection import AsyncSessionLocal
+        from src.db.postgres.models import Document as DBDocument
+
+        async with AsyncSessionLocal() as session:
+            result = await session.execute(
+                select(DBDocument).where(DBDocument.id == document_id)
+            )
+            return result.scalar_one_or_none()
+
+    @staticmethod
+    async def _default_fetch_blob(blob_name: str) -> bytes:
+        from src.storage.az_blob.az_blob import blob_storage
+
+        return await blob_storage.download_file(blob_name)
 
     async def introspect(self, location_ref: str) -> Source:
-        raise NotImplementedError
+        if not location_ref.startswith(_AZ_BLOB_PREFIX):
+            raise ValueError(
+                f"TabularIntrospector expects 'az_blob://...' location_ref, "
+                f"got {location_ref!r}"
+            )
+        rest = location_ref[len(_AZ_BLOB_PREFIX):]
+        user_id, _, document_id = rest.partition("/")
+        if not user_id or not document_id:
+            raise ValueError(
+                f"location_ref must be 'az_blob://{{user_id}}/{{document_id}}', "
+                f"got {location_ref!r}"
+            )
+
+        doc = await self._fetch_doc(document_id)
+        if doc is None:
+            raise ValueError(f"Document {document_id!r} not found")
+
+        logger.info(
+            "introspecting tabular source",
+            document_id=document_id,
+            file_type=doc.file_type,
+            filename=doc.filename,
+        )
+
+        content = await self._fetch_blob(doc.blob_name)
+
+        tables: list[Table] = await asyncio.to_thread(
+            self._introspect_sync, content, doc.file_type, doc.filename, document_id
+        )
+
+        return Source(
+            source_id=document_id,
+            source_type="tabular",
+            name=doc.filename,
+            description="",
+            location_ref=location_ref,
+            updated_at=datetime.now(UTC),
+            tables=tables,
+        )
+
+    def _introspect_sync(
+        self,
+        content: bytes,
+        file_type: str,
+        filename: str,
+        document_id: str,
+    ) -> list[Table]:
+        if file_type == "csv":
+            df = pd.read_csv(BytesIO(content))
+            return [self._build_table(df, document_id, Path(filename).stem, sheet_name=None)]
+        if file_type == "xlsx":
+            sheets: dict[str, pd.DataFrame] = pd.read_excel(BytesIO(content), sheet_name=None)
+            return [
+                self._build_table(df, document_id, sheet_name, sheet_name=sheet_name)
+                for sheet_name, df in sheets.items()
+            ]
+        if file_type == "parquet":
+            df = pd.read_parquet(BytesIO(content))
+            return [self._build_table(df, document_id, Path(filename).stem, sheet_name=None)]
+        raise ValueError(f"Unsupported file_type {file_type!r} for tabular introspection")
+
+    def _build_table(
+        self,
+        df: pd.DataFrame,
+        document_id: str,
+        table_name: str,
+        sheet_name: str | None,
+    ) -> Table:
+        id_parts = (document_id, sheet_name) if sheet_name else (document_id,)
+        columns = [
+            self._to_column(df[col], document_id, sheet_name, col)
+            for col in df.columns
+        ]
+        return Table(
+            table_id=_stable_id("t_", *id_parts),
+            name=table_name,
+            description="",
+            row_count=len(df),
+            columns=columns,
+            foreign_keys=[],
+        )
+
+    def _to_column(
+        self,
+        series: pd.Series,
+        document_id: str,
+        sheet_name: str | None,
+        col_name: str,
+    ) -> Column:
+        id_parts = (
+            (document_id, sheet_name, col_name) if sheet_name else (document_id, col_name)
+        )
+
+        sample_raw = series.dropna().head(5).tolist()
+        sample_values: list[Any] | None = [_normalize(v) for v in sample_raw] or None
+
+        is_numeric = pd.api.types.is_numeric_dtype(series)
+        is_dt = pd.api.types.is_datetime64_any_dtype(series)
+        non_null = series.dropna()
+        stats = ColumnStats(
+            min=_normalize(non_null.min()) if (is_numeric or is_dt) and len(non_null) > 0 else None,
+            max=_normalize(non_null.max()) if (is_numeric or is_dt) and len(non_null) > 0 else None,
+            distinct_count=int(series.nunique()),
+        )
+
+        column = Column(
+            column_id=_stable_id("c_", *id_parts),
+            name=col_name,
+            data_type=_map_pandas_type(series.dtype),
+            description="",
+            nullable=bool(series.isnull().any()),
+            pii_flag=False,
+            sample_values=sample_values,
+            stats=stats,
+        )
+        if self._pii.detect(column):
+            return column.model_copy(update={"pii_flag": True, "sample_values": None})
+        return column
+
+
+tabular_introspector = TabularIntrospector()

src/pipeline/triggers.py

@@ -36,17 +36,24 @@ async def on_db_registered(database_client_id: str, user_id: str) -> None:
 
 
 async def on_tabular_uploaded(document_id: str, user_id: str) -> None:
-    """Stub — implemented by TAB owner once `catalog/introspect/tabular.py` lands.
+    """Build an az_blob:// location_ref and run the structured pipeline.
 
-    Expected pattern (mirrors `on_db_registered`):
-
-        from src.catalog.introspect.tabular import tabular_introspector
-        from src.pipeline.structured_pipeline import default_structured_pipeline
-        location_ref = f"az_blob://{user_id}/{document_id}"
-        pipeline = default_structured_pipeline()
-        await pipeline.run(tabular_introspector, location_ref, user_id)
+    Called after a CSV/XLSX/Parquet file has been processed and its Parquet
+    blob(s) uploaded. The TabularIntrospector downloads the original blob,
+    profiles each column, and produces a Source. The CatalogEnricher then fills
+    in descriptions, the catalog is validated and upserted.
     """
-    raise NotImplementedError
+    from src.catalog.introspect.tabular import tabular_introspector
+    from src.pipeline.structured_pipeline import default_structured_pipeline
+
+    location_ref = f"az_blob://{user_id}/{document_id}"
+    logger.info(
+        "on_tabular_uploaded triggered",
+        user_id=user_id,
+        document_id=document_id,
+    )
+    pipeline = default_structured_pipeline()
+    await pipeline.run(tabular_introspector, location_ref, user_id)
 
 
 async def on_document_uploaded(document_id: str, user_id: str) -> None: