[KM-652] feat(query): single-level FK joins in QueryIR (T4)

The e2e "revenue by category" failed because the measure (order_items.line_total)
and the dimension (products.category) live in different tables and QueryIR was
single-table — the planner couldn't combine them and picked wrong tables.

- QueryIR gains `joins: [{target_table_id, left_column_id, right_column_id, type}]`.
select/group_by/filters/order_by may now reference columns from joined tables.
- SqlCompiler resolves each column_id to its owning table and emits table-qualified
refs + INNER/LEFT JOIN. Output is byte-identical to before when joins=[].
- IRValidator: joins are database-only and each column pair must match a foreign
key declared in the catalog; joined columns are added to the resolvable set.
- PandasCompiler rejects joins (cross-file merge deferred) — DB sources only in v1.
- retrieve_data spec + planner.md + query_planner.md teach the join pattern with a
revenue-by-category few-shot; prefer existing measure cols + single table when possible.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

src/agents/planner/registry.py

@@ -43,13 +43,20 @@ _DATA_ACCESS_SPEC_BODIES: tuple[ToolSpec, ...] = (
         input_schema={"required": ["ir"], "properties": {"ir": {"type": "object"}}},
         output_kind="table",
         description=(
-            "Run one validated, single-table query against a structured source (DB "
-            "schema or tabular file) and return rows. The `ir` argument is an inline "
-            "QueryIR (the JSON intent: source_id, table_id, select, filters, group_by, "
-            "order_by, limit) — never SQL. This is the data-access entry point: use it "
-            "to select, filter, and pull the rows the analytics (`analyze_*`) tools "
-            "then consume. It also does simple built-in aggregation the IR can express "
-            "(count/sum/avg/min/max/count_distinct). Do NOT use it for richer statistics "
+            "Run one validated query against a structured source and return rows. The "
+            "`ir` argument is an inline QueryIR (the JSON intent: source_id, table_id, "
+            "joins, select, filters, group_by, order_by, limit) — never SQL. This is the "
+            "data-access entry point: use it to select, filter, and pull the rows the "
+            "analytics (`analyze_*`) tools then consume. It also does simple built-in "
+            "aggregation the IR can express (count/sum/avg/min/max/count_distinct). "
+            "JOINS (database sources only): to group a measure in one table by a "
+            "dimension in a RELATED table, add a `joins` entry "
+            "({target_table_id, left_column_id, right_column_id}) along a declared "
+            "foreign key — e.g. sum order_items.line_total grouped by products.category "
+            "via order_items.product_id = products.id. Prefer an existing measure column "
+            "(e.g. line_total) over recomputing, and a single table when the measure and "
+            "dimension already live together. Joins are NOT supported on tabular/file "
+            "sources yet. Do NOT use this for richer statistics "
             "(median/percentile/mode/stddev/skew → analyze_descriptive), trends "
             "(analyze_trend), correlation, segmentation, or share-of-total; and do NOT "
             "use it to read documents (use retrieve_knowledge)."

src/config/prompts/planner.md

@@ -39,6 +39,17 @@ only a `TaskList` object that conforms to the provided schema.
   profiling — run `retrieve_data` to fetch the rows, then pass its output to
   the matching composite `analyze_*` tool via a `"${t<id>}"` `data` argument
   (referencing the upstream result's column aliases).
+- **Measure by a dimension in another table (joins).** When the number you are
+  aggregating and the grouping dimension live in DIFFERENT tables of the same
+  database source, add a `joins` entry to the `retrieve_data` IR along a foreign
+  key declared in the catalog — do NOT pick a table that lacks the measure, and do
+  NOT try to "combine" unrelated tables. Example — "revenue by category": the
+  measure `order_items.line_total` joined to `products` on
+  `order_items.product_id = products.id`, grouped by `products.category`. Prefer an
+  existing measure column over recomputing; use a single table (no join) when the
+  measure and dimension already live together (e.g. "revenue by region" from
+  `orders.region` + `orders.total_amount`). Joins are database-only — not available
+  for tabular/file sources.
 - **Mixing structured + unstructured.** If qualitative context helps, add a
   `retrieve_knowledge` task against an unstructured source listed in the catalog.
 - **CRISP-DM stages.** Tag each task with the stage it serves:

src/config/prompts/query_planner.md

@@ -15,7 +15,10 @@ A `QueryIR` object:
 {
   "ir_version": "1.0",
   "source_id":  "...",            // pick from catalog
-  "table_id":   "...",            // pick from chosen source
+  "table_id":   "...",            // base table
+  "joins": [                      // optional; DATABASE sources only; FK-backed
+    {"target_table_id": "...", "left_column_id": "...", "right_column_id": "...", "type": "inner"}
+  ],
   "select": [
     {"kind": "column", "column_id": "...", "alias": "..."},
     {"kind": "agg",    "fn": "count|count_distinct|sum|avg|min|max",
@@ -36,7 +39,7 @@ A `QueryIR` object:
 ## Hard constraints (a violation makes the IR invalid)
 
 1. `source_id`, `table_id`, `column_id` must come **verbatim** from the catalog. Never invent IDs or copy table/column **names** in their place.
-2. **Single-table only in v1.** Pick the table whose columns best answer the question. If the question genuinely needs a join, pick the table that yields the most useful answer alone and the user can refine.
+2. **Joins (database sources only).** Prefer a single table when the measure and the grouping dimension already live together. When the number you aggregate and the dimension are in DIFFERENT tables of the same DB source, add a `joins` entry along a **foreign key declared in the catalog** — `left_column_id` must already be in the query (base table or an earlier join), `right_column_id` is in `target_table_id`. Do NOT pick a table missing the measure, and do NOT join unrelated tables. Tabular/file sources are single-table only — no joins.
 3. Use only listed operators / aggregates. No window functions, no `CASE WHEN`, no subqueries — those are not part of v1.
 4. `value_type` must be compatible with the column's `data_type`:
    - `int` column ↔ value_type ∈ {int, decimal}
@@ -115,6 +118,28 @@ Output:
 }
 ```
 
+Question: "Which product category generates the most total revenue?"
+(catalog DB source `src_prod_db` also has: `order_items` [id=t_order_items] with `line_total` [decimal, id=c_oi_line_total] and `product_id` [int, id=c_oi_product_id, FK → products.id]; `products` [id=t_products] with `category` [string, id=c_products_category] and `id` [int, id=c_products_id])
+Output:
+```json
+{
+  "ir_version": "1.0",
+  "source_id": "src_prod_db",
+  "table_id": "t_order_items",
+  "joins": [
+    {"target_table_id": "t_products", "left_column_id": "c_oi_product_id", "right_column_id": "c_products_id", "type": "inner"}
+  ],
+  "select": [
+    {"kind": "column", "column_id": "c_products_category"},
+    {"kind": "agg", "fn": "sum", "column_id": "c_oi_line_total", "alias": "total_revenue"}
+  ],
+  "filters": [],
+  "group_by": ["c_products_category"],
+  "order_by": [{"column_id": "total_revenue", "dir": "desc"}],
+  "limit": 100
+}
+```
+
 Catalog excerpt (tabular source — XLSX sheet):
 
 ```

src/query/compiler/pandas.py

@@ -48,6 +48,13 @@ class PandasCompiler(BaseCompiler):
         self._catalog = catalog
 
     def compile(self, ir: QueryIR) -> CompiledPandas:
+        if ir.joins:
+            # Tabular joins need the executor to load + merge multiple parquet blobs
+            # (deferred). The validator already rejects joins on tabular sources;
+            # this is the defensive backstop. Joins are DB-only in v1 (KM-652 T4).
+            raise PandasCompilerError(
+                "joins are not supported on tabular sources yet (database sources only)"
+            )
         _, table, cols_by_id = self._lookup(ir)
         output_columns = _output_column_names(ir.select, cols_by_id)
 

src/query/compiler/sql.py

@@ -9,9 +9,10 @@ The output `CompiledSql.sql` uses SQLAlchemy-style named placeholders
 (`:p_0, :p_1, ...`) so it can be executed via `text(sql)` with a params
 dict on a sync SQLAlchemy engine.
 
-v1 supports the Postgres dialect only. Supabase reuses the same compiler
-output (Supabase = Postgres). MySQL / BigQuery / Snowflake compilers will
-be separate classes that implement `BaseCompiler`.
+Joins (KM-652 T4): a column_id resolves to its owning table across the base
+table + any joined tables, so every reference is emitted table-qualified
+(`"table"."col"`). With `joins=[]` the output is identical to the single-table
+form. v1 supports the Postgres dialect only (Supabase = Postgres).
 """
 
 from __future__ import annotations
@@ -36,6 +37,9 @@ from .base import BaseCompiler
 # `row_cap` and flags truncation.
 MAX_RESULT_ROWS = 10_000
 
+# A resolved column reference: the table that owns it + the column itself.
+ColRef = tuple[Table, Column]
+
 
 @dataclass
 class CompiledSql:
@@ -65,17 +69,15 @@ class SqlCompiler(BaseCompiler):
         self._dialect = dialect
 
     def compile(self, ir: QueryIR) -> CompiledSql:
-        _, table, cols_by_id = self._lookup(ir)
+        base_table, cols_by_id = self._lookup(ir)
         params: dict[str, Any] = {}
         param_seq = [0]
 
-        select_clause, select_aliases = self._build_select(ir.select, table, cols_by_id)
-        from_clause = self._build_from(table)
-        where_clause = self._build_where(ir.filters, table, cols_by_id, params, param_seq)
-        groupby_clause = self._build_groupby(ir.group_by, table, cols_by_id)
-        orderby_clause = self._build_orderby(
-            ir.order_by, table, cols_by_id, select_aliases
-        )
+        select_clause, select_aliases = self._build_select(ir.select, cols_by_id)
+        from_clause = self._build_from(base_table, ir, cols_by_id)
+        where_clause = self._build_where(ir.filters, cols_by_id, params, param_seq)
+        groupby_clause = self._build_groupby(ir.group_by, cols_by_id)
+        orderby_clause = self._build_orderby(ir.order_by, cols_by_id, select_aliases)
         limit_clause, row_cap = self._build_limit(ir.limit)
 
         parts: list[str] = [select_clause, from_clause]
@@ -86,23 +88,33 @@ class SqlCompiler(BaseCompiler):
         return CompiledSql(sql=" ".join(parts), params=params, row_cap=row_cap)
 
     # ------------------------------------------------------------------
-    # Catalog lookup
+    # Catalog lookup — column_id -> (owning Table, Column) across base + joins
     # ------------------------------------------------------------------
 
-    def _lookup(self, ir: QueryIR) -> tuple[Source, Table, dict[str, Column]]:
+    def _lookup(self, ir: QueryIR) -> tuple[Table, dict[str, ColRef]]:
         source = next(
             (s for s in self._catalog.sources if s.source_id == ir.source_id), None
         )
         if source is None:
             raise SqlCompilerError(f"source_id {ir.source_id!r} not in catalog")
-        table = next(
-            (t for t in source.tables if t.table_id == ir.table_id), None
-        )
+        base_table = self._find_table(source, ir.table_id)
+        cols_by_id: dict[str, ColRef] = {
+            c.column_id: (base_table, c) for c in base_table.columns
+        }
+        for j in ir.joins:
+            target = self._find_table(source, j.target_table_id)
+            for c in target.columns:
+                cols_by_id[c.column_id] = (target, c)
+        return base_table, cols_by_id
+
+    @staticmethod
+    def _find_table(source: Source, table_id: str) -> Table:
+        table = next((t for t in source.tables if t.table_id == table_id), None)
         if table is None:
             raise SqlCompilerError(
-                f"table_id {ir.table_id!r} not in source {ir.source_id!r}"
+                f"table_id {table_id!r} not in source {source.source_id!r}"
             )
-        return source, table, {c.column_id: c for c in table.columns}
+        return table
 
     # ------------------------------------------------------------------
     # Identifier quoting
@@ -113,7 +125,8 @@ class SqlCompiler(BaseCompiler):
         """Postgres-style double-quoted identifier with embedded-quote escape."""
         return '"' + name.replace('"', '""') + '"'
 
-    def _qcol(self, table: Table, col: Column) -> str:
+    def _qcol(self, ref: ColRef) -> str:
+        table, col = ref
         return f"{self._qident(table.name)}.{self._qident(col.name)}"
 
     # ------------------------------------------------------------------
@@ -121,17 +134,14 @@ class SqlCompiler(BaseCompiler):
     # ------------------------------------------------------------------
 
     def _build_select(
-        self,
-        items: list[SelectItem],
-        table: Table,
-        cols_by_id: dict[str, Column],
+        self, items: list[SelectItem], cols_by_id: dict[str, ColRef]
     ) -> tuple[str, set[str]]:
         if not items:
             raise SqlCompilerError("select clause cannot be empty")
         parts: list[str] = []
         aliases: set[str] = set()
         for i, item in enumerate(items):
-            expr, alias = self._select_item(item, table, cols_by_id, i)
+            expr, alias = self._select_item(item, cols_by_id, i)
             if alias:
                 parts.append(f"{expr} AS {self._qident(alias)}")
                 aliases.add(alias)
@@ -140,35 +150,27 @@ class SqlCompiler(BaseCompiler):
         return "SELECT " + ", ".join(parts), aliases
 
     def _select_item(
-        self,
-        item: SelectItem,
-        table: Table,
-        cols_by_id: dict[str, Column],
-        index: int,
+        self, item: SelectItem, cols_by_id: dict[str, ColRef], index: int
     ) -> tuple[str, str | None]:
         if isinstance(item, ColumnSelect):
-            col = self._require_col(cols_by_id, item.column_id, f"select[{index}]")
-            return self._qcol(table, col), item.alias
+            ref = self._require_col(cols_by_id, item.column_id, f"select[{index}]")
+            return self._qcol(ref), item.alias
         if not isinstance(item, AggSelect):
             raise SqlCompilerError(
                 f"select[{index}]: unknown SelectItem kind {type(item).__name__}"
             )
-        return self._compile_agg(item, table, cols_by_id, index), item.alias
+        return self._compile_agg(item, cols_by_id, index), item.alias
 
     def _compile_agg(
-        self,
-        item: AggSelect,
-        table: Table,
-        cols_by_id: dict[str, Column],
-        index: int,
+        self, item: AggSelect, cols_by_id: dict[str, ColRef], index: int
     ) -> str:
         if item.fn == "count_distinct":
             if item.column_id is None:
                 raise SqlCompilerError(
                     f"select[{index}].fn=count_distinct requires column_id"
                 )
-            col = self._require_col(cols_by_id, item.column_id, f"select[{index}]")
-            return f"COUNT(DISTINCT {self._qcol(table, col)})"
+            ref = self._require_col(cols_by_id, item.column_id, f"select[{index}]")
+            return f"COUNT(DISTINCT {self._qcol(ref)})"
         if item.column_id is None:
             if item.fn != "count":
                 raise SqlCompilerError(
@@ -176,24 +178,35 @@ class SqlCompiler(BaseCompiler):
                     "(only 'count' may omit it for COUNT(*))"
                 )
             return "COUNT(*)"
-        col = self._require_col(cols_by_id, item.column_id, f"select[{index}]")
-        return f"{item.fn.upper()}({self._qcol(table, col)})"
+        ref = self._require_col(cols_by_id, item.column_id, f"select[{index}]")
+        return f"{item.fn.upper()}({self._qcol(ref)})"
 
-    def _build_from(self, table: Table) -> str:
-        return f"FROM {self._qident(table.name)}"
+    def _build_from(
+        self, base_table: Table, ir: QueryIR, cols_by_id: dict[str, ColRef]
+    ) -> str:
+        sql = f"FROM {self._qident(base_table.name)}"
+        for i, j in enumerate(ir.joins):
+            left = self._require_col(cols_by_id, j.left_column_id, f"joins[{i}].left")
+            right = self._require_col(cols_by_id, j.right_column_id, f"joins[{i}].right")
+            target_table = right[0]
+            keyword = "INNER JOIN" if j.type == "inner" else "LEFT JOIN"
+            sql += (
+                f" {keyword} {self._qident(target_table.name)} "
+                f"ON {self._qcol(left)} = {self._qcol(right)}"
+            )
+        return sql
 
     def _build_where(
         self,
         filters: list[FilterClause],
-        table: Table,
-        cols_by_id: dict[str, Column],
+        cols_by_id: dict[str, ColRef],
         params: dict[str, Any],
         param_seq: list[int],
     ) -> str:
         if not filters:
             return ""
         parts = [
-            self._compile_filter(f, table, cols_by_id, params, param_seq, index=i)
+            self._compile_filter(f, cols_by_id, params, param_seq, index=i)
             for i, f in enumerate(filters)
         ]
         return "WHERE " + " AND ".join(parts)
@@ -201,14 +214,13 @@ class SqlCompiler(BaseCompiler):
     def _compile_filter(
         self,
         f: FilterClause,
-        table: Table,
-        cols_by_id: dict[str, Column],
+        cols_by_id: dict[str, ColRef],
         params: dict[str, Any],
         param_seq: list[int],
         index: int,
     ) -> str:
-        col = self._require_col(cols_by_id, f.column_id, f"filters[{index}]")
-        col_ref = self._qcol(table, col)
+        ref = self._require_col(cols_by_id, f.column_id, f"filters[{index}]")
+        col_ref = self._qcol(ref)
         op = f.op
 
         if op == "is_null":
@@ -248,15 +260,12 @@ class SqlCompiler(BaseCompiler):
         raise SqlCompilerError(f"filters[{index}]: unhandled op {op!r}")
 
     def _build_groupby(
-        self,
-        group_by: list[str],
-        table: Table,
-        cols_by_id: dict[str, Column],
+        self, group_by: list[str], cols_by_id: dict[str, ColRef]
     ) -> str:
         if not group_by:
             return ""
         parts = [
-            self._qcol(table, self._require_col(cols_by_id, col_id, f"group_by[{i}]"))
+            self._qcol(self._require_col(cols_by_id, col_id, f"group_by[{i}]"))
             for i, col_id in enumerate(group_by)
         ]
         return "GROUP BY " + ", ".join(parts)
@@ -264,8 +273,7 @@ class SqlCompiler(BaseCompiler):
     def _build_orderby(
         self,
         order_by: list[OrderByClause],
-        table: Table,
-        cols_by_id: dict[str, Column],
+        cols_by_id: dict[str, ColRef],
         select_aliases: set[str],
     ) -> str:
         if not order_by:
@@ -273,12 +281,12 @@ class SqlCompiler(BaseCompiler):
         parts: list[str] = []
         for i, ob in enumerate(order_by):
             if ob.column_id in cols_by_id:
-                ref = self._qcol(table, cols_by_id[ob.column_id])
+                ref = self._qcol(cols_by_id[ob.column_id])
             elif ob.column_id in select_aliases:
                 ref = self._qident(ob.column_id)
             else:
                 raise SqlCompilerError(
-                    f"order_by[{i}].column_id: {ob.column_id!r} not in table "
+                    f"order_by[{i}].column_id: {ob.column_id!r} not in query "
                     "columns or select aliases"
                 )
             parts.append(f"{ref} {ob.dir.upper()}")
@@ -312,9 +320,9 @@ class SqlCompiler(BaseCompiler):
 
     @staticmethod
     def _require_col(
-        cols_by_id: dict[str, Column], col_id: str, where: str
-    ) -> Column:
-        col = cols_by_id.get(col_id)
-        if col is None:
-            raise SqlCompilerError(f"{where}.column_id: {col_id!r} not in table")
-        return col
+        cols_by_id: dict[str, ColRef], col_id: str, where: str
+    ) -> ColRef:
+        ref = cols_by_id.get(col_id)
+        if ref is None:
+            raise SqlCompilerError(f"{where}.column_id: {col_id!r} not in query tables")
+        return ref

src/query/ir/models.py

@@ -2,8 +2,14 @@
 
 See ARCHITECTURE.md §7 for the schema.
 
-Initial scope: single-table; filter, group_by, agg, order_by, limit.
-Joins, having, offset, boolean tree filters are deferred to later versions.
+Scope: filter, group_by, agg, order_by, limit, plus a single-level `joins` to
+related tables in the SAME source (KM-652 T4). having, offset, and boolean tree
+filters are still deferred. Joins are supported on database (schema) sources only;
+the validator rejects them on tabular sources (cross-file merge is a later step).
+
+With joins, `select` / `group_by` / `filters` / `order_by` may reference a
+`column_id` from the base table OR any joined table (column_ids are globally
+unique), so a measure in one table can be grouped by a dimension in a related one.
 """
 
 from typing import Any, Literal
@@ -18,6 +24,21 @@ FilterOp = Literal[
 AggFn = Literal["count", "count_distinct", "sum", "avg", "min", "max"]
 ValueType = Literal["int", "decimal", "string", "datetime", "date", "bool"]
 SortDir = Literal["asc", "desc"]
+JoinType = Literal["inner", "left"]
+
+
+class Join(BaseModel):
+    """A single equi-join to another table in the same source.
+
+    `left_column_id` is a column already in the query (base table or an earlier
+    join); `right_column_id` is a column in `target_table_id`. The validator
+    requires the pair to match a foreign key declared in the catalog.
+    """
+
+    target_table_id: str
+    left_column_id: str
+    right_column_id: str
+    type: JoinType = "inner"
 
 
 class ColumnSelect(BaseModel):
@@ -52,6 +73,7 @@ class QueryIR(BaseModel):
     ir_version: str = "1.0"
     source_id: str
     table_id: str
+    joins: list[Join] = Field(default_factory=list)
     select: list[SelectItem]
     filters: list[FilterClause] = Field(default_factory=list)
     group_by: list[str] = Field(default_factory=list)

src/query/ir/validator.py

@@ -35,8 +35,12 @@ class IRValidator:
 
     def validate(self, ir: QueryIR, catalog: Catalog) -> None:
         source = self._find_source(catalog, ir.source_id)
-        table = self._find_table(source, ir.table_id)
-        columns_by_id: dict[str, Column] = {c.column_id: c for c in table.columns}
+        base_table = self._find_table(source, ir.table_id)
+        # Columns available to select/filter/group/order — grows as joins are added.
+        columns_by_id: dict[str, Column] = {c.column_id: c for c in base_table.columns}
+
+        if ir.joins:
+            self._validate_joins(ir, source, columns_by_id)
 
         select_aliases: set[str] = set()
         for i, item in enumerate(ir.select):
@@ -96,6 +100,51 @@ class IRValidator:
                     f"limit {ir.limit} exceeds hard cap {LIMIT_HARD_CAP}"
                 )
 
+    def _validate_joins(
+        self, ir: QueryIR, source: Source, columns_by_id: dict[str, Column]
+    ) -> None:
+        """Validate joins and grow `columns_by_id` with each joined table's columns.
+
+        Joins are DB-only in v1; each join's column pair must be a declared foreign
+        key, the left column must already be in the query, the right in the target.
+        """
+        if source.source_type != "schema":
+            raise IRValidationError(
+                f"joins are only supported on database sources in v1; source "
+                f"{ir.source_id!r} is {source.source_type!r}"
+            )
+        fk_pairs = self._fk_pairs(source)
+        for i, j in enumerate(ir.joins):
+            where = f"joins[{i}]"
+            target = self._find_table(source, j.target_table_id)
+            target_cols = {c.column_id: c for c in target.columns}
+            if j.left_column_id not in columns_by_id:
+                raise IRValidationError(
+                    f"{where}.left_column_id {j.left_column_id!r} is not in the query "
+                    "so far (base table or an earlier join)"
+                )
+            if j.right_column_id not in target_cols:
+                raise IRValidationError(
+                    f"{where}.right_column_id {j.right_column_id!r} not in target table "
+                    f"{j.target_table_id!r}"
+                )
+            if frozenset((j.left_column_id, j.right_column_id)) not in fk_pairs:
+                raise IRValidationError(
+                    f"{where}: ({j.left_column_id!r}, {j.right_column_id!r}) is not a "
+                    f"declared foreign key in source {ir.source_id!r} — only FK-backed "
+                    "joins are allowed"
+                )
+            columns_by_id.update(target_cols)
+
+    @staticmethod
+    def _fk_pairs(source: Source) -> set[frozenset[str]]:
+        """All declared FK column pairs in the source, as unordered {col, col} sets."""
+        pairs: set[frozenset[str]] = set()
+        for t in source.tables:
+            for fk in t.foreign_keys:
+                pairs.add(frozenset((fk.column_id, fk.target_column_id)))
+        return pairs
+
     @staticmethod
     def _find_source(catalog: Catalog, source_id: str) -> Source:
         for s in catalog.sources: