使用独立进程Azure Function向Synapse SQL Pool插入数据时出错

错误信息 #

System.Private.CoreLib: 执行函数时发生异常: Functions.GetWeather. Microsoft.Azure.WebJobs.Host: 函数返回后处理参数_binder时出错:. Microsoft.Azure.WebJobs.Extensions.Sql: upsert和回滚期间遇到异常。(第2行第54列解析错误: 'WITH'附近语法不正确。)(111214;尝试完成事务失败。未找到对应的事务。). Core Microsoft SqlClient Data Provider: 第2行第54列解析错误: 'WITH'附近语法不正确。

相关定义与代码 #

SQL表定义 #

CREATE TABLE WeatherDataV2 
(
    WeatherDataId BIGINT PRIMARY KEY NONCLUSTERED NOT ENFORCED, 
    createdDate DATETIME
);

独立进程Azure Function代码 #

[Function("GetWeather")]
[SqlOutput("dbo.WeatherDataV2", connectionStringSetting: "SqlConnectionString")]
public WeatherData Run([TimerTrigger("%_runEveryHourCron%")] MyInfo myTimer, FunctionContext context)
{
   var weatherDataItem = new WeatherData()
   {
      WeatherDataId = DateTime.Now.Ticks,
      createdDate = DateTime.Now
   };
   return weatherDataItem;
}

public class WeatherData
{
    [Key]
    public long WeatherDataId { get; set; }
    public DateTime createdDate { get; set; }
}

问题原因 #

Azure Functions的SQL Output绑定默认生成的Upsert语句使用了SQL Server特有的WITH (SERIALIZABLE)表提示语法，但Synapse SQL Pool不支持该语法，同时Synapse的事务模型与SQL Server存在差异，导致后续事务回滚失败。

解决方案 #

方案1：改用Insert模式（无主键冲突场景） #

如果业务逻辑能确保WeatherDataId不会重复，可直接将SQL Output绑定改为Insert模式，避免自动生成不兼容的Upsert语句：

[Function("GetWeather")]
// 添加SqlCommandType指定为Insert
[SqlOutput("dbo.WeatherDataV2", connectionStringSetting: "SqlConnectionString", SqlCommandType = SqlCommandType.Insert)]
public WeatherData Run([TimerTrigger("%_runEveryHourCron%")] MyInfo myTimer, FunctionContext context)
{
   var weatherDataItem = new WeatherData()
   {
      WeatherDataId = DateTime.Now.Ticks,
      createdDate = DateTime.Now
   };
   return weatherDataItem;
}

方案2：自定义Synapse兼容的Upsert逻辑（需主键冲突处理） #

若必须处理主键冲突，放弃SQL Output绑定的自动Upsert，手动使用SqlConnection执行Synapse支持的MERGE语句实现Upsert：

using System.Data.SqlClient;
using Microsoft.Azure.Functions.Worker;
using Microsoft.Extensions.Logging;

[Function("GetWeather")]
public void Run([TimerTrigger("%_runEveryHourCron%")] MyInfo myTimer, FunctionContext context)
{
    var logger = context.GetLogger("GetWeather");
    var weatherDataItem = new WeatherData()
    {
        WeatherDataId = DateTime.Now.Ticks,
        createdDate = DateTime.Now
    };

    var connectionString = Environment.GetEnvironmentVariable("SqlConnectionString");
    using var conn = new SqlConnection(connectionString);
    conn.Open();

    // Synapse兼容的MERGE语句
    var mergeSql = @"
MERGE INTO dbo.WeatherDataV2 AS Target
USING (VALUES (@WeatherDataId, @createdDate)) AS Source (WeatherDataId, createdDate)
ON Target.WeatherDataId = Source.WeatherDataId
WHEN MATCHED THEN
    UPDATE SET createdDate = Source.createdDate
WHEN NOT MATCHED THEN
    INSERT (WeatherDataId, createdDate) VALUES (Source.WeatherDataId, Source.createdDate);";

    using var cmd = new SqlCommand(mergeSql, conn);
    cmd.Parameters.AddWithValue("@WeatherDataId", weatherDataItem.WeatherDataId);
    cmd.Parameters.AddWithValue("@createdDate", weatherDataItem.createdDate);
    
    cmd.ExecuteNonQuery();
    logger.LogInformation("Upsert completed successfully");
}

public class WeatherData
{
    public long WeatherDataId { get; set; }
    public DateTime createdDate { get; set; }
}

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