基于Python提取PL/SQL代码中的源表、关联表及子查询表

需求概述 #

需要编写Python代码，从PL/SQL代码中精准识别并提取以下四类表：

• 源表（主查询FROM子句中的基础表）
• 关联表（主查询中通过JOIN或逗号分隔关联的表）
• 子查询表（子查询FROM子句中的基础表）
• 子查询内关联表（子查询内部通过JOIN关联的表）
  同时代码需要适配不同格式的PL/SQL代码变更。

预期输出格式 #

{'source_table': ['schema_name.table1 (rfslt)'], 
 'join_tables': ['schema_name.table2 (b)', 'table3 (doogal)', 'tab_time (b)'],
 'subquery_table': ['schema_name.table6 (e)'],
 'subquery_join_table': ['schema_name.table7 (h)']}
{'source_tables': ['table1 (a)'], 
 'join_tables': ['table2 (b)', 'table3 (c)', 'table4 (d)']}
{'source_tables': ['table1 (a)'], 
 'join_tables': ['table2 (b)', 'table3 (c)']}

待解析PL/SQL代码 #

Insert into tabs 
Select * from (Select * from 
schema_name.table1 rfslt
LEFT OUTER JOIN schema_name.table2 b ON rfslt.key = b.key
LEFT OUTER JOIN table3 doogal ON REPLACE(rfslt.code_c, ' ', '') = doogal.code_derived
LEFT OUTER JOIN (select *
    from schema_name.table6 e
    left join schema_name.table7 h on h.id = e.id order by h.change_name) muk ON rfslt.id=muk.id
) prems
WHERE choose1 = 1
) a,
tab_time b
WHERE TRUNC(a.ok_date) = b.g_date(+);

Insert into tab 
Select a.* 
from table1 a
left join table2 b on (a.col1 = b.col1)
left join table3 c on (a.col2 = c.col2)
inner join table4 d on (b.col3 = d.col3)
Where a.col4 = 'TEST';

Insert into tab 
Select case when a.col1 = 'text' then 'Next on top' end d_col1 
from (Select * from table1 tbl where tbl.col0 = 'sample') a, 
table2 b, 
(Select * from table3 tbl3 where col0 in (select col0 from table4 order by col8)) c 
Where a.col1 = b.col1(+) and a.col2 = 'TEST' and a.col3 = c.col3;

原尝试代码问题分析 #

原代码的正则表达式存在以下不足：

• 无法区分主查询和子查询中的表
• 正则分组逻辑错误，无法正确捕获表名和别名
• 未处理逗号分隔的隐式关联表
• 未递归解析子查询内部的表结构

改进后的Python实现 #

import re
from typing import Dict, List

def parse_table_ref(ref_str: str) -> str:
    """解析表引用字符串，返回`表名 (别名)`格式的字符串"""
    table_match = re.match(r'^\s*([\w\.]+)\s*(?:AS\s*)?(\w+)?\s*$', ref_str, re.IGNORECASE)
    if not table_match:
        return ""
    table_name = table_match.group(1).strip()
    alias = table_match.group(2).strip() if table_match.group(2) else ""
    return f"{table_name} ({alias})" if alias else table_name

