feat(semantic-layer): resolve alias/linked-bucket column types (workspace, auto by default in UI)

src/keboola_agent_cli/commands/semantic_layer.py

@@ -335,6 +335,13 @@ def _print_build_result(console: Console, data: dict) -> None:
             console.print(f"  [red]✗[/red] {e['type']} {e['item']} — {e['detail']}")
     if warns:
         console.print(f"\n[bold yellow]Validation: {len(warns)} warning(s)[/bold yellow]")
+    type_errs = data.get("type_resolution_errors") or []
+    if type_errs:
+        console.print(
+            f"\n[bold yellow]Type resolution: {len(type_errs)} table(s) unresolved[/bold yellow]"
+        )
+        for e in type_errs:
+            console.print(f"  [yellow]![/yellow] {e['table_id']} — {e['error']}")
     if data.get("created"):
         console.print(f"\nCreated: {data['created']}")
     elif data.get("dry_run"):
@@ -434,6 +441,24 @@ def semantic_layer_build(
     output: Path | None = typer.Option(
         None, "--output", help="Also write the generated JSON to this file."
     ),
+    types_workspace: int | None = typer.Option(
+        None,
+        "--types-workspace",
+        help=(
+            "Workspace ID used to read real column types from the warehouse "
+            "INFORMATION_SCHEMA for tables whose Storage metadata carries none "
+            "(alias / linked-bucket tables). Without it such tables reach the "
+            "heuristic untyped and every field becomes a dimension."
+        ),
+    ),
+    auto_types_workspace: bool = typer.Option(
+        False,
+        "--auto-types-workspace",
+        help=(
+            "Auto-pick a read-only workspace per backend for column-type "
+            "resolution (instead of passing --types-workspace explicitly)."
+        ),
+    ),
 ) -> None:
     """Build a semantic-layer model from a list of storage tables (non-interactive).
 
@@ -471,6 +496,8 @@ def semantic_layer_build(
         dry_run=dry_run,
         keep_on_failure=keep_on_failure,
         output_path=output,
+        types_workspace_id=types_workspace,
+        auto_resolve_types=auto_types_workspace,
     )
     formatter.output(result, _print_build_result)
 

src/keboola_agent_cli/server/routers/semantic_layer.py

@@ -188,6 +188,11 @@ class BuildRequest(BaseModel):
     name: str | None = None
     dry_run: bool = False
     keep_on_failure: bool = False
+    # Column-type resolution for alias / linked tables (empty Storage
+    # metadata). Provide a specific workspace, or leave auto-resolve on (the
+    # UI default) to have the server pick a read-only workspace per backend.
+    types_workspace_id: int | None = None
+    auto_resolve_types: bool = True
 
 
 class TokenEncryptRequest(BaseModel):
@@ -587,6 +592,8 @@ def build(body: BuildRequest, registry: ServiceRegistry = Depends(get_registry))
         model_name_or_uuid=body.model,
         dry_run=body.dry_run,
         keep_on_failure=body.keep_on_failure,
+        types_workspace_id=body.types_workspace_id,
+        auto_resolve_types=body.auto_resolve_types,
     )
 
 

src/keboola_agent_cli/services/_semantic_layer_internals.py

@@ -25,6 +25,8 @@
 
 from __future__ import annotations
 
+import csv
+import io
 import json
 import logging
 import os
@@ -807,6 +809,143 @@ def _worker(tid: str) -> WorkerResult:
     return schemas_by_tid, fetch_errors
 
 
+# ── build: warehouse type resolution (opt-in `--types-workspace`) ────
+
+# Keboola bucket/table identifiers are restricted to this character set. We
+# interpolate them into INFORMATION_SCHEMA SQL, so an identifier containing
+# anything else is rejected rather than escaped -- defense in depth even
+# though the Storage API would never mint such an id.
+_SAFE_IDENTIFIER_RE = re.compile(r"^[A-Za-z0-9_.\-]+$")
+
+
+def build_information_schema_sql(backend: str, bucket_id: str, table_name: str) -> str | None:
+    """Build an ``INFORMATION_SCHEMA.COLUMNS`` query for a table's real types.
+
+    Returns a query projecting ``(column_name, data_type)`` for the given
+    backend, or ``None`` when the backend is unsupported or an identifier
+    contains an unexpected character (see ``_SAFE_IDENTIFIER_RE``).
+    """
+    if not (
+        _SAFE_IDENTIFIER_RE.match(bucket_id or "") and _SAFE_IDENTIFIER_RE.match(table_name or "")
+    ):
+        return None
+    normalized_backend = (backend or "").lower()
+    if normalized_backend == "bigquery":
+        # Keboola maps a bucket to a BigQuery dataset by replacing `.` and `-`
+        # with `_` (e.g. `in.c-Foo-Bar` -> `in_c_Foo_Bar`).
+        dataset = bucket_id.replace(".", "_").replace("-", "_")
+        return (
+            "SELECT column_name, data_type "
+            f"FROM `{dataset}`.INFORMATION_SCHEMA.COLUMNS "
+            f"WHERE table_name = '{table_name}'"
+        )
+    if normalized_backend == "snowflake":
+        # Keboola's Snowflake mapping keeps the bucketId verbatim as the schema
+        # name inside the KEBOOLA database (mirrors ``_derive_fqn``).
+        return (
+            "SELECT COLUMN_NAME AS column_name, DATA_TYPE AS data_type "
+            'FROM "KEBOOLA".INFORMATION_SCHEMA.COLUMNS '
+            f"WHERE TABLE_SCHEMA = '{bucket_id}' AND TABLE_NAME = '{table_name}'"
+        )
+    return None
+
+
+def _parse_information_schema_result(result: dict[str, Any]) -> dict[str, str]:
+    """Extract a ``{column_name: data_type}`` map from an execute_query result.
+
+    Reads the first statement's synthesized ``csv_data`` (always present on the
+    fast inline path), which carries a ``column_name,data_type`` header.
+    """
+    statements = result.get("statements") or []
+    if not statements:
+        return {}
+    csv_data = statements[0].get("csv_data")
+    if not csv_data:
+        return {}
+    type_map: dict[str, str] = {}
+    for row in csv.DictReader(io.StringIO(csv_data)):
+        name = (row.get("column_name") or row.get("COLUMN_NAME") or "").strip()
+        dtype = (row.get("data_type") or row.get("DATA_TYPE") or "").strip()
+        if name and dtype:
+            type_map[name] = dtype
+    return type_map
+
+
+def enrich_schemas_with_workspace_types(
+    *,
+    schemas_by_tid: dict[str, dict[str, Any]],
+    alias: str,
+    resolve_workspace_id: Callable[[str], int | None],
+    execute_query: Callable[..., dict[str, Any]],
+) -> list[dict[str, str]]:
+    """Backfill missing column ``type``s from the warehouse INFORMATION_SCHEMA.
+
+    Alias / linked-bucket tables carry no per-column datatype metadata in
+    Storage -- the types live on the source table -- so the build heuristic
+    sees empty basetypes and classifies every field as ``dimension``. When a
+    workspace is available, query the warehouse for the real types and fill
+    them in on each ``column_details`` entry in place.
+
+    ``resolve_workspace_id`` maps a storage backend (e.g. ``"bigquery"``) to a
+    workspace ID that can run SQL against it -- either a fixed id supplied by
+    the caller, or one auto-picked per backend. Returning ``None`` means no
+    suitable workspace, and the table is reported as unresolved.
+
+    Returns a list of ``{"table_id", "error"}`` for tables that could not be
+    (fully) resolved. These are non-fatal and surfaced in the build result.
+    """
+    unresolved: list[dict[str, str]] = []
+    for tid, detail in schemas_by_tid.items():
+        cols = detail.get("column_details") or []
+        if not any(not col.get("type") for col in cols):
+            continue  # already fully typed (or no columns) -- nothing to do
+        backend = detail.get("backend", "")
+        sql = build_information_schema_sql(
+            backend,
+            detail.get("bucket_id", ""),
+            detail.get("name", "") or tid.split(".")[-1],
+        )
+        if sql is None:
+            unresolved.append(
+                {
+                    "table_id": tid,
+                    "error": f"type resolution unsupported for backend {backend!r}",
+                }
+            )
+            continue
+        workspace_id = resolve_workspace_id(backend)
+        if workspace_id is None:
+            unresolved.append(
+                {
+                    "table_id": tid,
+                    "error": f"no workspace available to read {backend!r} types",
+                }
+            )
+            continue
+        try:
+            result = execute_query(alias, workspace_id, sql)
+        except (KeboolaApiError, ConfigError) as exc:
+            unresolved.append({"table_id": tid, "error": str(exc)})
+            continue
+        type_map = _parse_information_schema_result(result)
+        if not type_map:
+            unresolved.append(
+                {"table_id": tid, "error": "INFORMATION_SCHEMA returned no column types"}
+            )
+            continue
+        for col in cols:
+            if not col.get("type"):
+                dtype = type_map.get(col.get("name", ""))
+                if dtype:
+                    col["type"] = dtype
+        missing = [col.get("name", "") for col in cols if not col.get("type")]
+        if missing:
+            unresolved.append(
+                {"table_id": tid, "error": f"no type found for column(s): {', '.join(missing)}"}
+            )
+    return unresolved
+
+
 def resolve_model_uuid(
     client: Any,
     model_name_or_uuid: str | None,

src/keboola_agent_cli/services/semantic_layer_service.py

@@ -39,6 +39,9 @@
 from ._semantic_layer_internals import collect_side_from_file
 from ._semantic_layer_internals import default_export_path as _default_export_path
 from ._semantic_layer_internals import diff_one_type as _diff_one_type_helper
+from ._semantic_layer_internals import (
+    enrich_schemas_with_workspace_types as _enrich_schemas_with_workspace_types,
+)
 from ._semantic_layer_internals import fetch_table_schemas as _fetch_table_schemas
 from ._semantic_layer_internals import heuristic_generate_model as _heuristic_generate_helper
 from ._semantic_layer_internals import push_built_model as _push_built_model
@@ -58,6 +61,7 @@
 from .base import BaseService, ClientFactory
 from .encrypt_service import EncryptService
 from .storage_service import StorageService
+from .workspace_service import WorkspaceService
 
 # Constraint name regex enforced by the metastore server.
 CONSTRAINT_NAME_RE = re.compile(r"^[a-z][a-z0-9_]*$")
@@ -1350,6 +1354,26 @@ def promote_model(
     # Phase 7 — build (AI-assisted / heuristic greenfield)
     # ------------------------------------------------------------------
 
+    @staticmethod
+    def _pick_types_workspace(workspace: WorkspaceService, alias: str, backend: str) -> int | None:
+        """Auto-pick a workspace able to read ``backend`` types via the Query
+        Service.
+
+        Prefers a Query-Service-compatible, read-only workspace whose backend
+        matches the table's (so ``INFORMATION_SCHEMA`` runs in the right
+        dialect). Returns its ID, or ``None`` if the project has none.
+        """
+        try:
+            listing = workspace.list_workspaces(aliases=[alias], qs_compatible_only=True)
+        except (KeboolaApiError, ConfigError):
+            return None
+        candidates = listing.get("workspaces", [])
+        for ws in candidates:
+            if not backend or (ws.get("backend", "") or "").lower() == backend.lower():
+                ws_id = ws.get("id")
+                return int(ws_id) if ws_id is not None else None
+        return None
+
     def build_model(
         self,
         alias: str,
@@ -1360,6 +1384,8 @@ def build_model(
         dry_run: bool = False,
         keep_on_failure: bool = False,
         output_path: Path | None = None,
+        types_workspace_id: int | None = None,
+        auto_resolve_types: bool = False,
     ) -> dict[str, Any]:
         """Build (or update) a semantic-layer model from a list of tableIds.
 
