SQLGlot库全面解析

SQLGlot，这是一个强大的 Python SQL 解析、生成和转换工具库。

一、SQLGlot 是什么？

SQLGlot 是一个纯 Python 实现的 SQL 解析器、生成器和优化器。它支持多种 SQL 方言，可以用于：

SQL 语法解析（将 SQL 字符串转换为 AST）
SQL 生成（将 AST 转换回 SQL 字符串）
SQL 转换（在不同 SQL 方言间转换）
SQL 优化（简化、重写 SQL）

二、核心特性

1. 多方言支持

支持 20+ 种 SQL 方言：

主要方言：SparkSQL, Hive, Presto, Trino, DuckDB, MySQL, PostgreSQL, SQLite, SQL Server, Oracle, Redshift, Snowflake, BigQuery
实验性支持：DB2, Databricks, Doris, ClickHouse, RisingWave, StarRocks, Tableau

2. 纯 Python 实现

无外部依赖
易于安装和部署
跨平台兼容

3. 功能丰富

完整的 SQL 解析和生成
表达式化简和优化
语义分析（作用域、类型推断）
SQL 重写和美化

三、安装与基础使用

pip install sqlglot

四、核心功能详解

1. SQL 解析（Parse）

import sqlglot

# 解析 SQL 为 AST
sql = "SELECT a, b FROM t WHERE c = 1"
ast = sqlglot.parse(sql)[0]

# 查看 AST 结构
print(ast.sql())

2. SQL 生成（Generate）

import sqlglot

# 从 AST 生成 SQL
ast = sqlglot.parse("SELECT * FROM t")[0]

# 生成不同方言的 SQL
print(ast.sql(dialect="spark"))    # SparkSQL
print(ast.sql(dialect="mysql"))    # MySQL
print(ast.sql(dialect="postgres")) # PostgreSQL

3. SQL 转换（Transpile）

import sqlglot

# 将 MySQL SQL 转换为 SparkSQL
sql = """
    SELECT DATE_FORMAT(date, '%Y-%m') as month,
           COUNT(*) as count
    FROM sales
    GROUP BY month
"""

transpiled = sqlglot.transpile(sql, 
                               read="mysql", 
                               write="spark")[0]
print(transpiled)
# 输出：SELECT DATE_FORMAT(date, 'yyyy-MM') AS month, COUNT(*) AS count FROM sales GROUP BY month

4. 表达式操作

import sqlglot
import sqlglot.expressions as exp

# 构建表达式
expr = exp.Column(this="id")
expr = exp.EQ(this=expr, expression=exp.Literal(this="1"))

# 遍历和修改表达式
for node in expr.walk():
    if isinstance(node, exp.Literal):
        node.set("this", "2")

print(expr.sql())

5. SQL 优化

import sqlglot
from sqlglot.optimizer import optimize

sql = """
SELECT * FROM (
    SELECT a, b FROM t
) subq
WHERE a > 10
"""

# 优化 SQL（如谓词下推）
optimized = optimize(sql)
print(optimized.sql())

五、高级功能

1. 自定义转换规则

import sqlglot
from sqlglot import exp

def remove_alias(node):
    """移除所有别名"""
    if isinstance(node, exp.Alias):
        return node.this

sql = "SELECT a AS col1, b AS col2 FROM t"
transformed = sqlglot.transpile(sql, 
                               identify=True,
                               transforms=[remove_alias])[0]
print(transformed)  # SELECT a, b FROM t

2. SQL 美化（格式化）

import sqlglot

sql = "select a,b,c from t where x=1"
pretty_sql = sqlglot.transpile(sql, pretty=True)[0]
print(pretty_sql)

3. 模式验证和语义分析

import sqlglot
from sqlglot.schema import Schema

# 定义模式
schema = {
    "users": {
        "id": "INT",
        "name": "VARCHAR",
        "created_at": "TIMESTAMP"
    }
}

# 验证 SQL
sql = "SELECT id, name, invalid_col FROM users"
try:
    sqlglot.validate(sql, schema=schema)
except Exception as e:
    print(f"验证错误: {e}")

4. SQL 方言差异处理

import sqlglot

# 处理方言特定的函数
sql = "SELECT GETDATE()"  # SQL Server 语法

# 转换为 PostgreSQL
pg_sql = sqlglot.transpile(sql, read="tsql", write="postgres")[0]
print(pg_sql)  # SELECT NOW()

六、实际应用场景

1. 数据仓库迁移

def migrate_redshift_to_bigquery(sql):
    """将 Redshift SQL 迁移到 BigQuery"""
    return sqlglot.transpile(
        sql,
        read="redshift",
        write="bigquery",
        identify=True,  # 自动标识符加引号
        pretty=True     # 美化输出
    )[0]

2. SQL 标准化

def standardize_sql(sql, target_dialect="spark"):
    """将各种 SQL 统一为标准格式"""
    return sqlglot.transpile(
        sql,
        write=target_dialect,
        pretty=True,
        normalize=True  # 标准化关键字大小写
    )[0]

3. SQL 质量检查

def check_sql_quality(sql):
    """检查 SQL 质量问题"""
    issues = []

    # 1. 检查 SELECT *
    if "SELECT *" in sql.upper():
        issues.append("避免使用 SELECT *")

    # 2. 使用 AST 进行更复杂的检查
    try:
        ast = sqlglot.parse(sql)[0]
        # 检查未使用 WHERE 的 DELETE
        if isinstance(ast, exp.Delete) and not ast.args.get("where"):
            issues.append("DELETE 语句缺少 WHERE 条件")
    except:
        issues.append("SQL 语法错误")

    return issues

七、与其他库的对比

特性	SQLGlot	sqlparse	ANTLR	sqloxide (Rust)
语言	Python	Python	多语言	Rust
依赖	无	无	Java	无
多方言	✅	❌	✅	✅
AST 操作	✅	有限	✅	✅
SQL 生成	✅	❌	✅	✅
转换	✅	❌	需要自定义	✅

八、最佳实践

1. 错误处理

import sqlglot

def safe_transpile(sql, read=None, write=None):
    try:
        return sqlglot.transpile(sql, read=read, write=write)[0]
    except sqlglot.errors.ParseError as e:
        print(f"解析错误: {e}")
        return sql
    except Exception as e:
        print(f"转换错误: {e}")
        return sql

2. 性能优化

# 对于大量 SQL 处理，使用解析器缓存
from sqlglot import parse

# 复用解析器实例
parser = sqlglot.Parser()

def batch_process(sql_list):
    results = []
    for sql in sql_list:
        # 使用缓存的解析器
        ast = parser.parse(sql)[0]
        # 处理 AST
        processed = ast.sql(dialect="spark")
        results.append(processed)
    return results

3. 自定义方言扩展

from sqlglot import exp
from sqlglot.dialects.dialect import Dialect
from sqlglot.generator import Generator

class CustomDialect(Dialect):
    class Generator(Generator):
        TRANSFORMS = {
            **Generator.TRANSFORMS,
            exp.CurrentDate: lambda self, e: "CURRENT_DATE()",
        }