如何用Python在SQL查询中重命名表与列（含复杂场景）

你当前基于sqlglot的实现仅能处理全局名称映射，但在同列名跨表需不同映射、嵌套子查询存在重复别名的场景下会失效。核心问题是没有追踪列所属的表（别名对应的原表），也未处理子查询的独立作用域。

1. 重构映射结构 #

将原扁平字典改为按表分组的结构，明确每个表对应的列映射规则：

mapping = {
    "tables": {
        "person": "individual",
        "job": "occupation"
    },
    "columns": {
        "person": {
            "name": "full_name",
            "age": "years_old",
            "job_id": "occupation_id"
        },
        "job": {
            "title": "role",
            "job_id": "identifier"
        }
    }
}

2. 完整实现代码 #

通过封装类来追踪查询的别名作用域，递归处理嵌套子查询，同时结合表别名完成精准的列名替换：

from sqlglot import parse
from sqlglot.expressions import Identifier, Column, Table, Select, Join, SubqueryAlias

class SQLNameTransformer:
    def __init__(self, mapping):
        self.table_mapping = mapping.get("tables", {})
        self.column_mapping = mapping.get("columns", {})
    
    def _collect_alias_mapping(self, node, alias_map):
        """递归收集当前查询中别名到原表名的映射，同时替换表名"""
        # 处理基础表
        if isinstance(node, Table):
            alias = node.args.get("alias")
            if alias:
                alias_name = alias.args.get("this")
                original_table = node.args.get("this")
                if alias_name and original_table:
                    alias_map[alias_name] = original_table
            # 替换表名
            original_table = node.args.get("this")
            if original_table in self.table_mapping:
                node.args["this"] = self.table_mapping[original_table]
            return
        
        # 处理JOIN子句
        if isinstance(node, Join):
            self._collect_alias_mapping(node.args.get("this"), alias_map)
            return
        
        # 处理子查询别名（如 (SELECT ...) AS t）
        if isinstance(node, SubqueryAlias):
            alias = node.args.get("alias")
            if alias:
                alias_name = alias.args.get("this")
                alias_map[alias_name] = alias_name
            # 递归处理子查询内部，使用独立的别名映射
            subquery = node.args.get("this")
            if isinstance(subquery, Select):
                self._collect_alias_mapping(subquery, {})
            return
        
        # 处理SELECT语句，遍历FROM/JOIN等子句
        if isinstance(node, Select):
            from_expr = node.args.get("from")
            if from_expr:
                self._collect_alias_mapping(from_expr, alias_map)
            for join in node.args.get("joins", []):
                self._collect_alias_mapping(join, alias_map)
            # 处理WHERE/HAVING中的子查询
            for key in ["where", "having"]:
                expr = node.args.get(key)
                if expr:
                    self._collect_alias_mapping(expr, {})
            return
        
        # 递归处理所有子节点
        for key, value in node.args.items():
            if isinstance(value, list):
                for item in value:
                    self._collect_alias_mapping(item, alias_map)
            elif hasattr(value, "args"):
                self._collect_alias_mapping(value, alias_map)
    
    def _transform_column(self, node, alias_map):
        """结合别名映射，替换列名"""
        prefix = node.args.get("this").args.get("prefix")
        column_name = node.args.get("this").args.get("this")
        
        if not prefix or not column_name:
            return node
        
        # 通过别名找到原表，再查对应列的映射
        original_table = alias_map.get(prefix)
        if not original_table:
            return node
        
        table_columns = self.column_mapping.get(original_table, {})
        if column_name in table_columns:
            node.args["this"].args["this"] = table_columns[column_name]
        return node
    
    def transform(self, sql):
        expressions = parse(sql)
        transformed_exprs = []
        
        for expr in expressions:
            alias_map = {}
            self._collect_alias_mapping(expr, alias_map)
            # 遍历替换所有列名
            expr = expr.transform(lambda node: self._transform_column(node, alias_map) if isinstance(node, Column) else node)
            transformed_exprs.append(expr)
        
        return "".join(e.sql() for e in transformed_exprs)

# 测试用例
if __name__ == "__main__":
    sql = '''SELECT p.name, p.age, j.title, p.job_id, j.job_id
FROM person AS p
LEFT JOIN job AS j ON p.job_id = j.job_id
WHERE p.age > 30
AND EXISTS (SELECT 1 FROM person AS p WHERE p.age < 18)'''
    
    transformer = SQLNameTransformer(mapping)
    print(transformer.transform(sql))

3. 代码说明 #

• _collect_alias_mapping：递归遍历AST，收集当前查询（含子查询）的别名-原表映射，同时直接完成表名替换。子查询会使用独立的别名映射，避免和外部查询冲突。
• _transform_column：根据列的前缀（别名）找到所属原表，再匹配对应的列映射规则完成替换。
• 嵌套子查询的作用域被独立处理，重复别名不会互相干扰。

测试输出的SQL为：

SELECT p.full_name, p.years_old, j.role, p.occupation_id, j.identifier FROM individual AS p LEFT JOIN occupation AS j ON p.occupation_id = j.identifier WHERE p.years_old > 30 AND EXISTS (SELECT 1 FROM individual AS p WHERE p.years_old < 18)

内容的提问来源于stack exchange，提问作者Heisen