Python循环校验Trino SQL语法如何添加3秒超时终止逻辑？

你原来的超时逻辑不生效，是因为cur.execute()是同步阻塞方法，调用后会一直等待Trino返回结果才会执行后续代码，计时判断逻辑永远在执行完成之后才触发，完全起不到超时终止的作用。

你核心需求仅为校验SQL语法，完全不需要实际执行查询拉取数据，优先用Trino原生的EXPLAIN语法校验即可，这个操作只会校验SQL语法合法性、生成执行计划，不会实际运行查询，耗时极短，本身就不会出现长耗时问题：

sql_results = []
for l in sql:
    sql_name, sql_code = l
    try:
        # 用EXPLAIN包裹原SQL，仅做语法和执行计划校验
        cur.execute(f"EXPLAIN {sql_code}")
        status = "OK"
    except Exception as e:
        status = str(e)
    sql_results.append([sql_name, sql_code, status])

如果一定要保留原执行逻辑+3秒超时规则，用以下两种方案实现：

方案1：使用Trino原生超时参数（最稳定） #

Trino Python客户端支持直接在执行查询时设置最大运行时间，超过时间Trino服务端会自动终止查询并抛出异常，你捕获异常后判断为超时即可标记为OK：

from trino.exceptions import TrinoQueryError

sql_results = []
for l in sql:
    sql_name, sql_code = l
    try:
        # 单个查询最大运行时间设为3秒，单位为毫秒
        cur.execute(sql_code, query_max_run_time=3000)
        rows = cur.fetchall()
        status = "OK"
    except TrinoQueryError as e:
        # 判断是否为超时错误
        if "Query exceeded maximum run time" in str(e):
            status = "OK"
        else:
            status = str(e)
    except Exception as e:
        status = str(e)
    sql_results.append([sql_name, sql_code, status])

方案2：Python层面实现超时终止（兼容所有数据库客户端） #

如果你的客户端不支持原生超时，用concurrent.futures.ThreadPoolExecutor实现执行逻辑的超时控制：

import concurrent.futures

def run_sql(sql_code):
    cur.execute(sql_code)
    return cur.fetchall()

sql_results = []
executor = concurrent.futures.ThreadPoolExecutor(max_workers=1)
for l in sql:
    sql_name, sql_code = l
    try:
        # 最多等待3秒
        future = executor.submit(run_sql, sql_code)
        rows = future.result(timeout=3)
        status = "OK"
    except concurrent.futures.TimeoutError:
        # 超时直接标记OK
        status = "OK"
    except Exception as e:
        status = str(e)
    sql_results.append([sql_name, sql_code, status])

内容的提问来源于stack exchange，提问作者Joaquín Navarro Mañas