如何快速批量列出Snowflake查询语句中访问的所有表？

方法一：正则表达式本地批量处理 #

适合快速处理本地存储的SQL文件，核心是通过正则匹配Snowflake表的典型引用格式（库.模式.表/模式.表/表），结合FROM/JOIN等上下文减少误判。

• 可以直接用Python脚本批量遍历文件，提取后去重输出：

import re
import os

def extract_tables(sql_content):
    # 匹配FROM/JOIN等关键字后的表名，覆盖三种格式
    pattern = re.compile(r'(?:FROM|JOIN|INTO|UPDATE|DELETE FROM)\s+([A-Za-z0-9_]+\.[A-Za-z0-9_]+\.[A-Za-z0-9_]+|[A-Za-z0-9_]+\.[A-Za-z0-9_]+|[A-Za-z0-9_]+)', re.IGNORECASE)
    matches = pattern.findall(sql_content)
    return list(set(matches))

def process_files(dir_path):
    result = {}
    for filename in os.listdir(dir_path):
        if filename.endswith('.sql'):
            with open(os.path.join(dir_path, filename), 'r', encoding='utf-8') as f:
                tables = extract_tables(f.read())
                result[filename] = tables
    
    # 输出结果到文件
    with open('extracted_tables.txt', 'w', encoding='utf-8') as out_f:
        for file, tables in result.items():
            out_f.write(f'文件: {file}\n')
            out_f.write('访问的表:\n')
            for table in tables:
                out_f.write(f'  - {table}\n')
            out_f.write('\n')

# 替换为你的SQL文件所在目录
process_files('./sql_queries')

• 注意：正则可能误匹配关键字，提取后建议快速校验，可根据实际查询风格调整正则的上下文匹配逻辑。

方法二：Snowflake元数据查询（针对已执行的查询） #

如果这些查询已经在Snowflake上运行过，直接利用系统视图提取更准确，避免正则的局限性：

SELECT DISTINCT
    qh.QUERY_ID,
    obj.OBJECT_DATABASE || '.' || obj.OBJECT_SCHEMA || '.' || obj.OBJECT_NAME AS FULL_TABLE_NAME
FROM
    TABLE(INFORMATION_SCHEMA.QUERY_HISTORY(END_TIME_RANGE_START=>DATEADD('day', -7, CURRENT_TIMESTAMP()))) qh
JOIN
    TABLE(INFORMATION_SCHEMA.QUERY_USAGE_HISTORY(QUERY_ID=>qh.QUERY_ID)) obj
WHERE
    obj.OBJECT_TYPE = 'TABLE'
    -- 可添加查询文本过滤，比如匹配特定文件中的查询特征
    -- AND qh.QUERY_TEXT LIKE '%your_query_keyword%'
ORDER BY
    qh.START_TIME DESC;

方法三：SQL语法解析库（精准处理复杂查询） #

针对嵌套查询、子查询较多的复杂SQL，用专业解析库提取AST（抽象语法树）来获取表名，比正则更可靠。比如用Python的sqlparse：

import sqlparse
from sqlparse.sql import IdentifierList, Identifier

def extract_tables_from_ast(sql):
    tables = set()
    parsed = sqlparse.parse(sql)[0]

    def traverse(token):
        if isinstance(token, IdentifierList):
            for item in token.get_identifiers():
                traverse(item)
        elif isinstance(token, Identifier):
            # 排除函数等非表对象
            if not token.token_first().is_keyword or token.token_first().value.upper() != 'FUNCTION':
                tables.add(token.get_real_name())
        elif hasattr(token, 'tokens'):
            for sub_token in token.tokens:
                traverse(sub_token)
    
    traverse(parsed)
    return list(tables)

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