feat(catalog): add schema command to inspect entry schemas without execution

Adds `xorq catalog schema <name-or-alias>` command that prints Schema In
(for unbound/partial expressions) and Schema Out (for all entries) by reading
expr.yaml directly from the tgz — no backend connection required.

- CatalogEntry.schema_out / .schema_in cached properties parse expr.yaml
- CatalogEntry._parse_schema_* staticmethods handle dtype dict → str conversion
- Catalog.get_entry() resolves both entry names and aliases
- --json flag for machine-readable output
- Shell completion covers both names and aliases

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: mesejo

python/xorq/catalog/catalog.py

@@ -160,6 +160,14 @@ def get_catalog_entry(self, name):
         catalog_entry = CatalogEntry(name, self)
         return catalog_entry
 
+    def get_entry(self, name_or_alias):
+        """Resolve an entry name or alias to a CatalogEntry."""
+        if name_or_alias in self.list():
+            return CatalogEntry(name_or_alias, self)
+        if name_or_alias in self.list_aliases():
+            return CatalogAlias.from_name(name_or_alias, self).catalog_entry
+        raise KeyError(f"No entry or alias named {name_or_alias!r}")
+
     def get_tgz(self, name, dir_path=None):
         catalog_entry = self.get_catalog_entry(name)
         return catalog_entry.get(dir_path)
@@ -470,6 +478,104 @@ def kind(self) -> str:
             return default_value
         return data.get("kind", default_value)
 
+    @cached_property
+    def schema_out(self) -> dict | None:
+        """Output schema as {column_name: dtype_str}, read directly from expr.yaml."""
+        data = self._read_tgz_yaml(DumpFiles.expr)
+        return CatalogEntry._parse_schema_out(data)
+
+    @cached_property
+    def schema_in(self) -> dict | None:
+        """For unbound expressions, the expected input schema as {column_name: dtype_str}."""
+        if self.kind != str(ExprKind.UnboundExpr):
+            return None
+        data = self._read_tgz_yaml(DumpFiles.expr)
+        return CatalogEntry._parse_schema_in(data)
+
+    @staticmethod
+    def _resolve_schema_ref(ref_val) -> str | None:
+        """Extract the schema key string from a schema_ref value.
+
+        In expr.yaml, schema_ref is stored as {'schema_ref': 'schema_<hash>'}.
+        """
+        if isinstance(ref_val, dict):
+            return ref_val.get("schema_ref")
+        if isinstance(ref_val, str):
+            return ref_val
+        return None
+
+    @staticmethod
+    def _dtype_dict_to_str(d) -> str:
+        """Convert a serialized DataType dict from expr.yaml to a human-readable string."""
+        import xorq.vendor.ibis.expr.datatypes as dt  # noqa: PLC0415
+
+        if not isinstance(d, dict) or d.get("op") != "DataType":
+            return str(d)
+        typ_cls = getattr(dt, d["type"], None)
+        if typ_cls is None:
+            return d["type"].lower()
+        kwargs = {
+            k: CatalogEntry._dtype_dict_to_str(v) if isinstance(v, dict) else v
+            for k, v in d.items()
+            if k not in ("op", "type")
+        }
+
+        # Re-parse nested dtype strings back to dt objects for the constructor
+        def _parse_val(k, v):
+            try:
+                arg_annotation = typ_cls.__dataclass_fields__.get(k)
+                if arg_annotation and isinstance(v, str):
+                    return dt.dtype(v)
+            except Exception:
+                pass
+            return v
+
+        kwargs = {k: _parse_val(k, v) for k, v in kwargs.items()}
+        try:
+            return str(typ_cls(**kwargs))
+        except Exception:
+            return d["type"].lower()
+
+    @staticmethod
+    def _schema_dict_to_str_dict(schema_dict) -> dict[str, str]:
+        return {
+            col: CatalogEntry._dtype_dict_to_str(dtype_val)
+            for col, dtype_val in schema_dict.items()
+        }
+
+    @staticmethod
+    def _parse_schema_out(data) -> dict | None:
+        if not isinstance(data, dict):
+            return None
+        schemas = data.get("definitions", {}).get("schemas", {})
+        schema_ref = CatalogEntry._resolve_schema_ref(
+            data.get("expression", {}).get("schema_ref")
+        )
+        if not schema_ref:
+            return None
+        schema_dict = schemas.get(schema_ref)
+        if not isinstance(schema_dict, dict):
+            return None
+        return CatalogEntry._schema_dict_to_str_dict(schema_dict)
+
+    @staticmethod
+    def _parse_schema_in(data) -> dict | None:
+        if not isinstance(data, dict):
+            return None
+        definitions = data.get("definitions", {})
+        nodes = definitions.get("nodes", {})
+        schemas = definitions.get("schemas", {})
+        for node_def in nodes.values():
+            if isinstance(node_def, dict) and node_def.get("op") == "UnboundTable":
+                schema_ref = CatalogEntry._resolve_schema_ref(
+                    node_def.get("schema_ref")
+                )
+                if schema_ref:
+                    schema_dict = schemas.get(schema_ref)
+                    if isinstance(schema_dict, dict):
+                        return CatalogEntry._schema_dict_to_str_dict(schema_dict)
+        return None
+
     @cached_property
     def backends(self) -> tuple:
         data = self._read_tgz_yaml(DumpFiles.profiles)

