feat(sql-plan): port SQLPlanGenerator consumers of the production model fields

docs/api/generators.md

@@ -7,3 +7,7 @@ Generate schemas in multiple formats from UMF specifications.
 ::: tablespec.schemas.generators.generate_pyspark_schema
 
 ::: tablespec.schemas.generators.generate_json_schema
+
+::: tablespec.schemas.sql_generator.generate_sql_plan
+
+::: tablespec.schemas.sql_generator.SQLPlanGenerator

docs/guide/excel.md

@@ -59,6 +59,7 @@ column-level fields written on the column's first row:
 | Order By | Window `ORDER BY` columns — JSON list in one cell |
 | Select Columns | Extra columns to carry through aggregation — JSON list |
 | Join Via | Multi-hop join through a lookup table — JSON object |
+| Union Value | Per-UNION-branch literal for synthetic columns (native type preserved: `'daily'`, `TRUE`, `1`) |
 | Reason | Why this source/priority was chosen |
 | Derivation Strategy | Top-level strategy: `primary_key`, `base_column`, `max_across_sources` |
 | Survivorship Strategy | Survivorship method (e.g. `highest_priority`, `most_recent`) |

docs/guide/happy-path.md

@@ -205,6 +205,11 @@ plan_sql = generate_sql_plan(
 The same UMF set also feeds the dbt DAG and LDP emitters, so the derived table
 stays aligned with the rest of the pipeline.
 
+Beyond simple joins, the plan generator supports base/final filters, UNION
+branch generation across source tables (with per-branch filters, literals, and
+window dedup), and alternative join paths — see
+[SQL Plan Generation](sql-plans.md).
+
 ## 6. Generate Spark, LDP, and dbt pipeline artifacts
 
 `compile_umfs(...)` is the current orchestration seam. It persists the ingest

docs/guide/sql-plans.md

@@ -0,0 +1,141 @@
+# SQL Plan Generation
+
+`generate_sql_plan` (and the underlying `SQLPlanGenerator`) compiles a
+*generated* table's UMF — its derivation candidates, relationships, and
+metadata — into an executable SQL plan: a sequence of
+`CREATE OR REPLACE TEMPORARY VIEW` statements (`mode="views"`) or a single
+`WITH ... SELECT` statement (`mode="cte"`, the form dbt and LDP gold models
+consume). The emitted SQL is engine-agnostic and executes verbatim on both
+DuckDB and Spark; the conformance corpus pins this on every gold path.
+
+```python
+from tablespec import generate_sql_plan
+
+sql = generate_sql_plan(target_umf, related_umfs, mode="cte")
+```
+
+See [Happy Path §5](happy-path.md) for the basic derived-table flow. This page
+documents the plan-shaping metadata controls.
+
+## Base-view strategies
+
+`metadata.base_table_strategy` selects how the plan's base view
+(`disposition_base`) is built:
+
+| Strategy | Base view |
+|----------|-----------|
+| *(unset)* | `SELECT <required columns> FROM base_table` |
+| `unpivot` | UNPIVOT `unpivot_columns` into rows (optionally deduped, see below) |
+| `union_sources` | Key-only universe: UNION of each source table's join key |
+| `union_branches` | One full SELECT branch per source table, combined with UNION ALL / UNION |
+
+### `union_branches`
+
+The base table plus each table in `union_base_tables` (falling back to
+`source_tables`) becomes one UNION branch. Unlike `union_sources` (which only
+unions *keys*), every branch projects the **target column set**, each column
+mapped through that source table's own derivation candidates:
+
+- a candidate with `union_value` emits `CAST(<literal> AS <type>)` — a
+  per-branch constant, typically a source discriminator;
+- otherwise the branch table's lowest-priority candidate supplies the
+  `expression` or `column`;
+- a column with no candidate for the branch emits `CAST(NULL AS <type>)`, so
+  the UNION stays column-aligned when sources have different columns.
+
+```yaml
+metadata:
+  base_table: bronze_inventory_detail
+  base_table_strategy: union_branches
+  union_base_tables: [bronze_halo_daily_inventory]
+  union_type: union_all          # or 'union' to dedupe exact rows
+  dedup_strategy: latest         # per-branch window dedup (see below)
+```
+
+**Per-branch filtering.** A branch's WHERE clause comes from the single
+distinct `row_filter` among that table's candidates — this is how generation
+cutovers are expressed (legacy feed `file_date < DATE '2026-07-20'`, daily feed
+`>=`). Candidates of one branch carrying *different* `row_filter` values is an
+error. `base_table_filter` additionally applies to the base branch (ANDed with
+its row_filter).
+
+**Per-branch dedup.** With `dedup_strategy: latest` and a candidate `order_by`,
+each branch is deduplicated before the union:
+
+```sql
+ROW_NUMBER() OVER (
+  PARTITION BY <target primary_key>
+  ORDER BY <order_by> DESC NULLS LAST
+) ... WHERE __rn = 1
+```
+
+Every `primary_key` column must be branch-projected; conflicting `order_by`
+lists within a branch raise. `NULLS LAST` is pinned because DuckDB and Spark
+default NULL placement differently.
+
+**Overlap handling.** `union_exclude_base: true` anti-joins each union branch
+against the base branch's *post-filter, post-dedup* rows on the target primary
+key (rows already present in the base are dropped). `union_coalesce_base: true`
+instead merges overlapping rows: base-only rows pass through, overlapping rows
+take `COALESCE(base.col, union.col)` (base wins; primary-key, meta, and
+`union_value` columns always come from the base side), union-only rows pass
+through. Coalesce supports exactly one union table — the overlap semantics are
+pairwise. Both modes require a primary key and raise without one.
+
+Joins to *other* tables still work after a union base view: join key columns
+are projected into every branch (typed NULL where a source lacks them).
+
+## Base and final filters
+
+- `base_table_filter` — WHERE on the base view, before any joins. Bare
+  base-table columns only; filters earliest and cheapest.
+- `final_filter` — WHERE applied *after* final assembly, so it can reference
+  derived output columns. The assembly is wrapped
+  (`SELECT ... FROM (<assembly>) _final WHERE ...`) because a same-level WHERE
+  cannot reference SELECT aliases.
+- `final_dedup: distinct` — emits `SELECT DISTINCT *` over the final assembly,
+  collapsing exact-duplicate rows produced by join fan-out.
+
+Both filters run through `{{template_var}}` substitution.
+
+## Join controls
+
+- `base_join_column` — overrides the auto-inferred base join key. Also
+  overwrites `source_column` on every relationship declared outgoing from the
+  base table: the field exists precisely when the auto-selected key is wrong,
+  and declared relationships carry that same wrong key. Set it only when every
+  join out of the base should use one key.
+- `ForeignKey.join_filter` — extra predicate ANDed into the JOIN ON clause.
+  Candidate-level `join_filter` (on `DerivationCandidate`) takes precedence
+  when both are present, because candidate filters are keyed by
+  `(table, table_instance)` and can disambiguate multi-instance joins;
+  FK-level filters fill the gaps.
+- `OutgoingRelationship.alternative_joins` — additional join paths tried in
+  declared priority order (the relationship's own `source_column/target_column`
+  is priority 1). Emitted as a **UNION-of-joins**, not `ON (a = b OR c = d)`:
+  Spark plans OR-joins as a BroadcastNestedLoopJoin, which is a known
+  performance hazard. Instead each path becomes an inner-join branch over the
+  distinct base keys, branches are UNIONed with a `__branch_priority` literal,
+  one match per base key survives (`ROW_NUMBER` ordered by branch priority),
+  and the result is joined back null-safely — spelled as
+  `(a = b OR (a IS NULL AND b IS NULL))` because `<=>` is Spark-only.
+
+```yaml
+relationships:
+  outgoing:
+    - target_table: payer_xref
+      source_column: payor_claim_number
+      target_column: pcn
+      alternative_joins:
+        - source_column: nsa_dispute_number
+          target_column: dispute_no
+```
+
+## Error handling philosophy
+
+Misconfiguration fails at *plan time* with `ValueError` (missing union tables,
+conflicting row_filters/order_bys, exclude/coalesce without a primary key,
+alternative-join columns that don't exist). Declarative fields that the
+selected strategy does not consume (e.g. `union_base_tables` without
+`base_table_strategy: union_branches`) log a warning and are ignored, so
+fork-authored specs load — the warning names the missing switch.

mkdocs.yml

@@ -29,6 +29,7 @@ nav:
     - Happy Path: guide/happy-path.md
     - UMF Format: guide/umf-format.md
     - Schema Generation: guide/schema-generation.md
+    - SQL Plans: guide/sql-plans.md
     - Great Expectations: guide/great-expectations.md
     - Profiling: guide/profiling.md
     - LLM Prompts: guide/llm-prompts.md

src/tablespec/__init__.py

@@ -18,6 +18,7 @@
     IngestionConfig,
     IngestionExclusionRule,
     JoinViaSpec,
+    MergeCondition,
     Nullable,
     OutgoingRelationship,
     OutputConfig,
@@ -28,6 +29,7 @@
     RelationshipSummary,
     Relationships,
     Survivorship,
+    TableReference,
     UMFColumn,
     UMFColumnDerivation,
     UMFMetadata,
@@ -147,6 +149,7 @@
     "IngestionConfig",
     "IngestionExclusionRule",
     "JoinViaSpec",
+    "MergeCondition",
     "Nullable",
     "OutgoingRelationship",
     "OutputConfig",
@@ -157,6 +160,7 @@
     "RelationshipSummary",
     "Relationships",
     "Survivorship",
+    "TableReference",
     "UMFColumnDerivation",
     "ValidationRule",
     "ValidationRules",

src/tablespec/compatibility.py

@@ -42,10 +42,16 @@ class CompatibilityReport:
 # ---------------------------------------------------------------------------
 
 
-def _nullable_contexts(n: Nullable | None) -> dict[str, bool]:
-    """Return a {context_key: bool} dict for a Nullable, treating None as empty."""
+def _nullable_contexts(n: bool | Nullable | None) -> dict[str, bool]:
+    """Return a {context_key: bool} dict for a Nullable, treating None as empty.
+
+    A plain boolean nullable applies to every context and is represented by
+    the synthetic "*" key so tightening (True -> False) is still detected.
+    """
     if n is None:
         return {}
+    if isinstance(n, bool):
+        return {"*": n}
     # Nullable uses model_config extra="allow", so extra fields are in __pydantic_extra__
     result: dict[str, bool] = {}
     for key in n.model_fields_set | set(n.__pydantic_extra__ or {}):

src/tablespec/excel_converter.py

@@ -490,8 +490,13 @@ def convert(self, umf: UMF) -> openpyxl.Workbook:
         self._create_readme_sheet(umf)
         self._create_schema_sheet(umf)
         self._create_columns_sheet(umf)
-        self._create_survivorship_sheet(umf)
-        self._create_derivations_sheet(umf)
+
+        # Derivation-related sheets only apply to generated/derived tables;
+        # source-layer specs without survivorship metadata get a lean workbook.
+        if any(col.derivation or col.provenance_policy for col in umf.columns):
+            self._create_survivorship_sheet(umf)
+        if any(col.derivation for col in umf.columns):
+            self._create_derivations_sheet(umf)
 
         if self._get_expectation_dicts_for_export(umf):
             self._create_validation_sheet(umf)
@@ -669,6 +674,7 @@ def _create_columns_sheet(self, umf: UMF) -> None:
             "Format",
             "Notes",
             "_Validation",
+            "Report Sheet",
         ]
         headers.extend(post_nullable_headers)
 
@@ -688,6 +694,7 @@ def col_letter(idx: int) -> str:
         reporting_idx = desc_idx + 5
         format_idx = desc_idx + 6
         notes_idx = desc_idx + 7
+        report_sheet_idx = desc_idx + 9  # desc_idx + 8 is the _Validation column
 
         # Data
         row = 2
@@ -722,9 +729,12 @@ def col_letter(idx: int) -> str:
             self._apply_font_to_cell(ws[f"G{row}"], default_font)
 
             # Nullable - write dynamic context columns
-            if col.nullable:
+            if col.nullable is not None:
                 nullable_data: dict[str, bool] = {}
-                if isinstance(col.nullable, Nullable):
+                if isinstance(col.nullable, bool):
+                    # Simple boolean applies to every context
+                    nullable_data = dict.fromkeys(context_keys, col.nullable)
+                elif isinstance(col.nullable, Nullable):
                     nullable_data = col.nullable.model_dump(exclude_none=True)
                 elif isinstance(col.nullable, dict):
                     nullable_data = col.nullable
@@ -770,12 +780,18 @@ def col_letter(idx: int) -> str:
                 ws[f"{col_letter(notes_idx)}{row}"] = ""
             self._apply_font_to_cell(ws[f"{col_letter(notes_idx)}{row}"], default_font)
 
+            # Report Sheet (workbook tab assignment for multi-sheet reports)
+            ws[f"{col_letter(report_sheet_idx)}{row}"] = col.report_sheet or ""
+            self._apply_font_to_cell(
+                ws[f"{col_letter(report_sheet_idx)}{row}"], default_font
+            )
+
             row += 1
 
         # Adjust column widths
         pre_nullable_widths = [15, 18, 20, 12, 10, 11, 8]
         nullable_widths = [12] * num_contexts
-        post_nullable_widths = [20, 20, 12, 18, 15, 12, 15, 30, 15]
+        post_nullable_widths = [20, 20, 12, 18, 15, 12, 15, 30, 15, 15]
         all_widths = pre_nullable_widths + nullable_widths + post_nullable_widths
         for i, width in enumerate(all_widths, 1):
             ws.column_dimensions[get_column_letter(i)].width = width
@@ -1301,6 +1317,7 @@ def _create_derivations_sheet(self, umf: UMF) -> None:
             "Order By",
             "Select Columns",
             "Join Via",
+            "Union Value",
             "Reason",
             "Derivation Strategy",
             "Survivorship Strategy",
@@ -1350,10 +1367,10 @@ def _json_or_blank(value: Any) -> str:
             def _write_row(candidate: Any, col_fields: dict[str, Any]) -> None:
                 nonlocal row
                 if candidate is None:
-                    # 11 candidate columns: Priority..Reason (must match the
+                    # 12 candidate columns: Priority..Reason (must match the
                     # populated branch below so column-level fields land in the
                     # right cells).
-                    cand_values = [""] * 11
+                    cand_values = [""] * 12
                 else:
                     cand_values = [
                         candidate.priority,
@@ -1370,6 +1387,9 @@ def _write_row(candidate: Any, col_fields: dict[str, Any]) -> None:
                             if candidate.join_via
                             else None
                         ),
+                        # Native value (str/int/float/bool) so the SQL literal
+                        # survives the round-trip untouched
+                        "" if candidate.union_value is None else candidate.union_value,
                         candidate.reason or "",
                     ]
                 values = [
@@ -1579,7 +1599,13 @@ def _create_metadata_sheet(self, umf: UMF) -> None:
         for key, value in metadata.items():
             ws[f"A{row}"] = key
             self._apply_font_to_cell(ws[f"A{row}"], self._get_default_font())
-            ws[f"B{row}"] = str(value)
+            # JSON-encode lists/dicts (e.g. union_base_tables, source_tables) so
+            # they survive the round-trip; str() would emit a Python repr the
+            # importer cannot parse back
+            if isinstance(value, (list, dict)):
+                ws[f"B{row}"] = json.dumps(value)
+            else:
+                ws[f"B{row}"] = str(value)
             self._apply_font_to_cell(ws[f"B{row}"], self._get_default_font())
             row += 1
 
@@ -1938,6 +1964,7 @@ def _get(row: tuple, idx: int | None) -> Any:
         derivation_expression_idx = header_map.get("derivation expression")
         format_idx = header_map.get("format")
         notes_idx = header_map.get("notes")
+        report_sheet_idx = header_map.get("report sheet")
 
         columns = []
         for row in ws.iter_rows(min_row=2, values_only=False):
@@ -2066,6 +2093,11 @@ def _get(row: tuple, idx: int | None) -> Any:
                         line.strip() for line in notes_val.split("\n") if line.strip()
                     ]
 
