如何在NpgsqlBatchedCommand中让INSERT...SELECT无结果时抛出异常

问题背景 #

当前代码通过INSERT ... SELECT复制dc_file_structure表中指定file_id的行，仅修改revision_id字段，ON CONFLICT子句保证了网络重试时的幂等性，但存在一个问题：当源行不存在时，SQL会静默执行成功，无法及时发现异常。同时需要保证一次性批量发送10000条此类命令时的性能不受影响。

解决方案 #

方案一：单语句WITH子句实现检测与插入 #

通过PostgreSQL的WITH子句，在同一语句中完成源行查询、插入操作和结果判断，避免多次数据库往返，保证批量性能。

修改后的SQL与C#代码：

var cmd = new NpgsqlBatchCommand(@"
WITH source AS (
    SELECT manual_code, file_path, file_size, file_modified
    FROM dc_file_structure
    WHERE file_id = $1
),
inserted AS (
    INSERT INTO dc_file_structure (manual_code, revision_id, file_path, file_size, file_modified)
    SELECT manual_code, $2, file_path, file_size, file_modified
    FROM source
    ON CONFLICT (manual_code, revision_id, file_path) DO NOTHING
    RETURNING 1
)
SELECT 
    CASE WHEN (SELECT COUNT(*) FROM source) = 0 THEN 'NOT_FOUND'
         WHEN (SELECT COUNT(*) FROM inserted) = 0 THEN 'CONFLICT'
         ELSE 'SUCCESS' END AS result;");
cmd.Parameters.AddWithValue(CloneExistingFilePk);
cmd.Parameters.AddWithValue(NewRevisionId);

// 执行并处理结果
using var reader = await cmd.ExecuteReaderAsync();
if (reader.Read())
{
    var result = reader.GetString(0);
    if (result == "NOT_FOUND")
    {
        throw new InvalidOperationException($"未找到file_id为{CloneExistingFilePk}的行");
    }
}

优势：无需提前创建数据库对象，每条命令独立完成逻辑，批量发送时性能损耗极低，NpgsqlBatch可高效处理这类批量命令。

方案二：PL/pgSQL函数封装逻辑 #

将检测与插入逻辑封装为数据库函数，批量调用时参数更简洁，性能更优。

1. 创建数据库函数

CREATE OR REPLACE FUNCTION clone_file_revision(p_file_id INT, p_new_revision_id INT)
RETURNS VOID AS $$
DECLARE
    source_count INT;
BEGIN
    SELECT COUNT(*) INTO source_count
    FROM dc_file_structure
    WHERE file_id = p_file_id;
    
    IF source_count = 0 THEN
        RAISE EXCEPTION '未找到file_id为%的行', p_file_id;
    END IF;
    
    INSERT INTO dc_file_structure (manual_code, revision_id, file_path, file_size, file_modified)
    SELECT manual_code, p_new_revision_id, file_path, file_size, file_modified
    FROM dc_file_structure
    WHERE file_id = p_file_id
    ON CONFLICT (manual_code, revision_id, file_path) DO NOTHING;
END;
$$ LANGUAGE plpgsql;

2. C#调用代码

var cmd = new NpgsqlBatchCommand(@"SELECT clone_file_revision($1, $2);");
cmd.Parameters.AddWithValue(CloneExistingFilePk);
cmd.Parameters.AddWithValue(NewRevisionId);

try
{
    await cmd.ExecuteNonQueryAsync();
}
catch (PostgresException ex) when (ex.SqlState == "P0001")
{
    throw new InvalidOperationException(ex.Message, ex);
}

优势：逻辑集中在数据库端，批量调用时参数传递更高效，函数内部操作原子性更强，适合长期复用的场景。

性能说明 #

两种方案均保证了单命令单次数据库交互，NpgsqlBatch批量发送10000条命令时，PostgreSQL可高效处理，不会出现明显性能下降。方案二的函数调用在批量场景下性能略优，因为减少了SQL语句的重复传输量。

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