def extract_tables_from_sql(sql_segment: str, is_subquery: bool = False) -> Dict[str, List[str]]:
    """从SQL片段中提取各类表，递归处理子查询"""
    result = {
        'source_table': [],
        'join_tables': [],
        'subquery_table': [],
        'subquery_join_table': []
    }
    
    # 先提取所有子查询，递归处理
    subquery_pattern = re.compile(r'\((SELECT\s+.+?)\)', re.IGNORECASE | re.DOTALL)
    subqueries = subquery_pattern.findall(sql_segment)
    for subquery in subqueries:
        sub_result = extract_tables_from_sql(subquery, is_subquery=True)
        # 合并子查询的结果到主结果
        result['subquery_table'].extend(sub_result['source_table'])
        result['subquery_join_table'].extend(sub_result['join_tables'])
        result['subquery_table'].extend(sub_result['subquery_table'])
        result['subquery_join_table'].extend(sub_result['subquery_join_table'])
        # 移除已处理的子查询，避免重复匹配
        sql_segment = sql_segment.replace(f"({subquery})", "")
    
    # 处理FROM子句中的源表（主表）
    from_pattern = re.compile(r'\bFROM\s+([\w\.\s]+?)(?:\s*(?:JOIN|,|WHERE|$))', re.IGNORECASE | re.DOTALL)
    from_matches = from_pattern.findall(sql_segment)
    for match in from_matches:
        table_str = parse_table_ref(match)
        if table_str and not is_subquery:
            result['source_table'].append(table_str)
        elif table_str and is_subquery:
            result['subquery_table'].append(table_str)
    
    # 处理显式JOIN的关联表
    join_pattern = re.compile(r'\b(?:LEFT|RIGHT|INNER|OUTER|CROSS)\s*JOIN\s+([\w\.\s]+?)\s+ON', re.IGNORECASE | re.DOTALL)
    join_matches = join_pattern.findall(sql_segment)
    for match in join_matches:
        table_str = parse_table_ref(match)
        if table_str and not is_subquery:
            result['join_tables'].append(table_str)
        elif table_str and is_subquery:
            result['subquery_join_table'].append(table_str)
    
    # 处理逗号分隔的隐式关联表
    comma_join_pattern = re.compile(r',\s+([\w\.\s]+?)(?:\s*(?:,|WHERE|$))', re.IGNORECASE | re.DOTALL)
    comma_matches = comma_join_pattern.findall(sql_segment)
    for match in comma_matches:
        table_str = parse_table_ref(match)
        if table_str and not is_subquery:
            result['join_tables'].append(table_str)
        elif table_str and is_subquery:
            result['subquery_join_table'].append(table_str)
    
    # 去重并过滤空字符串
    for key in result:
        result[key] = list(filter(None, list(set(result[key]))))
    
    return result

def process_plsql(plsql_code: str) -> List[Dict[str, List[str]]]:
    """处理完整的PL/SQL代码，按每个INSERT语句拆分处理"""
    # 按INSERT语句拆分PL/SQL代码
    insert_pattern = re.compile(r'INSERT\s+INTO\s+.+?(?=INSERT|$)', re.IGNORECASE | re.DOTALL)
    insert_statements = insert_pattern.findall(plsql_code)
    results = []
    for stmt in insert_statements:
        stmt_result = extract_tables_from_sql(stmt)
        # 兼容预期输出的键名（source_table/source_tables）
        if stmt_result['source_table']:
            if len(stmt_result['source_table']) > 1:
                stmt_result['source_tables'] = stmt_result.pop('source_table')
            else:
                stmt_result['source_table'] = stmt_result['source_table']
        results.append(stmt_result)
    return results

# 测试代码
if __name__ == "__main__":
    plsql_code = """
Insert into tabs 
Select * from (Select * from 
schema_name.table1 rfslt
LEFT OUTER JOIN schema_name.table2 b ON rfslt.key = b.key
LEFT OUTER JOIN table3 doogal ON REPLACE(rfslt.code_c, ' ', '') = doogal.code_derived
LEFT OUTER JOIN (select *
    from schema_name.table6 e
    left join schema_name.table7 h on h.id = e.id order by h.change_name) muk ON rfslt.id=muk.id
) prems
WHERE choose1 = 1
) a,
tab_time b
WHERE TRUNC(a.ok_date) = b.g_date(+);

Insert into tab 
Select a.* 
from table1 a
left join table2 b on (a.col1 = b.col1)
left join table3 c on (a.col2 = c.col2)
inner join table4 d on (b.col3 = d.col3)
Where a.col4 = 'TEST';

Insert into tab 
Select case when a.col1 = 'text' then 'Next on top' end d_col1 
from (Select * from table1 tbl where tbl.col0 = 'sample') a, 
table2 b, 
(Select * from table3 tbl3 where col0 in (select col0 from table4 order by col8)) c 
Where a.col1 = b.col1(+) and a.col2 = 'TEST' and a.col3 = c.col3;
    """
    results = process_plsql(plsql_code)
    for res in results:
        print(res)

代码说明 #

1. parse_table_ref函数：负责解析表引用字符串，将schema.table alias或schema.table AS alias格式转换为统一的表名 (别名)格式。
2. extract_tables_from_sql函数：递归处理SQL片段，先提取并解析所有子查询，再分别处理主查询/子查询中的源表、显式JOIN表、隐式逗号关联表，区分存入对应的结果键。
3. process_plsql函数：将完整的PL/SQL代码按INSERT语句拆分，逐个处理并返回结果列表，同时兼容预期输出的键名（source_table/source_tables）。
4. 正则表达式使用了IGNORECASE和DOTALL标志，适配不同大小写和换行格式的PL/SQL代码。

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