LanceDB 中文文档

外观

写入数据 ​

选择如何建表、追加记录或修改已有数据。

适用： Embedded ✅、Enterprise ✅

写入（ingestion）就是把数据带进一张 LanceDB 表。先确定 schema 并建表，然后根据新记录与已有数据的关系，选择对应的写操作。

选择写操作 ​

任务操作效果
创建表create_table / createTable用数据或显式 schema 建一张表。
追加记录add直接加行，不与已有键做匹配。
按 key 更新插入merge_insert / mergeInsert自行选择命中的行是更新、新键是插入。
更新选中的行update修改被谓词选中的行里的值。
删除选中的行delete从表的当前内容里移除命中的行。

同一批输入追加两次会产生重复记录。如果新数据应当更新已有记录，请用带稳定 key 的 merge insert。

连接并准备数据 ​

连接 Embedded 或 Enterprise 用连接指南。本节里的 SDK 写操作在两者上都能用。Enterprise 另外支持 SQL 写语句；而 SDK 的过滤表达式不需要另开一条 SQL 连接。

在开始大批量导入之前，先定好列类型、必填字段与向量维度。各 SDK 支持的输入形式不同，包括行对象与 Arrow 批次这些选择见建表指南。

跑一遍完整的写入流程 ​

这个示例从 Alice 和 Bob 开始，追加 Charlie，更新 Bob 的名字与登录次数，再合并一批包含 Charlie 和 Dana 的数据。最后把 Dana 那行查出来看一眼，然后删掉。

完整示例：创建、追加、更新、Upsert 与删除

每个程序都会建一个临时本地数据库，依次执行这五种操作，打印结果，跑完清理掉临时数据。

  • Python： 安装 lancedb，把代码存成 ingestion.py，执行 python ingestion.py。
  • TypeScript： 安装 @lancedb/lancedb，把代码存成 ingestion.mts，用 Node.js 22.18 或更高版本执行 node ingestion.mts。
  • Rust： 下载完整的 Rust 工程，解压后执行 cargo run --locked。压缩包里有 src/main.rs、Cargo manifest 以及示例依赖的 lockfile。代码见下。
python
from tempfile import TemporaryDirectory

import lancedb

with TemporaryDirectory(prefix="lancedb-ingestion-") as uri:
    db = lancedb.connect(uri)
    table = db.create_table(
        "users",
        [
            {"id": 1, "name": "Alice", "login_count": 10},
            {"id": 2, "name": "Bob", "login_count": 20},
        ],
    )
    table.add([{"id": 3, "name": "Charlie", "login_count": 5}])
    table.update(where="id = 2", values={"name": "Bobby"})
    table.update(where="id = 2", values_sql={"login_count": "login_count + 1"})
    (
        table.merge_insert("id")
        .when_matched_update_all()
        .when_not_matched_insert_all()
        .execute(
            [
                {"id": 3, "name": "Charlie", "login_count": 6},
                {"id": 4, "name": "Dana", "login_count": 1},
            ]
        )
    )
    print("Rows selected for deletion:", table.search().where("id = 4").to_list())
    table.delete("id = 4")
    rows = sorted(table.search().to_list(), key=lambda row: row["id"])
    print(rows)
typescript
const lancedb = await import("@lancedb/lancedb");
const { mkdtemp, rm } = await import("node:fs/promises");
const { tmpdir } = await import("node:os");
const { join } = await import("node:path");

const uri = await mkdtemp(join(tmpdir(), "lancedb-ingestion-"));
try {
  const db = await lancedb.connect(uri);
  const table = await db.createTable("users", [
    { id: 1, name: "Alice", login_count: 10 },
    { id: 2, name: "Bob", login_count: 20 },
  ]);
  await table.add([{ id: 3, name: "Charlie", login_count: 5 }]);
  await table.update({ where: "id = 2", values: { name: "Bobby" } });
  await table.update({
    where: "id = 2",
    valuesSql: { login_count: "login_count + 1" },
  });
  await table
    .mergeInsert("id")
    .whenMatchedUpdateAll()
    .whenNotMatchedInsertAll()
    .execute([
      { id: 3, name: "Charlie", login_count: 6 },
      { id: 4, name: "Dana", login_count: 1 },
    ]);
  console.log(
    "Rows selected for deletion:",
    await table.query().where("id = 4").toArray(),
  );
  await table.delete("id = 4");
  const rows = (await table.query().toArray()).sort((a, b) => a.id - b.id);
  console.log(rows);
} finally {
  await rm(uri, { recursive: true, force: true });
}
rust
use std::sync::Arc;