+            # Report Sheet (workbook tab assignment for multi-sheet reports)
+            report_sheet_val = _get(row, report_sheet_idx)
+            if report_sheet_val:
+                col_dict["report_sheet"] = str(report_sheet_val).strip()
+
             columns.append(col_dict)
 
         return columns
@@ -2440,6 +2472,11 @@ def _json_cell(row: tuple, name: str, col_name: str) -> Any:
                 candidate["select_columns"] = select_columns
             if join_via:
                 candidate["join_via"] = join_via
+            # union_value preserves its native type (str/int/float/bool) so the
+            # per-branch SQL literal round-trips unchanged (TRUE, 1, 'daily').
+            union_value = cell(row, "union value")
+            if union_value not in (None, ""):
+                candidate["union_value"] = union_value
             if reason:
                 candidate["reason"] = str(reason)
             entry["candidates"].append(candidate)
@@ -2555,6 +2592,12 @@ def _extract_metadata(self, workbook: openpyxl.Workbook) -> dict | None:
                     else None
                 )
 
+            # List/dict metadata fields (union_base_tables, source_tables,
+            # unpivot_columns, ...) are JSON-encoded on export
+            if isinstance(value, str) and value.startswith(("[", "{")):
+                with contextlib.suppress(json.JSONDecodeError):
+                    value = json.loads(value)
+
             metadata[field] = value
 
         return metadata if metadata else None