Static Query Analysis with Sqlglot

Minimal example: parsing and inspecting a query

from sqlglot import parse_one
from sqlglot import exp

sql = """
SELECT u.id, COUNT(*)
FROM mycatalog.analytics.users u
JOIN mycatalog.analytics.orders o ON o.user_id = u.id
WHERE u.country LIKE 'ES%'
GROUP BY 1
ORDER BY 2 DESC
"""

# Parse once (Athena dialect in this project)
tree = parse_one(sql, read="athena")

# List fully-qualified tables
tables = []
for t in tree.find_all(exp.Table):
    if fq := f"{t.db}.{t.name}":
        tables.append(fq)

print(f"Tables used in the query: {tables}")

What this shows

• parse_one(..., read="athena") builds an AST you can traverse.
• find_all(exp.Table) retrieves table nodes; you can assemble db.table easily.

Constructor

Parses once per query using the chosen dialect (Athena here). Everything else reads from the AST (self.tree).

class SqlAnalyzer:
    def __init__(self, query, engine="athena"):
        self.engine = engine
        self.tree = parse_one(query.strip(), read=self.engine)

Table lineage

What you learn: which tables the query reads and which tables it writes to (INSERT/CTAS). This lets you attribute cost and time to specific tables and build lineage.

    def get_cte_names(self):
        names = set()
        for w in self.tree.find_all(exp.With):
            for cte in w.find_all(exp.CTE):
                name = None
                if cte.alias:
                    name = cte.alias
                elif getattr(cte, "this", None) and getattr(cte.this, "alias", None):
                    name = cte.this.alias
                if isinstance(name, str):
                    names.add(name.lower())
        return names

    def get_tables(self, include_cte=False, include_insert_target=False, include_create_target=False):
        cte_names = self.get_cte_names()
        out = set()
        for table in self.tree.find_all(exp.Table):
            if not include_insert_target and self._is_insert_target(table):
                continue
            if not include_create_target and self._is_create_target(table):
                continue
            if not include_cte and table.name and table.name.lower() in cte_names:
                continue
            fq = self._fq_from_tablelike(table)
            if fq:
                out.add(fq)
        return sorted(out)

    def get_insert_target(self):
        for ins in self.tree.find_all(exp.Insert):
            return self._fq_from_tablelike(ins.this)
        return None

    def get_create_target(self):
        for cr in self.tree.find_all(exp.Create):
            return self._fq_from_tablelike(cr.this)
        return None

    def _is_create_target(self, table):
        node = table
        while node is not None:
            parent = node.parent
            if isinstance(parent, exp.Create):
                target = parent.this
                if node is target:
                    return True
                if isinstance(target, exp.Schema):
                    for x in target.find_all(exp.Table):
                        if node is x:
                            return True
            node = parent
        return False

    def _is_insert_target(self, table):
        node = table
        while node is not None:
            parent = node.parent
            if isinstance(parent, exp.Insert):
                target = parent.this
                if node is target:
                    return True
                if isinstance(target, exp.Schema):
                    for x in target.find_all(exp.Table):
                        if node is x:
                            return True
            node = parent
        return False

    def _fq_from_tablelike(self, node):
        if isinstance(node, exp.Table):
            fq = f"{node.db}.{node.name}".replace('"', "")
            return fq.lower() if fq else None
        if isinstance(node, exp.Schema):
            for t in node.find_all(exp.Table):
                return self._fq_from_tablelike(t)
        return None

Join & union metrics

What you learn: amount and type of data combination and deduplication. More joins → larger intermediate data; UNION DISTINCT forces dedupe.

    def get_joins(self, by_kind=True):
        total = 0
        kinds = {}
        for j in self.tree.find_all(exp.Join):
            total += 1
            kind = (j.args.get("kind") or "INNER").upper()
            kinds[kind] = kinds.get(kind, 0) + 1
        return total, (kinds if by_kind else {})

    def get_unions(self):
        total = 0
        union_all = 0
        for u in self.tree.find_all(exp.Union):
            total += 1
            is_distinct = u.args.get("distinct", True)
            if not is_distinct:
                union_all += 1
        return total, union_all, total - union_all

Structural complexity

What you learn: how deep/complex the query is (CTEs, subqueries, windows, nesting). Higher values often correlate with slower planning and heavier shuffles.

    def get_count_ctes_defined(self):
        return len(self.get_cte_names())

    def get_count_subqueries(self):
        return sum(1 for _ in self.tree.find_all(exp.Subquery))

    def get_count_exists_subqueries(self):
        return sum(1 for _ in self.tree.find_all(exp.Exists))

    def get_count_window_functions(self):
        return sum(1 for _ in self.tree.find_all(exp.Window))

    def get_count_max_ast_depth(self):
        maxd = 0
        stack = [(self.tree, 1)]
        seen = set()
        while stack:
            node, depth = stack.pop()
            nid = id(node)
            if nid in seen:
                continue
            seen.add(nid)
            if depth > maxd:
                maxd = depth
            for ch in node.iter_expressions():
                if ch is not None:
                    stack.append((ch, depth + 1))
        return maxd

