Dataflows Consumption Cli

Update Check — ONCE PER SESSION (mandatory)

The first time this skill is used in a session, run the check-updates skill before proceeding.

- GitHub Copilot CLI / VS Code: invoke the check-updates skill.

- Claude Code / Cowork / Cursor / Windsurf / Codex: compare local vs remote package.json version.

- Skip if the check was already performed earlier in this session.

CRITICAL NOTES

1. To find the workspace details (including its ID) from workspace name: list all workspaces and, then, use JMESPath filtering

2. To find a dataflow by name: list all dataflows in the workspace and filter by displayName client-side — there is no server-side name filter

3. getDefinition is a POST, not GET — even though it reads data

dataflows-consumption-cli — Dataflows Gen2 Consumption via CLI

Table of Contents

---

Tool Stack

Agent check — verify before first operation:

```bash

az account show >/dev/null 2>&1 || echo "RUN: az login"

command -v jq >/dev/null 2>&1 || echo "INSTALL: apt-get install jq OR brew install jq"

```

---

Connection

Resolve Workspace ID and Dataflow ID

Per COMMON-CLI.md Finding Workspaces and Items in Fabric:

# Find workspace ID by name
WS_ID=$(az rest --method get \
  --resource "https://api.fabric.microsoft.com" \
  --url "https://api.fabric.microsoft.com/v1/workspaces" \
  --query "value[?displayName=='My Workspace'].id" --output tsv)

# Find dataflow ID by name within workspace
DF_ID=$(az rest --method get \
  --resource "https://api.fabric.microsoft.com" \
  --url "https://api.fabric.microsoft.com/v1/workspaces/$WS_ID/dataflows" \
  --query "value[?displayName=='Sales Data Pipeline'].id" --output tsv)

Reusable Connection Variables

# Set once at script top
WS_ID="<workspaceId>"
DF_ID="<dataflowId>"
API="https://api.fabric.microsoft.com/v1"
AZ="az rest --resource https://api.fabric.microsoft.com"

---

Agentic Exploration ("Chat With My Dataflows")

Discovery Sequence

Run these in order to fully explore a workspace's dataflows. See references/discovery-queries.md for extended patterns.

# 1. List workspaces → find target
az rest --method get --resource "https://api.fabric.microsoft.com" \
  --url "$API/workspaces" --query "value[].{name:displayName, id:id}" -o table

# 2. List dataflows → enumerate all
az rest --method get --resource "https://api.fabric.microsoft.com" \
  --url "$API/workspaces/$WS_ID/dataflows" \
  --query "value[].{name:displayName, id:id, desc:description}" -o table

# 3. Get dataflow properties
az rest --method get --resource "https://api.fabric.microsoft.com" \
  --url "$API/workspaces/$WS_ID/dataflows/$DF_ID"

# 4. Discover parameters
az rest --method get --resource "https://api.fabric.microsoft.com" \
  --url "$API/workspaces/$WS_ID/dataflows/$DF_ID/parameters" \
  --query "value[].{name:name, type:type, required:isRequired, default:defaultValue}" -o table

# 5. Get definition → decode mashup.pq
RESPONSE=$(az rest --method post --resource "https://api.fabric.microsoft.com" \
  --url "$API/workspaces/$WS_ID/dataflows/$DF_ID/getDefinition")
echo "$RESPONSE" | jq -r '.definition.parts[] | select(.path=="mashup.pq") | .payload' | base64 --decode

# 6. Check job history
az rest --method get --resource "https://api.fabric.microsoft.com" \
  --url "$API/workspaces/$WS_ID/dataflows/$DF_ID/jobs/instances" \
  --query "value[].{status:status, type:invokeType, start:startTimeUtc, end:endTimeUtc, error:failureReason}" -o table

Agentic Workflow

1. Discover → Run Steps 1–3 to list and identify dataflows. 2. Parameters → Step 4 to understand inputs and defaults. 3. Definition → Step 5 to inspect M queries, connections, staging config. 4. Monitor → Step 6 for refresh history and error patterns. 5. Iterate → Drill into specific queries or connection details. 6. Present → Summarize findings or generate a reusable script (see script-templates.md).

---

Gotchas, Rules, Troubleshooting

For full platform gotchas: DATAFLOWS-CONSUMPTION-CORE.md Gotchas and Troubleshooting Reference and COMMON-CLI.md Gotchas & Troubleshooting (CLI-Specific).

MUST DO

• Always `az login` first — az rest uses the active session. No session → cryptic failure.
• Always `--resource "https://api.fabric.microsoft.com"` — wrong audience = 401.
• Handle pagination — repeat requests with continuationToken until absent/null.
• Handle LRO for `getDefinition` — may return 202 Accepted with Location header; poll until complete.
• Decode base64 before inspecting — definition parts are base64-encoded.
• Use POST for `getDefinition` — it is NOT a GET endpoint.

AVOID

• Hardcoded GUIDs — always discover via list-then-filter pattern.
• Assuming `getDefinition` is GET — it is POST (common mistake).
• Ignoring pagination — list endpoints may return partial results.
• Polling too aggressively — respect Retry-After headers on 429s.
• Expecting `getDefinition` with Viewer role — requires Read+Write (Contributor+).

PREFER

• `az rest` over raw `curl` — handles auth automatically.
• List-then-filter pattern — no server-side name filter for dataflows.
• Exponential backoff for job polling — 5s → 10s → 20s → 30s cap.
• `jq` for response parsing — cleaner than shell string manipulation.
• JMESPath `--query` for simple field extraction directly in az rest.
• Env vars (WS_ID, DF_ID, API) for script reuse.

TROUBLESHOOTING

---

Examples

Example 1: List All Dataflows in a Workspace

Prompt: "List all dataflows in this workspace."

Agent response:

az rest --method get \
  --url "https://api.fabric.microsoft.com/v1/workspaces/${WS_ID}/dataflows" \
  --resource "https://api.fabric.microsoft.com" \
  --query "value[].{Name:displayName, Id:id, Type:type}" -o table

Example 2: Decode a Dataflow Definition

Prompt: "Show me the Power Query M code inside this dataflow."

Agent response:

# Step 1: Request definition (POST returns 202 with Location header).
# az rest cannot expose response headers, so capture Location via curl + az-acquired token.
TOKEN=$(az account get-access-token --resource "https://api.fabric.microsoft.com" --query accessToken -o tsv)
LOCATION=$(curl -sS -X POST \
  -H "Authorization: Bearer $TOKEN" -H "Content-Length: 0" \
  "https://api.fabric.microsoft.com/v1/workspaces/${WS_ID}/dataflows/${DF_ID}/getDefinition" \
  -o /dev/null -D - | tr -d '\r' | grep -i "^location:" | awk '{print $2}')

# Step 2: Poll until definition is ready
DEF=$(az rest --method get --url "${LOCATION}" \
  --resource "https://api.fabric.microsoft.com")

