pip install sqlglot[all]  # 安裝完整功能包（含所有方言支持）

import sqlglot
from sqlglot.dialects import MySQL, Postgres
# 設(shè)置全局默認(rèn)方言
sqlglot.dialect = "mysql"
# 或針對特定會話設(shè)置
with sqlglot.dialect_context("postgres"):
    # 此代碼塊內(nèi)使用PostgreSQL方言
    pass

from sqlglot import parse_one
sql = """
WITH daily_metrics AS (
    SELECT 
        DATE_TRUNC('day', event_time) AS day,
        product_id,
        COUNT(DISTINCT user_id) AS dau
    FROM events
    WHERE event_type = 'click'
    GROUP BY 1, 2
)
SELECT 
    a.day,
    a.product_id,
    a.dau,
    b.sales,
    ROUND(a.dau / b.sales, 2) AS conversion_rate
FROM daily_metrics a
JOIN (
    SELECT 
        DATE_TRUNC('day', order_time) AS day,
        product_id,
        SUM(amount) AS sales
    FROM orders
    GROUP BY 1, 2
) b ON a.day = b.day AND a.product_id = b.product_id
WHERE a.day > CURRENT_DATE - INTERVAL '7' DAY
ORDER BY conversion_rate DESC
"""
ast = parse_one(sql)
print(f"AST節(jié)點(diǎn)數(shù): {len(ast.find_all())}")
print(f"CTE數(shù)量: {len(ast.args['with'].expressions)}")

import sqlglot
mysql_sql = """
SELECT 
    user_id,
    GROUP_CONCAT(DISTINCT product_id ORDER BY purchase_date SEPARATOR ',') AS products
FROM purchases
WHERE status = 'completed'
GROUP BY user_id
HAVING COUNT(DISTINCT order_id) > 3
"""
bq_sql = sqlglot.transpile(
    mysql_sql,
    read="mysql",
    write="bigquery",
    pretty=True
)[0]
print(bq_sql)

SELECT
    user_id,
    STRING_AGG(DISTINCT CAST(product_id AS STRING), ',' ORDER BY purchase_date) AS products
FROM
    purchases
WHERE
    status = 'completed'
GROUP BY
    user_id
HAVING
    COUNT(DISTINCT order_id) > 3

from sqlglot import parse_one, optimize
sql = """
SELECT 
    a.department,
    a.avg_salary,
    (SELECT AVG(salary) 
     FROM employees 
     WHERE department = a.department 
     AND hire_date > DATE_ADD(CURRENT_DATE, INTERVAL -5 YEAR)) AS junior_avg
FROM (
    SELECT 
        department, 
        AVG(salary) AS avg_salary
    FROM employees
    GROUP BY department
) a
"""
optimized = optimize(
    parse_one(sql),
    schema={
        "employees": {
            "columns": ["id", "name", "department", "salary", "hire_date"],
            "indexes": ["department", "hire_date"]
        }
    }
)
print(optimized.sql(pretty=True))

WITH anon_1 AS (
    SELECT
        department,
        AVG(salary) AS avg_salary
    FROM
        employees
    GROUP BY
        department
)
SELECT
    a.department,
    a.avg_salary,
    (
        SELECT
            AVG(salary)
        FROM
            employees
        WHERE
            department = a.department
            AND hire_date > DATE_ADD(CURRENT_DATE, INTERVAL -5 YEAR)
    ) AS junior_avg
FROM
    anon_1 a

from sqlglot import exp, select, func
def build_funnel_query(events, date_column="event_date"):
    query = select(
        f"DATE_TRUNC('day', {date_column}) AS day"
    ).with_alias("base_query")
    for i, event in enumerate(events):
        filter_expr = exp.condition(f"event_type = '{event['type']}'")
        if "filters" in event:
            filter_expr = filter_expr.and_(exp.condition(event["filters"]))
        count_expr = (
            exp.Count(distinct=True)
            .of(exp.Column("user_id"))
            .where(filter_expr)
            .alias(f"step_{i+1}_count")
        )
        query = query.add(count_expr)
    return (
        query.from_("events")
        .group_by("day")
        .order_by("day")
        .sql(dialect="snowflake")
    )
funnel_steps = [
    {"type": "page_view", "name": "頁面訪問"},
    {"type": "add_to_cart", "name": "加入購物車"},
    {"type": "checkout_start", "name": "開始結(jié)賬"},
    {"type": "purchase", "name": "完成購買", "filters": "status = 'success' AND amount > 0"}
]
print(build_funnel_query(funnel_steps))

from sqlglot import parse, transform, exp
def id_mask_transform(expression):
    if isinstance(expression, exp.Column) and expression.name == "id_card":
        return exp.func(
            "CONCAT",
            exp.Literal.string("****"),
            exp.func("SUBSTR", expression, 11, 4)
        ).alias("id_card")
    return expression