Predicate & projection metrics

What you learn: how selective and wide the query is. Large IN lists, long OR chains, function-wrapped columns, and leading-wildcard LIKE patterns often kill pruning and inflate scans.

    def get_columns_referenced_counts(self):
        total = 0
        distinct = set()
        for c in self.tree.find_all(exp.Column):
            total += 1
            t = getattr(c, "table", None)
            n = getattr(c, "name", None)
            if t and n:
                distinct.add(f"{t}.{n}".lower())
            elif n:
                distinct.add(str(n).lower())
            else:
                distinct.add(str(c).lower())
        return total, len(distinct)

    def get_count_projected_columns(self):
        cnt = 0
        for s in self.tree.find_all(exp.Select):
            exprs = s.expressions or []
            cnt += len(exprs)
        return cnt

    def get_count_or_chain_max(self):
        mx = 0
        for p in self._predicates():
            for node in p.find_all(exp.Or):
                mx = max(mx, self._flatten_or_terms(node))
        return mx

    def get_count_in_list_max(self):
        mx = 0
        for n in self.tree.find_all(exp.In):
            right = n.args.get("expressions") or n.args.get("query")
            if isinstance(right, exp.Tuple):
                mx = max(mx, len(right.expressions or []))
            elif isinstance(right, exp.Array):
                mx = max(mx, len(right.expressions or []))
        return mx

    def get_count_like_leading_wildcards(self):
        k = 0
        for n in self.tree.find_all(exp.Like):
            pat = n.args.get("expression") or n.args.get("right")
            if isinstance(pat, exp.Literal) and pat.is_string:
                s = (pat.this or "").strip().strip("'").strip('"')
                if s.startswith("%"):
                    k += 1
        return k

    def get_count_funcs_in_predicates(self):
        k = 0
        for p in self._predicates():
            for f in p.find_all(exp.Func):
                has_col = any(True for _ in f.find_all(exp.Column))
                if has_col:
                    k += 1
        return k

    def _predicates(self):
        preds = []
        for where in self.tree.find_all(exp.Where):
            if where.this is not None:
                preds.append(where.this)
        for j in self.tree.find_all(exp.Join):
            on = j.args.get("on")
            if on is not None:
                preds.append(on)
        return preds

    def _flatten_or_terms(self, node):
        if not isinstance(node, exp.Or):
            return 1
        return self._flatten_or_terms(node.left) + self._flatten_or_terms(node.right)

Best-practice flags

What you learn: quick red flags that typically inflate cost or latency.

• SELECT * reads every column (expensive on columnar stores).
• SELECT DISTINCT and ORDER BY without LIMIT trigger costly global operations.

    def get_has_select_star(self):
        return any(True for _ in self.tree.find_all(exp.Star))

    def get_has_select_distinct(self):
        for s in self.tree.find_all(exp.Select):
            if s.args.get("distinct"):
                return True
        return False

    def get_has_order_by_without_limit(self):
        for s in self.tree.find_all(exp.Select):
            has_order = s.args.get("order") is not None
            has_limit = s.args.get("limit") is not None
            if has_order and not has_limit:
                return True
        return False

Command classification

What you learn: whether a query performs reads only, or also DDL/DML (useful for governance and change tracking). exp.Command captures things like UNLOAD; explicit classes cover INSERT/CREATE/ALTER/DROP.

COMMANDS = {
    "INSERT": exp.Insert,
    "CREATE": exp.Create,
    "ALTER": exp.Alter,
    "DROP": exp.Drop,
}

class SqlAnalyzer:
    # ...
    def get_commands(self):
        kinds = {}
        for cmd in self.tree.find_all(exp.Command):
            kind = str(getattr(cmd, "this", "")).upper()
            if kind:
                kinds[kind] = kinds.get(kind, 0) + 1
        for name, expression in COMMANDS.items():
            total = sum(1 for _ in self.tree.find_all(expression))
            if total:
                kinds[name] = kinds.get(name, 0) + total
        total = sum(kinds.values())
        return total, kinds

Result schema (Pydantic) and analyze

We attach a query_id to every result so downstream systems can trace metrics back to the original query. Failures are represented with is_success=False.

from typing import Dict, List, Optional
from pydantic import BaseModel

class AnalysisResult(BaseModel):
    query_id: Optional[str] = None
    is_success: bool = True