# Step 3: Decode mashup.pq to see the Power Query M code
echo "$DEF" | python3 -c "
import json, base64, sys
parts = json.load(sys.stdin)['definition']['parts']
for p in parts:
    if p['path'] == 'mashup.pq':
        print(base64.b64decode(p['payload']).decode('utf-8'))
"

Example 3: Check Refresh Job History

Prompt: "Show me the recent refresh history for this dataflow."

Agent response:

# Get recent job instances for a dataflow
az rest --method get \
  --url "https://api.fabric.microsoft.com/v1/workspaces/${WS_ID}/dataflows/${DF_ID}/jobs/instances?limit=5" \
  --resource "https://api.fabric.microsoft.com" \
  --query "value[].{Status:status, Start:startTimeUtc, End:endTimeUtc, Id:id}" -o table

Example 4: Discover Parameters from Definition

Prompt: "What parameters does this dataflow accept?"

Agent response:

# After decoding the definition (see Example 2), extract parameters:
echo "$DEF" | python3 -c "
import json, base64, sys
parts = json.load(sys.stdin)['definition']['parts']
for p in parts:
    if p['path'] == 'queryMetadata.json':
        meta = json.loads(base64.b64decode(p['payload']).decode('utf-8'))
        for qname, qmeta in meta.get('queriesMetadata', {}).items():
            if qmeta.get('queryGroupId') == 'parameters' or 'IsParameterQuery' in str(qmeta):
                print(f'Parameter: {qname}')
"

---

Query Evaluation

Execute an individual query from a dataflow and inspect results. Responses are a raw Apache Arrow IPC stream with Content-Type: application/vnd.apache.arrow.stream — not a JSON envelope. The first four bytes of a valid stream are the IPC continuation marker ff ff ff ff. Parse with pyarrow.ipc.open_stream().

Wire format: executeQuery returns the raw Apache Arrow IPC byte stream (Content-Type: application/vnd.apache.arrow.stream) — not JSON. Don't try to parse it with jq — there is no JSON envelope to extract. Use --output-file to save the bytes and parse as Arrow (see Examples 5–7).

Failures return HTTP 200: executeQuery returns 200 OK with application/vnd.apache.arrow.stream even when the underlying source query fails (Kusto SEM0100, T-SQL syntax error, missing column, etc.). The error is embedded inside the stream's PQ Arrow Metadata section as {"Error":"..."} — see dataflows-authoring-cli § mashup-preview.md → Detecting failures inside the Arrow body for detector snippets. Naive HTTP-status checks will treat failures as success.

Intent split (canonical executeQuery reference is [mashup-preview.md](../dataflows-authoring-cli/references/mashup-preview.md)): the same executeQuery endpoint serves two distinct intents. This skill covers the consumption intents:

- (a) Execute a persisted query — body {"QueryName":"<saved-shared>"} only (no customMashupDocument).

- (b) Ad-hoc read-only `customMashupDocument` — preview a candidate section Section1; ... document with no intent to persist via updateDefinition (Example 7).

>

If you intend to persist the M, use `dataflows-authoring-cli` § Workflow C (Preview-Driven Authoring Loop) — it adds the bootstrap-bind rule (chicken-and-egg connection binding for new credentialed dataflows), auto-wrap rule, hard-avoid for unbounded preview, and the post-preview persistence steps.

Auto-wrap caveat: The Fabric REST API expects customMashupDocument to be a complete `section Section1; ... shared X = ...;` document. Raw let ... in ... expressions are not auto-wrapped server-side — send a full section document and ensure the QueryName request field matches a shared member declared inside it.

Body shape: send a flat object with a top-level QueryName (field name is case-insensitive on the wire; PascalCase canonical). The {"queries":[{...}]} array shape always returns 400 DataflowExecuteQueryError: Invalid query name regardless of inner casing. A wrong QueryName value returns QueryNotFound (different error code). See dataflows-authoring-cli § mashup-preview.md → Request body.

`TimedOut` recovery for heavy persisted queries: if the persisted shared <Q> member is too expensive for the server-side executeQuery budget (~90s observed; surfaces as 400 DataflowExecuteQueryError → InternalErrorCode: EvaluationError, Message: Evaluation result error code: TimedOut), do not retry the same body — it is deterministic. Instead, re-issue with customMashupDocument set to the full persisted mashup.pq (decoded via Example 2) plus an appended probe shared __PreviewCap = Table.FirstN(<Q>, 10); and QueryName: "__PreviewCap". The cap pushes down for most connectors (Kusto, SQL, OData, web tables) so the source returns only the sample. See dataflows-authoring-cli § mashup-preview.md → Observed `InternalErrorCode` values.

Prerequisites

• Dataflow must exist (verify via Example 1 above)
• Query name must exist in the dataflow (verify in mashup.pq via Example 2)
• User must have Contributor role or higher (read+write) on the dataflow
• Optional: Query parameters and custom M code modifications

Example 5: Execute a Query and Save Results

Prompt: "Execute the SalesData query in this dataflow and save the results."

Agent response:

# Step 1: Identify the query to execute (must be a `shared` member of the dataflow's mashup)
QUERY_NAME="SalesData"  # Replace with your query name

# Step 2: Build the request body. To execute the persisted query as-is, send ONLY
#   QueryName — omit customMashupDocument. (Do NOT pass a self-referential
#   `shared X = let Source = #shared[X] in Source` — that recurses on itself
#   inside the candidate document.) For custom M, see Example 7.
jq -n --arg q "$QUERY_NAME" '{QueryName: $q}' > req.json

# Step 3: Execute the query (raw Apache Arrow IPC stream is written directly to disk)
az rest --method post \
  --url "https://api.fabric.microsoft.com/v1/workspaces/${WS_ID}/dataflows/${DF_ID}/executeQuery" \
  --resource "https://api.fabric.microsoft.com" \
  --body "@req.json" \
  --output-file "${QUERY_NAME}_results.arrow"

# Step 4: Check for embedded source errors (HTTP 200 does NOT mean success)
if grep -q '"Error":"' "${QUERY_NAME}_results.arrow"; then
  echo "Query execution failed (embedded in Arrow body):"
  python3 -c "import re,sys; raw=open(sys.argv[1],'rb').read().decode('utf-8','replace'); m=re.search(r'\\{\"Error\":\"[^\"]+\"\\}', raw); print(m.group(0) if m else '(marker present, JSON not parsed)')" "${QUERY_NAME}_results.arrow"
  exit 1
fi

echo "Query results saved to ${QUERY_NAME}_results.arrow"

Why `--output-file`? Without it, az rest dumps raw bytes to stdout and prints the warning Not a json response, outputting to stdout. For binary data suggest use "--output-file" to write to a file. Always pass --output-file for executeQuery so the Arrow IPC stream is captured cleanly. The jq -r '.data' | base64 --decode pattern from earlier docs is incorrect for `az rest` — there is no JSON envelope to extract.

