Connectorx的cx.read_sql()并发更新表时偶现零行或崩溃

可能原因 #

0xC0000005错误本质是内存访问冲突，结合全零行的异常表现，大概率是ConnectorX为追求性能，底层Rust实现的并发数据读取逻辑未妥善处理数据库表的实时更新场景：

• ConnectorX默认的读取策略可能跳过了DBAPI标准的事务一致性保障，直接读取底层数据页，当并发写入修改数据页结构时，容易出现读取到不完整/无效数据（全零行），甚至触发内存访问错误。
• Pandas的read_sql_query()依赖Python DBAPI的事务机制，默认会保证读取数据的一致性，所以不会出现这类问题，但牺牲了速度。

可规避的解决方法 #

• 显式开启事务并设置隔离级别
  在读取前开启事务，利用数据库的事务隔离机制获取一致的数据快照，避免并发写入的干扰。示例代码（以SQL Server为例，其他数据库隔离级别语法略有差异）：

  with cx_con.cursor() as cur:
      cur.execute("SET TRANSACTION ISOLATION LEVEL REPEATABLE READ")
      cur.execute("BEGIN TRANSACTION")
      df = connectorx.read_sql(conn=cx_con, query=f"SELECT * FROM table;")
      cur.execute("COMMIT TRANSACTION")
  

  不同数据库的常用隔离级别：

  • MySQL: REPEATABLE READ（默认）、READ COMMITTED
  • PostgreSQL: REPEATABLE READ、SERIALIZABLE
  • SQL Server: REPEATABLE READ、READ COMMITTED SNAPSHOT

• 添加读取锁（按需使用）
  针对支持表级/行级锁的数据库，读取时加共享锁，避免写入操作修改正在读取的数据。示例（SQL Server）：

  df = connectorx.read_sql(conn=cx_con, query=f"SELECT * FROM table WITH (HOLDLOCK);")
  

  注意：加锁会降低写入性能，仅在数据一致性优先级高于写入吞吐量时使用。

• 异常捕获与重试机制
  针对偶发的全零行或崩溃问题，添加重试逻辑，检测异常结果后重新读取：

  import traceback
  
  def read_with_retry(max_retries=3):
      for _ in range(max_retries):
          try:
              df = connectorx.read_sql(conn=cx_con, query=f"SELECT * FROM table;")
              # 检查是否全零行（排除表本身无数据的情况）
              if not df.empty and (df == 0).all().all():
                  continue
              return df
          except Exception as e:
              tb_str = traceback.format_exc()
              if "Process finished with exit code -1073741819" in tb_str:
                  continue
              raise
      raise RuntimeError("多次重试读取仍失败")
  

• 升级ConnectorX到最新版本
  这类并发场景的bug可能在后续版本中被修复，执行pip install --upgrade connectorx更新到最新版后测试是否解决问题。

• 分批读取数据
  将大表拆分为多个小批次读取，减少单次读取时与并发写入的冲突概率，示例（按自增主键分块）：

  import pandas as pd
  
  batch_size = 10000
  max_id = connectorx.read_sql(conn=cx_con, query="SELECT MAX(id) FROM table;").iloc[0,0]
  dfs = []
  for start in range(0, max_id, batch_size):
      end = start + batch_size
      df_batch = connectorx.read_sql(conn=cx_con, query=f"SELECT * FROM table WHERE id BETWEEN {start} AND {end};")
      dfs.append(df_batch)
  df = pd.concat(dfs, ignore_index=True)
  

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