class SuccessfulAnalysisResult(AnalysisResult):
    tables: List[str]
    joins_by_kind: Dict[str, int]
    commands_by_kind: Dict[str, int]
    insert_target: Optional[str] = None
    create_target: Optional[str] = None

    count_joins_total: int
    count_unions_total: int
    count_union_all: int
    count_union_distinct: int
    count_commands_total: int
    count_ctes_defined: int
    count_subqueries: int
    count_exists_subqueries: int
    count_cross_joins: int
    count_distinct_calls: int
    count_group_by_cols: int
    count_order_by_cols: int
    count_window_functions: int
    count_projected_columns: int
    count_referenced_columns_total: int
    count_referenced_columns_distinct: int
    count_or_chain_max: int
    count_in_list_max: int
    count_like_leading_wildcards: int
    count_max_ast_depth: int

    has_select_star: bool
    has_select_distinct: bool
    has_order_by_without_limit: bool

The SqlAnalyzer.analyze method runs all get_* metrics and produces this schema in one step:

class SqlAnalyzer:
    def analyze(self, query_id=None):
        joins_total, joins_by_kind = self.get_joins()
        unions_total, union_all, union_distinct = self.get_unions()
        commands_total, commands_by_kind = self.get_commands()
        rc_total, rc_distinct = self.get_columns_referenced_counts()

        return SuccessfulAnalysisResult(
            query_id=query_id,
            tables=self.get_tables(),
            joins_by_kind=joins_by_kind,
            commands_by_kind=commands_by_kind,
            insert_target=self.get_insert_target(),
            create_target=self.get_create_target(),
            count_joins_total=joins_total,
            count_unions_total=unions_total,
            count_union_all=union_all,
            count_union_distinct=union_distinct,
            count_commands_total=commands_total,
            count_ctes_defined=self.get_count_ctes_defined(),
            count_subqueries=self.get_count_subqueries(),
            count_exists_subqueries=self.get_count_exists_subqueries(),
            count_cross_joins=self.get_count_cross_joins(),
            count_distinct_calls=self.get_count_distinct_calls(),
            count_group_by_cols=self.get_count_group_by_cols(),
            count_order_by_cols=self.get_count_order_by_cols(),
            count_window_functions=self.get_count_window_functions(),
            count_projected_columns=self.get_count_projected_columns(),
            count_referenced_columns_total=rc_total,
            count_referenced_columns_distinct=rc_distinct,
            count_or_chain_max=self.get_count_or_chain_max(),
            count_in_list_max=self.get_count_in_list_max(),
            count_like_leading_wildcards=self.get_count_like_leading_wildcards(),
            count_max_ast_depth=self.get_count_max_ast_depth(),
            has_select_star=self.get_has_select_star(),
            has_select_distinct=self.get_has_select_distinct(),
            has_order_by_without_limit=self.get_has_order_by_without_limit(),
        )

Batch extraction task (Prefect/ECS)

This task queries recent Athena statements, runs the analyzer for each, and writes the results to an Iceberg-backed Bronze table.

import pandas as pd
from prefect import task
from prefect import get_run_logger
from sqlglot.errors import ParseError, TokenError

from xxx.metadata.query_parser import AnalysisResult, SqlAnalyzer

def get_queries_to_process():
    # Placeholder: logic to retrieve the queries is out of scope
    # Typically this reads from a table or API with query_id + raw_query
    return {"query_id_1": "SELECT * FROM my_table", "query_id_2": "..."}

@task
def extract_batch():
    logger = get_run_logger()
    queries = get_queries_to_process()

    failures = 0
    data = []
    for i, (query_id, query) in enumerate(queries.items(), start=1):
        try:
            result = SqlAnalyzer(query).analyze(query_id=query_id)
        except (ParseError, TokenError):
            logger.warning(f"Unable to process {query_id=} ({query=})")
            result = AnalysisResult(query_id=query_id, is_success=False)
            failures += 1

        data.append(result.model_dump())

    df_out = pd.DataFrame(data)
    # Placeholder: write results to Iceberg or S3
    # athena.write_iceberg(df_out, DATABASE, TABLE, as_strings=True)

Each batch produces a DataFrame of analysis results, where each row represents one query and all computed metrics from SqlAnalyzer.analyze().

Analyze trends

Examine global patterns to understand how queries are written and evolve over time:

• Distribution of join types (INNER, LEFT, CROSS) to identify common patterns or misuse.
• Frequency of SELECT * to quantify non-selective queries.
• Typical AST depth and subquery counts to estimate structural complexity.
• Top N tables appearing across queries to detect heavily queried datasets.

Establish SQL quality guidelines

Use these findings to propose internal standards:

• Enforce limits on nesting or cross joins.
• Encourage explicit column selection.
• Track usage of complex CTEs and window functions for maintainability.

By periodically running this pipeline, you can detect regressions (e.g., more SELECT * or deeper nesting) and drive continuous improvements in your data engineering practices.

Future extensions

• Cost integration: correlate metrics with runtime cost (e.g., bytes scanned) for deeper efficiency analysis.
• Visualization: load metrics into Superset or other BI tools for dashboards and trend tracking.
• Quality scoring: create a query-level quality index combining static metrics and execution performance.