dbt™ Model Query Cost Optimizer — Snowflake

Automate Snowflake dbt™ cost optimization with a DinoAI agent that identifies expensive models, rewrites Snowflake-specific anti-patterns, opens a PR, and posts a cost summary to Slack.

Automatically detect your most expensive dbt™ models, diagnose Snowflake-specific SQL anti-patterns in the model source files, apply targeted rewrites, and open a pull request — all triggered by a single DinoAI agent session. The agent reads Snowflake query performance stats to understand where credits and execution time are being wasted, makes the fixes directly in your repository, and posts the PR link to Slack without any manual intervention.

Before You Start

Paradime

Your Paradime API endpoint, API key, and API secret — generate these under Workspace Settings → API. Make sure to enable DinoAI agent API capabilities. Requires Admin access.

Snowflake

Your Paradime workspace must have an active Snowflake connection with sufficient permissions to read query history metadata. The agent calls get_snowflake_query_performance_stats to retrieve performance stats for the Snowflake queries that each dbt™ model produced.

Model source

You need at least one of the following so the script knows which dbt™ models to analyse:

A Bolt schedule name — the script reads query IDs directly from the latest run's run_results.json artifact, which maps each dbt™ model to the Snowflake query it produced via adapter_response.query_id
One or more Snowflake query IDs supplied directly via CLI flag or environment variable, if you already know which models are expensive

Integrations

The following must already be connected in Paradime:

Slack — the agent posts the PR link and a cost summary to #finops via post_slack_message

What You'll Build

By the end of this guide you'll have:

A Poetry project wired to the Paradime SDK
A run_snowflake_query_optimizer.py trigger script that reads dbt™ model results from Bolt artifacts (or accepts Snowflake query IDs directly), filters to the top costly models by execution time, and hands the full context to the agent
A snowflake-query-optimizer DinoAI agent YAML that diagnoses Snowflake-specific anti-patterns, rewrites dbt™ model SQL and config, opens a PR, and notifies Slack

What the Agent Does

Once triggered, the agent works through five phases without stopping:

PHASE 1 — Fetch Snowflake performance stats for each dbt™ model's query ID
PHASE 2 — Locate the dbt™ model .sql and YAML files in the repository
PHASE 3 — Diagnose Snowflake SQL anti-patterns and build a concrete optimisation plan
PHASE 4 — Create a branch, apply all changes to the model files, commit and push
PHASE 5 — Open a PR and post a cost summary to #finops on Slack

The agent never drops columns referenced by downstream models, never pushes directly to main, and never guesses — if a business decision is required it leaves a TODO comment in the SQL and calls it out in the PR description.

Anti-Patterns the Agent Detects and Fixes

Anti-Pattern

What It Means

Fix Applied

WAREHOUSE_OVERSIZED

Model runs on a warehouse larger than required for its data volume, burning unnecessary credits

Adds a snowflake_warehouse config targeting a smaller warehouse size

NO_CLUSTER_KEY

Large table filtered on high-cardinality columns but no clustering key defined, causing full micro-partition scans

Adds cluster_by to the dbt™ config block with the appropriate filter columns

WRONG_MATERIALIZATION

Frequently queried view causing repeated full recomputation on every downstream query

Changes to table or incremental materialization

INCREMENTAL_OPPORTUNITY

Full table rebuild on every run despite having a natural time-based key, consuming credits unnecessarily

Converts to incremental with an is_incremental() filter on the timestamp column

RESULT_CACHE_BYPASS

Model query is not benefiting from Snowflake's result cache due to non-deterministic functions (e.g. CURRENT_TIMESTAMP(), RANDOM()) used unnecessarily

Removes or replaces non-deterministic expressions with deterministic equivalents

SELECT_STAR

All columns selected from a source, scanning micro-partitions that contain unneeded data

Enumerates only the columns actually used by downstream models

CARTESIAN_FAN_OUT_JOIN

JOIN on a non-unique key causing row multiplication and credit waste

Adds QUALIFY ROW_NUMBER() OVER (PARTITION BY ...) = 1 deduplication before the join

REPEATED_CTE

A CTE is referenced more than twice in the same query; Snowflake re-evaluates it each time

Materialises the CTE as a separate intermediate dbt™ model

UNFILTERED_LARGE_TABLE

A large source() or ref() pulled with no predicate, scanning all micro-partitions

Pushes a WHERE clause into the CTE that reads that table