sql = "SELECT name, id_card, phone FROM users WHERE age > 18"
ast = parse(sql)
transformed_ast = transform(ast, step=id_mask_transform)
print(transformed_ast.sql(dialect="mysql"))

SELECT name, CONCAT('****', SUBSTR(id_card, 11, 4)) AS id_card, phone FROM users WHERE age > 18

import sqlglot
from timeit import timeit
def compare_dialects(sql, dialects=["mysql", "postgres", "spark"]):
    results = {}
    for dialect in dialects:
        try:
            parsed = sqlglot.parse_one(sql)
            generated = parsed.sql(dialect=dialect)
            # 模擬執(zhí)行時間（實(shí)際應(yīng)連接數(shù)據(jù)庫執(zhí)行）
            exec_time = timeit(lambda: parse_one(generated), number=100)
            results[dialect] = {
                "sql": generated,
                "parse_time": exec_time,
                "length": len(generated)
            }
        except Exception as e:
            results[dialect] = {"error": str(e)}
    return results
query = """
SELECT 
    user_id,
    SUM(CASE WHEN event_type = 'click' THEN 1 ELSE 0 END) AS clicks,
    SUM(CASE WHEN event_type = 'view' THEN 1 ELSE 0 END) AS views
FROM events
GROUP BY user_id
"""
print(compare_dialects(query))

from sqlglot import parse_one
from collections import defaultdict
def analyze_sql_pattern(sql):
    ast = parse_one(sql)
    pattern_stats = defaultdict(int)
    # 統(tǒng)計(jì)JOIN類型
    for join in ast.find_all(exp.Join):
        join_type = join.args.get("join_type", "INNER").upper()
        pattern_stats[f"JOIN_{join_type}"] += 1
    # 統(tǒng)計(jì)聚合函數(shù)
    for func in ast.find_all(exp.Func):
        if func.name.upper() in ["SUM", "AVG", "COUNT", "MAX", "MIN"]:
            pattern_stats[f"AGG_{func.name.upper()}"] += 1
    return dict(pattern_stats)
complex_query = """
SELECT 
    u.id,
    u.name,
    COUNT(DISTINCT o.order_id) AS order_count,
    SUM(o.amount) AS total_amount,
    AVG(o.amount) AS avg_order_value
FROM users u
LEFT JOIN orders o ON u.id = o.user_id
WHERE u.status = 'active'
GROUP BY u.id, u.name
HAVING COUNT(DISTINCT o.order_id) > 5
"""
print(analyze_sql_pattern(complex_query))

import pandas as pd
from sqlglot import parse_one
def sql_to_dataframe(sql, data):
    ast = parse_one(sql)
    # 實(shí)際實(shí)現(xiàn)需要解析AST并轉(zhuǎn)換為Pandas操作
    # 此處僅為概念演示
    if "SELECT * FROM" in sql.upper():
        return pd.DataFrame(data)
    elif "WHERE" in sql.upper():
        condition = sql.split("WHERE")[1].split("GROUP BY")[0].strip()
        # 簡化處理，實(shí)際需解析條件表達(dá)式
        filtered_data = {k: v for k, v in data.items() if eval(condition, {}, v)}
        return pd.DataFrame(filtered_data)
    return pd.DataFrame()
sample_data = {
    "id": [1, 2, 3],
    "name": ["Alice", "Bob", "Charlie"],
    "age": [25, 30, 35]
}
df = sql_to_dataframe("SELECT name, age FROM sample WHERE age > 28", sample_data)
print(df)

from functools import lru_cache
from sqlglot import parse_one
@lru_cache(maxsize=1000)
def cached_parse(sql):
    return parse_one(sql)
# 重復(fù)解析相同SQL時將直接從緩存獲取

from sqlglot import exp
# 預(yù)定義常用表達(dá)式
COMMON_EXPRESSIONS = {
    "recent_7_days": exp.condition(
        "event_time >= DATE_SUB(CURRENT_DATE, INTERVAL 7 DAY)"
    ),
    "active_users": exp.condition(
        "last_active_date >= DATE_SUB(CURRENT_DATE, INTERVAL 30 DAY)"
    )
}
def build_query_with_patterns(base_sql, patterns):
    ast = parse_one(base_sql)
    for alias, expr in patterns.items():
        ast = ast.with_cte(exp.CTE(alias, exp.select().from_("dummy").where(expr)))
    return ast.sql()

from concurrent.futures import ThreadPoolExecutor
import sqlglot
def parallel_parse(sql_list, max_workers=4):
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        results = list(executor.map(sqlglot.parse_one, sql_list))
    return results
large_sql_batch = ["SELECT * FROM table{}".format(i) for i in range(100)]
parsed_asts = parallel_parse(large_sql_batch)