Redshift从stage表加载数据时table2被锁导致插入失败

• 分批插入，缩短锁持有时间
  大批次一次性插入会让表被长时间锁定，拆分成分批插入能有效减少锁的持有时长。同时修复原代码中未关闭数据库连接的问题（连接泄漏可能导致锁无法释放）：

  # 分批插入逻辑
  batch_size = 10000
  offset = 0
  while True:
      insert_sql = "INSERT INTO table2 SELECT c1,c2,c3,c4 FROM table_name LIMIT %s OFFSET %s"
      load_redshift_table(insert_sql, (batch_size, offset))
      # 检查是否还有剩余数据
      count_sql = "SELECT COUNT(*) FROM table_name OFFSET %s"
      remaining = get_redshift_count(count_sql, (offset,))
      if remaining == 0:
          break
      offset += batch_size
  
  # 改进后的加载函数，支持参数并关闭连接
  def load_redshift_table(sql, params=None):
      conn = None
      cur = None
      try:
          conn = psycopg2.connect(
              host="host",port="redshift_port",database="redshift_database",user="redshift_user",password="reshift_password")
          cur = conn.cursor()
          if params:
              cur.execute(sql, params)
          else:
              cur.execute(sql)
          conn.commit()
      except Exception as e:
          print("Error:", e)
          if conn:
              conn.rollback()
      finally:
          if cur:
              cur.close()
          if conn:
              conn.close()
  
  # 辅助函数：获取剩余数据量
  def get_redshift_count(sql, params=None):
      conn = None
      cur = None
      try:
          conn = psycopg2.connect(
              host="host",port="redshift_port",database="redshift_database",user="redshift_user",password="reshift_password")
          cur = conn.cursor()
          if params:
              cur.execute(sql, params)
          else:
              cur.execute(sql)
          return cur.fetchone()[0]
      except Exception as e:
          print("Count Error:", e)
          return 0
      finally:
          if cur:
              cur.close()
          if conn:
              conn.close()
  

• 改用COPY命令（Redshift最优批量加载方式）
  Redshift的COPY命令专为批量数据加载优化，相比INSERT效率更高，锁表时间极短。如果stage数据在S3，直接用COPY；如果stage是Redshift表，可以先UNLOAD到S3再COPY：

  # 先将stage表数据导出到S3
  unload_sql = """
  UNLOAD ('SELECT c1,c2,c3,c4 FROM table_name')
  TO 's3://your-bucket/path/to/stage-data/'
  IAM_ROLE 'arn:aws:iam::your-account-id:role/your-redshift-role'
  DELIMITER ','
  ALLOWOVERWRITE
  """
  load_redshift_table(unload_sql)
  
  # 再用COPY加载到table2
  copy_sql = """
  COPY table2 (c1,c2,c3,c4)
  FROM 's3://your-bucket/path/to/stage-data/'
  IAM_ROLE 'arn:aws:iam::your-account-id:role/your-redshift-role'
  CSV DELIMITER ','
  """
  load_redshift_table(copy_sql)
  

• 检查并调整事务隔离级别
  避免使用过高的事务隔离级别（如SERIALIZABLE），Redshift默认的READ COMMITTED已经足够，可在连接时显式设置：

  conn = psycopg2.connect(...)
  conn.set_session(isolation_level='READ COMMITTED')
  

• 排查并清理持有锁的长事务
  用以下SQL查询table2的锁持有情况，找出未提交的长事务：

  SELECT relname, locktype, mode, pid, query
  FROM pg_locks l
  JOIN pg_class c ON l.relation = c.oid
  WHERE relname = 'table2';
  

  若发现长时间运行的事务，执行以下命令终止对应的进程：

  SELECT pg_terminate_backend(pid);
  

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