SEARCH_OPTIMISATION_CANDIDATE

Model supports frequent point lookups or selective equality filters but search_optimization is not enabled

Adds post-hook: ALTER TABLE {{ this }} ADD SEARCH OPTIMIZATION to the config block

Architecture Overview

How It Works

run_snowflake_query_optimizer.py loads configuration from environment variables, optionally reads the latest Bolt run's run_results.json artifact to extract the Snowflake query ID produced by each dbt™ model from adapter_response.query_id, filters to the top-N slowest models above a configurable execution time threshold (default: 30 seconds), then triggers a single snowflake-query-optimizer DinoAI agent session with the full model context pre-embedded in the opening message. The script polls every 20 seconds until the agent completes or times out (default: 60 minutes), then prints the full optimisation report to stdout. An optional follow-up drill-down on a single model can be sent within the same session.

Unlike the BigQuery version which filters by bytes processed, the Snowflake version filters by execution time in seconds — this is the most reliable proxy for credit consumption available in run_results.json without needing to call the Snowflake query history API upfront.

Set Up the Poetry Project

Create the following pyproject.toml at the same directory level as your dbt_project.yml. This is required so the agent can locate and rewrite dbt™ model files correctly during the optimisation pass.

pyproject.toml

[tool.poetry]
name = "snowflake-query-optimizer"
version = "0.1.0"
description = "dbt Model Query Cost Optimizer (Snowflake) — DinoAI Agent Trigger"
authors = ["Your Name <you@example.com>"]
package-mode = false

[tool.poetry.dependencies]
python = ">=3.11,<3.13"
requests = "^2.28.1"
paradime-io = "^5.3.0"

[build-system]
requires = ["poetry-core>=1.0.0"]
build-backend = "poetry.core.masonry.api"

Install dependencies by running:

poetry install

The only two third-party dependencies are requests (for downloading Bolt artifacts) and paradime-io (the Paradime SDK). All other imports — argparse, os, sys, textwrap, time — are part of the Python standard library.

Create the Agent YAML

Create the following file in your repository at .dinoai/agents/snowflake-query-optimizer.yml. This defines the agent's role, five-phase goal, Snowflake-specific anti-pattern knowledge, guardrails, and Slack output channel.

.dinoai/agents/snowflake-query-optimizer.yml

name: snowflake-query-optimizer
version: 1

role: >
  Snowflake Query Optimization Engineer with deep expertise in dbt™ model
  performance tuning, Snowflake-specific SQL anti-patterns, and optimization
  strategies including warehouse right-sizing, clustering keys, search
  optimization, result cache utilization, incremental strategies, and
  micro-partition pruning.