python/xorq/catalog/cli.py

@@ -351,6 +351,63 @@ def clone(url, dest_name, dest_path):
         click.echo(f"Cloned to {catalog.repo_path}")
 
 
+def _complete_entry_or_alias_names(ctx, param, incomplete):
+    from click.shell_completion import CompletionItem
+
+    try:
+        catalog = _make_catalog_for_completion(ctx)
+        names = set(catalog.list()) | set(catalog.list_aliases())
+        return [CompletionItem(n) for n in sorted(names) if n.startswith(incomplete)]
+    except Exception:
+        return []
+
+
+@cli.command()
+@click.argument("name", shell_complete=_complete_entry_or_alias_names)
+@click.option("--json", "as_json", is_flag=True, default=False, help="Output as JSON.")
+@click.pass_context
+def schema(ctx, name, as_json):
+    """Show schema of a catalog entry (name or alias)."""
+    import json as json_mod
+
+    from xorq.ibis_yaml.compiler import ExprKind
+
+    with click_context_catalog(ctx):
+        catalog = ctx.obj.make_catalog(init=False)
+        try:
+            entry = catalog.get_entry(name)
+        except KeyError as e:
+            raise click.ClickException(str(e)) from e
+
+        kind = entry.kind
+        schema_out = entry.schema_out
+        schema_in = entry.schema_in
+
+        if as_json:
+            result = {"name": entry.name, "kind": kind, "schema_out": schema_out}
+            if schema_in is not None:
+                result["schema_in"] = schema_in
+            click.echo(json_mod.dumps(result, indent=2))
+            return
+
+        is_unbound = kind == str(ExprKind.UnboundExpr)
+        type_label = "Partial (unbound)" if is_unbound else "Source (bound)"
+        click.echo(f"Type: {type_label}\n")
+
+        if schema_in:
+            click.echo("Schema In:")
+            for col, dtype in schema_in.items():
+                click.echo(f"  {col:<24} {dtype}")
+            click.echo()
+
+        if schema_out:
+            click.echo("Schema Out:")
+            for col, dtype in schema_out.items():
+                click.echo(f"  {col:<24} {dtype}")
+        else:
+            click.echo("Schema Out: (unavailable)")
+
+
 @cli.command()
 @click.pass_context
 def check(ctx):

python/xorq/catalog/tests/test_catalog.py

@@ -349,3 +349,52 @@ def test_extract_kind_partial(catalog):
     expr = t.filter(t.a > 0)
     entry = catalog.add(expr)
     assert entry.kind == ExprKind.UnboundExpr
+
+
+def test_schema_out_bound(catalog):
+    expr = xo.memtable({"col_a": [1, 2], "col_b": ["x", "y"]})
+    entry = catalog.add(expr)
+    assert entry.schema_out == {"col_a": "int64", "col_b": "string"}
+
+
+def test_schema_in_none_for_bound(catalog):
+    expr = xo.memtable({"a": [1, 2, 3]})
+    entry = catalog.add(expr)
+    assert entry.schema_in is None
+
+
+def test_schema_out_unbound(catalog):
+    t = xo.table(schema={"amount": "float64", "currency": "string"})
+    expr = t.mutate(amount_usd=t.amount * 1.2)
+    entry = catalog.add(expr)
+    assert entry.schema_out == {
+        "amount": "float64",
+        "currency": "string",
+        "amount_usd": "float64",
+    }
+
+
+def test_schema_in_unbound(catalog):
+    t = xo.table(schema={"amount": "float64", "currency": "string"})
+    expr = t.filter(t.amount > 0)
+    entry = catalog.add(expr)
+    assert entry.schema_in == {"amount": "float64", "currency": "string"}
+
+
+def test_get_entry_by_name(catalog):
+    expr = xo.memtable({"x": [1]})
+    entry = catalog.add(expr)
+    resolved = catalog.get_entry(entry.name)
+    assert resolved.name == entry.name
+
+
+def test_get_entry_by_alias(catalog):
+    expr = xo.memtable({"x": [1]})
+    entry = catalog.add(expr, aliases=("my-alias",))
+    resolved = catalog.get_entry("my-alias")
+    assert resolved.name == entry.name
+
+
+def test_get_entry_unknown_raises(catalog):
+    with pytest.raises(KeyError, match="no-such"):
+        catalog.get_entry("no-such")