Snowflake Python Connector使用executemany插入时SQL编译错误求助

表结构 #

CREATE TABLE "VALIDATION_RULES_RESULTS" (
    id INT AUTOINCREMENT,
    created_on TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    mongo_data_ingestion_id STRING,
    table_name STRING,
    col_name STRING,
    rule_name STRING,
    success STRING,
    element_count INT,
    missing_count INT,
    failed_records_indexes VARIANT,
    failed_records_values VARIANT
);

Python批量插入脚本 #

insert_statement = f"""
INSERT INTO "VALIDATION_RULES_RESULTS" (
    mongo_data_ingestion_id,
    table_name,
    col_name,
    rule_name,
    success,
    element_count,
    missing_count,
    failed_records_indexes,
    failed_records_values
) select 
    mongo_data_ingestion_id,
    table_name,
    col_name,
    rule_name,
    success,
    element_count,
    missing_count,
    parse_json(failed_records_indexes),
    parse_json(failed_records_values)
from VALUES (
    %(mongo_data_ingestion_id)s,
    %(table_name)s,
    %(col_name)s,
    %(rule_name)s,
    %(success)s,
    %(element_count)s,
    %(missing_count)s,
    %(failed_records_indexes)s,
    %(failed_records_values)s    
    )
"""

# Prepare the data for executemany
rows_to_insert = []
for rcd in snowflake_validation_rule_row:
    row = {
        'mongo_data_ingestion_id': rcd['mongo_data_ingestion_id'],
        'table_name': rcd['table_name'],
        'col_name': rcd['col_name'],
        'rule_name': rcd['rule_name'],
        'success': str(rcd['success']).lower(),
        'element_count': rcd['element_count'],
        'missing_count': rcd['missing_count'],
        'failed_records_indexes': json.dumps(rcd['failed_records_indexes'], default=str),
        'failed_records_values': json.dumps(rcd['failed_records_values'], default=str)
    }
    rows_to_insert.append(row)

snoflake_conn.cursor().executemany(
    insert_statement,
    rows_to_insert
)

报错信息 #

20:44:13.63 !!! snowflake.connector.errors.ProgrammingError: 000904 (42000): SQL compilation error: error line 12 at position 8
20:44:13.63 !!! invalid identifier 'MONGO_DATA_INGESTION_ID'
20:44:13.63 !!! When calling: snoflake_conn.cursor().executemany(
20:44:13.63                       insert_statement,
20:44:13.63                       rows_to_insert
20:44:13.63                   )
20:44:13.63 !!! Call ended by exception

示例插入数据 #

{'mongo_data_ingestion_id': '1728370271', 'table_name': 'yest', 'col_name': 'test', 'rule_name': 'basic_rule_expect_column_value_lengths_to_be_between', 'success': 'false', 'element_count': 10, 'missing_count': 0, 'failed_records_indexes': '[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]', 'failed_records_values': '["1982-04-20 00:00:00", "1984-12-15 00:00:00", "1981-07-10 00:00:00", "1977-12-24 00:00:00", "1965-05-13 00:00:00", "1957-05-17 00:00:00", "1991-12-29 00:00:00", "1913-03-15 00:00:00", "1983-08-27 00:00:00", "1957-04-07 00:00:00"]'}

1. 字段名大小写不匹配：使用VALUES子句生成临时数据集时，Snowflake会自动将字段名转为大写（如mongo_data_ingestion_id变为MONGO_DATA_INGESTION_ID），但SELECT语句中引用的是小写字段名，导致无法匹配，抛出标识符无效错误。
2. 写法不适配executemany：当前的INSERT...SELECT...VALUES结构冗余，executemany本身会自动处理批量参数，无需手动用VALUES子句包裹。

推荐使用第一种方式，适配executemany的批量插入逻辑：

方式一：标准INSERT...VALUES参数化写法 #

修改插入语句为标准格式，去掉多余的SELECT层，让executemany自动处理批量数据：

insert_statement = """
INSERT INTO "VALIDATION_RULES_RESULTS" (
    mongo_data_ingestion_id,
    table_name,
    col_name,
    rule_name,
    success,
    element_count,
    missing_count,
    failed_records_indexes,
    failed_records_values
) VALUES (
    %(mongo_data_ingestion_id)s,
    %(table_name)s,
    %(col_name)s,
    %(rule_name)s,
    %(success)s,
    %(element_count)s,
    %(missing_count)s,
    parse_json(%(failed_records_indexes)s),
    parse_json(%(failed_records_values)s)
)
"""

数据准备部分代码保持不变，直接执行executemany即可。

方式二：给VALUES子句显式指定字段别名（不推荐） #

若坚持使用INSERT...SELECT结构，需为VALUES子句的每一列指定别名，确保SELECT语句能正确匹配：

insert_statement = f"""
INSERT INTO "VALIDATION_RULES_RESULTS" (
    mongo_data_ingestion_id,
    table_name,
    col_name,
    rule_name,
    success,
    element_count,
    missing_count,
    failed_records_indexes,
    failed_records_values
) select 
    mongo_data_ingestion_id,
    table_name,
    col_name,
    rule_name,
    success,
    element_count,
    missing_count,
    parse_json(failed_records_indexes),
    parse_json(failed_records_values)
from VALUES (
    %(mongo_data_ingestion_id)s,
    %(table_name)s,
    %(col_name)s,
    %(rule_name)s,
    %(success)s,
    %(element_count)s,
    %(missing_count)s,
    %(failed_records_indexes)s,
    %(failed_records_values)s    
) AS t(
    mongo_data_ingestion_id,
    table_name,
    col_name,
    rule_name,
    success,
    element_count,
    missing_count,
    failed_records_indexes,
    failed_records_values
)
"""

此写法复杂度高，不适合executemany的批量场景，仅作备选。

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