goal: >
  You will be given one or more Snowflake query IDs, each corresponding to
  a dbt™ model execution. Your mission is not to report problems — it is to
  fix them. Locate the dbt™ model source files, diagnose the inefficiencies,
  rewrite the SQL and/or YAML config directly in the repository, open a pull
  request with all changes, and post the PR link to Slack.

  ── PHASE 1: GATHER PERFORMANCE DATA ──────────────────────────────────────

  For each provided query ID, call get_snowflake_query_performance_stats to
  retrieve execution_time_ms, credits_used, bytes_scanned, bytes_spilled_to_disk,
  partitions_scanned, partitions_total, percentage_scanned_from_cache,
  warehouse_name, warehouse_size, and query_text. Rank models by
  execution_time_ms descending. Skip any query where execution_time_ms < 30000
  (30 seconds).

  ── PHASE 2: LOCATE THE dbt™ MODELS ──────────────────────────────────────

  For each model above the threshold, find the .sql file that produces the
  destination table. Then find the corresponding YAML documentation file
  (e.g. _<model_name>.yml or schema.yml) so config blocks and descriptions
  can be updated together.

  ── PHASE 3: DIAGNOSE AND PLAN ───────────────────────────────────────────

  Read the full SQL of each model. Identify which anti-patterns apply,
  citing exact line numbers. For each model, write a concise plan with:
    - Current cost metrics (execution_time_ms, credits_used, partitions_scanned
      vs partitions_total, bytes_spilled_to_disk, warehouse_size)
    - List of issues found (label them by anti-pattern name)
    - Proposed changes (exact config block additions, SQL rewrites)
    - Expected impact (estimated credit reduction or execution time saved)

  Key signals to look for:
    - partitions_scanned close to partitions_total → missing cluster key or
      no partition pruning
    - bytes_spilled_to_disk > 0 → warehouse undersized for this query or
      query needs refactoring to reduce data movement
    - percentage_scanned_from_cache = 0 and query is deterministic →
      result cache bypass due to non-deterministic functions
    - warehouse_size is X-Large or above for a model with low row counts →
      WAREHOUSE_OVERSIZED
    - materialized='view' with high execution_time_ms and multiple refs →
      WRONG_MATERIALIZATION

  ── PHASE 4: IMPLEMENT THE CHANGES ───────────────────────────────────────

  Create a new branch via run_terminal_command:
    git checkout -b perf/snowflake-optimizer-<YYYYMMDD>-<5-random-chars>

  Apply every planned change directly to the repository files via
  read_file + write_file. For each model, update both the config block
  and the SQL body as needed. After modifying any .sql file, update the
  corresponding YAML documentation if columns, config, or descriptions
  have changed. Stage and commit all changes.

  ── PHASE 5: OPEN A PR AND NOTIFY SLACK ──────────────────────────────────

  Create the pull request via `gh pr create` with title:
    perf: optimise Snowflake credit usage — <comma-separated model names>

  After the PR is created, post to #finops via post_slack_message:
    🚀 *Snowflake Query Optimizer — Optimisation PR Opened*
    *Models optimised:* `<model_1>`, `<model_2>`, ...
    *Est. total credit reduction:* ~<X> credits per run
    *Changes made:* one-line summary per model
    *Pull Request:* <PR URL>

  Do not stop after Phase 4. The run is only complete when the PR is
  open and the Slack message has been posted.

  ── GUARDRAILS ────────────────────────────────────────────────────────────

  - NEVER drop columns referenced by downstream models. Before removing any
    column, check all downstream .sql files for usage via ripgrep_search.
  - NEVER change a model's grain unless explicitly fixing a fan-out join —
    and always call this out in the PR body.
  - NEVER push directly to main. Always use a feature branch and PR.
  - NEVER change the warehouse_size in a dbt™ config without confirming the
    model's typical row count justifies it — check via run_sql_query.
  - If a model cannot be safely converted to incremental without more
    business context, change only the config block and leave a TODO
    comment in the SQL explaining what the team needs to decide.
  - If execution_time_ms < 30000 for a given query, skip it and say so.
  - Be surgical. Do not reformat the entire file — only touch what needs
    to change, preserving existing style (uppercase keywords, snake_case
    aliases, CTE structure, column grouping comments).

backstory: >
  You are a senior Snowflake performance engineer embedded in the analytics
  team. You have a bias for action — you do not write reports, you write
  code. When you identify an inefficiency you fix it in the file, commit
  it to a branch, open a PR, and notify the team on Slack — all without
  human intervention.

  You think in credits and partition pruning ratios, not abstract
  performance scores. You always quantify the expected impact of each
  change. You never make a change that could silently break a downstream
  model — you check first.

  You know the following Snowflake anti-patterns and their fixes:

    WAREHOUSE_OVERSIZED — the model runs on a warehouse larger than
    necessary for its data volume. Fix: add a snowflake_warehouse config
    targeting a smaller warehouse, or use query_tag to route to an
    appropriately sized warehouse.

    NO_CLUSTER_KEY — a large table is filtered on high-cardinality columns
    but no clustering key is defined, causing full micro-partition scans
    (partitions_scanned ≈ partitions_total). Fix: add cluster_by to the
    dbt™ config block using the most selective filter column(s).

    WRONG_MATERIALIZATION — a model is a view but is queried frequently by
    multiple downstream models, causing repeated full recomputation. Fix:
    change to materialized='table' or materialized='incremental'.

    INCREMENTAL_OPPORTUNITY — a model is a full table rebuild every run but
    has a natural time-based incremental key (e.g. created_at, event_date).
    Fix: convert to incremental with an is_incremental() filter.

    RESULT_CACHE_BYPASS — the model does not benefit from Snowflake's result
    cache because it uses non-deterministic functions (CURRENT_TIMESTAMP(),
    CURRENT_DATE(), RANDOM(), UUID_STRING()) unnecessarily.
    Fix: replace with deterministic equivalents or move non-deterministic
    expressions outside the cached layer.

    SELECT_STAR — the model selects all columns from a source, scanning
    micro-partitions that contain unneeded data. Fix: enumerate only the
    columns actually used downstream (check ref() usages via ripgrep_search).

    CARTESIAN_FAN_OUT_JOIN — a JOIN on a non-unique key causing row
    multiplication and excess credit use. Fix: add QUALIFY ROW_NUMBER()
    OVER (PARTITION BY <join_key> ORDER BY <tie_break>) = 1 before the join.

    REPEATED_CTE — a CTE is referenced more than twice in the same query;
    Snowflake re-evaluates it each time. Fix: materialise the CTE as a
    separate intermediate dbt™ model.

    UNFILTERED_LARGE_TABLE — a source() or ref() model pulls a very large
    table with no predicate, scanning all micro-partitions. Fix: push a
    WHERE clause into the CTE that reads that table.

    SEARCH_OPTIMISATION_CANDIDATE — the model supports frequent point
    lookups or selective equality/LIKE filters but search_optimization is
    not enabled. Fix: add a post-hook to the config block:
    post-hook: "ALTER TABLE {{ this }} ADD SEARCH OPTIMIZATION"