use arrow_array::{Int64Array, RecordBatch, RecordBatchIterator, RecordBatchReader, StringArray};
use arrow_schema::{DataType, Field, Schema};
use futures::TryStreamExt;
use lancedb::query::{ExecutableQuery, QueryBase};

fn make_users_reader(rows: &[(i64, &str, i64)]) -> Box<dyn RecordBatchReader + Send> {
    let schema = Arc::new(Schema::new(vec![
        Field::new("id", DataType::Int64, false),
        Field::new("name", DataType::Utf8, false),
        Field::new("login_count", DataType::Int64, false),
    ]));
    let batch = RecordBatch::try_new(
        schema.clone(),
        vec![
            Arc::new(Int64Array::from_iter_values(rows.iter().map(|r| r.0))),
            Arc::new(StringArray::from_iter_values(rows.iter().map(|r| r.1))),
            Arc::new(Int64Array::from_iter_values(rows.iter().map(|r| r.2))),
        ],
    )
    .unwrap();
    Box::new(RecordBatchIterator::new(
        vec![Ok(batch)].into_iter(),
        schema,
    ))
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let directory = tempfile::tempdir()?;
    let db = lancedb::connect(directory.path().to_str().unwrap())
        .execute()
        .await?;
    let table = db
        .create_table(
            "users",
            make_users_reader(&[(1, "Alice", 10), (2, "Bob", 20)]),
        )
        .execute()
        .await?;
    table
        .add(make_users_reader(&[(3, "Charlie", 5)]))
        .execute()
        .await?;
    table
        .update()
        .only_if("id = 2")
        .column("name", "'Bobby'")
        .execute()
        .await?;
    table
        .update()
        .only_if("id = 2")
        .column("login_count", "login_count + 1")
        .execute()
        .await?;
    let mut merge = table.merge_insert(&["id"]);
    merge
        .when_matched_update_all(None)
        .when_not_matched_insert_all();
    merge
        .execute(make_users_reader(&[(3, "Charlie", 6), (4, "Dana", 1)]))
        .await?;
    let selected: Vec<RecordBatch> = table
        .query()
        .only_if("id = 4")
        .execute()
        .await?
        .try_collect()
        .await?;
    println!("Rows selected for deletion: {selected:?}");
    table.delete("id = 4").await?;
    let rows: Vec<RecordBatch> = table.query().execute().await?.try_collect().await?;
    println!("Remaining rows: {rows:?}");
    assert_eq!(table.count_rows(None).await?, 3);
    assert_eq!(
        table
            .count_rows(Some(
                "id = 2 AND name = 'Bobby' AND login_count = 21".into()
            ))
            .await?,
        1
    );
    assert_eq!(
        table
            .count_rows(Some("id = 3 AND login_count = 6".into()))
            .await?,
        1
    );
    Ok(())
}

这些操作跑完之后，Alice 有 10 次登录，Bobby 有 21 次，Charlie 有 6 次。Dana 被选中删除，最终结果里没有它。

校验结果 ​

一次写入完成后，检查受影响的键与字段，并核对行数。SDK 的返回值随操作和语言而异；如果你要依赖返回的计数或版本元数据，请查对应的 API 参考。

持有旧快照的读取方，可能不会立刻看到另一个连接句柄的写入，详见一致性。一张被 checkout 到历史版本的表，必须先回到最新版本才能被修改。

加载更大的数据集 ​

当数据集大到放不进内存时，改用批次或文件型输入。插入指南里讲了输入格式、进度上报与写入并行度。也要注意重试语义：重复一次追加，和重试一次基于 key 的 upsert，并不是一回事。