Find the SQL that will not run on your next warehouse before you migrate

id: sqlfluff-dialect-portability-preflight
namespace: company.team

inputs:
  - id: source_dialect
    type: SELECT
    displayName: Current warehouse dialect
    description: The dialect the SQL is written for today. Every file must parse
      cleanly here, otherwise the corpus itself is broken.
    values:
      [
        ansi,
        bigquery,
        databricks,
        duckdb,
        mysql,
        postgres,
        redshift,
        snowflake,
        sparksql,
        sqlite,
        trino,
        tsql
      ]
    defaults: snowflake

  - id: target_dialect
    type: SELECT
    displayName: Migration target dialect
    description: The warehouse you plan to move to. The run fails while any file
      still cannot be parsed in this dialect.
    values:
      [
        ansi,
        bigquery,
        databricks,
        duckdb,
        mysql,
        postgres,
        redshift,
        snowflake,
        sparksql,
        sqlite,
        trino,
        tsql
      ]
    defaults: bigquery

  - id: compare_dialects
    type: ARRAY
    itemType: STRING
    displayName: Other dialects to compare
    description: Extra candidate targets for the portability matrix. They are
      reported, never gated.
    defaults: [ postgres, tsql, duckdb, databricks ]

  - id: notify_slack
    type: BOOL
    displayName: Notify Slack
    description: Post the verdict to Slack. Needs the SLACK_WEBHOOK_URL secret.
    defaults: false

variables:
  report_path: "migration/sql-portability-{{ inputs.source_dialect }}-to-{{
    inputs.target_dialect }}.md"

triggers:
  - id: weekly
    type: io.kestra.plugin.core.trigger.Schedule
    description: Track migration progress every Monday morning. The blocker count
      should only go down.
    cron: "0 7 * * 1"