tools:
  mode: allowlist
  list:
    - get_snowflake_query_performance_stats
    - run_sql_query
    - read_file
    - write_file
    - search_files_and_directories
    - ripgrep_search
    - run_terminal_command
    - post_slack_message

slack:
  channel: "#finops"

tools.mode: allowlist restricts the agent to only the tools listed. This is important for a write-capable agent — it prevents accidental access to tools outside the intended optimisation workflow.

The Slack channel is set to #finops by default. Update slack.channel before committing if your team routes cost alerts to a different channel.

Create the Trigger Script

Create scripts/run_snowflake_query_optimizer.py. This script reads dbt™ model results from Bolt artifacts (or accepts Snowflake query IDs directly via CLI), extracts adapter_response.query_id from each result, filters to the top costly models by execution time, builds a structured opening message with all pre-fetched metadata, triggers the snowflake-query-optimizer agent, and polls until completion.

In Snowflake's run_results.json, the query ID is found at adapter_response.query_id — for example "01c4683c-0003-4ff0-0002-731a00bda6ea". This is different from BigQuery where the job ID is at adapter_response.job_id. The script reads this field specifically when parsing Bolt artifacts.

scripts/run:snowflake:query:optimizer.py

"""
run_snowflake_query_optimizer.py
---------------------------------
Trigger script for the `snowflake-query-optimizer` DinoAI agent.

Reads dbt™ model results from the latest Bolt run's run_results.json,
extracts adapter_response.query_id per model, filters to the top-N
slowest models above an execution time threshold, and triggers a single
DinoAI agent session to diagnose and fix the anti-patterns.

Environment variables (required)
=================================
  PARADIME_API_ENDPOINT   – Paradime API endpoint
  PARADIME_API_KEY        – Paradime API key
  PARADIME_API_SECRET     – Paradime API secret

  At least ONE of the following must also be set (or passed via CLI):
  BOLT_SCHEDULE_NAME          – Bolt schedule to pull query IDs from automatically
  QUERY_OPTIMIZER_QUERY_IDS   – Comma-separated Snowflake query IDs (skips Bolt fetch)

Environment variables (optional)
=================================
  QUERY_OPTIMIZER_TOP_N           – Max models to optimise in one run (default: 5)
  QUERY_OPTIMIZER_MIN_EXEC_SECS   – Skip models below this execution time in seconds
                                    (default: 30)
  QUERY_OPTIMIZER_DRILL_MODEL     – dbt™ model name to drill into after the main pass
  QUERY_OPTIMIZER_TIMEOUT         – Max seconds to wait for the agent (default: 3600)
  QUERY_OPTIMIZER_POLL_INTERVAL   – Polling cadence in seconds (default: 20)
"""

import argparse
import os
import sys
import textwrap
import time

import requests

from paradime import Paradime
from paradime.apis.dinoai_agents.exception import DinoaiAgentRunFailedException
from paradime.apis.dinoai_agents.types import DinoaiAgentRunStatus

AGENT_NAME    = "snowflake-query-optimizer"
ARTIFACT_PATH = "target/run_results.json"

TOP_N         = int(os.environ.get("QUERY_OPTIMIZER_TOP_N", 5))
MIN_EXEC_SECS = float(os.environ.get("QUERY_OPTIMIZER_MIN_EXEC_SECS", 30))
DRILL_MODEL   = os.environ.get("QUERY_OPTIMIZER_DRILL_MODEL", "").strip()
TIMEOUT       = int(os.environ.get("QUERY_OPTIMIZER_TIMEOUT", 3600))
POLL_INTERVAL = int(os.environ.get("QUERY_OPTIMIZER_POLL_INTERVAL", 20))


