From 9616f7e9171afcedbf83b00cd2bd3a15909df829 Mon Sep 17 00:00:00 2001 From: Gaurav Bajpai Date: Sun, 10 May 2026 08:24:29 +0530 Subject: [PATCH] feat: add explaining-batch-data-transform skill @W-22196528@ A new skill that helps Salesforce Data Cloud users understand existing Batch Data Transform (BDT) JSON definitions. Given a BDT JSON file (via path or pasted content), the skill produces a progressive-disclosure explanation and answers lineage / logic / structure questions. Contents: - SKILL.md: frontmatter + instructions + worked examples + troubleshooting. - scripts/bdt_analyze.py: generic typed-DAG parser over BDT JSON. Python 3.9+, stdlib only. Subcommands: summary, stages, nodes, node, lineage, field-trace, sources, outputs, formula, definitions. - references/: curated Markdown grounding synthesized from official BDT help documentation (no raw XML shipped). - assets/sample_bdts/: synthetic BDTs covering common action types. --- .../explaining-batch-data-transform/SKILL.md | 191 +++ .../assets/sample_bdts/append_and_split.json | 77 + .../assets/sample_bdts/joins_and_filters.json | 76 + .../sample_bdts/minimal_dmo_to_dmo.json | 58 + .../sample_bdts/window_and_aggregate.json | 97 ++ .../references/bdt-function-catalog.md | 124 ++ .../references/bdt-node-catalog.md | 763 +++++++++ .../references/bdt-reference.md | 179 +++ .../references/bdt-window-functions.md | 73 + .../scripts/bdt_analyze.py | 1381 +++++++++++++++++ 10 files changed, 3019 insertions(+) create mode 100644 skills/explaining-batch-data-transform/SKILL.md create mode 100644 skills/explaining-batch-data-transform/assets/sample_bdts/append_and_split.json create mode 100644 skills/explaining-batch-data-transform/assets/sample_bdts/joins_and_filters.json create mode 100644 skills/explaining-batch-data-transform/assets/sample_bdts/minimal_dmo_to_dmo.json create mode 100644 skills/explaining-batch-data-transform/assets/sample_bdts/window_and_aggregate.json create mode 100644 skills/explaining-batch-data-transform/references/bdt-function-catalog.md create mode 100644 skills/explaining-batch-data-transform/references/bdt-node-catalog.md create mode 100644 skills/explaining-batch-data-transform/references/bdt-reference.md create mode 100644 skills/explaining-batch-data-transform/references/bdt-window-functions.md create mode 100644 skills/explaining-batch-data-transform/scripts/bdt_analyze.py diff --git a/skills/explaining-batch-data-transform/SKILL.md b/skills/explaining-batch-data-transform/SKILL.md new file mode 100644 index 0000000..c4b3b20 --- /dev/null +++ b/skills/explaining-batch-data-transform/SKILL.md @@ -0,0 +1,191 @@ +--- +name: explaining-batch-data-transform +description: "Explain and investigate an existing Salesforce Data Cloud Batch Data Transform (BDT) JSON definition. Use when the user provides or references a BDT JSON export and asks to understand, summarize, walk through, trace lineage in, or debug that specific BDT — for example, 'explain this BDT', 'what does this batch data transform do', 'where does field X come from in this BDT', 'which nodes feed OUTPUT5', or 'what does this computeRelative formula mean'. Accepts a path to a BDT JSON file or pasted JSON content. Does not author, edit, run, or fetch BDTs from an org." +allowed-tools: Bash Read Write +metadata: + version: "1.0" +--- + +## 1. When to Use This Skill + +Use this skill when the user asks you to understand a Salesforce Data Cloud **Batch Data Transform (BDT)** JSON export. The BDT's JSON is a dependency graph of 3 to 100+ nodes — loads, joins, filters, formulas, window functions, aggregates, and outputs — that together describe how data flows from source DMOs/DLOs to target DMOs/DLOs. + +Trigger on prompts like: + +- "Explain this BDT." / "What does this BDT do?" +- "Walk me through this batch data transform." +- "Summarize this BDT JSON." +- "Which nodes feed OUTPUT5?" +- "Where does field `ssot__ProductAmount__c` come from in this BDT?" +- "Trace the lineage of `TotalAmount__c`." +- "What does this computeRelative formula mean?" / "Why would this field be zero?" +- "What DMOs does this BDT read from?" / "What does this filter do?" + +The skill accepts both shapes a BDT can appear in: the **editor export** (what customers download from the BDT viewer — `{version, nodes, ui}`) and the **Connect API create payload** (what developers POST when authoring via the API — with outer wrapper `{name, label, type, dataSpaceName, definition: {...}}`). See `references/bdt-reference.md` for details. Developers and customers alike can use the skill to understand a BDT — validation is out of scope for v1 but planned as a follow-up. + +## 2. Prerequisites + +- **Python 3.9+** on `PATH`. The parser script uses only the standard library — no `pip install` required. +- A **BDT JSON** — either a file path the user gives you, content the user pastes into the chat, or a file attachment you can read locally. + +Do NOT assume the user has Salesforce CLI, `gh`, `jq`, Node, or any Salesforce org auth. The skill runs entirely locally against a static JSON file. + +## 3. Input Handling + +Decision tree for getting the BDT JSON into a local file the script can read: + +1. **User gives a file path** (absolute or relative) → use it directly. +2. **User pastes JSON inline** → write it to `/tmp/bdt-.json` with the Write tool, then use that path. +3. **User attaches a file** → the path is in the attachment metadata; use it. +4. **No BDT provided yet** → ask: *"Please paste the BDT JSON, give me a file path, or attach the export."* + +Before any further work, validate by running `python bdt_analyze.py summary `. If it exits 3 with "Invalid JSON", tell the user: + +> "This file doesn't appear to be valid JSON. Could you re-export from the BDT viewer and try again? The exact error was: \." + +### 3.1 Multi-definition payloads + +**Multi-definition payloads.** If the user provides a Connect API payload with a `definitions[]` array containing multiple entries, first run `python bdt_analyze.py definitions ` to list them. Then offer the user a choice — *"The payload has 3 definitions: MyTransform, OtherTransform, ThirdTransform. Which do you want me to explain?"* — and invoke subsequent subcommands with `--definition N` (0-indexed). For example: `python bdt_analyze.py summary --definition 1`. All analysis subcommands (summary, sources, outputs, stages, nodes, node, lineage, field-trace, formula) accept the `--definition` flag. Editor exports and single-definition payloads can be explained directly without this step; running `definitions` on them still succeeds and shows one row, so it is safe to run on any input if you are unsure. + +## 4. Explanation Flow (Job A — Progressive Disclosure) + +The default flow for "explain this BDT": + +**Step 1.** Run `python bdt_analyze.py summary ` and render the **Executive Summary (Mode A)** in plain English to the user. Target ≤ 200 words. Name every source DMO/DLO and every output target. Identify at least one business-domain cue from the DMO names (e.g., *"looks like a sales-orders pipeline"*). + +**Step 2.** Offer the four explanation modes: + +> Want to go deeper? I can give you: +> +> - **(A) Executive summary** — reprint of what I just gave you. +> - **(B) Layered breakdown** — per-output lineage + per-stage mini-explanations. +> - **(C) Node-by-node walkthrough** — every node in topological order. +> - **(D) Business-intent read** — what this BDT is *trying to do*, in business language (marked as inference). + +**Step 3.** Based on the user's pick: + +- **Mode A** — reprint the summary; no further script calls. +- **Mode B** — run `python bdt_analyze.py stages `, then for each sink run `python bdt_analyze.py lineage `. Narrate: one section per output (its lineage), then a per-stage narration using the topo order. +- **Mode C** — run `python bdt_analyze.py nodes ` (add `--limit 50` if the BDT is large). Walk through each node in topo order, translating the digest into plain English. Consult `references/bdt-node-catalog.md` before narrating any action type the conversation hasn't already covered. +- **Mode D** — narrate the inferred business purpose from the `summary` digest + DMO/DLO names + output structure. **Always prefix with** *"Based on the structure, this appears to…"*. Explicit uncertainty marker is mandatory. + +**Override rule.** If the user's first message names a mode (e.g., *"walk me through every node"* = Mode C), skip Step 2 entirely and go straight to the chosen mode. + +## 5. Q&A Flow (Job B — Investigation) + +When the user asks a follow-up question, route it to the right subcommand using this table. **Always narrate the output in plain English — never dump raw script output to the user.** + +| User question pattern | Subcommand to run | +|---|---| +| "What does node X do?" / "Explain X" (L2) | `node X` | +| "Which nodes feed X?" / "Upstream of X?" (L1) | `lineage X` | +| "Where does field F come from?" (L3) | `field-trace F` | +| "What DMOs/DLOs does this read from?" (L4) | `sources ` | +| "What fields does OUTPUT N produce?" (L5) | `node OUTPUT_N` | +| "What does this formula mean?" (I1) | `formula X` | +| "Why would F be null/zero/unexpected?" (I2) | `field-trace F` to locate the defining node(s), **then** `formula ` on the formula or computeRelative node that computes the field (not the output mapping — pick the earliest formula/computeRelative node in the field-trace output). | +| "What does this filter do?" (I3) | `node X` | + +After running the subcommand(s), translate the output into plain English. For I1/I2 you **must** consult `references/bdt-function-catalog.md` (and `references/bdt-window-functions.md` if it's a `computeRelative` node) before narrating. + +## 6. Output Style Guidance + +- **Plain language first, technical detail second.** Lead with what the BDT or node does in business terms; only then mention the action type / parameter shape. +- **Use UI labels when available.** Say *"Join Users (JOIN0)"* rather than just *"JOIN0"* whenever `ui.nodes[X].label` exists. +- **Prefer field purpose over expression.** *"Product amount, zeroed except on the first ranked product per order"* rather than echoing the `case when ... end`. +- **Explicit uncertainty markers** on Mode D and I2 answers: *"Based on the structure, this appears to…"*. +- **Never fabricate lineage or definitions.** If `field-trace` returns no match, say the field isn't defined in this BDT. Don't invent a source. +- **Redact Salesforce record IDs in narration** (15/18-char IDs starting `0`) unless the user asks about them specifically. This keeps conversations shareable. +- **Don't fabricate features that aren't in the JSON.** If the user asks about scheduling, run history, or streaming behavior and those aren't in the JSON, say so plainly. +- **Don't dump raw JSON.** If the user wants to see raw params, run `node --json ` and *then* explain what's interesting in the result. + +## 7. Reference Consultation (mandatory) + +- **Always-loaded:** `references/bdt-reference.md` — the top-level overview. Assume this is in context every conversation. +- **Consult `references/bdt-node-catalog.md`** via the Read tool **before**: + - Explaining any node type you haven't already explained in this conversation. + - Narrating any of: `bucket`, `flatten`, `flattenJson`, `split`, `extractGrains`, `extractTable`, `appendV2`, `multidefinitionMerge`, `cdpPredict`, `extension`, `extensionFunction`, `jsonAggregate`, `typeCast`, `formatDate` — these are rarer action types where the parameter shape matters. +- **Consult `references/bdt-function-catalog.md`** **before** interpreting any `formulaExpression` (I1 or I2). Look up each function named in the expression. +- **Consult `references/bdt-window-functions.md`** **whenever** you narrate a `computeRelative` node. +- **Cite the source when narrating deep detail.** Good: *"Per the node catalog, `computeRelative` runs a window function over the rows partitioned by `` and ordered by ``."* Bad: asserting SQL semantics with no grounding. +- **If a node type or function isn't in any of the reference files**, narrate what the JSON shows (action name + pretty-printed parameters) and **explicitly flag that the item is not in the documentation materials shipped with this skill**. Do not guess at semantics. + +## 8. Error Handling Guidance (what to tell the user) + +| Script outcome | What to say | +|---|---| +| Exit 3: `Invalid JSON at line X` | "This file isn't valid JSON at line X. Could you re-export from the BDT viewer?" Then paste the exact error. | +| Exit 3: `Expected top-level 'nodes' object` | "This doesn't look like a BDT export — a BDT has a top-level `nodes` object. Is this the right file?" | +| Exit 3: `Cycle or broken reference detected` | "This BDT has a dependency cycle or references a non-existent node. You'll need to edit the BDT in the Data Cloud viewer to fix the circular dependency or broken reference — re-exporting won't help because the cycle exists in the BDT definition itself. The problematic nodes are: \." | +| Exit 2: `No node named 'X'` | "I don't see a node named `X` in this BDT. Did you mean one of: \?" | +| Exit 2: `Field 'F' is not defined by any node` | "That field isn't defined anywhere in this BDT. Closest matches from the fields I can see: \. Is it possibly a field from the source DMO that's not pulled in?" | +| User's BDT uses an action type not in the catalog | "This BDT uses ``, which isn't in my reference material. Here's what its parameters look like: \. I can describe the graph structure around it, but I can't vouch for the exact semantics of this action type." | + +## 9. Non-Goals (explicit "Do not" directives) + +**Do not** author, create, or edit BDTs. If the user asks to modify a BDT, explain that this skill only explains BDTs and point them at the BDT viewer in Data Cloud. + +**Do not** run, schedule, or trigger BDTs. Execution happens in Data Cloud; this skill works on the JSON export only. + +**Do not** fetch BDTs from a live org. v1 accepts only a user-provided file path or pasted JSON. If the user asks, say that live-org fetch is planned for v2. + +**Do not** validate BDTs — don't claim a BDT is "correct" or "will run successfully". Validation (required-field checks, enum checks, dangling-ref detection) is planned as a follow-up v1.1 and is explicitly out of scope. If the user asks "is my BDT valid?", reply with *"This skill explains BDTs but doesn't validate them; validation is planned for a follow-up release. Want me to explain what the BDT does so you can spot issues manually?"* + +**Do not** guess at node-type or function semantics not covered by the reference materials. Say "this isn't in the reference material shipped with this skill" and surface the raw parameters. + +**Do not** invent business intent. Mode D and I2 answers must be prefixed with *"Based on the structure, this appears to…"* — never claim business purpose as fact. + +**Do not** echo Salesforce record IDs verbatim (15/18-char IDs) in narration unless the user asks specifically about an ID. + +**Do not** output raw Python script output to the user — always narrate. + +**Do not** modify the BDT JSON file on disk. This skill is read-only. + +**Do not** visualize the BDT as a graph image or diagram. Skill output is text only. + +**Do not** compare multiple BDTs. This is a v1 limitation (single-BDT focus). If asked, suggest running the skill on each BDT separately. + +## 10. Troubleshooting + +Common issues and how to address them: + +| Issue | Solution | +|---|---| +| `python: command not found` or `python3: command not found` | Ask the user to install Python 3.9+. Point them at https://www.python.org/downloads/. The skill's parser uses only the standard library, so `pip install` is **not** needed. | +| `ModuleNotFoundError: No module named 'bdt_analyze'` | The script path is wrong. The script lives at `scripts/bdt_analyze.py` inside the skill directory. Run it by absolute path (the Bash tool invocation should include the full path from the repo root). | +| BDT file path contains spaces (e.g. `"default B2C (Prod).json"`) | Quote the path when calling the script: `python bdt_analyze.py summary "default B2C (Prod).json"`. | +| Script output is very large for a big BDT (Mode C) | Use `--limit 50` on the `nodes` subcommand, or recommend Mode B (layered) instead of Mode C. | +| Field trace returns no match despite the field clearly appearing in the BDT | The field may only appear qualified (e.g., `SalesOrder.ssot__Id__c`). Try the qualified form. If still no match, the field may be passed through from a source DMO but not explicitly defined in the BDT — in that case field-trace cannot locate it; narrate "this field originates from the load node's source DMO" instead. Do NOT attempt to trace it by stripping the `__c` suffix — that would risk matching a different field with a similar name. | +| User says output "doesn't match what I see in BDT viewer" | Verify the version in the JSON header matches the org's current release. The canonical schema reference in `references/bdt-node-catalog.md` may predate the user's BDT, so older/newer BDTs may use variants not yet documented. Flag as "version drift" and narrate best-effort. | + +## 11. Example Interactions + +**Example 1 — "Explain this BDT" on a small DMO-to-DMO BDT.** + +User: *"Explain this BDT."* (attaches `minimal_dmo_to_dmo.json`) + +Skill: +1. Runs `summary` → gets: 3 nodes (1 load, 1 formula, 1 output), reads from `ssot__Account__dlm`, writes to `Account_Upper__dlm`. +2. Narrates in plain English: + > This BDT takes accounts from the `ssot__Account__dlm` DMO, uppercases the account name, and writes the result to `Account_Upper__dlm`. Three nodes: **Account** (load), **Uppercase Name** (formula), **Account Upper** (output). +3. Offers modes: *"Want the layered breakdown, node-by-node walkthrough, or business-intent read?"* + +**Example 2 — Field lineage on a larger BDT.** + +User: *"Where does `TotalAmount__c` come from in this BDT?"* (after already loading `window_and_aggregate.json`) + +Skill: +1. Runs `field-trace window_and_aggregate.json TotalAmount__c`. +2. Receives: defined by `AGG_BY_ACCOUNT` (action `aggregate`), specifically as `SUM` over `ssot__GrandTotalAmount__c`, grouping by `ssot__AccountId__c`. Upstream chain: `LOAD_ORDERS → RANK_ORDERS → AGG_BY_ACCOUNT`. +3. Consults `references/bdt-node-catalog.md` for the `aggregate` section. +4. Narrates: + > `TotalAmount__c` is produced by the **Totals per account (`AGG_BY_ACCOUNT`)** aggregate node. It's a `SUM` of `ssot__GrandTotalAmount__c`, grouped by `ssot__AccountId__c` (one row per account). The amount values come from the original Sales Orders DMO via `LOAD_ORDERS`. The `RANK_ORDERS` node between them adds a per-account order rank but doesn't alter the amount. + +## 12. Next Steps + +After completing an explanation or Q&A, if the conversation seems to be wrapping up, offer: + +- "Want to trace another field or explain another node?" +- "Need this BDT's summary in a different format (e.g., as a bulleted list for a doc)?" +- "If you want to edit or run this BDT, use the BDT viewer in Data Cloud — this skill explains BDTs but doesn't author or execute them." +- If the user found a bug or limitation, point them at the [afv-library GitHub issues](https://github.com/forcedotcom/afv-library/issues) to report it. diff --git a/skills/explaining-batch-data-transform/assets/sample_bdts/append_and_split.json b/skills/explaining-batch-data-transform/assets/sample_bdts/append_and_split.json new file mode 100644 index 0000000..5e6053f --- /dev/null +++ b/skills/explaining-batch-data-transform/assets/sample_bdts/append_and_split.json @@ -0,0 +1,77 @@ +{ + "version": "66.0", + "nodes": { + "LOAD_WEB_ORDERS": { + "action": "load", + "sources": [], + "parameters": { + "dataset": {"name": "WebOrders__dlo", "type": "dataLakeObject"}, + "fields": ["OrderId__c", "Amount__c", "ChannelOrderKey__c"], + "sampleDetails": {"type": "TopN", "sortBy": []} + } + }, + "LOAD_STORE_ORDERS": { + "action": "load", + "sources": [], + "parameters": { + "dataset": {"name": "StoreOrders__dlo", "type": "dataLakeObject"}, + "fields": ["OrderId__c", "Amount__c", "ChannelOrderKey__c"], + "sampleDetails": {"type": "TopN", "sortBy": []} + } + }, + "APPEND_ALL_ORDERS": { + "action": "appendV2", + "sources": ["LOAD_WEB_ORDERS", "LOAD_STORE_ORDERS"], + "parameters": { + "allowImplicitDisjointSchema": false, + "fieldMappings": [ + {"targetField": "OrderId__c", "sources": [{"node": "LOAD_WEB_ORDERS", "field": "OrderId__c"}, {"node": "LOAD_STORE_ORDERS", "field": "OrderId__c"}]}, + {"targetField": "Amount__c", "sources": [{"node": "LOAD_WEB_ORDERS", "field": "Amount__c"}, {"node": "LOAD_STORE_ORDERS", "field": "Amount__c"}]}, + {"targetField": "ChannelOrderKey__c", "sources": [{"node": "LOAD_WEB_ORDERS", "field": "ChannelOrderKey__c"}, {"node": "LOAD_STORE_ORDERS", "field": "ChannelOrderKey__c"}]} + ] + } + }, + "SPLIT_CHANNEL_KEY": { + "action": "split", + "sources": ["APPEND_ALL_ORDERS"], + "parameters": { + "sourceField": "ChannelOrderKey__c", + "delimiter": "-", + "targetFields": [ + {"name": "Channel__c", "label": "Channel"}, + {"name": "OrderNumber__c", "label": "Order Number"} + ] + } + }, + "OUTPUT_ORDERS": { + "action": "outputD360", + "sources": ["SPLIT_CHANNEL_KEY"], + "parameters": { + "name": "OrdersWithChannel__dlm", + "type": "dataModelObject", + "writeMode": "OVERWRITE", + "fieldsMappings": [ + {"sourceField": "OrderId__c", "targetField": "OrderId__c"}, + {"sourceField": "Amount__c", "targetField": "Amount__c"}, + {"sourceField": "Channel__c", "targetField": "Channel__c"}, + {"sourceField": "OrderNumber__c", "targetField": "OrderNumber__c"} + ] + } + } + }, + "ui": { + "nodes": { + "LOAD_WEB_ORDERS": {"label": "Web Orders", "type": "LOAD_DATASET", "top": 100, "left": 100}, + "LOAD_STORE_ORDERS": {"label": "Store Orders", "type": "LOAD_DATASET", "top": 260, "left": 100}, + "APPEND_ALL_ORDERS": {"label": "Union", "type": "APPEND", "top": 180, "left": 260}, + "SPLIT_CHANNEL_KEY": {"label": "Split channel key", "type": "SPLIT", "top": 180, "left": 420}, + "OUTPUT_ORDERS": {"label": "Orders + channel", "type": "OUTPUT", "top": 180, "left": 580} + }, + "connectors": [ + {"source": "LOAD_WEB_ORDERS", "target": "APPEND_ALL_ORDERS"}, + {"source": "LOAD_STORE_ORDERS", "target": "APPEND_ALL_ORDERS"}, + {"source": "APPEND_ALL_ORDERS", "target": "SPLIT_CHANNEL_KEY"}, + {"source": "SPLIT_CHANNEL_KEY", "target": "OUTPUT_ORDERS"} + ] + } +} diff --git a/skills/explaining-batch-data-transform/assets/sample_bdts/joins_and_filters.json b/skills/explaining-batch-data-transform/assets/sample_bdts/joins_and_filters.json new file mode 100644 index 0000000..1db8bae --- /dev/null +++ b/skills/explaining-batch-data-transform/assets/sample_bdts/joins_and_filters.json @@ -0,0 +1,76 @@ +{ + "version": "66.0", + "nodes": { + "LOAD_ORDERS": { + "action": "load", + "sources": [], + "parameters": { + "dataset": {"name": "ssot__SalesOrder__dlm", "type": "dataModelObject"}, + "fields": ["ssot__Id__c", "ssot__AccountId__c", "ssot__Status__c", "ssot__GrandTotalAmount__c"], + "sampleDetails": {"type": "TopN", "sortBy": []} + } + }, + "LOAD_ACCOUNTS": { + "action": "load", + "sources": [], + "parameters": { + "dataset": {"name": "ssot__Account__dlm", "type": "dataModelObject"}, + "fields": ["ssot__Id__c", "ssot__Name__c", "ssot__Industry__c"], + "sampleDetails": {"type": "TopN", "sortBy": []} + } + }, + "FILTER_ACTIVE_ORDERS": { + "action": "filter", + "sources": ["LOAD_ORDERS"], + "parameters": { + "filterExpressions": [ + {"type": "TEXT", "field": "ssot__Status__c", "operator": "EQUAL", "operands": ["Active"]}, + {"type": "NUMBER", "field": "ssot__GrandTotalAmount__c", "operator": "GREATER_THAN", "operands": ["0"]}, + {"type": "NUMBER", "field": "ssot__GrandTotalAmount__c", "operator": "IS_NOT_NULL", "operands": []} + ], + "filterBooleanLogic": "1 AND 2 AND 3" + } + }, + "JOIN_ORDER_ACCOUNT": { + "action": "join", + "sources": ["FILTER_ACTIVE_ORDERS", "LOAD_ACCOUNTS"], + "parameters": { + "joinType": "LEFT_OUTER", + "leftKeys": ["ssot__AccountId__c"], + "rightQualifier": "Account", + "rightKeys": ["ssot__Id__c"] + } + }, + "OUTPUT_ENRICHED": { + "action": "outputD360", + "sources": ["JOIN_ORDER_ACCOUNT"], + "parameters": { + "name": "Order_Enriched__dlm", + "type": "dataModelObject", + "writeMode": "OVERWRITE", + "fieldsMappings": [ + {"sourceField": "ssot__Id__c", "targetField": "OrderId__c"}, + {"sourceField": "ssot__AccountId__c", "targetField": "AccountId__c"}, + {"sourceField": "ssot__GrandTotalAmount__c", "targetField": "Amount__c"}, + {"sourceField": "Account.ssot__Name__c", "targetField": "AccountName__c"}, + {"sourceField": "Account.ssot__Industry__c", "targetField": "Industry__c"} + ] + } + } + }, + "ui": { + "nodes": { + "LOAD_ORDERS": {"label": "Sales Orders", "type": "LOAD_DATASET", "top": 100, "left": 100}, + "LOAD_ACCOUNTS": {"label": "Accounts", "type": "LOAD_DATASET", "top": 260, "left": 100}, + "FILTER_ACTIVE_ORDERS": {"label": "Active, >0", "type": "FILTER", "top": 100, "left": 260}, + "JOIN_ORDER_ACCOUNT": {"label": "Join Account info", "type": "JOIN", "top": 180, "left": 420}, + "OUTPUT_ENRICHED": {"label": "Order Enriched", "type": "OUTPUT", "top": 180, "left": 580} + }, + "connectors": [ + {"source": "LOAD_ORDERS", "target": "FILTER_ACTIVE_ORDERS"}, + {"source": "FILTER_ACTIVE_ORDERS", "target": "JOIN_ORDER_ACCOUNT"}, + {"source": "LOAD_ACCOUNTS", "target": "JOIN_ORDER_ACCOUNT"}, + {"source": "JOIN_ORDER_ACCOUNT", "target": "OUTPUT_ENRICHED"} + ] + } +} diff --git a/skills/explaining-batch-data-transform/assets/sample_bdts/minimal_dmo_to_dmo.json b/skills/explaining-batch-data-transform/assets/sample_bdts/minimal_dmo_to_dmo.json new file mode 100644 index 0000000..4506b24 --- /dev/null +++ b/skills/explaining-batch-data-transform/assets/sample_bdts/minimal_dmo_to_dmo.json @@ -0,0 +1,58 @@ +{ + "version": "66.0", + "nodes": { + "LOAD_ACCOUNT": { + "action": "load", + "sources": [], + "parameters": { + "dataset": {"name": "ssot__Account__dlm", "type": "dataModelObject"}, + "fields": ["ssot__Id__c", "ssot__Name__c", "ssot__Industry__c"], + "sampleDetails": {"type": "TopN", "sortBy": []} + } + }, + "FORMULA_NAME_UPPER": { + "action": "formula", + "sources": ["LOAD_ACCOUNT"], + "parameters": { + "expressionType": "SQL", + "fields": [ + { + "name": "AccountNameUpper__c", + "label": "Account Name (upper)", + "formulaExpression": "upper(ssot__Name__c)", + "type": "TEXT", + "businessType": "Text", + "precision": 255, + "defaultValue": "" + } + ] + } + }, + "OUTPUT_ACCOUNT_UPPER": { + "action": "outputD360", + "sources": ["FORMULA_NAME_UPPER"], + "parameters": { + "name": "Account_Upper__dlm", + "type": "dataModelObject", + "writeMode": "OVERWRITE", + "fieldsMappings": [ + {"sourceField": "ssot__Id__c", "targetField": "ssot__Id__c"}, + {"sourceField": "AccountNameUpper__c", "targetField": "Name_Upper__c"}, + {"sourceField": "ssot__Industry__c", "targetField": "Industry__c"} + ] + } + } + }, + "ui": { + "nodes": { + "LOAD_ACCOUNT": {"label": "Account", "type": "LOAD_DATASET", "top": 100, "left": 100}, + "FORMULA_NAME_UPPER": {"label": "Uppercase Name", "type": "FORMULA", "top": 100, "left": 260}, + "OUTPUT_ACCOUNT_UPPER": {"label": "Account Upper", "type": "OUTPUT", "top": 100, "left": 420} + }, + "connectors": [ + {"source": "LOAD_ACCOUNT", "target": "FORMULA_NAME_UPPER"}, + {"source": "FORMULA_NAME_UPPER", "target": "OUTPUT_ACCOUNT_UPPER"} + ], + "hiddenColumns": [] + } +} diff --git a/skills/explaining-batch-data-transform/assets/sample_bdts/window_and_aggregate.json b/skills/explaining-batch-data-transform/assets/sample_bdts/window_and_aggregate.json new file mode 100644 index 0000000..4c9a127 --- /dev/null +++ b/skills/explaining-batch-data-transform/assets/sample_bdts/window_and_aggregate.json @@ -0,0 +1,97 @@ +{ + "version": "66.0", + "nodes": { + "LOAD_ORDERS": { + "action": "load", + "sources": [], + "parameters": { + "dataset": {"name": "ssot__SalesOrder__dlm", "type": "dataModelObject"}, + "fields": ["ssot__Id__c", "ssot__AccountId__c", "ssot__CreatedDate__c", "ssot__GrandTotalAmount__c"], + "sampleDetails": {"type": "TopN", "sortBy": []} + } + }, + "RANK_ORDERS": { + "action": "computeRelative", + "sources": ["LOAD_ORDERS"], + "parameters": { + "partitionBy": ["ssot__AccountId__c"], + "orderBy": [{"fieldName": "ssot__CreatedDate__c", "direction": "ASC"}], + "expressionType": "SQL", + "fields": [ + { + "name": "OrderRank__c", + "label": "Order Rank", + "formulaExpression": "row_number()", + "type": "NUMBER", + "businessType": "NUMBER", + "precision": 18, + "scale": 0, + "defaultValue": "" + } + ] + } + }, + "FIRST_ORDER_AMOUNT": { + "action": "formula", + "sources": ["RANK_ORDERS"], + "parameters": { + "expressionType": "SQL", + "fields": [ + { + "name": "FirstOrderAmount__c", + "label": "First Order Amount (per account)", + "formulaExpression": "case when OrderRank__c = 1 then ssot__GrandTotalAmount__c else 0 end", + "type": "NUMBER", + "businessType": "NUMBER", + "precision": 18, + "scale": 2, + "defaultValue": "0" + } + ] + } + }, + "AGG_BY_ACCOUNT": { + "action": "aggregate", + "sources": ["FIRST_ORDER_AMOUNT"], + "parameters": { + "groupings": ["ssot__AccountId__c"], + "aggregations": [ + {"action": "SUM", "name": "TotalAmount__c", "source": "ssot__GrandTotalAmount__c"}, + {"action": "SUM", "name": "FirstOrderTotal__c", "source": "FirstOrderAmount__c"}, + {"action": "COUNT", "name": "OrderCount__c", "source": "ssot__Id__c"} + ], + "nodeType": "STANDARD" + } + }, + "OUTPUT_SUMMARY": { + "action": "outputD360", + "sources": ["AGG_BY_ACCOUNT"], + "parameters": { + "name": "Account_Summary__dlm", + "type": "dataModelObject", + "writeMode": "OVERWRITE", + "fieldsMappings": [ + {"sourceField": "ssot__AccountId__c", "targetField": "AccountId__c"}, + {"sourceField": "TotalAmount__c", "targetField": "TotalAmount__c"}, + {"sourceField": "FirstOrderTotal__c", "targetField": "FirstOrderAmount__c"}, + {"sourceField": "OrderCount__c", "targetField": "OrderCount__c"} + ] + } + } + }, + "ui": { + "nodes": { + "LOAD_ORDERS": {"label": "Sales Orders", "type": "LOAD_DATASET", "top": 100, "left": 100}, + "RANK_ORDERS": {"label": "Rank by account", "type": "COMPUTE_RELATIVE", "top": 100, "left": 260}, + "FIRST_ORDER_AMOUNT": {"label": "Extract first order amount", "type": "FORMULA", "top": 100, "left": 420}, + "AGG_BY_ACCOUNT": {"label": "Totals per account", "type": "AGGREGATE", "top": 100, "left": 580}, + "OUTPUT_SUMMARY": {"label": "Account Summary", "type": "OUTPUT", "top": 100, "left": 740} + }, + "connectors": [ + {"source": "LOAD_ORDERS", "target": "RANK_ORDERS"}, + {"source": "RANK_ORDERS", "target": "FIRST_ORDER_AMOUNT"}, + {"source": "FIRST_ORDER_AMOUNT", "target": "AGG_BY_ACCOUNT"}, + {"source": "AGG_BY_ACCOUNT", "target": "OUTPUT_SUMMARY"} + ] + } +} diff --git a/skills/explaining-batch-data-transform/references/bdt-function-catalog.md b/skills/explaining-batch-data-transform/references/bdt-function-catalog.md new file mode 100644 index 0000000..89906b5 --- /dev/null +++ b/skills/explaining-batch-data-transform/references/bdt-function-catalog.md @@ -0,0 +1,124 @@ +# SFSQL Function Catalog — for `formula` reasoning + +> **Last synced:** 2026-04-23 from the Data Processing Engine PDF (2025-06-24 version) and +> the BDT help DITA XMLs (`c360_a_batch_transform_numeric.xml`, `_string.xml`, +> `_date_functions.xml`, `_boolean_functions.xml`, `_multivalue_functions.xml`, +> `_additionalfunctions.xml`). Consult this file before interpreting any `formulaExpression`. + +## How formulas appear in BDT JSON + +```jsonc +"action": "formula", +"parameters": { + "expressionType": "SQL", // or "DCSQL" + "fields": [ + { + "name": "OutputField__c", + "label": "Output Label", + "type": "TEXT", // TEXT | NUMBER | BOOLEAN | DATE_ONLY | DATETIME + "businessType": "Text", // user-visible type name + "formulaExpression": "upper(coalesce(SourceField__c, ''))", + "precision": 255, + "scale": 0, // only for NUMBER + "defaultValue": "" + } + ] +} +``` + +## Functions by family + +### Math / Numeric + +| Function | Signature | Purpose | +|---|---|---| +| `abs(n)` | number → number | Absolute value (strips sign). | +| `ceiling(n)` | number → number | Round up, away from zero for negatives. | +| `floor(n)` | number → number | Round toward negative infinity. Always rounds down on the number line: `floor(2.7) = 2`, `floor(-2.3) = -3`. | +| `exp(n)` | number → number | e raised to n. | +| `log(base, n)` | number → number | Logarithm of n in the given base. | +| `max(a, b, …)` | number → number | Largest value. | +| `min(a, b, …)` | number → number | Smallest value. | +| `mod(a, b)` | number → number | Remainder after dividing a by b. | +| `power(a, b)` | number → number | a raised to the power b. | +| `round(n, digits)` | number → number | Round to `digits` places. | +| `sqrt(n)` | number → number | Positive square root. | +| `trunc(n, digits)` | number → number | Truncate to `digits` places (doesn't round). | + +### String + +| Function | Signature | Purpose | +|---|---|---| +| `begins(s, prefix)` | text → boolean | True if s starts with prefix. | +| `concat(a, b, …)` | text → text | Concatenate. | +| `contains(haystack, needle)` | text → boolean | True if haystack contains needle. | +| `ends(s, suffix)` | text → boolean | True if s ends with suffix. | +| `length(s)` | text → number | Character count. | +| `lower(s)` | text → text | Lowercase (locale-aware if a locale is provided). | +| `ltrim(s)` / `ltrim(s, substring)` | text → text | Remove leading whitespace (or a specific substring). | +| `rtrim(s)` / `rtrim(s, substring)` | text → text | Remove trailing whitespace (or substring). | +| `substitute(s, old, new)` | text → text | Replace old with new in s. | +| `substr(s, start, length)` | text → text | Extract a substring. | +| `text(n)` | number → text | Convert a number to its text form. | +| `trim(s)` / `trim(s, substring)` | text → text | Remove leading + trailing whitespace or substring. | +| `upper(s)` | text → text | Uppercase. | +| `uuid()` | → text | Newly generated unique ID. | +| `value(s)` | text → number | Parse a text representation of a number into a numeric value. | + +### Date + +| Function | Signature | Purpose | +|---|---|---| +| `adddays(date, n)` | date → date | Add n days. | +| `addmonths(date, n)` | date → date | Add n months. | +| `datediff(start, end)` | date × date → number | Days between two dates. | +| `datetimevalue(text_or_date)` | → datetime | GMT/UTC date+time value. | +| `datevalue(text_or_datetime)` | → date | Extract the date part. | +| `day(date)` | date → number | 1-31. | +| `monthdiff(start, end)` | date × date → number | Months between two dates. | +| `now()` | → datetime | Current moment (UTC). | +| `today()` | → date | Current date. | +| `weekday(date)` | date → number | 1=Sunday, 2=Monday, … 7=Saturday. | + +### Boolean / Logical + +| Function | Signature | Purpose | +|---|---|---| +| `and(a, b, …)` | bool → bool | True when all are true. | +| `or(a, b, …)` | bool → bool | True when any is true. | +| `if(cond, then_val, else_val)` | → any | Ternary. | +| `case when then else end` | — | Multi-branch (SQL CASE). | +| `blankvalue(expr, substitute)` | → any | Substitute when expr is blank. | +| `isblank(expr)` | → bool | True when expr is blank. | +| `isnull(expr)` | → bool | True when expr is null. | +| `nullvalue(expr, substitute)` | → any | Substitute when expr is null; returns expr otherwise. | + +### Multivalue + +| Function | Signature | Purpose | +|---|---|---| +| `sequence(start, end, step?)` | → array | Array of numbers/dates between start and end (step defaults to 1 / 1 day). | +| `explode(array)` | → row per element | Convert multivalue data into one row per element. **Cannot be nested inside another function.** | + +### Additional / Null-handling (commonly seen) + +| Function | Signature | Purpose | +|---|---|---| +| `coalesce(a, b, …)` | → any | First non-null argument (SQL standard). | + +## Patterns to recognize in narration + +- **`coalesce(X, Y)`** — "use X when present, otherwise Y." Very common for default-value + handling. +- **`case when RANK = 1 then X else 0 end`** — the "first-occurrence extract" idiom paired + with a `computeRelative` `row_number()` node. +- **`case when FLAG = 'Y' then 'Y' else 'N' end`** — boolean-like text normalization. +- **`concat(coalesce(A, ''), coalesce(B, ''))`** — safe string concatenation that defaults + NULLs to empty strings (producing a synthetic composite key). + +## Sources + +- Data Processing Engine reference PDF (captured at `research/bdt-doc-10-data_processing_engine_6-24-2025.pdf.md`) +- BDT help XML: `c360_a_batch_transform_numeric.xml`, `_string.xml`, `_date_functions.xml`, + `_boolean_functions.xml`, `_multivalue_functions.xml`, `_additionalfunctions.xml` (captured + at `research/help-xml/`). diff --git a/skills/explaining-batch-data-transform/references/bdt-node-catalog.md b/skills/explaining-batch-data-transform/references/bdt-node-catalog.md new file mode 100644 index 0000000..0e06c21 --- /dev/null +++ b/skills/explaining-batch-data-transform/references/bdt-node-catalog.md @@ -0,0 +1,763 @@ +# BDT Node Catalog — per-action reference (on-demand) + +> **Source of truth:** `research/bdt-schema-canonical.md`, extracted from the BDT Connect +> API Java sources (`cdp-connect-api` module, release 264) on 2026-04-23. The entries +> below are curated to answer the skill's job — explaining BDT JSON. For the full +> extraction (all fields, minVersion annotations, polymorphism map), see the research doc. + +Consult this file before explaining any node type. Each section covers: +- **JSON action name** and UI name(s). +- **Purpose** in plain English. +- **Key parameters** with types (and enum values where applicable). +- **How it affects lineage** — rename? add? drop? change row cardinality? +- **Common gotchas**. +- **Example JSON snippet**. +- **Source citation** back to the canonical Java input-rep class. + +--- + +## `action: "load"` — UI: "Load" / "Data Source" + +**Purpose.** Reads rows from a DMO or DLO into the pipeline. Every BDT has at least one +load node (they are the graph roots — `sources: []`). + +**Key parameters** (class `LoadParametersInputRepresentation`): + +| Param | Type | Notes | +|---|---|---| +| `dataset` | `LoadDatasetInputRepresentation` | `{ name: string, type: "dataModelObject" \| "dataLakeObject" }` | +| `fields` | `string[]` | Field names to pull from the source. Not pulling a field means it's unavailable downstream. | +| `sampleDetails` | `{ type: "TopN" \| "Custom" \| "Unique", sortBy: string[] }` | Editor-only sampling behavior; doesn't affect runtime. | + +**Lineage effect.** Defines the *initial* set of field names available downstream. No row +cardinality change (all rows are loaded). + +**Gotchas.** +- A field referenced by a downstream node must appear in some load's `fields` array, **or** + be generated later by a formula/aggregate. If you can't trace a field to a load or a + derivation, the BDT is broken. +- `dataset.type` was historically called `dataLakeObject` in some older dialects; per + canonical schema both `dataModelObject` and `dataLakeObject` are valid. +- `sampleDetails.sortBy` is a list of strings — may be empty. + +**Example.** +```jsonc +{ + "action": "load", + "sources": [], + "parameters": { + "dataset": {"name": "ssot__SalesOrder__dlm", "type": "dataModelObject"}, + "fields": ["ssot__Id__c", "ssot__AccountId__c", "ssot__GrandTotalAmount__c"], + "sampleDetails": {"type": "TopN", "sortBy": []} + } +} +``` + +**Source.** `LoadNodeInputRepresentation` + `LoadParametersInputRepresentation` + +`LoadDatasetInputRepresentation`. + +--- + +## `action: "join"` — UI: "Join" + +**Purpose.** Joins two upstream streams by key. Always has exactly two `sources`; field names +on the right-hand side get a qualifier prefix so they don't collide with left-hand names. + +**Key parameters** (class `JoinParametersInputRepresentation`): + +| Param | Type | Notes | +|---|---|---| +| `joinType` | enum `JoinType`: `INNER`, `OUTER`, `LEFT_OUTER`, `RIGHT_OUTER`, `LOOKUP`, `MULTI_VALUE_LOOKUP`, `CROSS` | | +| `leftKeys` | `string[]` | Join keys on the left (first) source. | +| `rightKeys` | `string[]` | Join keys on the right (second) source. | +| `leftQualifier` | `string` (optional) | Prefix for left-side field names in the output. Often omitted. | +| `rightQualifier` | `string` (required) | Prefix for right-side fields (e.g., `SalesOrder` → right-side field `ssot__Id__c` becomes `SalesOrder.ssot__Id__c`). | + +**Optional node-level** `schema.slice`: +- `{ mode: "DROP" | "SELECT", fields: string[], ignoreMissingFields: bool }` +- Used to trim unwanted fields from the joined output. + +**Lineage effect.** Combines two field sets. Fields from the right side get renamed with +`rightQualifier` prefix. Row cardinality depends on join type: +- `INNER` — only matched pairs. +- `LEFT_OUTER` — all left rows + matching right. +- `RIGHT_OUTER` — all right rows + matching left. +- `OUTER` — all rows from both sides. +- `LOOKUP` — 1:1 left-side preserving. +- `MULTI_VALUE_LOOKUP` — left-side preserving; multi-valued right. +- `CROSS` — Cartesian product; every left paired with every right. + +**Gotchas.** +- Data types of joined keys should match. Type mismatches cause silent no-match or errors + depending on pair. +- `rightQualifier` is the *only* way to disambiguate same-named fields from two sources. + Always surface it when narrating. +- `schema.slice` on a join is a post-join projection, not a pre-join filter. + +**Example.** +```jsonc +{ + "action": "join", + "sources": ["JOIN7", "FILTER4"], + "parameters": { + "joinType": "LEFT_OUTER", + "leftKeys": ["ssot__Id__c"], + "rightQualifier": "SalesOrder", + "rightKeys": ["ssot__SalesOrderId__c"] + }, + "schema": { + "slice": { + "mode": "DROP", + "ignoreMissingFields": true, + "fields": ["SalesOrder.ssot__InternalOrganizationId__c"] + } + } +} +``` + +**Source.** `JoinNodeInputRepresentation` + `JoinParametersInputRepresentation`. + +--- + +## `action: "filter"` — UI: "Filter" + +**Purpose.** Keep only rows satisfying filter criteria combined by boolean logic. + +**Key parameters** (class `FilterParametersInputRepresentation`): + +| Param | Type | Notes | +|---|---|---| +| `filterExpressions` | `FilterExpression[]` | Each expression is `{ field, operator, type, operands[] }`. | +| `filterBooleanLogic` | `string` | A formula like `"1 AND (2 OR 3)"` indexed 1..N into `filterExpressions`. If omitted, default is all AND. | + +**FilterExpression fields** (class `FilterExpressionInputRepresentation`): +- `field` — the field the filter examines. +- `operator` — `string` (no typed enum in the canonical input-rep; free-form in the API). + Commonly observed values: `EQUAL`, `NOT_EQUAL`, `GREATER_THAN`, `LESS_THAN`, + `GREATER_OR_EQUAL`, `LESS_OR_EQUAL`, `IN_RANGE`, `LIKE`, `IS_NULL`, `IS_NOT_NULL`. +- `type` — enum `DataType` (same as elsewhere in BDT): `TEXT`, `NUMBER`, `BOOLEAN`, + `DATE_ONLY`, `DATETIME`. +- `operands` — array of operand values (strings in JSON; interpreted per `type`). + +**Lineage effect.** Does not change field names; only reduces rows. + +**Gotchas.** +- `filterBooleanLogic` operands are **1-indexed** into `filterExpressions`. Be careful when + narrating which expression is "expression 1". +- If a referenced field is dropped upstream, the filter becomes invalid at runtime. + +**Example.** +```jsonc +{ + "action": "filter", + "sources": ["LOAD_DATASET0"], + "parameters": { + "filterExpressions": [ + {"type": "TEXT", "field": "MobilePhone_Formatted_Flag__c", "operator": "EQUAL", "operands": ["Y"]}, + {"type": "TEXT", "field": "IsActive__c", "operator": "EQUAL", "operands": ["true"]} + ], + "filterBooleanLogic": "1 AND 2" + } +} +``` + +**Source.** `FilterNodeInputRepresentation` + `FilterParametersInputRepresentation` + +`FilterExpressionInputRepresentation`. + +--- + +## `action: "sqlFilter"` — UI: "SQL Filter" + +**Purpose.** A filter whose predicate is a raw SQL expression — more expressive than the +structured `filter` node, at the cost of being harder to validate statically. + +**Key parameters** (class `SqlFilterParametersInputRepresentation`): + +| Param | Type | Notes | +|---|---|---| +| `sqlFilterExpression` | `string` | A SQL WHERE-clause-like predicate referring to fields by name. | + +**Lineage effect.** Same as `filter` — reduces rows only. + +**Example.** +```jsonc +{ + "action": "sqlFilter", + "sources": ["JOIN1"], + "parameters": { + "sqlFilterExpression": "ssot__CreatedDate__c >= current_date() - interval '30' day" + } +} +``` + +**Source.** `SqlFilterNodeInputRepresentation` + `SqlFilterParametersInputRepresentation`. + +--- + +## `action: "formula"` — UI: "Formula" + +**Purpose.** Add one or more derived columns via per-row SQL formulas. No window / cross-row +semantics — for that, see `computeRelative`. + +**Key parameters** (class `FormulaParametersInputRepresentation`): + +| Param | Type | Notes | +|---|---|---| +| `expressionType` | enum `FormulaExpressionType`: `SQL`, `DCSQL` | | +| `fields` | `SqlFormulaFieldInputRepresentation[]` | One entry per derived field. | + +**Each field** carries: +- `name` — the output field name. +- `label` — user-visible label. +- `formulaExpression` — SFSQL (or DCSQL) expression. See `bdt-function-catalog.md`. +- `type` — enum `DataType`: `TEXT`, `NUMBER`, `BOOLEAN`, `DATE_ONLY`, `DATETIME`. +- **`businessType`** — a user-facing business-semantic type name. Enum `BusinessTypeEnum` — + **canonical values** (complete list; matches `BusinessTypeEnum.java` as of the capture date + at the top of this file): + - `"TEXT"`, `"NUMBER"`, `"BOOLEAN"` + - `"EMAIL"`, `"PHONE"`, `"URL"` — text-valued with semantic meaning + - `"PERCENT"`, `"CURRENCY"` — number-valued with semantic meaning + - `"DATE"` — semantically a date, stored as datetime at the underlying `type` level + - `"DATE_ONLY"` — date-only value (no time component) + - `"DATETIME"` — full datetime + + Note: `businessType` values map to underlying `type` (`DataType`) values. E.g., + `businessType: "PERCENT"` is stored as `type: "NUMBER"`; `businessType: "DATE"` is stored + as `type: "DATETIME"`; `businessType: "EMAIL"` is stored as `type: "TEXT"`. If the skill + ever encounters a `businessType` value outside this list, that is a sign the upstream BDT + schema has evolved — surface the raw value in narration and flag it as undocumented. +- `precision` — integer precision (default 10 for numbers; characters for text). +- `scale` — decimal places; only for NUMBER. +- `defaultValue` — value when the expression yields NULL. + +**Lineage effect.** Adds new columns to the downstream row stream. Original columns pass +through unchanged unless dropped later by a `schema` node. Row cardinality unchanged. + +**Gotchas.** +- If a field referenced in `formulaExpression` is removed upstream, the formula fails at + runtime. +- **`type` and `businessType`** — both use UPPER wire form per canonical enums + (`type: "NUMBER"`, `businessType: "NUMBER"`). Some user-authored BDT JSON + may show capitalized forms (`"Number"`) — the runtime is case-insensitive + per `BusinessTypeEnum.valueOfInternal()`, but the canonical wire form is + UPPER. +- Concrete sub-classes exist for typed fields (`SqlFormulaNumericFieldInputRepresentation`, + etc.) but the JSON shape is the same. + +**Example.** +```jsonc +{ + "action": "formula", + "sources": ["JOIN5"], + "parameters": { + "expressionType": "SQL", + "fields": [ + { + "name": "ssot__SalesOrderProductConcat__c", + "label": "SalesOrderProductConcat", + "formulaExpression": "concat(coalesce(\"SalesOrder.ssot__Id__c\",'NULL'),coalesce(\"SalesOrder.ssot__Id__c\",''))", + "type": "TEXT", + "businessType": "TEXT", + "precision": 60, + "defaultValue": "" + } + ] + } +} +``` + +**Source.** `FormulaNodeInputRepresentation` + `FormulaParametersInputRepresentation` + +`SqlFormulaFieldInputRepresentation`. + +--- + +## `action: "computeRelative"` — UI: "Window Transform" + +**Purpose.** A formula that evaluates a **window function** over partitioned, ordered rows. +Used for ranking, lead/lag, running totals, etc. + +**Key parameters** (class `ComputeRelativeParametersInputRepresentation`): + +| Param | Type | Notes | +|---|---|---| +| `partitionBy` | `string[]` | Column(s) to partition rows by. Empty = whole stream. | +| `orderBy` | `ComputeRelativeSortParametersInputRepresentation[]` | Each `{ fieldName, direction: "ASC" \| "DESC" }`. | +| `expressionType` | enum `FormulaExpressionType`: `SQL`, `DCSQL` | | +| `fields` | `SqlFormulaFieldInputRepresentation[]` | Same shape as `formula` fields. | + +**Lineage effect.** Adds one or more columns. Original columns pass through. Row cardinality +unchanged. + +**Gotchas.** +- The documentation states "A formula can include only one Compute Relative function" — if + you see multiple compute-relative calls in one expression, flag as unusual. +- If `partitionBy` is empty, the function runs over the whole stream (one big window). +- `orderBy` is required for order-dependent functions (`row_number`, `rank`, `lag`, etc.). + Without it, results are non-deterministic. + +**Example.** +```jsonc +{ + "action": "computeRelative", + "sources": ["LOAD_ORDERS"], + "parameters": { + "partitionBy": ["ssot__AccountId__c"], + "orderBy": [{"fieldName": "ssot__CreatedDate__c", "direction": "ASC"}], + "expressionType": "SQL", + "fields": [ + { + "name": "OrderRank__c", + "label": "Order Rank", + "formulaExpression": "row_number()", + "type": "NUMBER", + "businessType": "NUMBER", + "precision": 18, + "scale": 0, + "defaultValue": "" + } + ] + } +} +``` + +**Source.** `ComputeRelativeNodeInputRepresentation` + +`ComputeRelativeParametersInputRepresentation` + +`ComputeRelativeSortParametersInputRepresentation`. For the function catalog, see +`bdt-window-functions.md`. + +--- + +## `action: "aggregate"` — UI: "Aggregate" / "Group and Aggregate" + +**Purpose.** Group-by aggregation. Also supports a hierarchical mode for parent/child aggregation. + +**Key parameters** (class `AggregateParametersInputRepresentation`): + +| Param | Type | Notes | +|---|---|---| +| `groupings` | `string[]` | Group-by field names. **Not** `groupBy`. | +| `aggregations` | `AggregateInputRepresentation[]` | Each `{ action: AggregateType, name, source, label? }`. | +| `nodeType` | enum `AggregateNodeEnum`: `STANDARD`, `HIERARCHICAL` | | +| `selfField`, `parentField`, `percentageField` | `string` (hierarchical-only) | | +| `pivot_v2` | `PivotV2InputRepresentation` (optional) | Advanced; pivot in the same node. | + +**Aggregation functions** (enum `AggregateType`): +`UNIQUE`, `SUM`, `AVG`, `COUNT`, `MAX`, `MIN`, `MEDIAN`, `STDDEVP`, `STDDEV`, `VARP`, `VAR`. + +**Lineage effect.** +- Output columns = `groupings` (pass-through) + each `aggregation.name` (new derived column). +- Row cardinality: one output row per unique combination of `groupings`. + +**Gotchas.** +- Group-by column is called `groupings` not `groupBy` in the JSON. +- `aggregation.source` is the field being aggregated; `aggregation.name` is the output column name. +- In `HIERARCHICAL` mode, the three `*Field` parameters carry parent-child semantics; the + aggregations roll up across the hierarchy. + +**Example.** +```jsonc +{ + "action": "aggregate", + "sources": ["FILTER0"], + "parameters": { + "groupings": ["ssot__AccountId__c"], + "aggregations": [ + {"action": "SUM", "name": "TotalAmount__c", "source": "ssot__GrandTotalAmount__c"}, + {"action": "COUNT", "name": "OrderCount__c", "source": "ssot__Id__c"} + ], + "nodeType": "STANDARD" + } +} +``` + +**Source.** `AggregateNodeInputRepresentation` + `AggregateParametersInputRepresentation` + +`AggregateInputRepresentation`. + +--- + +## `action: "schema"` — UI: "Edit Attributes" / "Drop Fields" + +**Purpose.** Modify column-level schema: rename columns, change properties, or slice the set +of columns. + +**Key parameters** (class `SchemaParametersInputRepresentation`): + +| Param | Type | Notes | +|---|---|---| +| `fields` | `SchemaFieldParametersInputRepresentation[]` | Per-field: `{ name, newProperties: { name?, label? } }`. | +| `slice` | `SchemaSliceInputRepresentation` (optional) | `{ mode: "DROP" \| "SELECT", fields: string[], ignoreMissingFields: bool }`. | + +**Slice semantics.** +- `mode: DROP` — remove the listed fields from the output. +- `mode: SELECT` — keep only the listed fields. +- `ignoreMissingFields: true` means listed fields that aren't present are silently ignored. + Without this flag, missing fields would error. + +**Lineage effect.** Schema-only: renames or drops columns. Row cardinality unchanged. + +**Gotchas.** +- `ignoreMissingFields: true` plus a typo = silent failure. When explaining a schema node + that uses it, flag the risk ("these field names are silently skipped if missing"). +- `fields[].newProperties.name` is where a rename lands; the original `name` is the key. + +**Example (drop).** +```jsonc +{ + "action": "schema", + "sources": ["FORMULA43"], + "parameters": { + "slice": { + "mode": "DROP", + "ignoreMissingFields": true, + "fields": ["ssot__SalesOrderProductConcat__c", "FirstPurchase__c"] + } + } +} +``` + +**Example (rename).** +```jsonc +{ + "action": "schema", + "sources": ["AGGREGATE0"], + "parameters": { + "fields": [ + {"name": "sum_amount", "newProperties": {"name": "TotalAmount__c", "label": "Total Amount"}} + ] + } +} +``` + +**Source.** `SchemaNodeInputRepresentation` + `SchemaParametersInputRepresentation` + +`SchemaFieldParametersInputRepresentation` + `SchemaSliceInputRepresentation`. + +--- + +## `action: "outputD360"` — UI: "Output" / "Writeback" + +**Purpose.** Writes the row stream to a target DMO or DLO. Every BDT has at least one +outputD360 node — these are the graph sinks. + +**Key parameters** (class `OutputD360ParametersInputRepresentation`): + +| Param | Type | Notes | +|---|---|---| +| `name` | `string` | Target object's API name (e.g., `Account_Upper__dlm`). | +| `type` | enum `D360OutputTypeEnum`: `dataModelObject`, `dataLakeObject` | | +| `writeMode` | enum `WriteModeEnum`: `APPEND`, `MERGE`, `OVERWRITE`, `MERGE_UPSERT_DELETE`, `DELETE_ONLY` | Write semantics. | +| `fieldsMappings` | `OutputD360FieldsMappingInputRepresentation[]` | Each `{ sourceField, targetField }` pair. | +| `dedupOrder` | `SortSpecificationRepresentation[]` | Tiebreaker for deduplicating records with the same primary key. | +| `streaming` | `StreamingParametersInputRepresentation` (optional) | Streaming-specific; not relevant for BDT. | + +**Write modes.** +- `APPEND` — add rows; no primary-key checks. +- `OVERWRITE` — replace the target with the dataset. +- `MERGE` — merge on PK: update matching rows (only for columns present in the input), insert + new rows. +- `MERGE_UPSERT_DELETE` — merge with per-row UPSERT/DELETE markers. +- `DELETE_ONLY` — delete matching rows. + +The underlying DaaS library supports more modes (`OVERWRITE_PARTITIONS`, +`OVERWRITE_PARTITION_FILTER`, `SECONDARY_INDEX_INCREMENTAL_WRITE`), but the BDT Connect API +exposes only the five above. + +**Lineage effect.** Terminal node. Maps source-stream fields to target-object fields; any +unmapped source field is discarded. + +**Gotchas.** +- `OUTPUT0 nodes must have DataModelObject type` is a known restriction in some frameworks + (data-kit templates). If a BDT has `type: dataLakeObject` and is failing to install in a + template context, that may be the cause. +- If multiple input rows have the same primary key, `dedupOrder` decides which wins. +- Fields not listed in `fieldsMappings` are not written. + +**Example.** +```jsonc +{ + "action": "outputD360", + "sources": ["DROP_FIELDS0"], + "parameters": { + "name": "Kohler_Internal_Users__dlm", + "type": "dataModelObject", + "writeMode": "MERGE", + "fieldsMappings": [ + {"sourceField": "Formatted_MobilePhone__c", "targetField": "Formatted_MobilePhone__c"}, + {"sourceField": "Kohler_Internal_User_Flag", "targetField": "Kohler_Internal_User_Flag__c"} + ] + } +} +``` + +**Source.** `OutputD360NodeInputRepresentation` + `OutputD360ParametersInputRepresentation` + +`OutputD360FieldsMappingInputRepresentation`. + +--- + +## `action: "appendV2"` — UI: "Append" + +**Purpose.** Union rows from two or more upstream streams into one output. + +**Key parameters** (class `AppendParametersInputRepresentation`): + +| Param | Type | Notes | +|---|---|---| +| `columnMappings` | `Map` | Map of source-stream-node-name → column mapping. | +| `fieldMappings` | `AppendMappingInputRepresentation[]` | Explicit per-field mappings. | +| `allowImplicitDisjointSchema` | `boolean` | When true, sources with different fields are merged; missing fields become NULL. | + +**Lineage effect.** +- Rows: union of input streams. +- Columns: the merged schema (union if `allowImplicitDisjointSchema`; otherwise intersection). + +**Gotchas.** +- Source nodes must have the same column count *and* matching column names (in order), unless + `allowImplicitDisjointSchema: true`. +- Append can accept up to 200 fields total from its sources (per help docs). + +**Example.** +```jsonc +{ + "action": "appendV2", + "sources": ["LOAD_A", "LOAD_B"], + "parameters": { + "fieldMappings": [ + {"targetField": "Id__c", "sources": [{"node": "LOAD_A", "field": "ssot__Id__c"}, + {"node": "LOAD_B", "field": "ssot__Id__c"}]} + ] + } +} +``` + +**Source.** `AppendV2NodeInputRepresentation` + `AppendParametersInputRepresentation` + +`AppendMappingInputRepresentation`. + +--- + +## `action: "split"` — UI: "Split" + +**Purpose.** Split the value of one source string field into multiple target columns based on a delimiter. One row in, one row out — each row's `sourceField` is split into the named `targetFields`. + +**Key parameters** (class `SplitParametersInputRepresentation`): + +| Param | Type | Notes | +|---|---|---| +| `sourceField` | `string` | Name of the field whose value will be split. | +| `delimiter` | `string` | Delimiter used to split the source value. | +| `targetFields` | `{name, label}[]` | One entry per column the split produces. Order matches the left-to-right order of the split parts. | + +**Lineage effect.** +- Rows: unchanged. Row cardinality is preserved — split does not route rows into branches. +- Columns: adds each `targetFields[i].name` as a new column. The original `sourceField` passes through unchanged. + +**Gotchas.** +- `split` is string-splitting, not row-routing. If you need to route rows into multiple branches based on predicates, use `filter` nodes downstream of a common source, not `split`. +- If a row's `sourceField` has fewer delimited parts than the `targetFields` length, the remaining target columns are populated with NULL (no error). +- If a row's `sourceField` has more delimited parts than `targetFields` length, the extra parts are discarded. + +**Example.** +```jsonc +{ + "action": "split", + "sources": ["LOAD_RAW"], + "parameters": { + "sourceField": "FullName__c", + "delimiter": " ", + "targetFields": [ + {"name": "FirstName__c", "label": "First Name"}, + {"name": "LastName__c", "label": "Last Name"} + ] + } +} +``` + +**Source.** `SplitNodeInputRepresentation` + `SplitParametersInputRepresentation` + `NameLabelInputRepresentation`. + +--- + +## `action: "flatten"` / `"flattenJson"` — UI: "Flatten" / "Flatten JSON" + +**Purpose.** +- `flatten` — flatten a nested (array-valued or structured) field into additional rows. +- `flattenJson` — flatten a JSON string field by parsing it and emitting its fields as + columns (or extracting array elements as rows via a subsequent `extractTable`). + +**Key parameters** (`FlattenParametersInputRepresentation`, +`FlattenJsonParametersInputRepresentation`): + +- `fields` — list of `FlattenFieldInputRepresentation { name, attributePath?, label? }`. +- For JSON, a schema description may be embedded. + +**Lineage effect.** May increase rows (array-explode) or add columns (object-flatten). + +**Source.** `FlattenNodeInputRepresentation`, `FlattenJsonNodeInputRepresentation`. + +--- + +## `action: "extractGrains"` — UI: "Extract Grains" + +**Purpose.** Expand rows across time grains — e.g., a date column + list of grains produces +one row per source row per grain. + +**Key parameters** (class `ExtractGrainParametersInputRepresentation`): + +- `grainExtractions` — each `{ source, targets: [{ name, label, grainType }] }`. +- `dateConfigurationName` — which date configuration (fiscal calendar, etc.) to use. + +**Valid `grainType` values** (enum `DateGrain`): +`YEAR, QUARTER, MONTH, WEEK, DAY, HOUR, MINUTE, SECOND, DAY_EPOCH, SEC_EPOCH, FISCAL_YEAR, +FISCAL_QUARTER, FISCAL_MONTH, FISCAL_WEEK`. + +**Lineage effect.** Usually adds one column per grain type; may or may not multiply rows +depending on configuration. + +**Source.** `ExtractGrainNodeInputRepresentation` + `ExtractGrainParametersInputRepresentation`. + +--- + +## `action: "extractTable"` — UI: "Extract Table" + +**Purpose.** Pairs with `flattenJson`: extract a named table (from a JSON array) as a +separate output stream. + +**Key parameters.** See `ExtractTableParametersInputRepresentation`. + +**Source.** `ExtractTableNodeInputRepresentation`. + +--- + +## `action: "typeCast"` — UI: "Type Cast" + +**Purpose.** Cast one or more fields to new types. + +**Key parameters.** See `TypecastParametersInputRepresentation`, +`SchemaTypePropertiesCastInputRepresentation`. + +**Lineage effect.** Schema-only; row cardinality unchanged. + +--- + +## `action: "bucket"` — UI: "Bucket Date/Dimension/Measure" + +**Purpose.** Assign bucket labels to values of a source field, per a bucket setup. + +**Key parameters** (class `BucketParametersInputRepresentation`): + +- `fields` — list of `BucketFieldInputRepresentation`, polymorphic on the field type: + - Boolean: `BucketBooleanFieldInputRepresentation` + - DateOnly: `BucketDateOnlyFieldInputRepresentation` + - DateTime: `BucketDateTimeFieldInputRepresentation` + - Dimension: `BucketDimensionFieldInputRepresentation` + - Measure: `BucketMeasureFieldInputRepresentation` + +Each sub-class includes a `setup` object describing the buckets (ranges, algorithms, labels). + +**Algorithm type** (enum `BucketAlgorithmType`): `TYPOGRAPHIC_CLUSTERING` (and potentially +others — verify against sample if needed). + +**Lineage effect.** Adds a derived bucket-label column. Row cardinality unchanged. + +--- + +## `action: "formatDate"` — UI: "Format Dates" + +**Purpose.** Reformat date fields — e.g., parse a custom string format into a date type, +or produce a formatted text representation. + +**Key parameters.** See `FormatDateParametersInputRepresentation`, +`FormatDatePatternInputRepresentation`. + +--- + +## `action: "update"` — UI: "Update" + +**Purpose.** Update records in place in the pipeline (used in specific update workflows). + +**Key parameters.** See `UpdateParametersInputRepresentation`. + +--- + +## `action: "extension"` / `"extensionFunction"` — UI: custom extensions + +**Purpose.** Run a custom extension node / function registered with Data Cloud. + +**Key parameters.** See `ExtensionParametersInputRepresentation`, +`ExtensionFunctionParametersInputRepresentation`, +`ExtensionFunctionOutputFieldInputRepresentation`. + +**Gotcha.** Extensions are user-defined; explanation must rely on parameter content since +the semantics are defined outside the BDT spec. + +--- + +## `action: "cdpPredict"` — UI: "Predict" + +**Purpose.** Apply a prediction model (CDP Predict / Einstein). + +**Key parameters.** See `CdpPredictNodeInputRepresentation` + +`CdpPredictParametersInputRepresentation` + `PredictionFieldInputRepresentation` + +`PredictSourceInputRepresentation`. + +--- + +## `action: "jsonAggregate"` — UI: "JSON Aggregate" + +**Purpose.** Aggregate JSON-valued fields into structured output. + +**Key parameters.** See the `JsonAggregateEnum` enum and the relevant input representations +(currently sparsely documented — inspect raw parameters when encountered). + +--- + +## `action: "save"` — UI: "Save" + +**Purpose.** Save an intermediate result to a checkpoint (not a final output). Less common. + +--- + +## `action: "recommendation"` — UI: "Recommendation" + +**Purpose.** Apply a recommendation model (product recommendations). + +**Key parameters.** See `PredictionContributorInputRepresentation`. + +--- + +## Common `schema.slice` block (can appear on most node types) + +Many nodes accept an optional top-level `schema.slice` block to trim fields at the node +boundary. Its shape is always the same: + +```jsonc +"schema": { + "slice": { + "mode": "DROP" | "SELECT", + "fields": ["Field1", "Field2"], + "ignoreMissingFields": true + } +} +``` + +- `DROP` — remove the listed fields. +- `SELECT` — keep only the listed fields. +- `ignoreMissingFields: true` — silent skip when a listed field doesn't exist. + +--- + +## Polymorphism map (for JSON parsers) + +Several input reps are polymorphic; the discriminator is a JSON property: + +- `AbstractBucketAlgorithmInputRepresentation` discriminated by `"type"`: + - `"TYPOGRAPHIC_CLUSTERING"` → `TypographicClusterInputRepresentation`. + +Other polymorphic classes (e.g., `SqlFormulaFieldInputRepresentation` → numeric / text / +boolean / date variants) are resolved at deserialization based on `type`. When narrating, +the field-level JSON usually carries the concrete fields directly — no special handling +needed. + +## Source citations + +Every section above traces back to one or more classes in the BDT Connect API +`cdp-connect-api` module, packages `sfdc.cdp.connect.api.{input,enums}.datatransform`. +See `research/bdt-schema-canonical.md` for the full machine-extracted schema plus +version-drift verification across releases 260, 262, and 264. diff --git a/skills/explaining-batch-data-transform/references/bdt-reference.md b/skills/explaining-batch-data-transform/references/bdt-reference.md new file mode 100644 index 0000000..d00a50e --- /dev/null +++ b/skills/explaining-batch-data-transform/references/bdt-reference.md @@ -0,0 +1,179 @@ +# BDT Reference — Overview (always loaded) + +> **Last synced:** 2026-04-23 from canonical BDT Connect API Java sources +> (`cdp-connect-api` module, `sfdc.cdp.connect.api.{input,enums}.datatransform` packages, +> release 264). Cross-verified identical across releases 260, 262, and 264. See +> `research/bdt-schema-canonical.md` for the full machine-extracted schema. + +## What a BDT is + +**Batch Data Transform (BDT)** — a repeatable, scheduled or on-demand data pipeline inside +Salesforce Data Cloud. It reads from one or more source objects, applies a DAG of +transformations (joins, filters, formulas, aggregations, etc.), and writes one or more +outputs. BDTs run over batches; their streaming counterpart is **SDT** (Streaming Data +Transform). + +- Salesforce object type: `MktDataTransform`. +- Typical export: a JSON file from the BDT viewer, or via Workbench + `/ssot/data-transforms?htmlEncode=false`. +- Underlying runtime: the Data Processing Engine (DPE) / DCSQL. + +## Top-level JSON shape + +A BDT JSON can arrive in **three** shapes. The skill's parser accepts all three +transparently (since commit `ee2485c`); the inner node graph is identical in every case. + +### 1. Editor export (the default "download" shape) + +What you get when you export a BDT from the BDT editor UI, or fetch it via +`/ssot/data-transforms?htmlEncode=false` on Workbench. The outer object is the +**definition** itself, with no wrapper metadata: + +```jsonc +{ + "version": "66.0", // Schema version string + "nodes": { /* NodeName → Node */ }, + "ui": { // Optional: layout hints + human-readable labels + "nodes": { /* NodeName → { label, description, type, top, left } */ }, + "connectors": [ /* { source, target } edges */ ], + "hiddenColumns": [] + } +} +``` + +**Note:** editor exports have no top-level `type` field. The definition-type +discriminator (`"STL"`) only appears when the definition is nested inside a Connect +API payload (see below). + +### 2. Connect API create payload — single definition + +What developers POST to the `MktDataTransform` Connect API when authoring +programmatically. The outer object is a `DataTransformInputRepresentation`; the +editor-export shape is nested inside under `definition`: + +```jsonc +{ + "name": "MyTransform", // Required — API name of the transform + "label": "My Transform", // Required — user-visible label + "description": "…", // Optional + "type": "BATCH", // DataTransformType: "BATCH" | "STREAMING" + "dataSpaceName": "default", // Data space the transform lives in + "definition": { + "version": "66.0", + "type": "STL", // DataTransformDefinitionType: "STL" | "SQL" | "DCSQL" | … + "nodes": { /* … */ }, + "ui": { /* … */ } + } +} +``` + +Two distinct `type` fields live at different nesting levels: +- Outer `type` = `DataTransformType` (batch vs streaming). +- Inner `definition.type` = `DataTransformDefinitionType` (STL for batch BDTs; SQL / + DCSQL for raw-SQL transforms; hidden variants exist but are not surfaced to users). + +### 3. Connect API create payload — multi-definition variant + +For transforms with multiple definitions. Same outer fields; instead of a single +`definition` object there's a `definitions` array: + +```jsonc +{ + "name": "…", "label": "…", "type": "BATCH", "dataSpaceName": "default", + "definitions": [ + { "version": "66.0", "type": "STL", "nodes": { /* … */ }, "ui": { /* … */ } }, + /* … additional definitions … */ + ] +} +``` + +v1 of this skill reads the first definition and explains only that one; multi-definition +explanations are out of scope. + +### Each node (inner shape, identical in all three outer shapes) + +```jsonc +{ + "action": "", // One of the 25 documented action strings below + "sources": ["NODE_A", "NODE_B"], // Array of node names (empty for load/roots) + "parameters": { /* action-specific */ }, + "schema": { /* optional — slice/rename at node boundary */ } +} +``` + +## The 25 canonical action types + +| `action` string | Purpose (one line) | +|---|---| +| `load` | Read rows from a DMO or DLO | +| `filter` | Keep rows matching structured filter criteria | +| `sqlFilter` | Keep rows matching a raw SQL predicate | +| `join` | Join two upstream streams by key | +| `formula` | Add derived columns via per-row SQL formulas | +| `computeRelative` | Window function over partitioned / ordered rows | +| `aggregate` | Group-by aggregation | +| `extractGrains` | Fan out rows across time grains | +| `extractTable` | Extract a table from a flatten-JSON output | +| `schema` | Rename / drop / reshape columns | +| `outputD360` | Write rows to a target DMO or DLO | +| `appendV2` | Union rows from multiple upstream streams (v2 variant) | +| `flatten` | Flatten nested structure | +| `flattenJson` | Flatten a JSON field into rows/columns | +| `split` | Split rows into multiple downstream branches | +| `typeCast` | Cast field types | +| `update` | Update records | +| `bucket` | Assign bucket labels to field values | +| `formatDate` | Reformat date values | +| `extension` | Run a custom extension node | +| `extensionFunction` | Run a custom extension function | +| `cdpPredict` | Apply a prediction model | +| `jsonAggregate` | Aggregate JSON fields | +| `save` | Save an intermediate result | +| `recommendation` | Apply a recommendation model | + +**For full parameter shapes, enums, and examples: see `references/bdt-node-catalog.md`.** +**For SFSQL functions used in `formula` expressions: see `references/bdt-function-catalog.md`.** +**For window functions used in `computeRelative`: see `references/bdt-window-functions.md`.** + +## Terminology + +- **DMO** — Data Model Object. Canonical modeled entity. Suffix convention: `__dlm`. +- **DLO** — Data Lake Object. Raw-lake table. Suffix convention: `__dll`. +- **SFSQL** — Salesforce's internal SQL dialect used in formula expressions and SQL filters. +- **DCSQL** — a related SQL dialect used for Spark push-down execution (you may see it as + an `expressionType` value). +- **`ssot__` prefix** — Salesforce Standard Schema field, canonical / semantic. +- **`__c` suffix** — Custom field. +- **`KQ_` prefix** — Key-qualifier pattern (external ID / composite-key helper). Observed in + practice; not always documented. +- **UI label vs. JSON action name** — The JSON uses concise strings (e.g., `computeRelative`); + the BDT editor shows friendlier labels (e.g., "Window Transform"). The `ui.nodes[X].label` + field in the JSON carries the user-specified human-readable name per node. **Always use + the UI label in explanations when present.** + +## Key enums (for quick orientation) + +- **`DataTransformType`**: `BATCH`, `STREAMING` +- **`JoinType`**: `INNER`, `OUTER`, `LEFT_OUTER`, `RIGHT_OUTER`, `LOOKUP`, `MULTI_VALUE_LOOKUP`, `CROSS` +- **`WriteModeEnum`** (outputD360): `APPEND`, `MERGE`, `OVERWRITE`, `MERGE_UPSERT_DELETE`, `DELETE_ONLY` +- **`D360OutputTypeEnum`**: `dataLakeObject`, `dataModelObject` +- **`SliceMode`**: `SELECT`, `DROP` +- **`AggregateType`**: `UNIQUE`, `SUM`, `AVG`, `COUNT`, `MAX`, `MIN`, `MEDIAN`, `STDDEVP`, `STDDEV`, `VARP`, `VAR` +- **`DataType`**: `TEXT`, `NUMBER`, `DATE_ONLY`, `DATETIME`, `BOOLEAN` +- **`SortDirection`**: `ASC`, `DESC` + +For exhaustive enum lists with semantics: `bdt-node-catalog.md` or (for the raw extraction) +`research/bdt-schema-canonical.md`. + +## How this skill uses these references + +- `bdt-reference.md` (this file) — always in context. +- `bdt-node-catalog.md` — consulted before explaining any node type not previously covered + in the current conversation. +- `bdt-function-catalog.md` — consulted before interpreting any `formulaExpression` (job I1) + or reasoning about conditional logic (I2). +- `bdt-window-functions.md` — consulted whenever a `computeRelative` node is narrated. + +**If a node type or function isn't in any of the reference files**, the skill narrates what's +visible in the JSON and explicitly flags the item as undocumented in the materials available. +It does not guess at semantics. diff --git a/skills/explaining-batch-data-transform/references/bdt-window-functions.md b/skills/explaining-batch-data-transform/references/bdt-window-functions.md new file mode 100644 index 0000000..664a630 --- /dev/null +++ b/skills/explaining-batch-data-transform/references/bdt-window-functions.md @@ -0,0 +1,73 @@ +# Window Functions — for `computeRelative` nodes + +> **Last synced:** 2026-04-23 from the SFSQL window-functions reference and the BDT +> canonical schema. Consult this file whenever narrating a `computeRelative` node or +> explaining a window-function expression. + +## When this applies + +`computeRelative` nodes evaluate a **window function** over rows. The `parameters`: + +```jsonc +{ + "partitionBy": ["ssot__AccountId__c"], // → SQL `PARTITION BY` + "orderBy": [ // → SQL `ORDER BY` + {"fieldName": "ssot__CreatedDate__c", "direction": "ASC"} + ], + "expressionType": "SQL", // or "DCSQL" + "fields": [ + { + "name": "OrderRank__c", + "formulaExpression": "row_number()", // the window function call + "type": "NUMBER", "businessType": "Number", + "precision": 18, "scale": 0 + } + ] +} +``` + +The `formulaExpression` names the window function; partitioning and ordering come from the +top-level `partitionBy` and `orderBy`. **A computeRelative node may include at most one +compute-relative function per expression** (per upstream BDT docs). + +## Available window functions + +| Function | Returns | What it does | +|---|---|---| +| `row_number()` | NUMBER | 1, 2, 3… for each row in its partition, in the given order. Non-deterministic when sort keys tie. | +| `rank()` | NUMBER | Like `row_number` but peers share a rank; next rank after N peers is N+1 (gaps). | +| `dense_rank()` | NUMBER | Like `rank` but no gaps — consecutive integers even with ties. | +| `percent_rank()` | NUMBER | `(rank - 1) / (partition_rows - 1)` — relative rank within partition, 0 to 1. | +| `cume_dist()` | NUMBER | Cumulative distribution: fraction of partition rows at or before current. | +| `ntile(n)` | NUMBER | Bucket number 1..n, dividing partition rows as evenly as possible. | +| `lag(value)` / `lag(value, offset)` / `lag(value, offset, default)` | same as value | Value at offset rows *before* current (default offset=1; default if no such row is NULL unless a default is supplied). | +| `lead(value)` / `lead(value, offset)` / `lead(value, offset, default)` | same as value | Symmetric with `lag` but looks forward. | +| `first_value(value)` | same as value | Value at the first row of the current window frame. | +| `last_value(value)` | same as value | Value at the last row of the frame. Default frame ends at "current + peers", which is often *not* what users want. | +| `nth_value(value, n)` | same as value | Value at the nth row of the frame (counting from 1). NULL if no such row. | +| Any aggregate with `OVER(...)` | depends | Runs the aggregate over the window (running sum, etc.). | + +## How this maps to BDT JSON + +- `partitionBy` is the SQL `PARTITION BY` — the columns that group rows into windows. +- `orderBy` is the SQL `ORDER BY` — the ordering within each partition. +- **Peers** are rows with identical sort keys. +- Default **frame** (when not otherwise specified): rows from the first row of the partition + through the current row's last peer. For `last_value` and `nth_value` this is often not + the user's intent — narrate accordingly. + +## Common narration patterns + +- **`row_number()` partitioned by X** → "numbers each row within the same X, in the order + given by `orderBy`." +- **`rank() partitioned by X order by Y`** → "ranks rows within each X group by Y; ties + share a rank and the next rank has gaps." +- **`case when row_number()=1 then VAL else 0 end`** → "keeps VAL only on the first ranked + row per partition; everything else is 0. This is the canonical 'first-occurrence extract' + idiom." + +## Sources + +- SFSQL window-functions reference (internal Data Cloud / SDB SFSQL docs). +- BDT canonical schema: `ComputeRelativeParametersInputRepresentation`, + `ComputeRelativeSortParametersInputRepresentation`. diff --git a/skills/explaining-batch-data-transform/scripts/bdt_analyze.py b/skills/explaining-batch-data-transform/scripts/bdt_analyze.py new file mode 100644 index 0000000..3ed71f1 --- /dev/null +++ b/skills/explaining-batch-data-transform/scripts/bdt_analyze.py @@ -0,0 +1,1381 @@ +#!/usr/bin/env python3 +"""Analyze Salesforce Data Cloud Batch Data Transform (BDT) JSON. + +This module provides a generic typed-DAG parser over BDT JSON definitions. +Grounded against the canonical BDT Connect API InputRepresentation classes. +Design principle: structural facts come from deterministic graph walks here; +narrative explanations are built by the host LLM. + +Usage (CLI lands in a later plan): + python bdt_analyze.py [...] + +Requirements: + - Python 3.9+ (standard library only). +""" +from __future__ import annotations + +import argparse +import json +import pathlib +import re +import sys +from collections import Counter +from dataclasses import dataclass, field +from typing import Any, Dict, List, Mapping, Optional, Set, Tuple + + +# --------- Exit codes --------------------------------------------------------- +EXIT_OK = 0 +EXIT_NOT_FOUND = 2 # unknown node / field +EXIT_BAD_INPUT = 3 # malformed JSON / bad shape / cycle + + +# --------- Exceptions -------------------------------------------------------- +class BdtInputError(Exception): + """Raised on any malformed input. CLI maps this to exit code 3.""" + + +class BdtNotFoundError(Exception): + """Raised when a queried node or field doesn't exist. CLI maps to exit 2.""" + + +@dataclass +class Node: + """One entry in BDT `nodes{}`.""" + name: str + action: str + sources: List[str] = field(default_factory=list) + parameters: Dict[str, Any] = field(default_factory=dict) + schema: Optional[Dict[str, Any]] = None + ui_label: Optional[str] = None + ui_description: Optional[str] = None + + @property + def display_name(self) -> str: + """UI label if present and non-empty; else node name.""" + if self.ui_label: + return f"{self.ui_label} ({self.name})" + return self.name + + +@dataclass +class Definition: + """One parsed STL block from a BDT payload. + + A `DataTransform` always holds at least one of these (length 1 for the + editor-export and single-definition Connect API shapes, length N for the + multi-definition Connect API shape). + """ + index: int + version: Optional[str] + nodes: Dict[str, Node] + ui: Dict[str, Any] + name: Optional[str] = None + label: Optional[str] = None + + +@dataclass +class DataTransform: + """Parsed BDT JSON, with a typed DAG view. + + The parser accepts both the editor-export shape (`nodes` at top level) and + the Connect API create-payload shapes (`definition` or `definitions[]` + wrappers around the editor-export content). When a wrapper is detected the + parser descends into the STL definition(s) transparently; outer metadata + (name, label, type, dataSpaceName) is preserved as optional attributes. + + For multi-definition payloads, every definition is parsed and kept on + `self.definitions`. The currently-selected definition (default index 0) + drives the pass-through properties `nodes`, `version`, and `ui`, so all + existing methods operate on the selected block with no signature changes. + Callers can hop between definitions with `select_definition(index)`. + """ + definitions: List[Definition] + # Outer-wrapper metadata (only populated for Connect API input shapes). + name: Optional[str] = None + label: Optional[str] = None + data_transform_type: Optional[str] = None + data_space_name: Optional[str] = None + # Which definition the accessor properties return; 0 by default. + selected_index: int = 0 + + # --- Backward-compat pass-through properties --------------------------- + # Existing callers (tests + CLI handlers) read `.nodes`, `.version`, `.ui` + # as simple attributes; keep that shape by delegating to the selected + # definition. + + @property + def nodes(self) -> Dict[str, Node]: + return self.definitions[self.selected_index].nodes + + @property + def version(self) -> Optional[str]: + return self.definitions[self.selected_index].version + + @property + def ui(self) -> Dict[str, Any]: + return self.definitions[self.selected_index].ui + + @property + def definition_count(self) -> int: + return len(self.definitions) + + @property + def definition_index(self) -> int: + return self.selected_index + + def select_definition(self, index: int) -> None: + """Pick which definition the accessor properties (and the rest of the + API surface) operate on. Raises `BdtNotFoundError` if `index` is out + of range.""" + if type(index) is not int or index < 0 or index >= len(self.definitions): + raise BdtNotFoundError( + f"Definition index {index} out of range; payload has " + f"{len(self.definitions)} definitions " + f"(indices 0-{len(self.definitions) - 1})." + ) + self.selected_index = index + + @classmethod + def from_path(cls, path: pathlib.Path) -> "DataTransform": + path = pathlib.Path(path) + try: + raw = json.loads(path.read_text()) + except OSError as e: + raise BdtInputError(f"Cannot read {path}: {e}") from e + except json.JSONDecodeError as e: + raise BdtInputError( + f"Invalid JSON in {path} at line {e.lineno}, col {e.colno}: {e.msg}" + ) from e + return cls.from_dict(raw) + + @classmethod + def from_dict(cls, raw: Mapping[str, Any]) -> "DataTransform": + if not isinstance(raw, Mapping): + raise BdtInputError("Top-level BDT JSON must be an object") + + # --- Shape detection ------------------------------------------------- + # 1. Editor-export shape: `nodes` at top level — exactly one definition. + # 2. Connect API single-definition shape: `definition` object with + # `nodes` — one definition, plus outer wrapper metadata. + # 3. Connect API multi-definition shape: non-empty `definitions` array + # whose entries each have `nodes` — N definitions, plus outer + # wrapper metadata. Every definition is parsed. + outer_name: Optional[str] = None + outer_label: Optional[str] = None + outer_type: Optional[str] = None + outer_data_space: Optional[str] = None + raw_definitions: List[Mapping[str, Any]] + + if isinstance(raw.get("nodes"), Mapping): + raw_definitions = [raw] + elif (isinstance(raw.get("definition"), Mapping) + and isinstance(raw["definition"].get("nodes"), Mapping)): + raw_definitions = [raw["definition"]] + outer_name = raw.get("name") if isinstance(raw.get("name"), str) else None + outer_label = raw.get("label") if isinstance(raw.get("label"), str) else None + outer_type = raw.get("type") if isinstance(raw.get("type"), str) else None + outer_data_space = ( + raw.get("dataSpaceName") if isinstance(raw.get("dataSpaceName"), str) else None + ) + elif (isinstance(raw.get("definitions"), list) + and len(raw["definitions"]) > 0 + and all( + isinstance(d, Mapping) and isinstance(d.get("nodes"), Mapping) + for d in raw["definitions"] + )): + raw_definitions = list(raw["definitions"]) + outer_name = raw.get("name") if isinstance(raw.get("name"), str) else None + outer_label = raw.get("label") if isinstance(raw.get("label"), str) else None + outer_type = raw.get("type") if isinstance(raw.get("type"), str) else None + outer_data_space = ( + raw.get("dataSpaceName") if isinstance(raw.get("dataSpaceName"), str) else None + ) + else: + raise BdtInputError( + "Expected 'nodes' at top level, or 'definition.nodes', " + "or 'definitions[].nodes'; none found" + ) + + definitions: List[Definition] = [] + for idx, def_raw in enumerate(raw_definitions): + definitions.append( + cls._parse_definition( + idx, def_raw, + fallback_name=outer_name, + fallback_label=outer_label, + ) + ) + + return cls( + definitions=definitions, + name=outer_name, + label=outer_label, + data_transform_type=outer_type, + data_space_name=outer_data_space, + selected_index=0, + ) + + @classmethod + def _parse_definition( + cls, + index: int, + definition: Mapping[str, Any], + fallback_name: Optional[str] = None, + fallback_label: Optional[str] = None, + ) -> Definition: + """Parse a single STL block into a `Definition`. + + `name`/`label` come from the definition itself when present; otherwise + fall back to the outer wrapper's values (so a single-definition API + payload still carries the transform's name on `definitions[0]`). + """ + nodes_raw = definition.get("nodes") + if not isinstance(nodes_raw, Mapping): + # Defensive: the shape-detection step should keep this out. + raise BdtInputError( + "Expected 'nodes' at top level, or 'definition.nodes', " + "or 'definitions[].nodes'; none found" + ) + ui = definition.get("ui") or {} + ui_nodes = (ui.get("nodes") if isinstance(ui, Mapping) else {}) or {} + + nodes: Dict[str, Node] = {} + for name, body in nodes_raw.items(): + if not isinstance(body, Mapping): + raise BdtInputError(f"Node {name!r} body is not an object") + action = body.get("action") + if not isinstance(action, str) or not action: + raise BdtInputError(f"Node {name!r} missing string 'action'") + srcs = body.get("sources") or [] + if not isinstance(srcs, list): + raise BdtInputError(f"Node {name!r} 'sources' is not a list") + ui_entry = ui_nodes.get(name) or {} + nodes[name] = Node( + name=name, + action=action, + sources=list(srcs), + parameters=dict(body.get("parameters") or {}), + schema=body.get("schema"), + ui_label=(ui_entry.get("label") if isinstance(ui_entry, Mapping) else None) or None, + ui_description=(ui_entry.get("description") if isinstance(ui_entry, Mapping) else None) or None, + ) + + def_name = definition.get("name") if isinstance(definition.get("name"), str) else None + def_label = definition.get("label") if isinstance(definition.get("label"), str) else None + + return Definition( + index=index, + version=definition.get("version"), + nodes=nodes, + ui=dict(ui) if isinstance(ui, Mapping) else {}, + name=def_name or fallback_name, + label=def_label or fallback_label, + ) + + def warnings(self) -> List[str]: + """Non-fatal advisories about the parsed BDT. + + The multi-definition case used to emit an advisory here ("payload + contains N definitions; this skill explains the first only"). With + full multi-definition support that warning is gone; the method is + preserved for future advisories and currently returns an empty list. + """ + return [] + + def roots(self) -> List[str]: + """Node names with no sources (data enters here).""" + return [name for name, n in self.nodes.items() if not n.sources] + + def sinks(self) -> List[str]: + """Node names that are not listed as a source by any other node (data exits here).""" + referenced = set() + for n in self.nodes.values(): + referenced.update(n.sources) + return [name for name in self.nodes if name not in referenced] + + def topo_order(self) -> List[str]: + """Kahn's algorithm. Raises BdtInputError on cycles.""" + from collections import deque + in_degree: Dict[str, int] = {name: 0 for name in self.nodes} + forward: Dict[str, List[str]] = {name: [] for name in self.nodes} + for name, n in self.nodes.items(): + for s in n.sources: + if s not in self.nodes: + # Broken reference — count as in-edge so the dependent node stays blocked + in_degree[name] += 1 + continue + forward[s].append(name) + in_degree[name] += 1 + + # Sort zero-in-degree nodes once for deterministic starting order + ready = deque(sorted([name for name, d in in_degree.items() if d == 0])) + order: List[str] = [] + while ready: + current = ready.popleft() + order.append(current) + # Collect newly-ready nodes, sort them once, append + newly_ready = [] + for consumer in forward[current]: + in_degree[consumer] -= 1 + if in_degree[consumer] == 0: + newly_ready.append(consumer) + for node in sorted(newly_ready): + ready.append(node) + + if len(order) != len(self.nodes): + stuck = [n for n in self.nodes if n not in order] + raise BdtInputError( + f"Cycle or broken reference detected; unresolved nodes: {sorted(stuck)}" + ) + return order + + def _topo_order_subset(self, subset: Set[str]) -> List[str]: + """Kahn's algorithm restricted to `subset`. + + Only edges whose endpoints are both in `subset` are considered. This + tolerates broken references that sit outside the queried sub-DAG and + avoids crashing on unrelated cycles. Assumes the sub-DAG itself is + acyclic (caller guarantees this by restricting to a reachable set + via a walk that skips broken refs). + """ + in_degree: Dict[str, int] = {name: 0 for name in subset} + forward: Dict[str, List[str]] = {name: [] for name in subset} + for name in subset: + n = self.nodes[name] + for s in n.sources: + if s in subset: + forward[s].append(name) + in_degree[name] += 1 + ready = sorted([name for name, d in in_degree.items() if d == 0]) + order: List[str] = [] + while ready: + ready.sort() + current = ready.pop(0) + order.append(current) + for consumer in sorted(forward[current]): + in_degree[consumer] -= 1 + if in_degree[consumer] == 0: + ready.append(consumer) + return order + + def upstream(self, node_name: str) -> List[str]: + """All nodes transitively feeding `node_name`, in topological order.""" + if node_name not in self.nodes: + raise BdtNotFoundError( + f"No node named {node_name!r}. Available: " + f"{sorted(list(self.nodes.keys()))[:20]}" + ) + visited: Set[str] = set() + stack = list(self.nodes[node_name].sources) + while stack: + cur = stack.pop() + if cur in visited: + continue + if cur not in self.nodes: + # Broken reference — skip (surfaced via broken_references()) + continue + visited.add(cur) + stack.extend(self.nodes[cur].sources) + # Sort only the reachable sub-DAG (+ the queried node itself), which is + # guaranteed clean. Drop the queried node from the returned list so + # upstream() returns just the ancestors. + sub = visited | {node_name} + topo = self._topo_order_subset(sub) + return [n for n in topo if n in visited] + + def downstream(self, node_name: str) -> List[str]: + """All nodes transitively consuming `node_name`, in topological order.""" + if node_name not in self.nodes: + raise BdtNotFoundError( + f"No node named {node_name!r}. Available: " + f"{sorted(list(self.nodes.keys()))[:20]}" + ) + # Build reverse adjacency on the fly + consumers: Dict[str, List[str]] = {name: [] for name in self.nodes} + for name, n in self.nodes.items(): + for s in n.sources: + if s in consumers: + consumers[s].append(name) + visited: Set[str] = set() + stack = list(consumers[node_name]) + while stack: + cur = stack.pop() + if cur in visited: + continue + visited.add(cur) + stack.extend(consumers[cur]) + # Sort only the reachable sub-DAG; tolerate unrelated broken refs/cycles. + sub = visited | {node_name} + topo = self._topo_order_subset(sub) + return [n for n in topo if n in visited] + + def broken_references(self) -> List[Tuple[str, str]]: + """List of (node_name, missing_source) pairs. Empty if BDT is well-formed.""" + out = [] + for name, n in self.nodes.items(): + for s in n.sources: + if s not in self.nodes: + out.append((name, s)) + return sorted(out) + + +# --------- CLI --------------------------------------------------------------- + + +def _build_parser() -> argparse.ArgumentParser: + p = argparse.ArgumentParser( + prog="bdt_analyze", + description="Explain and investigate Salesforce Data Cloud BDT JSON.", + ) + sub = p.add_subparsers(dest="subcommand", required=True) + + # Common: --json flag + path positional + --definition selector. + # `--definition N` picks which entry in a multi-definition Connect API + # payload the subcommand operates on. Default 0 (the first definition, + # matching the historical behavior). + def _common(sp, with_definition: bool = True): + sp.add_argument("path", help="Path to BDT JSON file") + sp.add_argument("--json", action="store_true", dest="as_json", + help="Emit machine-readable JSON instead of Markdown") + if with_definition: + sp.add_argument( + "--definition", type=int, default=0, dest="definition", + help=( + "Which definition to operate on in a multi-definition " + "Connect API payload (0-indexed, default 0). Use the " + "'definitions' subcommand to list them." + ), + ) + + for name, help_text in [ + ("summary", "High-level digest: counts, sources, outputs"), + ("sources", "All graph-root nodes with their dataset details"), + ("outputs", "All graph-sink nodes with their target details"), + ("stages", "Topologically ordered list of nodes with action + hint"), + ("nodes", "Every node in topo order with a one-line parameter digest"), + ]: + s = sub.add_parser(name, help=help_text) + _common(s) + if name == "nodes": + s.add_argument("--limit", type=int, default=0, + help="Cap row count (0 = no limit)") + + # `node ` + n = sub.add_parser("node", help="Full detail of one node") + _common(n) + n.add_argument("node_name", help="The node name (key in `nodes` map)") + + # Lineage / tracing subcommands + lg = sub.add_parser("lineage", help="All upstream nodes feeding a target node") + _common(lg) + lg.add_argument("node_name", help="The target node name") + + ft = sub.add_parser("field-trace", help="Trace a field back through the DAG") + _common(ft) + ft.add_argument("field_name", help="Field name (qualified or unqualified)") + + fm = sub.add_parser("formula", help="Focused formula expression + upstream field deps") + _common(fm) + fm.add_argument("node_name", help="A formula or computeRelative node name") + + # `definitions ` — list every definition in a payload. Always + # operates on the whole set, so no --definition flag here. + dfs = sub.add_parser("definitions", + help="List every definition in the payload (one row per definition)") + _common(dfs, with_definition=False) + + return p + + +def main(argv: Optional[List[str]] = None) -> int: + """CLI entry point. Returns exit code.""" + parser = _build_parser() + args = parser.parse_args(argv) + + try: + bdt = DataTransform.from_path(args.path) + except BdtInputError as e: + print(str(e), file=sys.stderr) + return EXIT_BAD_INPUT + + # Select the requested definition (default 0). `definitions` subcommand + # doesn't expose the flag — it always reports on the full list. + definition_idx = getattr(args, "definition", 0) + if definition_idx != 0 or args.subcommand != "definitions": + # Only validate/select when the flag is present (it's missing on + # `definitions`, which always walks the full list). The 0-default + # also goes through here for consistency; `select_definition(0)` is + # always safe given we guarantee at least one definition exists. + try: + bdt.select_definition(definition_idx) + except BdtNotFoundError as e: + # Spec: out-of-range `--definition` exits 2 with a message that + # points the user at the `definitions` subcommand. + msg = ( + f"Definition index {definition_idx} out of range; payload " + f"has {bdt.definition_count} definitions " + f"(indices 0-{bdt.definition_count - 1}). " + f"Use 'definitions ' to list them." + ) + # Preserve the underlying error type for completeness, but surface + # the user-facing message on stderr. + _ = e + print(msg, file=sys.stderr) + return EXIT_NOT_FOUND + + dispatch = { + "summary": cmd_summary, + "sources": cmd_sources, + "outputs": cmd_outputs, + "stages": cmd_stages, + "nodes": cmd_nodes, + "node": cmd_node, + "lineage": cmd_lineage, + "field-trace": cmd_field_trace, + "formula": cmd_formula, + "definitions": cmd_definitions, + } + handler = dispatch.get(args.subcommand) + if handler is None: + print(f"Unknown subcommand: {args.subcommand!r}", file=sys.stderr) + return EXIT_BAD_INPUT + try: + output = handler(bdt, args) + except BdtNotFoundError as e: + print(str(e), file=sys.stderr) + return EXIT_NOT_FOUND + except BdtInputError as e: + print(str(e), file=sys.stderr) + return EXIT_BAD_INPUT + print(output) + return EXIT_OK + + +# --------- Field-discovery helpers ------------------------------------------ + +# Matches an unquoted identifier reasonable as a field name: optional qualifier +# (e.g., "SalesOrder.ssot__Id__c") followed by the identifier itself. +_FIELD_NAME_RE = re.compile(r'\b(?:[A-Za-z_][A-Za-z0-9_]*\.)?[A-Za-z_][A-Za-z0-9_]*\b') + +# Reserved identifiers we exclude when scraping a formula expression for field refs. +_SQL_RESERVED = { + "case", "when", "then", "else", "end", "and", "or", "not", "null", "true", "false", + "if", "is", "in", "like", "between", "cast", "as", + # Common SFSQL functions — callers can add more if needed. + "abs", "ceiling", "exp", "floor", "log", "max", "min", "mod", "power", "round", "sqrt", "trunc", + "concat", "contains", "ends", "length", "lower", "ltrim", "rtrim", "substitute", "substr", + "text", "trim", "upper", "uuid", "value", "begins", + "adddays", "addmonths", "datediff", "datetimevalue", "datevalue", "day", "monthdiff", + "now", "today", "weekday", + "blankvalue", "isblank", "isnull", "nullvalue", + "sequence", "explode", + "row_number", "rank", "dense_rank", "lag", "lead", "first_value", "last_value", + "nth_value", "ntile", "percent_rank", "cume_dist", + "coalesce", + # Type names + "number", "text", "boolean", "date", "datetime", +} + + +def fields_produced(node: "Node") -> List[str]: + """Which field names this node emits into its output stream. + + Heuristic per action type. Unknown actions fall back to [] (caller must cope). + """ + p = node.parameters if isinstance(node.parameters, dict) else {} + if node.action == "load": + return list(p.get("fields") or []) + if node.action in ("formula", "computeRelative"): + return [f.get("name") for f in (p.get("fields") or []) + if isinstance(f, dict) and f.get("name")] + if node.action == "outputD360": + return [m.get("targetField") for m in (p.get("fieldsMappings") or []) + if isinstance(m, dict) and m.get("targetField")] + if node.action == "schema": + # Schema nodes rename/drop; renamed names come from slice.fields[].newProperties.name + fs = (p.get("fields") or []) + out = [] + for f in fs: + if isinstance(f, dict): + np_ = f.get("newProperties") or {} + out.append(np_.get("name") or f.get("name")) + return [x for x in out if x] + if node.action == "aggregate": + # Aggregations add their output names; groupings pass through as-is + out = list(p.get("groupings") or []) + for a in (p.get("aggregations") or []): + if isinstance(a, dict) and a.get("name"): + out.append(a["name"]) + return out + # Default: we can't guess — caller must inspect parameters directly. + return [] + + +def fields_consumed(node: "Node") -> List[str]: + """Which upstream field names this node reads. + + For pass-through nodes we return [] — callers that need "what flows through + this node" should look at `fields_produced` on the source(s). + """ + p = node.parameters if isinstance(node.parameters, dict) else {} + if node.action == "load": + return [] + if node.action == "join": + out = list(p.get("leftKeys") or []) + list(p.get("rightKeys") or []) + return out + if node.action == "filter": + return [e.get("field") for e in (p.get("filterExpressions") or []) + if isinstance(e, dict) and e.get("field")] + if node.action in ("formula", "computeRelative"): + consumed = set() + for f in (p.get("fields") or []): + if isinstance(f, dict): + expr = f.get("formulaExpression") or "" + consumed.update(_scrape_field_refs(expr)) + # computeRelative also consumes partitionBy + orderBy fields + if node.action == "computeRelative": + consumed.update(p.get("partitionBy") or []) + for ob in (p.get("orderBy") or []): + if isinstance(ob, dict) and ob.get("fieldName"): + consumed.add(ob["fieldName"]) + return sorted(consumed) + if node.action == "aggregate": + out = list(p.get("groupings") or []) + for a in (p.get("aggregations") or []): + if isinstance(a, dict) and a.get("source"): + out.append(a["source"]) + # Also hierarchical fields + for f_key in ("selfField", "parentField", "percentageField"): + v = p.get(f_key) + if v: + out.append(v) + return out + if node.action == "outputD360": + return [m.get("sourceField") for m in (p.get("fieldsMappings") or []) + if isinstance(m, dict) and m.get("sourceField")] + if node.action == "schema": + return [f.get("name") for f in (p.get("fields") or []) + if isinstance(f, dict) and f.get("name")] + return [] + + +def _scrape_field_refs(expression: str) -> List[str]: + """Pull likely field identifiers out of a SQL-ish formula expression. + + This is a heuristic: it returns identifiers that aren't reserved keywords or + function names. Callers should still treat results as a *candidate* set, not + a verified one. In particular, we also pick up quoted identifiers. + """ + if not isinstance(expression, str): + return [] + # Also capture "Quoted.Name" style identifiers + quoted = re.findall(r'"([A-Za-z_][\w.]*)"', expression) + bare = _FIELD_NAME_RE.findall(expression) + candidates = set(quoted) | set(bare) + # Drop reserved tokens (case-insensitive) and numeric-looking tokens + out = [] + for c in candidates: + last = c.rsplit(".", 1)[-1] + if last.lower() in _SQL_RESERVED: + continue + if last.isdigit(): + continue + out.append(c) + # Deduplicate while preserving order + seen = set() + dedup = [] + for c in out: + if c not in seen: + seen.add(c) + dedup.append(c) + return dedup + + +def _field_specific_upstream_deps(node: "Node", field: str) -> List[str]: + """Deps relevant to `field` specifically, not all deps the node consumes. + + `fields_consumed(node)` gives the union of every field the node reads + across *all* of its outputs. For field-trace narration we want a much + tighter set: only the upstream fields that feed *this one* field. This + keeps the trace output small and focused, which matters for real BDTs + where a single `outputD360` may have 40+ mappings or a `formula` node + may define a dozen unrelated output fields. + """ + p = node.parameters if isinstance(node.parameters, dict) else {} + action = node.action + if action in ("formula", "computeRelative"): + out: List[str] = [] + for f in (p.get("fields") or []): + if isinstance(f, dict) and f.get("name") == field: + out.extend(_scrape_field_refs(f.get("formulaExpression") or "")) + break + if action == "computeRelative": + out.extend(p.get("partitionBy") or []) + for ob in (p.get("orderBy") or []): + if isinstance(ob, dict) and ob.get("fieldName"): + out.append(ob["fieldName"]) + # Deduplicate, preserve order + seen: set = set() + dedup: List[str] = [] + for x in out: + if x not in seen: + seen.add(x) + dedup.append(x) + return dedup + if action == "outputD360": + for m in (p.get("fieldsMappings") or []): + if isinstance(m, dict) and m.get("targetField") == field: + src = m.get("sourceField") + return [src] if src else [] + return [] + if action == "aggregate": + # If the field is one of the aggregations, return just its source. + for a in (p.get("aggregations") or []): + if isinstance(a, dict) and a.get("name") == field: + src = a.get("source") + return [src] if src else [] + # If it's one of the groupings (pass-through), the dep is itself. + if field in (p.get("groupings") or []): + return [field] + return [] + if action == "schema": + # Renames: if this node renamed some X to `field`, the dep is X. + for f in (p.get("fields") or []): + if isinstance(f, dict): + np_ = f.get("newProperties") or {} + if np_.get("name") == field: + return [f.get("name")] if f.get("name") else [] + return [] + if action == "load": + return [] + # Fallback for unknown / unhandled actions — reuse the generic consumed set + return fields_consumed(node) + + +def cmd_definitions(bdt: "DataTransform", args) -> str: + """List every definition in the payload — one row per definition. + + Safe to call on editor-export or single-definition inputs: they always + have exactly one row. On multi-definition payloads this is the entry + point for the SKILL.md flow that offers the user a choice. + """ + rows = [] + for d in bdt.definitions: + rows.append({ + "index": d.index, + "name": d.name, + "label": d.label, + "version": d.version, + "node_count": len(d.nodes), + }) + + if getattr(args, "as_json", False): + return json.dumps(rows, indent=2) + + lines = [f"# Definitions ({len(rows)})", ""] + lines.append("| Index | Name | Label | Version | Nodes |") + lines.append("|---|---|---|---|---|") + for r in rows: + lines.append( + f"| {r['index']} | " + f"{r['name'] or '(no name)'} | " + f"{r['label'] or '(no label)'} | " + f"{r['version'] or '(unknown)'} | " + f"{r['node_count']} |" + ) + return "\n".join(lines) + + +def cmd_summary(bdt: "DataTransform", args) -> str: + """High-level digest of the BDT.""" + action_counts = Counter(n.action for n in bdt.nodes.values()) + roots = bdt.roots() + sinks = bdt.sinks() + + if getattr(args, "as_json", False): + data = { + "version": bdt.version, + "total_nodes": len(bdt.nodes), + "action_counts": dict(action_counts), + "sources": [_source_digest(bdt, r) for r in roots], + "outputs": [_output_digest(bdt, s) for s in sinks], + } + return json.dumps(data, indent=2) + + # Markdown + lines = [] + lines.append("# BDT Summary") + lines.append(f"Version: {bdt.version or '(unknown)'}") + lines.append(f"Total nodes: {len(bdt.nodes)}") + if action_counts: + counts_str = ", ".join(f"{a}: {c}" for a, c in action_counts.most_common()) + lines.append(f"By action: {counts_str}") + lines.append("") + if roots: + lines.append(f"## Sources ({len(roots)} nodes)") + lines.append("| Node | Label | Dataset | Type | Field count |") + lines.append("|---|---|---|---|---|") + for r in roots: + d = _source_digest(bdt, r) + lines.append( + f"| {r} | {d['label'] or '(no label)'} | " + f"{d['dataset_name'] or '(none)'} | " + f"{d['dataset_type'] or '(none)'} | " + f"{d['field_count']} |" + ) + lines.append("") + if sinks: + lines.append(f"## Outputs ({len(sinks)} nodes)") + lines.append("| Node | Label | Target | Type | Mapping count |") + lines.append("|---|---|---|---|---|") + for s in sinks: + d = _output_digest(bdt, s) + lines.append( + f"| {s} | {d['label'] or '(no label)'} | " + f"{d['target_name'] or '(none)'} | " + f"{d['target_type'] or '(none)'} | " + f"{d['mapping_count']} |" + ) + return "\n".join(lines) + + +def _source_digest(bdt: "DataTransform", name: str) -> dict: + n = bdt.nodes[name] + ds = (n.parameters.get("dataset") or {}) if isinstance(n.parameters, dict) else {} + fields = n.parameters.get("fields") or [] if isinstance(n.parameters, dict) else [] + return { + "name": name, + "label": n.ui_label, + "dataset_name": ds.get("name") if isinstance(ds, dict) else None, + "dataset_type": ds.get("type") if isinstance(ds, dict) else None, + "field_count": len(fields) if isinstance(fields, list) else 0, + } + + +def _output_digest(bdt: "DataTransform", name: str) -> dict: + n = bdt.nodes[name] + p = n.parameters if isinstance(n.parameters, dict) else {} + mappings = p.get("fieldsMappings") or [] + return { + "name": name, + "label": n.ui_label, + "target_name": p.get("name"), + "target_type": p.get("type"), + "write_mode": p.get("writeMode"), + "mapping_count": len(mappings) if isinstance(mappings, list) else 0, + } + + +def cmd_sources(bdt: "DataTransform", args) -> str: + roots = bdt.roots() + enriched = [] + for r in roots: + n = bdt.nodes[r] + p = n.parameters if isinstance(n.parameters, dict) else {} + ds = p.get("dataset") or {} + enriched.append({ + "name": r, + "label": n.ui_label, + "dataset_name": ds.get("name") if isinstance(ds, dict) else None, + "dataset_type": ds.get("type") if isinstance(ds, dict) else None, + "fields": list(p.get("fields") or []), + "action": n.action, + }) + if getattr(args, "as_json", False): + return json.dumps(enriched, indent=2) + lines = [f"# Sources ({len(enriched)} node(s))", ""] + for e in enriched: + lines.append(f"## {e['name']}" + (f" — {e['label']}" if e['label'] else "")) + lines.append(f"- action: `{e['action']}`") + lines.append(f"- dataset: `{e['dataset_name']}` ({e['dataset_type']})") + if e['fields']: + lines.append(f"- fields ({len(e['fields'])}): " + ", ".join(f"`{f}`" for f in e['fields'])) + else: + lines.append("- fields: _(none declared)_") + lines.append("") + return "\n".join(lines).rstrip() + "\n" + + +def cmd_outputs(bdt: "DataTransform", args) -> str: + sinks = bdt.sinks() + enriched = [] + for s in sinks: + n = bdt.nodes[s] + p = n.parameters if isinstance(n.parameters, dict) else {} + mappings = p.get("fieldsMappings") or [] + enriched.append({ + "name": s, + "label": n.ui_label, + "action": n.action, + "target_name": p.get("name"), + "target_type": p.get("type"), + "write_mode": p.get("writeMode"), + "dedup_order": list(p.get("dedupOrder") or []), + "mappings": [ + { + "source": m.get("sourceField") if isinstance(m, dict) else None, + "target": m.get("targetField") if isinstance(m, dict) else None, + } + for m in (mappings if isinstance(mappings, list) else []) + ], + }) + if getattr(args, "as_json", False): + return json.dumps(enriched, indent=2) + lines = [f"# Outputs ({len(enriched)} node(s))", ""] + for e in enriched: + lines.append(f"## {e['name']}" + (f" — {e['label']}" if e['label'] else "")) + lines.append(f"- action: `{e['action']}`") + lines.append(f"- target: `{e['target_name']}` ({e['target_type']})") + lines.append(f"- writeMode: `{e['write_mode']}`") + if e['dedup_order']: + lines.append(f"- dedupOrder: {e['dedup_order']}") + if e['mappings']: + lines.append("") + lines.append(f"### Field mappings ({len(e['mappings'])})") + lines.append("| Source field | Target field |") + lines.append("|---|---|") + for m in e['mappings']: + lines.append(f"| `{m['source']}` | `{m['target']}` |") + lines.append("") + return "\n".join(lines).rstrip() + "\n" + + +# One-line purpose hint per known action. Unknown actions fall back to the raw action name. +_ACTION_HINT = { + "load": "Load data from a DMO or DLO", + "filter": "Keep rows matching filter criteria", + "sqlFilter": "Keep rows matching a raw SQL predicate", + "join": "Join two upstream streams", + "formula": "Add derived columns via per-row formulas", + "computeRelative": "Compute a window function over partitioned/ordered rows", + "aggregate": "Group-by aggregation", + "extractGrains": "Fan out rows across time grains", + "extractTable": "Extract a table from flattened JSON", + "schema": "Rename/drop/reshape columns", + "outputD360": "Write rows to a target DMO/DLO", + "appendV2": "Union rows from multiple upstream streams", + "flatten": "Flatten nested structure", + "flattenJson": "Flatten a JSON field into rows/columns", + "split": "Split rows across multiple outputs", + "typeCast": "Cast field types", + "update": "Update records", + "bucket": "Bucket field values", + "formatDate": "Reformat date values", + "extension": "Run custom extension logic", + "extensionFunction":"Run custom extension function", + "cdpPredict": "Apply a prediction model", + "jsonAggregate": "Aggregate JSON fields", + "save": "Save intermediate result", + "recommendation": "Apply a recommendation model", +} + + +def cmd_stages(bdt: "DataTransform", args) -> str: + order = bdt.topo_order() + rows = [] + for name in order: + n = bdt.nodes[name] + hint = _ACTION_HINT.get(n.action, f"({n.action} — not in known catalog)") + rows.append({ + "name": name, + "label": n.ui_label, + "action": n.action, + "sources": list(n.sources), + "hint": hint, + }) + if getattr(args, "as_json", False): + return json.dumps(rows, indent=2) + lines = [f"# Stages ({len(rows)} node(s), topological order)", ""] + for i, r in enumerate(rows, 1): + srcs = ", ".join(f"`{s}`" for s in r["sources"]) if r["sources"] else "_(root)_" + label_part = f" — {r['label']}" if r['label'] else "" + lines.append(f"{i}. **{r['name']}**{label_part} `{r['action']}`") + lines.append(f" - sources: {srcs}") + lines.append(f" - {r['hint']}") + return "\n".join(lines) + + +def cmd_nodes(bdt: "DataTransform", args) -> str: + order = bdt.topo_order() + limit = getattr(args, "limit", 0) or 0 + limited = limit > 0 and len(order) > limit + shown = order[:limit] if limited else order + + rows = [] + for name in shown: + n = bdt.nodes[name] + rows.append({ + "name": name, + "label": n.ui_label, + "action": n.action, + "sources": list(n.sources), + "param_digest": _param_digest(n), + }) + + if getattr(args, "as_json", False): + return json.dumps({"total": len(order), "shown": len(rows), "nodes": rows}, indent=2) + + lines = [f"# Nodes ({len(order)} total)"] + if limited: + lines.append(f"_Output limited to {limit} of {len(order)} — re-run without --limit for full list._") + lines.append("") + lines.append("| Node | Label | Action | Sources | Digest |") + lines.append("|---|---|---|---|---|") + for r in rows: + srcs = ", ".join(r["sources"]) if r["sources"] else "_(root)_" + label = r["label"] or "" + digest = r["param_digest"].replace("|", "\\|") + lines.append(f"| {r['name']} | {label} | {r['action']} | {srcs} | {digest} |") + return "\n".join(lines) + + +def _param_digest(n: "Node") -> str: + """One-line Markdown-safe parameter summary. Never exceeds ~120 chars.""" + p = n.parameters if isinstance(n.parameters, dict) else {} + if n.action == "load": + ds = p.get("dataset") or {} + ds_name = ds.get("name") if isinstance(ds, dict) else "?" + fc = len(p.get("fields") or []) + return f"dataset=`{ds_name}`, {fc} field(s)" + if n.action == "join": + return (f"type=`{p.get('joinType')}` " + f"leftKeys={p.get('leftKeys')} rightKeys={p.get('rightKeys')} " + f"rightQualifier=`{p.get('rightQualifier')}`") + if n.action == "filter": + exprs = p.get("filterExpressions") or [] + return f"{len(exprs)} expression(s), logic=`{p.get('filterBooleanLogic') or 'all AND'}`" + if n.action in ("formula", "computeRelative"): + fs = p.get("fields") or [] + names = ", ".join(f.get("name", "?") for f in fs if isinstance(f, dict)) + extra = "" + if n.action == "computeRelative": + extra = f" partitionBy={p.get('partitionBy')} orderBy={p.get('orderBy')}" + return f"{len(fs)} output field(s): {names}{extra}" + if n.action == "aggregate": + aggs = p.get("aggregations") or [] + groupings = p.get("groupings") or [] + return f"{len(aggs)} aggregation(s), groupBy={groupings}" + if n.action == "outputD360": + mc = len(p.get("fieldsMappings") or []) + return f"target=`{p.get('name')}` ({p.get('type')}) writeMode=`{p.get('writeMode')}` {mc} mapping(s)" + if n.action == "schema": + slc = p.get("slice") or {} + return f"slice.mode=`{slc.get('mode')}` {len(slc.get('fields') or [])} field(s)" + # Fallback: show param keys only + keys = list(p.keys())[:5] + return f"parameters keys: {keys}" + + +def cmd_node(bdt: "DataTransform", args) -> str: + name = args.node_name + if name not in bdt.nodes: + available = sorted(bdt.nodes.keys())[:20] + raise BdtNotFoundError( + f"No node named {name!r}. Available (first 20): {available}" + ) + n = bdt.nodes[name] + # Compute consumers (nodes that list this one as a source) + consumers = [m for m, other in bdt.nodes.items() if name in other.sources] + + if getattr(args, "as_json", False): + return json.dumps({ + "name": n.name, + "label": n.ui_label, + "description": n.ui_description, + "action": n.action, + "sources": n.sources, + "consumers": consumers, + "parameters": n.parameters, + "schema": n.schema, + }, indent=2) + + lines = [f"# {n.name}" + (f" — {n.ui_label}" if n.ui_label else "")] + if n.ui_description: + lines.append(f"> {n.ui_description}") + lines.append("") + lines.append(f"- **action:** `{n.action}`") + lines.append(f"- **hint:** {_ACTION_HINT.get(n.action, '(not in known catalog)')}") + lines.append(f"- **sources:** {n.sources or '_(root)_'}") + lines.append(f"- **direct consumers:** {consumers or '_(sink — no downstream)_'}") + lines.append("") + lines.append("## Parameters") + lines.append("```json") + lines.append(json.dumps(n.parameters, indent=2, default=str)) + lines.append("```") + if n.schema: + lines.append("") + lines.append("## Schema block") + lines.append("```json") + lines.append(json.dumps(n.schema, indent=2, default=str)) + lines.append("```") + return "\n".join(lines) + + +def cmd_lineage(bdt: "DataTransform", args) -> str: + target = args.node_name + if target not in bdt.nodes: + available = sorted(bdt.nodes.keys())[:20] + raise BdtNotFoundError( + f"No node named {target!r}. Available (first 20): {available}" + ) + upstream_names = bdt.upstream(target) # topo-ordered + + rows = [] + for name in upstream_names: + n = bdt.nodes[name] + rows.append({ + "name": name, + "action": n.action, + "label": n.ui_label, + "sources": list(n.sources), + }) + + if getattr(args, "as_json", False): + return json.dumps({"target": target, "upstream": rows}, indent=2) + + lines = [f"# Lineage of {target}"] + t = bdt.nodes[target] + if t.ui_label: + lines.append(f"_{t.ui_label}_") + lines.append("") + if not upstream_names: + lines.append(f"`{target}` is a graph root (no upstream).") + return "\n".join(lines) + lines.append(f"{len(upstream_names)} upstream node(s), topological order:") + lines.append("") + for r in rows: + arrow = " ← " if r["sources"] else " " + label_part = f" — {r['label']}" if r['label'] else "" + lines.append(f"{arrow}**{r['name']}** `{r['action']}`{label_part}") + if r['sources']: + srcs = ", ".join(f"`{s}`" for s in r['sources']) + lines.append(f" (fed by {srcs})") + lines.append("") + lines.append(f" → **{target}** `{t.action}`" + (f" — {t.ui_label}" if t.ui_label else "")) + return "\n".join(lines) + + +def cmd_field_trace(bdt: "DataTransform", args) -> str: + field = args.field_name + # Cache topo order once — `bdt.topo_order()` walks the whole graph. + topo = bdt.topo_order() + # 1. Find all nodes that "produce" this field (per fields_produced heuristic). + definers = [] + for name in topo: + n = bdt.nodes[name] + produced = fields_produced(n) + if field in produced: + definers.append(n) + + # 2. If nothing matched, also try matching the unqualified part (drop any "Qualifier." prefix) + if not definers: + unq = field.rsplit(".", 1)[-1] + for name in topo: + n = bdt.nodes[name] + produced = fields_produced(n) + if unq in produced: + definers.append(n) + field = unq # report with the unqualified name we actually matched + + if not definers: + available_sample = [] + for name in list(bdt.nodes.keys())[:10]: + available_sample.extend(fields_produced(bdt.nodes[name])) + raise BdtNotFoundError( + f"Field {args.field_name!r} is not defined by any node in this BDT. " + f"Sample of visible field names (from the first 10 nodes): " + f"{sorted(set(available_sample))[:20]}" + ) + + # 3. For each definer, compute: (a) the formula or mapping that defines it, if any; + # (b) its direct upstream field dependencies (`fields_consumed`). + def _definition_of(n: "Node", fname: str) -> dict: + p = n.parameters if isinstance(n.parameters, dict) else {} + detail = {"kind": n.action} + if n.action in ("formula", "computeRelative"): + for f in (p.get("fields") or []): + if isinstance(f, dict) and f.get("name") == fname: + detail["formulaExpression"] = f.get("formulaExpression") + detail["type"] = f.get("type") + detail["label"] = f.get("label") + break + elif n.action == "outputD360": + for m in (p.get("fieldsMappings") or []): + if isinstance(m, dict) and m.get("targetField") == fname: + detail["sourceField"] = m.get("sourceField") + break + elif n.action == "load": + ds = p.get("dataset") or {} + detail["dataset"] = {"name": ds.get("name"), "type": ds.get("type")} + elif n.action == "aggregate": + for a in (p.get("aggregations") or []): + if isinstance(a, dict) and a.get("name") == fname: + detail["aggregation"] = { + "action": a.get("action"), + "source": a.get("source"), + } + break + elif n.action == "schema": + for f in (p.get("fields") or []): + if isinstance(f, dict): + np_ = f.get("newProperties") or {} + if (np_.get("name") or f.get("name")) == fname: + detail["renamed_from"] = f.get("name") + detail["renamed_to"] = np_.get("name") + break + return detail + + result = [] + for n in definers: + entry = { + "node": n.name, + "label": n.ui_label, + "action": n.action, + "definition": _definition_of(n, field), + # Narrowed to deps of *this specific field*, not every field the + # node consumes. For output/formula/aggregate nodes with many + # unrelated output fields this dramatically reduces trace size + # and noise. The broader node-level context is still available + # via `upstream_chain` below. + "upstream_deps": _field_specific_upstream_deps(n, field), + "sources": list(n.sources), + } + result.append(entry) + + # Upstream node chain for the first definer (for narration convenience) + chain = [] + if definers: + chain = bdt.upstream(definers[0].name) + [definers[0].name] + + if getattr(args, "as_json", False): + return json.dumps({ + "field": field, + "definitions": result, + "upstream_chain": chain, + }, indent=2) + + lines = [f"# Field trace: `{field}`"] + if len(definers) > 1: + lines.append(f"> Field is defined in {len(definers)} places (first in topological order listed first).") + lines.append("") + for i, entry in enumerate(result, 1): + header = f"## {i}. Defined by {entry['node']}" + if entry['label']: + header += f" — {entry['label']}" + lines.append(header) + lines.append(f"- action: `{entry['action']}`") + d = entry['definition'] + if d.get("formulaExpression"): + lines.append(f"- formula: `{d['formulaExpression']}`") + if d.get("type"): + lines.append(f"- type: `{d['type']}`") + if d.get("sourceField"): + lines.append(f"- sourced from: `{d['sourceField']}`") + if d.get("dataset"): + lines.append(f"- loaded from: `{d['dataset']['name']}` ({d['dataset']['type']})") + if d.get("aggregation"): + a = d['aggregation'] + lines.append(f"- aggregation: `{a['action']}` over `{a['source']}`") + if d.get("renamed_from"): + lines.append(f"- renamed: `{d['renamed_from']}` → `{d['renamed_to']}`") + if entry['upstream_deps']: + lines.append(f"- upstream field deps: {entry['upstream_deps']}") + lines.append(f"- upstream nodes: {entry['sources'] or '_(root)_'}") + lines.append("") + if chain: + lines.append("## Upstream node chain") + lines.append(" → ".join(f"`{c}`" for c in chain)) + return "\n".join(lines) + + +def cmd_formula(bdt: "DataTransform", args) -> str: + name = args.node_name + if name not in bdt.nodes: + raise BdtNotFoundError( + f"No node named {name!r}. Available (first 20): " + f"{sorted(bdt.nodes.keys())[:20]}" + ) + n = bdt.nodes[name] + if n.action not in ("formula", "computeRelative"): + raise BdtInputError( + f"Node {name!r} has action {n.action!r}; `formula` subcommand " + f"only applies to `formula` and `computeRelative` nodes. " + f"Use `node {name}` instead." + ) + p = n.parameters if isinstance(n.parameters, dict) else {} + fields_out = p.get("fields") or [] + consumed = fields_consumed(n) + + # For each consumed field, find the node that defines it (if any) — give + # the LLM ready context for I2-style reasoning. Hoist `bdt.upstream(name)` + # out of the loop — it's a pure function of `name` and walks the graph. + # Pre-compute fields_produced for every upstream node once (was O(consumed × upstream)) + upstream_names = bdt.upstream(name) + upstream_fields_map = {up_name: fields_produced(bdt.nodes[up_name]) for up_name in upstream_names} + upstream_defs = {} + for f in consumed: + matches = [] + for up_name in upstream_names: + up = bdt.nodes[up_name] + prod = upstream_fields_map[up_name] + if f in prod or f.rsplit(".", 1)[-1] in prod: + matches.append({ + "node": up_name, + "action": up.action, + "label": up.ui_label, + }) + if matches: + upstream_defs[f] = matches[-1] + + payload = { + "node": name, + "label": n.ui_label, + "action": n.action, + "formula_fields": [ + { + "name": f.get("name"), + "label": f.get("label"), + "type": f.get("type"), + "formulaExpression": f.get("formulaExpression"), + } + for f in fields_out if isinstance(f, dict) + ], + "consumed_fields": consumed, + "upstream_field_definitions": upstream_defs, + } + if n.action == "computeRelative": + payload["partitionBy"] = list(p.get("partitionBy") or []) + payload["orderBy"] = p.get("orderBy") or [] + + if getattr(args, "as_json", False): + return json.dumps(payload, indent=2) + + lines = [f"# Formula extract: {name}" + (f" — {n.ui_label}" if n.ui_label else "")] + lines.append(f"- action: `{n.action}`") + if n.action == "computeRelative": + lines.append(f"- partitionBy: {payload['partitionBy']}") + lines.append(f"- orderBy: {payload['orderBy']}") + lines.append("") + lines.append("## Output fields") + for f in payload["formula_fields"]: + lines.append(f"### `{f['name']}`" + (f" _{f['label']}_" if f['label'] else "")) + lines.append(f"- type: `{f['type']}`") + lines.append(f"- expression: `{f['formulaExpression']}`") + lines.append("") + lines.append("## Upstream field dependencies") + if not consumed: + lines.append("_(none detected — expression may reference only literals)_") + else: + for f in consumed: + defn = upstream_defs.get(f) + if defn: + label = f" _{defn['label']}_" if defn.get('label') else "" + lines.append(f"- `{f}` ← **{defn['node']}** (`{defn['action']}`){label}") + else: + lines.append(f"- `{f}` ← (no upstream definition found — may be a literal or a passthrough)") + return "\n".join(lines) + + +if __name__ == "__main__": + sys.exit(main(sys.argv[1:]))