tasks:
  - id: port_check
    type: io.kestra.plugin.scripts.python.Script
    description: >
      Parse every SQL file once per dialect with SQLFluff and keep only parse
      errors (code PRS), which mark syntax the dialect does not accept. Style
      rules are ignored. Prints the portability matrix and the blockers for the
      target as outputs.
    containerImage: python:3.12-slim
    beforeCommands:
      - pip install -q --root-user-action=ignore sqlfluff==4.2.1
    env:
      SOURCE: "{{ inputs.source_dialect }}"
      TARGET: "{{ inputs.target_dialect }}"
      COMPARE: "{{ inputs.compare_dialects | toJson }}"
    # Demo corpus written for Snowflake. Replace with a git.Clone in a WorkingDirectory to check your own repository.
    inputFiles:
      sql/01_top_customers.sql: |
        SELECT TOP 10 customer_id, SUM(amount) AS revenue
        FROM orders
        GROUP BY customer_id
        ORDER BY revenue DESC
      sql/02_latest_order.sql: |
        SELECT customer_id, order_id, created_at
        FROM orders
        QUALIFY ROW_NUMBER() OVER (PARTITION BY customer_id ORDER BY created_at DESC) = 1
      sql/03_daily_revenue.sql: |
        SELECT CAST(created_at AS DATE) AS order_day, COUNT(*) AS orders, SUM(amount) AS revenue
        FROM orders
        GROUP BY CAST(created_at AS DATE)
      sql/04_json_field.sql: |
        SELECT order_id, payload:customer.email::string AS email
        FROM raw_orders
      sql/05_email_domain.sql: |
        SELECT customer_id FROM customers WHERE email ILIKE '%@example.com'
    outputFiles:
      - report.md
    script: |
      import json, os, subprocess, sys

      source, target = os.environ["SOURCE"], os.environ["TARGET"]
      compare = [d for d in json.loads(os.environ["COMPARE"]) if d not in (source, target)]
      dialects = list(dict.fromkeys([source, target] + compare))

      def parse_errors(dialect):
          # LT01 is a cheap rule so the lint pass runs; only PRS violations are kept.
          # --nofail zeroes the violation exit code but not a usage error, so a bad dialect still fails here.
          run = subprocess.run(["sqlfluff", "lint", "sql", "--dialect", dialect, "--rules", "LT01",
                                "--format", "json", "--nofail"], capture_output=True, text=True)
          if run.returncode != 0:
              sys.exit(f"sqlfluff failed for dialect {dialect}: {run.stderr.strip()[-500:]}")
          errors = {}
          for f in json.loads(run.stdout):
              name = os.path.basename(f["filepath"])
              errors[name] = [f"line {v['start_line_no']}: {v['description'].split(': ', 1)[-1][:80]}"
                              for v in f["violations"] if v["code"] == "PRS"]
          return errors

      results = {d: parse_errors(d) for d in dialects}
      files = sorted(results[source])
      source_broken = [f for f in files if results[source][f]]
      blockers = [f for f in files if results[target][f] and f not in source_broken]

      header = "| file | " + " | ".join(dialects) + " |"
      lines = [header, "|" + "---|" * (len(dialects) + 1)]
      for f in files:
          lines.append(f"| `{f}` | " + " | ".join("FAIL" if results[d][f] else "ok" for d in dialects) + " |")
      ranking = sorted(((sum(1 for f in files if results[d][f]), d) for d in [target] + compare))
      lines.append("")
      lines.append("Files that do not parse, fewest first: " + ", ".join(f"{d} {n}" for n, d in ranking))
      lines.append("")
      lines.append(f"Blockers for {target}:")
      lines += [f"- `{f}`: {results[target][f][0]}" for f in blockers] or ["- none"]
      matrix = "\n".join(lines)
      print(matrix)

      verdict = "SOURCE_INVALID" if source_broken else ("BLOCKED" if blockers else "READY")
      with open("report.md", "w") as fh:
          fh.write(f"# SQL portability: {source} to {target}\n\nVerdict: {verdict}, {len(blockers)} of {len(files)} files blocked\n\n{matrix}\n")
      print("::" + json.dumps({"outputs": {
          "verdict": verdict,
          "files": len(files),
          "blocker_count": len(blockers),
          "blockers": ", ".join(blockers),
          "source_broken": ", ".join(source_broken),
          "matrix": matrix,
      }}) + "::")

  - id: publish_report
    type: io.kestra.plugin.core.namespace.UploadFiles
    description: Keep the latest matrix as a namespace file, so the migration team
      reads one page that updates every week.
    namespace: "{{ flow.namespace }}"
    filesMap:
      "{{ render(vars.report_path) }}": "{{ outputs.port_check.outputFiles['report.md'] }}"

  - id: gate
    type: io.kestra.plugin.core.flow.Switch
    value: "{{ outputs.port_check.vars.verdict }}"
    cases:
      READY:
        - id: log_ready
          type: io.kestra.plugin.core.log.Log
          message: "All {{ outputs.port_check.vars.files }} SQL files parse in {{
            inputs.target_dialect }}. The corpus is ready to move from {{
            inputs.source_dialect }}."
      SOURCE_INVALID:
        - id: source_invalid
          type: io.kestra.plugin.core.execution.Fail
          description: A file that does not parse in its own dialect would make every
            comparison meaningless.
          errorMessage: "These files do not parse in the source dialect {{
            inputs.source_dialect }}, fix them first: {{
            outputs.port_check.vars.source_broken }}"
    defaults:
      - id: notify_blocked
        type: io.kestra.plugin.core.flow.If
        condition: "{{ inputs.notify_slack }}"
        then:
          - id: slack_blocked
            type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
            url: "{{ secret('SLACK_WEBHOOK_URL') }}"
            payload: |
              {
                "text": "SQL migration {{ inputs.source_dialect }} to {{ inputs.target_dialect }}: {{ outputs.port_check.vars.blocker_count }} of {{ outputs.port_check.vars.files }} files still do not parse ({{ outputs.port_check.vars.blockers }}). Report: {{ render(vars.report_path) }}. Execution {{ execution.id }}."
              }
      - id: blocked
        type: io.kestra.plugin.core.execution.Fail
        errorMessage: "{{ outputs.port_check.vars.blocker_count }} of {{
          outputs.port_check.vars.files }} files do not parse in {{
          inputs.target_dialect }}: {{ outputs.port_check.vars.blockers }}"