def _parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Trigger the snowflake-query-optimizer DinoAI agent."
    )
    source = parser.add_mutually_exclusive_group()
    source.add_argument(
        "--query-ids", nargs="+", metavar="QUERY_ID",
        default=_split_env("QUERY_OPTIMIZER_QUERY_IDS"),
        help="One or more Snowflake query IDs. (env: QUERY_OPTIMIZER_QUERY_IDS)",
    )
    source.add_argument(
        "--schedule-name",
        default=os.environ.get("BOLT_SCHEDULE_NAME", "").strip(),
        help="Bolt schedule name — query IDs fetched from latest run artifact. (env: BOLT_SCHEDULE_NAME)",
    )
    parser.add_argument("--top-n", type=int, default=TOP_N)
    parser.add_argument("--min-exec-secs", type=float, default=MIN_EXEC_SECS)
    parser.add_argument("--drill-model", default=DRILL_MODEL)
    return parser.parse_args()


def _split_env(var: str) -> list[str]:
    raw = os.environ.get(var, "").strip()
    return [v.strip() for v in raw.split(",") if v.strip()] if raw else []


def _build_client() -> Paradime:
    required = {
        "PARADIME_API_ENDPOINT": os.environ.get("PARADIME_API_ENDPOINT"),
        "PARADIME_API_KEY":      os.environ.get("PARADIME_API_KEY"),
        "PARADIME_API_SECRET":   os.environ.get("PARADIME_API_SECRET"),
    }
    missing = [k for k, v in required.items() if not v]
    if missing:
        print(f"ERROR: Missing env vars: {', '.join(missing)}", file=sys.stderr)
        raise SystemExit(1)
    return Paradime(
        api_endpoint=required["PARADIME_API_ENDPOINT"],
        api_key=required["PARADIME_API_KEY"],
        api_secret=required["PARADIME_API_SECRET"],
    )


def _fetch_run_results_from_bolt(paradime: Paradime, schedule_name: str) -> list[dict]:
    """
    Fetch run_results.json from the latest Bolt run and extract one record
    per dbt™ model. For Snowflake, the query ID is at adapter_response.query_id.
    """
    print(f"[snowflake-optimizer] 🔍  Fetching latest Bolt run for '{schedule_name}' …")
    try:
        latest_runs = paradime.bolt.list_runs(schedule_name=schedule_name, limit=1)
        latest_run  = latest_runs.runs[0] if latest_runs.runs else None
    except Exception as exc:
        print(f"[snowflake-optimizer] ⚠️   Could not fetch run list: {exc}", file=sys.stderr)
        return []

    if not latest_run:
        print("[snowflake-optimizer] ⚠️   No runs found for this schedule.", file=sys.stderr)
        return []

    print(f"[snowflake-optimizer] 🏃  Run ID: {latest_run.id} | status: {latest_run.state}")

    try:
        run_status      = paradime.bolt.get_run_status(latest_run.id)
        commands        = getattr(run_status, "commands", None) or []
        command_indices = list(range(len(commands))) if commands else list(range(5))
    except Exception:
        command_indices = list(range(5))

    all_model_results: list[dict] = []
    seen_urls: set[str] = set()

    for command_index in command_indices:
        try:
            url = paradime.bolt.get_latest_artifact_url(
                schedule_name=schedule_name,
                artifact_path=ARTIFACT_PATH,
                command_index=command_index,
            )
        except Exception:
            continue

        if not url or url in seen_urls:
            continue
        seen_urls.add(url)

        try:
            resp = requests.get(url, timeout=30)
            resp.raise_for_status()
            data = resp.json()
        except Exception as exc:
            print(f"[snowflake-optimizer] ⚠️   Failed to download artifact: {exc}", file=sys.stderr)
            continue

        for result in data.get("results", []):
            unique_id = result.get("unique_id", "")
            if not unique_id.startswith("model."):
                continue

            adapter  = result.get("adapter_response") or {}
            query_id = adapter.get("query_id", "—")   # Snowflake: adapter_response.query_id

            all_model_results.append({
                "name":             unique_id.split(".")[-1],
                "unique_id":        unique_id,
                "status":           result.get("status", "—"),
                "execution_time_s": round(result.get("execution_time") or 0, 2),
                "query_id":         query_id,
                "rows_affected":    adapter.get("rows_affected", "—"),
                "command_index":    command_index,
                "generated_at":     data.get("metadata", {}).get("generated_at", "unknown"),
            })

    print(f"[snowflake-optimizer] 📊  Total model results collected: {len(all_model_results)}")
    return all_model_results