Example 5b: Render Query Results as a Markdown Table

Prompt: "Show me the first 10 rows of those results."

Agent response — read the saved Arrow file and render head(N) as a markdown table in chat. Always do this after `executeQuery` (default N=10). The visual preview catches silent-success bugs that the embedded-error detector misses: filter dropped all rows, wrong column referenced, wrong cast, off-by-one filter, join key typo.

N=${N:-10}; ARROW_FILE="${QUERY_NAME}_results.arrow"
python3 - <<EOF
import json, sys, pyarrow as pa, pyarrow.ipc as ipc

with open("$ARROW_FILE", "rb") as f:
    try:
        reader = ipc.open_stream(f)
    except pa.lib.ArrowInvalid:
        f.seek(0); reader = ipc.open_file(f)
    # Surface any embedded engine error before rendering (mirrors mashup-preview.md § Error handling — A)
    for v in (reader.schema.metadata or {}).values():
        s = v.decode("utf-8", errors="replace")
        if '"Error"' in s:
            try:
                msg = json.loads(s)["Error"]
            except (ValueError, KeyError):
                msg = s[:200]
            sys.exit(f"Preview failed: {msg}")
    table = reader.read_all()

rows, cols = table.num_rows, table.num_columns
print(f"**{rows} rows × {cols} columns** · {', '.join(table.schema.names)}\n")
if rows == 0:
    sys.exit(0)

# Slice at the Arrow level before converting — avoids materialising the whole result.
df = table.slice(0, $N).to_pandas().copy()
# Truncate long string cells so the chat table stays readable.
for c in df.select_dtypes(include=["object", "string"]).columns:
    df[c] = df[c].astype(str).str.slice(0, 50)
try:
    print(df.to_markdown(index=False))           # requires the `tabulate` package
except ImportError:
    print(df.to_string(index=False))             # fallback: fixed-width text
EOF

Optional dep: pandas.DataFrame.to_markdown() requires the tabulate package — install once via your environment's standard Python tooling, the same way you installed pyarrow / pandas for Example 6. If tabulate is absent, the snippet falls back to fixed-width to_string() so rendering still works.

When to skip rendering: render head(N) by default. Skip only when (a) the user explicitly asked for the raw Arrow file with no preview, (b) the result row count exceeds ~1000 — in that case still render head(N) and add a one-line note about the total row count, or (c) the result is being streamed straight into another tool that will consume the Arrow file directly. In every other case, render — the saved file alone is not visible to the user in chat.

Example 6: Convert Arrow Results to CSV

Prompt: "Convert my Arrow query results to CSV so I can open them in Excel."

Agent response:

Prerequisite (one-time client-side setup): This example uses pyarrow to decode the Arrow IPC stream and pandas to write CSV. Install them once via your environment's standard Python tooling. These are local client dependencies, not part of the Fabric recipe.

python3 <<EOF
import pyarrow as pa
import pandas as pd
import sys

ARROW_FILE = "${QUERY_NAME}_results.arrow"
CSV_FILE = "${QUERY_NAME}_results.csv"

# Open the Arrow IPC stream directly (raw on the wire, no envelope)
with open(ARROW_FILE, "rb") as f:
    reader = pa.ipc.open_stream(f)

    # Defensive: surface any error embedded in the stream's PQ Arrow Metadata
    md = reader.schema.metadata or {}
    for k, v in md.items():
        s = v.decode("utf-8", errors="replace")
        if '"Error"' in s:
            print(f"Preview failed: {s}", file=sys.stderr)
            sys.exit(1)

    table = reader.read_all()

# Convert to pandas and export as CSV
df = table.to_pandas()
df.to_csv(CSV_FILE, index=False)

print(f"Converted {len(df)} rows to CSV")
print("Columns:", list(df.columns))
EOF

Example 7: Query with Custom M Code

Prompt: "Run a one-off ad-hoc M query against this dataflow without saving it."

Intent: ad-hoc read-only execution. The customMashupDocument is not persisted. If you intend to save the M via updateDefinition, use `dataflows-authoring-cli` § Workflow C instead — it adds bootstrap-bind, auto-wrap, and post-preview persistence rules.

Agent response:

# Execute a query with custom M code (e.g., filter, aggregate, transform).
# The customMashupDocument must be a complete `section` document; az rest does NOT auto-wrap raw expressions.
CUSTOM_M='section Section1;

shared CustomQuery = let
    Source = Table.FromRecords({[id=1, name="Alice"], [id=2, name="Bob"]}),
    Filtered = Table.SelectRows(Source, each [id] > 0)
in
    Filtered;'

jq -n --arg m "$CUSTOM_M" '{QueryName: "CustomQuery", customMashupDocument: $m}' > req.json

az rest --method post \
  --url "https://api.fabric.microsoft.com/v1/workspaces/${WS_ID}/dataflows/${DF_ID}/executeQuery" \
  --resource "https://api.fabric.microsoft.com" \
  --body "@req.json" \
  --output-file custom_results.arrow

# Always check for embedded errors before treating the file as a success
if grep -q '"Error":"' custom_results.arrow; then
    echo "Custom query failed; inspect custom_results.arrow for the embedded {\"Error\":...} block."
    exit 1
fi

---

Output Expectations

When this skill completes a task, the agent should return:

Consumption CLI Quick Reference

Concise az rest one-liners for all Dataflows Gen2 consumption operations. For full API details, see DATAFLOWS-CONSUMPTION-CORE.md. For full reusable scripts, see script-templates.md.

All examples assume reusable connection variables are set:

WS_ID="<workspaceId>"
DF_ID="<dataflowId>"
API="https://api.fabric.microsoft.com/v1"
AZ="az rest --resource https://api.fabric.microsoft.com"

Listing and Discovery

# List all dataflows in a workspace
$AZ --method get --url "$API/workspaces/$WS_ID/dataflows" \
  --query "value[].{name:displayName, id:id}" -o table

# Find dataflow by name
$AZ --method get --url "$API/workspaces/$WS_ID/dataflows" \
  --query "value[?displayName=='Sales Data Pipeline'].id" -o tsv

# Get dataflow properties
$AZ --method get --url "$API/workspaces/$WS_ID/dataflows/$DF_ID"

# List dataflows across all workspaces (iterate)
for ws in $(az rest --method get --resource "https://api.fabric.microsoft.com" \
  --url "$API/workspaces" --query "value[].id" -o tsv); do
  echo "--- Workspace: $ws ---"
  $AZ --method get --url "$API/workspaces/$ws/dataflows" \
    --query "value[].{name:displayName, id:id}" -o table 2>/dev/null
done

Parameters

# Discover all parameters
$AZ --method get --url "$API/workspaces/$WS_ID/dataflows/$DF_ID/parameters" \
  --query "value[].{name:name, type:type, required:isRequired, default:defaultValue}" -o table

