Ingesting data

Applies to: Embedded ✅, Enterprise ✅ Ingestion brings data into a LanceDB table. Start by choosing a schema and creating the table, then choose a write operation according to how incoming records relate to data already stored.

Choose a write operation

Task	Operation	What happens
Create a table	`create_table` / `createTable`	Create a table from data or an explicit schema.
Append records	`add`	Add rows without matching existing keys.
Upsert by key	`merge_insert` / `mergeInsert`	Choose whether matching rows are updated and new keys inserted.
Update selected rows	`update`	Change values in rows selected by a predicate.
Delete selected rows	`delete`	Remove matching rows from the current table contents.

Appending the same input twice can create duplicate records. Use merge insert with a stable key when the incoming data should update existing records.

Connect and prepare the data

Use the connection guide to connect to Embedded or Enterprise. The SDK write operations in this section work with both. Enterprise also supports SQL write statements; an SDK filter expression does not require a separate SQL connection. Choose the column types, required fields and vector dimensions before a large load. SDKs support different input representations, including row objects and Arrow batches. The table creation guide covers those choices.

Try a complete write workflow

This example starts with Alice and Bob, appends Charlie, updates Bob’s name and login count, then merges a batch containing Charlie and Dana. Finally, it selects Dana’s row for inspection and deletes it.

Complete examples: create, append, update, upsert and delete

Each program creates a temporary local database, performs the five operations, prints the result, and removes its temporary data when finished.

Python: install lancedb, save the code as ingestion.py, and run python ingestion.py.
TypeScript: install @lancedb/lancedb, save the code as ingestion.mts, and run node ingestion.mts with Node.js 22.18 or later.
Rust: download the complete Rust project, extract it, and run cargo run --locked. It includes src/main.rs, the Cargo manifest and a lockfile for the example’s dependencies. The code is shown below.

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)

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 });
}

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(())
}

After the operations, Alice has 10 logins, Bobby has 21, and Charlie has 6. Dana is selected for deletion and is absent from the final result.

Verify the result

After a write completes, inspect the affected keys and fields and check the row count. SDK return values differ by operation and language; use the relevant API reference when relying on returned counts or version metadata. Readers with older snapshots may not see another handle’s writes immediately. See Consistency. A table checked out at a historical version must return to the latest version before it can be modified.

Load larger datasets

Use batches or file-backed input when the dataset is too large to hold in memory. The insert guide covers input formats, progress reporting and write parallelism. Keep retry behavior in mind: repeating an append is not the same as retrying a key-based upsert.

Introduction

Getting Started

Concepts

Data Ingestion

Search

Data Curation

Feature Engineering

Training

Management

Maintenance

Security

Deployment

SQL Reference

Choose a write operation

Connect and prepare the data

Try a complete write workflow

Verify the result

Load larger datasets

​Choose a write operation

​Connect and prepare the data

​Try a complete write workflow

​Verify the result

​Load larger datasets

Choose a write operation

Connect and prepare the data

Try a complete write workflow

Verify the result

Load larger datasets