def _top_slow_models(
    all_results: list[dict],
    top_n: int,
    min_exec_secs: float,
) -> list[dict]:
    """
    Deduplicate by model name (keep slowest instance), apply execution time
    threshold, sort by execution_time_s descending, return top N.
    Only includes models where a Snowflake query_id was captured.
    """
    deduped: dict[str, dict] = {}
    for row in all_results:
        name = row["name"]
        if name not in deduped or row["execution_time_s"] > deduped[name]["execution_time_s"]:
            deduped[name] = row

    filtered = [
        r for r in deduped.values()
        if r["execution_time_s"] >= min_exec_secs and r["query_id"] != "—"
    ]
    return sorted(filtered, key=lambda r: r["execution_time_s"], reverse=True)[:top_n]


def _rows_from_query_ids(query_ids: list[str]) -> list[dict]:
    """Build minimal model result dicts from raw Snowflake query IDs."""
    return [
        {
            "name":             f"query_{i + 1}",
            "unique_id":        "—",
            "status":           "unknown",
            "execution_time_s": 0,
            "query_id":         qid,
            "rows_affected":    "—",
            "command_index":    0,
            "generated_at":     "—",
        }
        for i, qid in enumerate(query_ids)
    ]


def _format_table(rows: list[dict]) -> str:
    if not rows:
        return "(no model results to display)"
    headers = {
        "name":             "Model",
        "execution_time_s": "Time (s)",
        "rows_affected":    "Rows Affected",
        "query_id":         "Snowflake Query ID",
        "command_index":    "Cmd",
    }
    widths = {col: len(label) for col, label in headers.items()}
    for row in rows:
        for col in headers:
            widths[col] = max(widths[col], len(str(row.get(col, ""))))

    def _row_str(values):
        return "  ".join(str(values.get(col, "")).ljust(widths[col]) for col in headers)

    header_line = _row_str(headers)
    return "\n".join([header_line, "-" * len(header_line)] + [_row_str(r) for r in rows])


def _build_opening_message(rows: list[dict], source_label: str) -> str:
    query_id_lines = "\n".join(
        f"  {i + 1}. model=`{r['name']}`  "
        f"query_id=`{r['query_id']}`  "
        f"execution_time={r['execution_time_s']}s  "
        f"rows_affected={r['rows_affected']}"
        for i, r in enumerate(rows)
    )
    return textwrap.dedent(f"""
        You are being triggered to analyse and fix costly dbt™ models running on Snowflake.

        SOURCE: {source_label}

        ── PRE-FETCHED MODEL PERFORMANCE DATA ────────────────────────────────────

        {_format_table(rows)}

        ── QUERY IDs TO OPTIMISE ─────────────────────────────────────────────────

        {query_id_lines}

        Work through all five phases (GATHER → LOCATE → DIAGNOSE → IMPLEMENT →
        PR + SLACK) in order without stopping. The run is only complete when the
        PR is open and the Slack message has been posted to #finops.
    """).strip()


def _build_drill_message(drill_model: str, rows: list[dict]) -> str:
    drill_row = next((r for r in rows if r["name"] == drill_model), None)
    query_hint = (
        f"Its Snowflake query ID is `{drill_row['query_id']}` "
        f"({drill_row['execution_time_s']}s execution time). "
        f"Use get_snowflake_query_performance_stats to retrieve the full "
        f"query profile including partition pruning ratios and spill details. "
        if drill_row and drill_row["query_id"] != "—"
        else f"No pre-fetched query ID for `{drill_model}` — analyse the .sql file directly. "
    )
    return (
        f"Now drill deeper into `{drill_model}` specifically. {query_hint}"
        f"Show the exact partition pruning ratio, spill-to-disk bytes, and "
        f"warehouse size for this query. Identify whether the bottleneck is a "
        f"missing cluster key, warehouse sizing, or a data movement issue, and "
        f"apply any additional targeted fixes not already covered in the main pass. "
        f"Commit changes to the existing branch, update the PR description, and "
        f"post a follow-up to #finops with the updated PR link and a summary of "
        f"the extra changes."
    )