# Get parameters as JSON (for scripting)
$AZ --method get --url "$API/workspaces/$WS_ID/dataflows/$DF_ID/parameters" \
  --query "value" -o json

Definition Exploration

⚠ LRO caveat: the getDefinition one-liners below are happy-path only — they assume a synchronous 200 response. The Fabric API may return 202 + Location instead (large definitions, server load), in which case the snippet decodes garbage. For production code, use the LRO-aware pattern in authoring-cli-quickref.md § Validate All Connections in a Dataflow, or copy the Fabric LRO Polling Pattern bash branch into your script.

# Get definition (returns base64 parts; happy-path 200 only — see caveat)
$AZ --method post --url "$API/workspaces/$WS_ID/dataflows/$DF_ID/getDefinition"

# Decode mashup.pq (Power Query M code)
$AZ --method post --url "$API/workspaces/$WS_ID/dataflows/$DF_ID/getDefinition" | \
  jq -r '.definition.parts[] | select(.path=="mashup.pq") | .payload' | base64 --decode

# Decode queryMetadata.json (query config and connections)
$AZ --method post --url "$API/workspaces/$WS_ID/dataflows/$DF_ID/getDefinition" | \
  jq -r '.definition.parts[] | select(.path=="queryMetadata.json") | .payload' | base64 --decode | jq .

# Decode .platform (item metadata)
$AZ --method post --url "$API/workspaces/$WS_ID/dataflows/$DF_ID/getDefinition" | \
  jq -r '.definition.parts[] | select(.path==".platform") | .payload' | base64 --decode | jq .

Job and Refresh Monitoring

# Recent job instances (all)
$AZ --method get --url "$API/workspaces/$WS_ID/dataflows/$DF_ID/jobs/instances" \
  --query "value[].{status:status, type:invokeType, start:startTimeUtc, end:endTimeUtc, error:failureReason}" -o table

# Last job status only
$AZ --method get --url "$API/workspaces/$WS_ID/dataflows/$DF_ID/jobs/instances" \
  --query "value[0].{status:status, start:startTimeUtc, end:endTimeUtc}" -o table

# Failed jobs only
$AZ --method get --url "$API/workspaces/$WS_ID/dataflows/$DF_ID/jobs/instances" \
  --query "value[?status=='Failed'].{id:id, start:startTimeUtc, error:failureReason}" -o table

# Poll a running operation
OP_ID="<operationId>"
$AZ --method get --url "$API/operations/$OP_ID"

PowerShell Equivalents

# List dataflows
az rest --method get --resource "https://api.fabric.microsoft.com" `
  --url "https://api.fabric.microsoft.com/v1/workspaces/$WS_ID/dataflows" `
  --query "value[].{name:displayName, id:id}" -o table

# Discover parameters
az rest --method get --resource "https://api.fabric.microsoft.com" `
  --url "https://api.fabric.microsoft.com/v1/workspaces/$WS_ID/dataflows/$DF_ID/parameters" `
  --query "value[].{name:name, type:type, required:isRequired}" -o table

