批量INSERT操作日志报错未返回所有涉及表名的问题求助

原代码存在两个核心问题导致无法定位全部出错表：

1. executemany调用时机错误：循环内每添加一条SQL就执行一次executemany，会重复执行之前所有已添加的语句，且报错时只能捕获当前循环到的schema.table，而非真正出错的语句对应表。
2. 批量执行无绑定表信息：即使正确批量执行，默认错误信息不会关联具体表名，无法直接映射到出错的schema.table。

方案一：单条执行（简单直接，便于定位） #

将每条INSERT语句单独执行，报错时直接记录当前操作的schema和表名：

sql = "SELECT * FROM TABLE"
cur.execute(sql)
df = pd.DataFrame.from_records(iter(cur), columns=[x[0] for x in cur.description])
my_dict = dict()

for i in df['col1'].unique().tolist():
    df_x = df[df['col1'] == i]
    my_dict[i] = df_x['col_table'].tolist()

# 遍历每个schema和表，单独执行INSERT
for schema, tables in my_dict.items():
    for table in tables:
        insert_sql = f"INSERT INTO {schema}.{table} SELECT * FROM {schema}.{table} where col2 = 1;"
        try:
            cur.execute(insert_sql)
        except snowflake.connector.errors.ProgrammingError as e:
            logging.error(f"Insert failed on {schema}.{table}: {str(e)}")

conn.close()

方案二：批量执行并绑定表信息（适合性能要求高的场景） #

如果需要批量执行，可将SQL语句与对应的schema.table绑定，执行时捕获错误并匹配对应表：

sql = "SELECT * FROM TABLE"
cur.execute(sql)
df = pd.DataFrame.from_records(iter(cur), columns=[x[0] for x in cur.description])
my_dict = dict()

for i in df['col1'].unique().tolist():
    df_x = df[df['col1'] == i]
    my_dict[i] = df_x['col_table'].tolist()

# 收集SQL和对应的表信息
sql_with_table = []
for schema, tables in my_dict.items():
    for table in tables:
        insert_sql = f"INSERT INTO {schema}.{table} SELECT * FROM {schema}.{table} where col2 = 1;"
        sql_with_table.append( (insert_sql, f"{schema}.{table}") )

# 批量执行，出错时定位具体表
try:
    # 提取所有SQL语句用于executemany
    sql_list = [item[0] for item in sql_with_table]
    cur.executemany('EXECUTE IMMEDIATE %s', sql_list)
except snowflake.connector.errors.ProgrammingError as e:
    logging.error(f"Batch insert failed. Error: {str(e)}")
    # 记录本次批量涉及的所有表
    logging.error(f"Tables involved in this batch: {', '.join([item[1] for item in sql_with_table])}")

    # 精准定位：拆分批量为单条重试排查
    for sql, table in sql_with_table:
        try:
            cur.execute(sql)
        except snowflake.connector.errors.ProgrammingError as e:
            logging.error(f"Insert failed on {table}: {str(e)}")

conn.close()

关键优化点 #

• 移除循环内的executemany调用：避免重复执行语句，提升效率同时减少混乱。
• 绑定表信息与SQL：确保报错时能准确关联到对应的schema.table。
• 优先单条执行排查：调试阶段优先用单条执行快速定位错误表，稳定后再考虑批量。

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