def _poll_until_done(paradime: Paradime, session_id: str, label: str) -> object:
    start = time.time()
    while True:
        run = paradime.dinoai_agents.get_run(agent_session_id=session_id)
        if run.status == DinoaiAgentRunStatus.COMPLETED:
            print(f"[snowflake-optimizer] ✅  {label} completed")
            return run
        if run.status == DinoaiAgentRunStatus.FAILED:
            last_msg = run.messages[-1].content if run.messages else "no messages"
            raise DinoaiAgentRunFailedException(f"{label} FAILED. Last: {last_msg}")
        elapsed = time.time() - start
        if elapsed > TIMEOUT:
            raise TimeoutError(f"Timed out after {TIMEOUT}s for session {session_id}")
        print(f"[snowflake-optimizer]    {label} — {run.status.value} ({int(elapsed)}s) … {POLL_INTERVAL}s")
        time.sleep(POLL_INTERVAL)


def main() -> None:
    args     = _parse_args()
    paradime = _build_client()

    if args.query_ids:
        rows         = _rows_from_query_ids(args.query_ids)
        source_label = f"explicit query IDs: {', '.join(args.query_ids)}"
    elif args.schedule_name:
        all_results  = _fetch_run_results_from_bolt(paradime, args.schedule_name)
        rows         = _top_slow_models(all_results, args.top_n, args.min_exec_secs)
        source_label = (
            f"Bolt schedule '{args.schedule_name}' — top {len(rows)} models "
            f"above {args.min_exec_secs}s execution time"
        )
        if not rows:
            print("[snowflake-optimizer] ⚠️   No models above threshold or missing query IDs. Exiting.")
            raise SystemExit(0)
    else:
        print("ERROR: Provide --query-ids or --schedule-name.", file=sys.stderr)
        raise SystemExit(1)

    print("\n" + "=" * 72)
    print(_format_table(rows))
    print("=" * 72 + "\n")

    trigger    = paradime.dinoai_agents.trigger_run(
        agent=AGENT_NAME,
        message=_build_opening_message(rows, source_label),
    )
    session_id = trigger.agent_session_id
    print(f"[snowflake-optimizer] 📋  Session ID: {session_id}")

    main_run = _poll_until_done(paradime, session_id, "Main optimisation pass")
    print(main_run.messages[-1].content if main_run.messages else "(no report)")

    if args.drill_model:
        paradime.dinoai_agents.send_message(
            agent_session_id=session_id,
            message=_build_drill_message(args.drill_model, rows),
        )
        drill_run = _poll_until_done(paradime, session_id, f"Drill-down on '{args.drill_model}'")
        print(drill_run.messages[-1].content if drill_run.messages else "(no drill report)")

    print("[snowflake-optimizer] ✅  All done.")


if __name__ == "__main__":
    main()

The script accepts --schedule-name or --query-ids but not both — they are mutually exclusive. When --schedule-name is used, the script reads run_results.json from the latest Bolt run and extracts adapter_response.query_id from each dbt™ model result. Models where query_id is absent (e.g. tests or snapshots) are automatically excluded.

When no models are found above the execution time threshold, or none have a captured query_id, the script exits cleanly with code 0 — safe to run on a schedule without generating noise.

Set Your Environment Variables

Store your secrets in Paradime before scheduling. Go to Workspace Settings → Environment Variables and add the following:

Variable

Description

PARADIME_API_ENDPOINT

Your Paradime API endpoint

PARADIME_API_KEY

Your Paradime API key

PARADIME_API_SECRET

Your Paradime API secret

BOLT_SCHEDULE_NAME

Name of the Bolt schedule whose run_results.json maps dbt™ models to their Snowflake query IDs

QUERY_OPTIMIZER_TOP_N

(optional) Max models per run — defaults to 5

QUERY_OPTIMIZER_MIN_EXEC_SECS

(optional) Execution time threshold in seconds — defaults to 30

QUERY_OPTIMIZER_DRILL_MODEL

(optional) Model name to drill into after the main pass

QUERY_OPTIMIZER_TIMEOUT

(optional) Max seconds to wait per session — defaults to 3600

QUERY_OPTIMIZER_QUERY_IDS

(optional) Comma-separated Snowflake query IDs — bypasses Bolt artifact fetch when you already know which dbt™ models are slow

Your Paradime API endpoint, key, and secret are available under Workspace Settings → API. Make sure the key has DinoAI agent API capabilities enabled.