# Decode mashup.pq (PowerShell)
$response = az rest --method post --resource "https://api.fabric.microsoft.com" `
  --url "https://api.fabric.microsoft.com/v1/workspaces/$WS_ID/dataflows/$DF_ID/getDefinition" | ConvertFrom-Json
$mashup = $response.definition.parts | Where-Object { $_.path -eq "mashup.pq" }
[System.Text.Encoding]::UTF8.GetString([System.Convert]::FromBase64String($mashup.payload))

Agent Integration Notes

• GitHub Copilot CLI: use gh copilot suggest -t shell for az rest one-liners; ensure --resource in output.
• Claude Code / Cowork: run az rest via bash tool; follow the Agentic Workflow in SKILL.md; produce scripts using script-templates.md.
• Always verify az login session before first REST operation.

Query Execution

# Execute a persisted query — send QueryName only (omit customMashupDocument).
# Raw Apache Arrow IPC stream is written directly to a file via --output-file.
QUERY_NAME="SalesData"
$AZ --method post --url "$API/workspaces/$WS_ID/dataflows/$DF_ID/executeQuery" \
  --body "{\"QueryName\": \"$QUERY_NAME\"}" \
  --output-file "${QUERY_NAME}.arrow"

# Check for embedded errors on EVERY arrow file before consuming it — executeQuery
# returns HTTP 200 even when the source query fails; the error JSON is embedded inside
# the Arrow stream's `PQ Arrow Metadata` block. Feeding a failed stream to pyarrow
# produces confusing parse errors or silent garbage.
if grep -q '"Error":"' "${QUERY_NAME}.arrow"; then echo "$QUERY_NAME query failed (see ${QUERY_NAME}.arrow)"; exit 1; fi

# Execute with custom M code (full section document; az rest does NOT auto-wrap a raw `let ... in ...` expression)
CUSTOM_M='section Section1;

shared Custom = let Source = Table.FromRecords({[id=1, val="A"]}) in Source;'
jq -n --arg m "$CUSTOM_M" '{QueryName: "Custom", customMashupDocument: $m}' > req.json
$AZ --method post --url "$API/workspaces/$WS_ID/dataflows/$DF_ID/executeQuery" \
  --body "@req.json" --output-file Custom.arrow

if grep -q '"Error":"' Custom.arrow; then echo "Custom query failed (see Custom.arrow)"; exit 1; fi

# Convert successful Arrow results to CSV (requires pyarrow + pandas)
python3 -c "
import pyarrow as pa
table = pa.ipc.open_stream(open('${QUERY_NAME}.arrow', 'rb')).read_all()
print(table.to_pandas().to_csv(index=False))
"

Discovery Patterns and Queries

Extended discovery patterns beyond the basics in SKILL.md Agentic Exploration. All patterns use az rest with JMESPath queries and jq for JSON processing.

Listing and Filtering Dataflows

List All Dataflows with Full Details

# All dataflows with description and type
az rest --method get --resource "https://api.fabric.microsoft.com" \
  --url "$API/workspaces/$WS_ID/dataflows" \
  --query "value[].{id:id, name:displayName, desc:description, type:type}" -o table

List with Pagination (Handling continuationToken)

# Paginate through all dataflows
URL="$API/workspaces/$WS_ID/dataflows"
while [ -n "$URL" ]; do
  RESPONSE=$(az rest --method get --resource "https://api.fabric.microsoft.com" --url "$URL")
  echo "$RESPONSE" | jq -r '.value[] | [.id, .displayName] | @tsv'
  TOKEN=$(echo "$RESPONSE" | jq -r '.continuationToken // empty')
  if [ -n "$TOKEN" ]; then
    URL="$API/workspaces/$WS_ID/dataflows?continuationToken=$TOKEN"
  else
    URL=""
  fi
done

Cross-Workspace Dataflow Inventory

# Find dataflows across all accessible workspaces
for ws in $(az rest --method get --resource "https://api.fabric.microsoft.com" \
  --url "$API/workspaces" --query "value[].id" -o tsv); do
  WS_NAME=$(az rest --method get --resource "https://api.fabric.microsoft.com" \
    --url "$API/workspaces/$ws" --query "displayName" -o tsv 2>/dev/null)
  DATAFLOWS=$(az rest --method get --resource "https://api.fabric.microsoft.com" \
    --url "$API/workspaces/$ws/dataflows" --query "value[].displayName" -o tsv 2>/dev/null)
  if [ -n "$DATAFLOWS" ]; then
    echo "=== $WS_NAME ($ws) ==="
    echo "$DATAFLOWS"
  fi
done

Definition Decoding

Decode All Definition Parts

RESPONSE=$(az rest --method post --resource "https://api.fabric.microsoft.com" \
  --url "$API/workspaces/$WS_ID/dataflows/$DF_ID/getDefinition")

# Decode each part
for PART_PATH in "mashup.pq" "queryMetadata.json" ".platform"; do
  echo "=== $PART_PATH ==="
  echo "$RESPONSE" | jq -r ".definition.parts[] | select(.path==\"$PART_PATH\") | .payload" | base64 --decode
  echo ""
done

Handle LRO for getDefinition

# getDefinition may return 202 with Location header
HTTP_CODE=$(curl -s -o /tmp/df_def.json -w "%{http_code}" -X POST \
  -H "Authorization: Bearer $(az account get-access-token --resource "https://api.fabric.microsoft.com" --query accessToken -o tsv)" \
  "$API/workspaces/$WS_ID/dataflows/$DF_ID/getDefinition")

if [ "$HTTP_CODE" = "202" ]; then
  LOCATION=$(grep -i "location:" /tmp/df_def_headers.txt | tr -d '\r' | awk '{print $2}')
  echo "LRO started. Polling $LOCATION ..."
  while true; do
    STATUS=$(az rest --method get --resource "https://api.fabric.microsoft.com" \
      --url "$LOCATION" --query "status" -o tsv)
    echo "Status: $STATUS"
    [ "$STATUS" = "Succeeded" ] || [ "$STATUS" = "Failed" ] && break
    sleep 5
  done
  # Get the result
  az rest --method get --resource "https://api.fabric.microsoft.com" \
    --url "${LOCATION}/result"
fi

Extract Query Names from mashup.pq

# Decode mashup.pq and list all shared queries
az rest --method post --resource "https://api.fabric.microsoft.com" \
  --url "$API/workspaces/$WS_ID/dataflows/$DF_ID/getDefinition" | \
  jq -r '.definition.parts[] | select(.path=="mashup.pq") | .payload' | \
  base64 --decode | grep -oP '(?<=shared )\w+(?= =)'

Analyze Connections from queryMetadata.json

# Decode queryMetadata.json and list connections
az rest --method post --resource "https://api.fabric.microsoft.com" \
  --url "$API/workspaces/$WS_ID/dataflows/$DF_ID/getDefinition" | \
  jq -r '.definition.parts[] | select(.path=="queryMetadata.json") | .payload' | \
  base64 --decode | jq '.connections[] | {path, kind, connectionId}'

Identify Load-Enabled vs Helper Queries

# Decode queryMetadata.json and classify queries
az rest --method post --resource "https://api.fabric.microsoft.com" \
  --url "$API/workspaces/$WS_ID/dataflows/$DF_ID/getDefinition" | \
  jq -r '.definition.parts[] | select(.path=="queryMetadata.json") | .payload' | \
  base64 --decode | jq '.queriesMetadata | to_entries[] | {
    name: .key,
    loadEnabled: .value.loadEnabled,
    isHidden: .value.isHidden,
    role: (if .value.loadEnabled then "OUTPUT" elif .value.isHidden then "HELPER (hidden)" else "STAGING" end)
  }'

Parameter Discovery

Format Parameters as Report

# Tabular parameter report
az rest --method get --resource "https://api.fabric.microsoft.com" \
  --url "$API/workspaces/$WS_ID/dataflows/$DF_ID/parameters" | \
  jq -r '.value[] | "Parameter: \(.name)\n  Type:     \(.type)\n  Required: \(.isRequired)\n  Default:  \(.defaultValue // "none")\n  Desc:     \(.description // "none")\n"'

Find Parameters in mashup.pq

# Extract parameter definitions from M code
az rest --method post --resource "https://api.fabric.microsoft.com" \
  --url "$API/workspaces/$WS_ID/dataflows/$DF_ID/getDefinition" | \
  jq -r '.definition.parts[] | select(.path=="mashup.pq") | .payload' | \
  base64 --decode | grep "IsParameterQuery"

Job History Analysis

Summarize Recent Job Results

# Count jobs by status
az rest --method get --resource "https://api.fabric.microsoft.com" \
  --url "$API/workspaces/$WS_ID/dataflows/$DF_ID/jobs/instances" | \
  jq '[.value[] | .status] | group_by(.) | map({status: .[0], count: length})'

Calculate Average Refresh Duration

# Average duration of completed jobs (in seconds)
az rest --method get --resource "https://api.fabric.microsoft.com" \
  --url "$API/workspaces/$WS_ID/dataflows/$DF_ID/jobs/instances" | \
  jq '[.value[] | select(.status=="Completed" and .endTimeUtc != null) |
    (((.endTimeUtc | fromdateiso8601) - (.startTimeUtc | fromdateiso8601)))] |
    if length > 0 then (add / length | round) else 0 end' | \
  xargs -I{} echo "Average refresh duration: {} seconds"

Extract Failure Reasons

# List all failures with reasons
az rest --method get --resource "https://api.fabric.microsoft.com" \
  --url "$API/workspaces/$WS_ID/dataflows/$DF_ID/jobs/instances" | \
  jq -r '.value[] | select(.status=="Failed") | "\(.startTimeUtc) | \(.failureReason // "unknown")"'

Script Templates

Self-contained templates for dataflow consumption via CLI. Copy-paste and customize.

Bash — List Dataflows with Pagination

#!/usr/bin/env bash
set -euo pipefail

# --- Configuration ---
WS_ID="${WS_ID:?Set WS_ID env var (workspace ID)}"
API="https://api.fabric.microsoft.com/v1"

# --- Prerequisites ---
az account show >/dev/null 2>&1 || { echo "ERROR: Not logged in. Run: az login"; exit 1; }
command -v jq >/dev/null 2>&1 || { echo "ERROR: jq not found. Install: apt-get install jq"; exit 1; }

# --- Paginated listing ---
echo "Dataflows in workspace $WS_ID:"
echo "---"
URL="$API/workspaces/$WS_ID/dataflows"
COUNT=0
while [ -n "$URL" ]; do
  RESPONSE=$(az rest --method get --resource "https://api.fabric.microsoft.com" --url "$URL")
  echo "$RESPONSE" | jq -r '.value[] | "\(.displayName)\t\(.id)"'
  PAGE_COUNT=$(echo "$RESPONSE" | jq '.value | length')
  COUNT=$((COUNT + PAGE_COUNT))
  TOKEN=$(echo "$RESPONSE" | jq -r '.continuationToken // empty')
  if [ -n "$TOKEN" ]; then
    URL="$API/workspaces/$WS_ID/dataflows?continuationToken=$TOKEN"
  else
    URL=""
  fi
done
echo "---"
echo "✓ Total: $COUNT dataflow(s)"

Bash — Full Definition Decode and Report

#!/usr/bin/env bash
set -euo pipefail

# --- Configuration ---
WS_ID="${WS_ID:?Set WS_ID env var}"
DF_ID="${DF_ID:?Set DF_ID env var (dataflow ID)}"
API="https://api.fabric.microsoft.com/v1"
OUTPUT_DIR="${1:-.}"

# --- Prerequisites ---
az account show >/dev/null 2>&1 || { echo "ERROR: Run 'az login' first."; exit 1; }
command -v jq >/dev/null 2>&1 || { echo "ERROR: jq not found."; exit 1; }
mkdir -p "$OUTPUT_DIR"

# --- Get dataflow name ---
DF_NAME=$(az rest --method get --resource "https://api.fabric.microsoft.com" \
  --url "$API/workspaces/$WS_ID/dataflows/$DF_ID" --query "displayName" -o tsv)
echo "=== Dataflow: $DF_NAME ==="

# --- Get definition ---
echo "Fetching definition..."
RESPONSE=$(az rest --method post --resource "https://api.fabric.microsoft.com" \
  --url "$API/workspaces/$WS_ID/dataflows/$DF_ID/getDefinition")

# --- Decode and save each part ---
for PART in "mashup.pq" "queryMetadata.json" ".platform"; do
  FILENAME=$(basename "$PART")
  echo "$RESPONSE" | jq -r ".definition.parts[] | select(.path==\"$PART\") | .payload" | \
    base64 --decode > "$OUTPUT_DIR/$FILENAME"
  echo "✓ Decoded $PART → $OUTPUT_DIR/$FILENAME"
done

# --- Report: Queries ---
echo ""
echo "=== Queries ==="
grep -oP '(?<=shared )\w+(?= =)' "$OUTPUT_DIR/mashup.pq" | while read -r QUERY; do
  echo "  - $QUERY"
done

# --- Report: Connections ---
echo ""
echo "=== Connections ==="
jq -r '.connections[] | "  - [\(.kind)] \(.path)"' "$OUTPUT_DIR/queryMetadata.json"

# --- Report: Load-enabled queries ---
echo ""
echo "=== Output Queries (loadEnabled=true) ==="
jq -r '.queriesMetadata | to_entries[] | select(.value.loadEnabled==true) | "  - \(.key)"' "$OUTPUT_DIR/queryMetadata.json"

echo ""
echo "✓ Definition report complete. Files saved to $OUTPUT_DIR/"

Bash — Monitor Refresh with Polling Loop

#!/usr/bin/env bash
set -euo pipefail

# --- Configuration ---
WS_ID="${WS_ID:?Set WS_ID env var}"
DF_ID="${DF_ID:?Set DF_ID env var}"
API="https://api.fabric.microsoft.com/v1"
MAX_WAIT="${MAX_WAIT:-3600}"  # 60 minutes default
POLL_INTERVAL=5

# --- Prerequisites ---
az account show >/dev/null 2>&1 || { echo "ERROR: Run 'az login' first."; exit 1; }

# --- Get latest job instance ---
echo "Checking latest job for dataflow $DF_ID..."
LATEST=$(az rest --method get --resource "https://api.fabric.microsoft.com" \
  --url "$API/workspaces/$WS_ID/dataflows/$DF_ID/jobs/instances" \
  --query "value[0]")

STATUS=$(echo "$LATEST" | jq -r '.status')
JOB_ID=$(echo "$LATEST" | jq -r '.id')
echo "Job $JOB_ID — Status: $STATUS"

if [ "$STATUS" = "Completed" ] || [ "$STATUS" = "Failed" ] || [ "$STATUS" = "Cancelled" ]; then
  echo "Job already in terminal state."
  echo "$LATEST" | jq '{status, invokeType, startTimeUtc, endTimeUtc, failureReason}'
  exit 0
fi

# --- Poll until terminal ---
ELAPSED=0
while [ "$ELAPSED" -lt "$MAX_WAIT" ]; do
  sleep "$POLL_INTERVAL"
  ELAPSED=$((ELAPSED + POLL_INTERVAL))

  STATUS=$(az rest --method get --resource "https://api.fabric.microsoft.com" \
    --url "$API/workspaces/$WS_ID/dataflows/$DF_ID/jobs/instances" \
    --query "value[0].status" -o tsv)

  echo "[${ELAPSED}s] Status: $STATUS"

  case "$STATUS" in
    Completed|Failed|Cancelled)
      echo "✓ Job reached terminal state: $STATUS"
      az rest --method get --resource "https://api.fabric.microsoft.com" \
        --url "$API/workspaces/$WS_ID/dataflows/$DF_ID/jobs/instances" \
        --query "value[0]"
      exit 0
      ;;
  esac

  # Exponential backoff: 5 → 10 → 20 → 30 (cap)
  if [ "$POLL_INTERVAL" -lt 30 ]; then
    POLL_INTERVAL=$((POLL_INTERVAL * 2))
    [ "$POLL_INTERVAL" -gt 30 ] && POLL_INTERVAL=30
  fi
done

echo "⚠ Timeout after ${MAX_WAIT}s. Last status: $STATUS"
exit 1

Bash — Parameter Discovery Report

#!/usr/bin/env bash
set -euo pipefail

# --- Configuration ---
WS_ID="${WS_ID:?Set WS_ID env var}"
DF_ID="${DF_ID:?Set DF_ID env var}"
API="https://api.fabric.microsoft.com/v1"

# --- Prerequisites ---
az account show >/dev/null 2>&1 || { echo "ERROR: Run 'az login' first."; exit 1; }
command -v jq >/dev/null 2>&1 || { echo "ERROR: jq not found."; exit 1; }

# --- Get dataflow name ---
DF_NAME=$(az rest --method get --resource "https://api.fabric.microsoft.com" \
  --url "$API/workspaces/$WS_ID/dataflows/$DF_ID" --query "displayName" -o tsv)
echo "=== Parameters for: $DF_NAME ==="
echo ""

# --- Discover parameters ---
PARAMS=$(az rest --method get --resource "https://api.fabric.microsoft.com" \
  --url "$API/workspaces/$WS_ID/dataflows/$DF_ID/parameters" 2>&1) || {
  echo "No parameters found (dataflow may not be parametric)."
  exit 0
}

PARAM_COUNT=$(echo "$PARAMS" | jq '.value | length')
if [ "$PARAM_COUNT" -eq 0 ]; then
  echo "No parameters defined."
  exit 0
fi

echo "$PARAMS" | jq -r '.value[] | "Parameter: \(.name)\n  Type:     \(.type)\n  Required: \(.isRequired)\n  Default:  \(.defaultValue // "none")\n  Desc:     \(.description // "none")\n"'

echo "---"
echo "✓ $PARAM_COUNT parameter(s) discovered"

PowerShell — List Dataflows

#Requires -Version 5.1
param(
    [Parameter(Mandatory)][string]$WorkspaceId
)

$null = az account show 2>$null
if ($LASTEXITCODE -ne 0) { Write-Error "Not logged in. Run: az login"; exit 1 }

$API = "https://api.fabric.microsoft.com/v1"
$url = "$API/workspaces/$WorkspaceId/dataflows"
$allDataflows = @()

do {
    $response = az rest --method get --resource "https://api.fabric.microsoft.com" --url $url | ConvertFrom-Json
    $allDataflows += $response.value
    if ($response.continuationToken) {
        $url = "$API/workspaces/$WorkspaceId/dataflows?continuationToken=$($response.continuationToken)"
    } else {
        $url = $null
    }
} while ($url)

$allDataflows | ForEach-Object {
    [PSCustomObject]@{
        Name = $_.displayName
        Id   = $_.id
        Desc = $_.description
    }
} | Format-Table -AutoSize

Write-Host "Total: $($allDataflows.Count) dataflow(s)"

PowerShell — Definition Decode

⚠ LRO caveat: the az rest --method post .../getDefinition call below assumes a synchronous 200 response. The Fabric API may return 202 + Location (large definitions, server load). For production code, use the Invoke-WebRequest LRO branch shown immediately below, or copy the Fabric LRO Polling Pattern PowerShell branch into your script.

#Requires -Version 5.1
param(
    [Parameter(Mandatory)][string]$WorkspaceId,
    [Parameter(Mandatory)][string]$DataflowId,
    [string]$OutputDir = "."
)

$null = az account show 2>$null
if ($LASTEXITCODE -ne 0) { Write-Error "Not logged in. Run: az login"; exit 1 }

$API = "https://api.fabric.microsoft.com/v1"
$resource = "https://api.fabric.microsoft.com"
New-Item -ItemType Directory -Path $OutputDir -Force | Out-Null

# getDefinition is an LRO. az rest cannot surface response headers, so use Invoke-WebRequest
# to capture the 202/Location case, then poll per Fabric LRO contract.
$token = az account get-access-token --resource $resource --query accessToken -o tsv
$headers = @{ Authorization = "Bearer $token"; "Content-Length" = "0" }
try {
    $resp = Invoke-WebRequest -Method Post -Uri "$API/workspaces/$WorkspaceId/dataflows/$DataflowId/getDefinition" `
        -Headers $headers -UseBasicParsing
} catch {
    Write-Error "getDefinition failed: $($_.Exception.Message)"; exit 1
}
if ($resp.StatusCode -eq 202) {
    $location = $resp.Headers["Location"]
    if ($location -is [array]) { $location = $location[0] }
    $retryRaw = $resp.Headers["Retry-After"]
    if ($retryRaw -is [array]) { $retryRaw = $retryRaw[0] }
    $retryAfter = 5; [void][int]::TryParse([string]$retryRaw, [ref]$retryAfter)
    $response = $null
    while ($null -eq $response) {
        Start-Sleep -Seconds $retryAfter
        $op = az rest --method get --resource $resource --url $location | ConvertFrom-Json
        switch ($op.status) {
            'Succeeded' { $response = az rest --method get --resource $resource --url "$($location.TrimEnd('/'))/result" | ConvertFrom-Json }
            'Failed'    { Write-Error "getDefinition failed: $($op.error.message)"; exit 1 }
            'Cancelled' { Write-Error "getDefinition cancelled"; exit 1 }
        }
    }
} else {
    $response = $resp.Content | ConvertFrom-Json
}