@@ -1397,6 +1423,37 @@ def build_model(
         )
         schemas_by_tid, fetch_errors = _fetch_table_schemas(storage, alias, table_ids)
 
+        # Opt-in: backfill column types that Storage does not carry locally
+        # (alias / linked-bucket tables) by querying the warehouse
+        # INFORMATION_SCHEMA via the supplied workspace. Without this, such
+        # tables reach the heuristic with empty basetypes and every field is
+        # classified as `dimension`.
+        type_resolution_errors: list[dict[str, str]] = []
+        if types_workspace_id is not None or auto_resolve_types:
+            workspace = WorkspaceService(
+                config_store=self._config_store, client_factory=self._client_factory
+            )
+            if types_workspace_id is not None:
+                # Explicit workspace: use it for every backend.
+                def resolve_workspace_id(_backend: str) -> int | None:
+                    return types_workspace_id
+            else:
+                # Auto mode (UI / server default): pick a Query-Service-capable
+                # read-only workspace per backend, cached across tables.
+                ws_cache: dict[str, int | None] = {}
+
+                def resolve_workspace_id(backend: str) -> int | None:
+                    if backend not in ws_cache:
+                        ws_cache[backend] = self._pick_types_workspace(workspace, alias, backend)
+                    return ws_cache[backend]
+
+            type_resolution_errors = _enrich_schemas_with_workspace_types(
+                schemas_by_tid=schemas_by_tid,
+                alias=alias,
+                resolve_workspace_id=resolve_workspace_id,
+                execute_query=workspace.execute_query,
+            )
+
         # Generate model JSON (heuristic). The internals helper takes the
         # FQN-derivation and role-classification callables as kwargs so its
         # module stays free of import-time coupling to this module.
@@ -1430,6 +1487,7 @@ def build_model(
             "keep_on_failure": keep_on_failure,
             "fallback_used": "heuristic",  # no AI endpoint shipped yet
             "fetch_errors": fetch_errors,
+            "type_resolution_errors": type_resolution_errors,
             "generated": generated,
             "validation": {"errors": errors, "warnings": warnings},
             "validated": len(errors) == 0,

src/keboola_agent_cli/services/storage_service.py

@@ -595,6 +595,10 @@ def get_table_detail(
             "name": table.get("name", ""),
             "display_name": table.get("displayName", ""),
             "bucket_id": table.get("bucket", {}).get("id", ""),
+            # Storage backend of the owning bucket (e.g. "snowflake",
+            # "bigquery"); needed to pick the right INFORMATION_SCHEMA dialect
+            # when resolving column types for alias / linked tables.
+            "backend": table.get("bucket", {}).get("backend", ""),
             "description": description,
             "columns": columns,
             "column_details": column_details,