Run the Optimizer Manually

Option A — Automatically detect costly dbt™ models from a Bolt schedule:

poetry run python scripts/run_snowflake_query_optimizer.py \
  --schedule-name "my_dbt_production_schedule" \
  --top-n 5 \
  --min-exec-secs 30

Option B — Supply Snowflake query IDs directly (when you already know which models are slow):

poetry run python scripts/run_snowflake_query_optimizer.py \
  --query-ids 01c4683c-0003-4ff0-0002-731a00bda6ea 01c4683c-0003-4ff0-0002-731a00bda6eb

Option C — Include a follow-up drill-down on a specific dbt™ model:

poetry run python scripts/run_snowflake_query_optimizer.py \
  --schedule-name "my_dbt_production_schedule" \
  --drill-model fct_orders

You should see output like:

========================================================================
Model              Time (s)  Rows Affected  Snowflake Query ID                       Cmd
------------------------------------------------------------------------
fct_orders         142.30    84921          01c4683c-0003-4ff0-0002-731a00bda6ea     0
dim_sessions       98.50     12043          01c4683c-0003-4ff0-0002-731a00bda6eb     0
========================================================================

[snowflake-optimizer] 📋  Session ID: agt_sess_abc123xyz
[snowflake-optimizer]    Main optimisation pass — running (0s) … 20s
[snowflake-optimizer]    Main optimisation pass — running (20s) … 20s
[snowflake-optimizer] ✅  Main optimisation pass completed

The agent then posts its full optimisation plan and PR link to stdout, and a credit reduction summary to #finops on Slack.

Schedule with Bolt

Run the optimizer automatically on a recurring cadence so costly dbt™ models are caught and fixed without any manual intervention. The schedule reads the latest Bolt run's run_results.json, identifies the top offending models by execution time, and triggers the agent to fix and PR them.

Add Environment Variables to Paradime

Go to Workspace Settings → Environment Variables and confirm the variables from the previous step are saved. Bolt schedules read from the same environment variable store, so no additional setup is needed.

Create the Bolt Schedule

Go to Bolt → Schedules and click New Schedule. Name it something like Snowflake Query Cost Optimizer and add the following two commands in order:

poetry install

poetry run python scripts/run_snowflake_query_optimizer.py --schedule-name "your_dbt_production_schedule" --top-n 5 --min-exec-secs 30

Replace your_dbt_production_schedule with the exact name of the Bolt schedule that runs your dbt™ models in production. The script reads that schedule's run_results.json artifact, which contains a record per dbt™ model including the Snowflake query ID captured at adapter_response.query_id.

poetry install runs first on every execution so that any dependency updates committed to pyproject.toml are picked up automatically — no manual intervention needed after a dependency bump.

Choose a Schedule Frequency

The right cadence depends on how frequently your production dbt™ models run and how quickly Snowflake credit costs accumulate. The table below covers the most common options:

Cadence

Cron expression

When it runs

Every day at 8 AM

0 8 * * *

Monday–Sunday, 8:00 AM

Weekdays at 8 AM

0 8 * * 1-5

Monday–Friday, 8:00 AM

Once a week (Monday 8 AM)

0 8 * * 1

Every Monday, 8:00 AM

Twice a week (Mon & Thu)

0 8 * * 1,4

Monday and Thursday, 8:00 AM

Every 12 hours

0 */12 * * *

Midnight and noon daily

For most teams, weekdays at 8 AM (0 8 * * 1-5) is a sensible default — it runs after overnight production dbt™ models have completed, so the latest run_results.json artifacts and Snowflake query history are always available. The script exits cleanly with code 0 when no dbt™ models exceed the execution time threshold, so there is no cost or noise on quiet days.

File Structure

Your repository should look like this after completing the setup:

your-repo/
├── dbt_project.yml
├── pyproject.toml                              ← same level as dbt_project.yml
├── .dinoai/
│   └── agents/
│       └── snowflake-query-optimizer.yml
└── scripts/
    └── run_snowflake_query_optimizer.py        ← trigger script

pyproject.toml must sit at the same directory level as dbt_project.yml. The agent uses run_terminal_command, read_file, and write_file to navigate and rewrite your dbt™ project files, so the working directory at session start must be the repo root.

PreviousBolt Pipeline Healer Nextdbt™ Model Query Cost Optimizer — BigQuery

Last updated 28 days ago

Was this helpful?