foreach ($part in $response.definition.parts) {
    $filename = Split-Path $part.path -Leaf
    $decoded = [System.Text.Encoding]::UTF8.GetString([System.Convert]::FromBase64String($part.payload))
    Set-Content -Path (Join-Path $OutputDir $filename) -Value $decoded -Encoding UTF8
    Write-Host "Decoded $($part.path) -> $OutputDir\$filename"
}

Write-Host "`nDone. Files saved to $OutputDir"

PowerShell — Job History Report

#Requires -Version 5.1
param(
    [Parameter(Mandatory)][string]$WorkspaceId,
    [Parameter(Mandatory)][string]$DataflowId
)

$null = az account show 2>$null
if ($LASTEXITCODE -ne 0) { Write-Error "Not logged in. Run: az login"; exit 1 }

$API = "https://api.fabric.microsoft.com/v1"
$jobs = (az rest --method get --resource "https://api.fabric.microsoft.com" `
  --url "$API/workspaces/$WorkspaceId/dataflows/$DataflowId/jobs/instances" | ConvertFrom-Json).value

if (-not $jobs) { Write-Host "No job history found."; exit 0 }

$jobs | ForEach-Object {
    [PSCustomObject]@{
        Status    = $_.status
        Type      = $_.invokeType
        Start     = $_.startTimeUtc
        End       = $_.endTimeUtc
        Error     = $_.failureReason
    }
} | Format-Table -AutoSize

$completed = $jobs | Where-Object { $_.status -eq "Completed" }
$failed = $jobs | Where-Object { $_.status -eq "Failed" }
Write-Host "Summary: $($completed.Count) completed, $($failed.Count) failed, $($jobs.Count) total"

Bash — Query Execution and Arrow Conversion

Execute a query and convert Apache Arrow results to CSV:

#!/bin/bash

# Configuration
WORKSPACE_ID="${1:?Usage: $0 <workspace_id> <dataflow_id> <query_name>}"
DATAFLOW_ID="${2:?Usage: $0 <workspace_id> <dataflow_id> <query_name>}"
QUERY_NAME="${3:?Usage: $0 <workspace_id> <dataflow_id> <query_name>}"

API="https://api.fabric.microsoft.com/v1"
AZ="az rest --resource https://api.fabric.microsoft.com"

echo "Executing query: $QUERY_NAME from dataflow: $DATAFLOW_ID"

# Execute the query — raw Apache Arrow IPC stream is captured via --output-file (NOT a JSON envelope).
# For persisted queries, send QueryName only — omit customMashupDocument.
ARROW_PATH="${QUERY_NAME}_results.arrow"
$AZ --method post \
  --url "$API/workspaces/$WORKSPACE_ID/dataflows/$DATAFLOW_ID/executeQuery" \
  --body "{\"QueryName\": \"$QUERY_NAME\"}" \
  --output-file "$ARROW_PATH"

if [ ! -s "$ARROW_PATH" ]; then
  echo "ERROR: Empty response from executeQuery"
  exit 1
fi

# Detect embedded source errors — executeQuery returns HTTP 200 even on Kusto/SQL failures;
# the error JSON sits inside the stream's PQ Arrow Metadata block.
if grep -q '"Error":"' "$ARROW_PATH"; then
  echo "ERROR: Query execution failed (embedded in Arrow body):"
  python3 -c "import re,sys; raw=open(sys.argv[1],'rb').read().decode('utf-8','replace'); m=re.search(r'\\{\"Error\":\"[^\"]+\"\\}', raw); print(m.group(0) if m else '(error marker present, JSON not parsed)')" "$ARROW_PATH"
  exit 1
fi

echo "Arrow data saved to: $ARROW_PATH"

# Convert to CSV if pyarrow is available
if command -v python3 >/dev/null && python3 -c "import pyarrow, pandas" 2>/dev/null; then
  python3 << EOF
import pyarrow as pa
import pandas as pd

table = pa.ipc.open_stream(open("${QUERY_NAME}_results.arrow", "rb")).read_all()
df = table.to_pandas()
df.to_csv("${QUERY_NAME}_results.csv", index=False)

print(f"Converted {len(df)} rows to: ${QUERY_NAME}_results.csv")
print(f"Columns: {list(df.columns)}")
EOF
else
  echo "Note: Install pyarrow+pandas for automatic CSV conversion: pip install pyarrow pandas"
fi

PowerShell — Query Execution and Arrow Conversion

Execute a query and convert Apache Arrow results to CSV:

param(
    [Parameter(Mandatory)]
    [string]$WorkspaceId,
    
    [Parameter(Mandatory)]
    [string]$DataflowId,
    
    [Parameter(Mandatory)]
    [string]$QueryName,
    
    [string]$CustomMashup = ""
)

$API = "https://api.fabric.microsoft.com/v1"
$AZ = "az rest --resource https://api.fabric.microsoft.com"

Write-Host "Executing query: $QueryName from dataflow: $DataflowId"

# Build request body — omit customMashupDocument for persisted queries
if ($CustomMashup) {
    $body = @{
        QueryName = $QueryName
        customMashupDocument = $CustomMashup
    } | ConvertTo-Json -Compress
} else {
    $body = @{
        QueryName = $QueryName
    } | ConvertTo-Json -Compress
}

$bodyFile = [IO.Path]::GetTempFileName()
[IO.File]::WriteAllText($bodyFile, $body, [Text.UTF8Encoding]::new($false))

$arrowPath = "${QueryName}_results.arrow"

# Execute the query — raw Apache Arrow IPC stream is captured via --output-file (NOT a JSON envelope).
try {
    az rest --method post `
      --resource "https://api.fabric.microsoft.com" `
      --url "$API/workspaces/$WorkspaceId/dataflows/$DataflowId/executeQuery" `
      --body "@$bodyFile" `
      --output-file $arrowPath
} catch {
    Write-Error "Failed to execute query: $_"
    exit 1
} finally {
    Remove-Item $bodyFile -ErrorAction SilentlyContinue
}

if (-not (Test-Path $arrowPath) -or (Get-Item $arrowPath).Length -eq 0) {
    Write-Error "Empty response from executeQuery"
    exit 1
}

# Detect embedded source errors — executeQuery returns HTTP 200 even on Kusto/SQL failures;
# the error JSON sits inside the stream's PQ Arrow Metadata block.
$arrowText = [Text.Encoding]::UTF8.GetString([IO.File]::ReadAllBytes($arrowPath))
$embeddedErr = [Regex]::Match($arrowText, '\{"Error":"[^"]+"\}').Value
if ($embeddedErr) {
    Write-Error "Query execution failed (embedded in Arrow body): $embeddedErr"
    exit 1
}

Write-Host "Arrow data saved to: $arrowPath"

# Convert to CSV if pyarrow is available
$pythonScript = @"
import pyarrow as pa
import pandas as pd
import sys

try:
    table = pa.ipc.open_stream(open('$arrowPath', 'rb')).read_all()
    df = table.to_pandas()
    df.to_csv('${QueryName}_results.csv', index=False)
    print(f'Converted {len(df)} rows to: ${QueryName}_results.csv')
    print(f'Columns: {list(df.columns)}')
except Exception as e:
    print(f'Conversion failed: {e}', file=sys.stderr)
    sys.exit(1)
"@

$pythonScript | python3
# Native commands don't throw on non-zero exit — guard on $LASTEXITCODE so a missing
# pyarrow/pandas (ModuleNotFoundError) or absent python3 surfaces the install hint
# instead of silently leaving a half-finished conversion.
if ($LASTEXITCODE -ne 0) {
    Write-Host "Note: Install pyarrow+pandas for automatic CSV conversion: pip install pyarrow pandas"
}

Batch Query Execution with Error Handling

Execute multiple queries and collect results:

#!/bin/bash

# Configuration
WORKSPACE_ID="${1:?Usage: $0 <workspace_id> <dataflow_id>}"
DATAFLOW_ID="${2:?Usage: $0 <workspace_id> <dataflow_id>}"
QUERIES=("SalesData" "CustomerList" "ProductCatalog")

API="https://api.fabric.microsoft.com/v1"
AZ="az rest --resource https://api.fabric.microsoft.com"

mkdir -p query_results
FAILED_QUERIES=()

for QUERY in "${QUERIES[@]}"; do
    echo "Processing: $QUERY"

    ARROW_PATH="query_results/${QUERY}.arrow"
    $AZ --method post \
      --url "$API/workspaces/$WORKSPACE_ID/dataflows/$DATAFLOW_ID/executeQuery" \
      --body "{\"QueryName\": \"$QUERY\"}" \
      --output-file "$ARROW_PATH" 2>/dev/null

    # executeQuery returns HTTP 200 even on source-layer failures; the error JSON is embedded
    # inside the Arrow stream's PQ Arrow Metadata block. Treat absence of "Error":" as success.
    if [ -s "$ARROW_PATH" ] && ! grep -q '"Error":"' "$ARROW_PATH"; then
        echo "  ✓ Saved to $ARROW_PATH"
    else
        echo "  ✗ Failed"
        FAILED_QUERIES+=("$QUERY")
    fi
done

echo ""
echo "Summary: $((${#QUERIES[@]} - ${#FAILED_QUERIES[@]}))/${#QUERIES[@]} queries succeeded"

if [ ${#FAILED_QUERIES[@]} -gt 0 ]; then
    echo "Failed queries: ${FAILED_QUERIES[*]}"
    exit 1
fi