如何手动实现Try Insert和Insert Or Update

Posted 2021-03-22 ripley

在日常开发中，我们有时会需要对数据的插入操作进行定制。比如，如果表里已有某某记录就不写入新纪录，或者表里没该记录就插入，否则就更新。前者我们称为TryInsert，后者为InsertOrUpdate(也叫做upsert)。一般来说，很多orm框架都会附带这样的函数，但是如果你要批量插入数据，orm自带的函数就不太够用了。下面我们从手动拼SQL的角度来实现TryInsert和InsertOrUpdate。

考虑到现在流行的两大开源RDBMS对SQL标准支持比较落后，而早期的标准并没有这方面的标准语法，所以我们分成mysql篇和Postgres篇来分别使用它们各自的方言解决上面提到的两个问题。

MySQL篇

原理解析

insert ignore into

插入如果报错（主键或者Unique键重复），会把错误转成警告，此时返回的影响行数为0，可以用来实现TryInsert()。

replace into

replace跟insert语法基本一致，是Mysql的扩展语法，官方的InsertOrUpdate，replace语句的基本逻辑如下：

ok:=Insert()
if !ok {
  if duplicate-key {  // key重复就删掉重新插入
    Delete()
    Insert()
  }
}

从这里我们可以看出replace语句的影响行数，如果是插入，影响行数为1；如果是更新，删除再插入，影响行数为2。

Insert into ... on duplicate key update

也是MySQL扩展语法。... on duplicate key update的逻辑与replace差不多，唯一的区别就是如果插入的新值与旧值一样，默认返回的影响行数为0，所以这里的逻辑是如果新值和旧值相同就不作处理。

代码示例

下面是以golang为例，给出示例：

type User struct {
  UserID   int64  `gorm:"user_id"`
  Username string   `gorm:"username"`
  Password string   `gorm:"password"`
  Address  string   `gorm:"address"`
}

func BulkTryInsert(data []*User) error{
  str:=make([]string, 0, len(data))
  param:=make([]interface{},0,len(data)*4)  // 4个属性
  for _,d:=range data {
    str=append(str,"(?,?,?,?)")
    param=append(d.UserID)
    param=append(d.Username)
    param=append(d.Password)
    param=append(d.Address)
  }
  stmt:=fmt.Sprintf("INSERT IGNORE INTO table_name(user_id,username,password,address) VALUES %s",strings.Join(str,",") )
  return DB.Exec(stmt, param...).Error
}

func BulkUpsert(data []*User) error{
  str:=make([]string, 0, len(data))
  param:=make([]interface{},0,len(data)*4)  // 4个属性
  for _,d:=range data {
    str=append(str,"(?,?,?,?)")
    param=append(d.UserID)
    param=append(d.Username)
    param=append(d.Password)
    param=append(d.Address)
  }
  stmt:=fmt.Sprintf("REPLACE INTO table_name(user_id,username,password,address) VALUES %s",strings.Join(str,",") )    // 与上面的区别仅在这行的SQL
  return DB.Exec(stmt, param...).Error
}

Postgres篇

原理解析

Insert into ... on conflict (...) do nothing

on conflict后面需要带上冲突的键，比如主键或者Unique约束。这条SQL的意思就如字面所示，当某某键存在重复冲突的时候，什么也不做，即TryInsert。

Insert into ... on conflict (...) do update set (...)

这条SQL就比较复杂了，Postgres这个语法表面上看比MySQL自由度更高，实际上非常繁琐笨重，不如MySQL务实。set的意思是，冲突时需要指定更新哪些属性，这是强制的，必须具体地说明每个字段，真是不友好啊。大概是要写成这样，其中EXCLUDED指代要插入的那条记录：

INSERT INTO ... on conflict (user_id, address) do update set password=EXCLUDED.password and username=EXCLUDED.username

代码示例

这次我们设想一种实用的场景，python经常被用作科学计算，pandas是大家偏爱的计算包，pandas的io部分提供了傻瓜式的读写文件和数据库里数据的函数，比如写数据库的to_sql，但是这个函数有局限性，它只能做到TryInsert和清空表数据再插入，对于upsert则无能为力。目前来说，我们只能手动实现它。

按照上面的解析，我们需要给每张表设置好UniqueConstraint才能使用这个语法。下面给出一个例子：

# 使用的是sqlalchemy
Base = declarative_base()

# 将一个list分割成m个大小为n的list
def chunks(a, n):
    return [a[i:i + n] for i in range(0, len(a), n)]

class DBUser(Base):
  __tablename__ = 'user' # UniqueConstraint和PrimaryKey至少要有一个
  __table_args__ = (UniqueConstraint('user_id', 'address'), 
                   {'schema': 'db'})
  user_id = Column(BigInteger)
  username = Column(String(200))
  password = Column(String(200))
  address = Column(String(200))
  
  def dtype(self): # pandas需要的dtype
    d = {c.name: c.type for c in self.__table__.c}
    if 'id' in d:
        el d['id']   # 一般id都是自动生成的，提供给pandas的dtype应该剔除id
    return d
  
  def fullname(self):
    return self.__table_args__[-1]['schema'] + '.' + self.__tablename__
  
  # 只要DBUser再提供一个Unique Constraint的属性列表，下面这两个函数就可以写成通用的函数
  # 这里只是给出例子，点到为止
  def bulk_try_insert(self, engine, data):
    col = self.dtype().keys()
    col_str = ','.join(col)
    col_str = '(' + col_str + ')'
    update_col = []
    for c in col:
      update_str = '{0}=EXCLUDED.{1}'.format(c, c)
      update_col.append(update_str)
    value_str = []
    value_args = []
    for d in data:
      tmp_str = '(' + col.__len__() * '%s,'
      tmp_str = tmp_str[:-1] + ')'
      value_str.append(tmp_str)
      for k in col:
        value_args.append(d[k])
    
    stmt= 'insert into ' + self.fullname() + col_str + 'values ' + ','.join(
      value_str) + 'on conflict (user_id, address) do update set ' + ",".join(update_col)
    engine.execute(stmt, value_args)
  
  def bulk_insert_chunk(self, engine, data, n=1000):
    d_list = chunks(data, n)
    for a in d_list:
      self.bulk_insert(engine, a)