如何在Slick中批量插入或更新多行，无主键时按外键匹配执行upsert

前置准备 #

首先在数据库层为container、containerOrder（对应模型里的order字段）、vendor三个字段添加唯一联合索引，避免并发场景下插入重复数据，SQL参考（MySQL为例）：

ALTER TABLE order_calculation 
ADD UNIQUE KEY uk_container_order_vendor (container, containerOrder, vendor);

方案1：逐行处理（适合小数据量，逻辑简单） #

该方案逻辑易懂，无需适配不同数据库语法，适合单次处理数据量不超过百级的场景：

// 替换为你实际使用的数据库对应的Profile
import slick.jdbc.MySQLProfile.api._

def insertOrUpdateCalculations(calculations: Seq[OrderCalculationSummary]): Future[Seq[OrderCalculation]] = {
  val actions = calculations.map { calc =>
    // 构造三个唯一字段的匹配规则
    val matchQuery = orderCalculationTbl.filter { row =>
      row.container === calc.containerId &&
      row.order === calc.orderId &&
      row.vendor === calc.vendorId
    }

    // 先执行更新操作
    matchQuery.map(r => (r.volume, r.weight))
      .update((calc.volume, calc.weight))
      .flatMap { updateCnt =>
        if (updateCnt > 0) {
          // 更新成功直接返回匹配到的记录
          matchQuery.result.head
        } else {
          // 无匹配记录执行插入，返回带自增ID的新记录
          val newRecord = OrderCalculation(
            id = 0, // 自增ID占位，插入后会替换为真实值
            container = calc.containerId,
            order = calc.orderId,
            vendor = calc.vendorId,
            volume = calc.volume,
            weight = calc.weight
          )
          (orderCalculationTbl returning orderCalculationTbl.map(_.id)) += newRecord
            .map(generatedId => newRecord.copy(id = generatedId))
        }
      }
  }
  // 所有操作组合成事务执行，避免部分成功部分失败
  db.run(DBIO.sequence(actions).transactionally)
}

注意：该方案存在竞态条件，并发场景下如果两个请求同时查询到无匹配记录，同时插入会触发唯一索引冲突报错，对并发要求高的场景建议使用方案2。

方案2：数据库原生批量Upsert（适合大数据量、高并发场景） #

利用数据库原生的Upsert语法实现原子操作，性能和安全性更高，以下分别给出主流数据库的实现参考：

MySQL版本（基于ON DUPLICATE KEY UPDATE） #

import slick.jdbc.MySQLProfile.api._
import scala.concurrent.ExecutionContext.Implicits.global

def insertOrUpdateCalculations(calculations: Seq[OrderCalculationSummary]): Future[Seq[OrderCalculation]] = {
  if (calculations.isEmpty) return Future.successful(Nil)
  
  // 批量执行Upsert
  val upsertAction = sqlu"""
    INSERT INTO order_calculation (container, containerOrder, vendor, volume, weight)
    VALUES #${calculations.map(c => s"(${c.containerId}, ${c.orderId}, ${c.vendorId}, ${c.volume}, ${c.weight})").mkString(",")}
    ON DUPLICATE KEY UPDATE volume = VALUES(volume), weight = VALUES(weight)
  """

  // 执行完Upsert后查询所有匹配的记录返回
  val queryAction = orderCalculationTbl.filter { row =>
    calculations.map { calc =>
      row.container === calc.containerId &&
      row.order === calc.orderId &&
      row.vendor === calc.vendorId
    }.reduce(_ || _)
  }.result

  // 两个操作放在同一个事务中执行
  db.run((upsertAction >> queryAction).transactionally)
}

如果担心SQL拼接有注入风险，可替换为Slick预编译批量参数绑定写法。

PostgreSQL版本（基于ON CONFLICT DO UPDATE） #

只需要替换上方代码中的upsertAction部分即可：

val upsertAction = sqlu"""
  INSERT INTO order_calculation (container, containerOrder, vendor, volume, weight)
  VALUES #${calculations.map(c => s"(${c.containerId}, ${c.orderId}, ${c.vendorId}, ${c.volume}, ${c.weight})").mkString(",")}
  ON CONFLICT (container, containerOrder, vendor) DO UPDATE 
  SET volume = EXCLUDED.volume, weight = EXCLUDED.weight
"""

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