如何在DuckDB中以列方式追加数据且不加载至内存

我有需要存储到磁盘数据库的输入数据，示例数据生成函数如下：

# 示例数据生成函数
incoming_data <- function(ncol=5){
  dat <- sample(1:10,100,replace = T) |> matrix(ncol = ncol) |> as.data.frame()
  random_names <- sapply(1:ncol(dat),\(x) paste0(sample(letters,1), sample(1:100,1)))
  colnames(dat) <- random_names
  dat
}

# 生成示例数据
incoming_data()

实际场景中，单批输入数据包含5k行、约50k列，最终数据总规模达200-400GB。我希望实现类似my_dat <- cbind("my_dat", incoming_data())的横向拼接效果，即在DuckDB的my_dat表中追加新列，且无需将整个表加载至内存。现有初始化代码如下：

# 现有初始化代码
path <- "D:\\R_scripts\\new\\duckdb\\data\\DB.duckdb"
library(duckdb)
con <- dbConnect(duckdb(), dbdir = path, read_only = FALSE)
# 写入初始数据到数据库
dbWriteTable(con, "my_dat", incoming_data())

实现步骤 #

由于直接加载大表到内存执行cbind不现实，我们可以利用DuckDB的磁盘级操作特性，通过临时表中转+分步骤增列更新的方式实现需求，核心是避免全量加载原表。

1. 初始化表时添加行号主键

首先修改初始化代码，给表添加row_id列作为行唯一标识，确保后续新列能准确对应到每一行：

path <- "D:\\R_scripts\\new\\duckdb\\data\\DB.duckdb"
library(duckdb)
con <- dbConnect(duckdb(), dbdir = path, read_only = FALSE)

# 生成初始数据并添加行号
initial_dat <- incoming_data()
initial_dat$row_id <- seq_len(nrow(initial_dat))

# 写入数据库，并设置行号为主键（加速后续关联）
dbWriteTable(con, "my_dat", initial_dat)
dbExecute(con, "ALTER TABLE my_dat ADD PRIMARY KEY (row_id)")

2. 批量追加新列的通用流程

每次获取新批次数据后，按以下步骤操作：

# 1. 生成新批次数据并添加行号（必须和原表行数完全匹配）
new_dat <- incoming_data()
new_dat$row_id <- seq_len(nrow(new_dat))

# 2. 将新数据写入临时表（仅加载新批次数据到内存，原表无需动）
dbWriteTable(con, "temp_new_cols", new_dat, temporary = TRUE)

# 3. 获取新表中除row_id外的所有列名
new_cols <- dbGetQuery(con, "SELECT column_name FROM information_schema.columns WHERE table_name = 'temp_new_cols' AND column_name != 'row_id'")$column_name

# 4. 给目标表逐个添加新列（自动匹配数据类型）
for (col in new_cols) {
  col_type <- dbGetQuery(con, sprintf("SELECT data_type FROM information_schema.columns WHERE table_name = 'temp_new_cols' AND column_name = '%s'", col))$data_type
  dbExecute(con, sprintf("ALTER TABLE my_dat ADD COLUMN %s %s", col, col_type))
}

# 5. 通过行号关联，将临时表的列值更新到目标表
update_queries <- lapply(new_cols, function(col) {
  sprintf("UPDATE my_dat SET %s = temp_new_cols.%s FROM temp_new_cols WHERE my_dat.row_id = temp_new_cols.row_id", col, col)
})
lapply(update_queries, function(q) dbExecute(con, q))

# 6. 删除临时表释放资源
dbExecute(con, "DROP TABLE temp_new_cols")

关键说明 #

• 全程无需加载原表到内存，所有操作由DuckDB在磁盘上高效执行，适合超大规模数据
• row_id是核心关联键，必须保证每次新批次数据的行数和行号与原表完全一致
• 临时表自动在会话结束后销毁，也可手动删除释放空间
• 分步骤增列和更新，避免一次性操作大表带来的性能压力

内容的提问来源于stack exchange，提问作者mr.T