本页尚未翻译

本页内容仍为英文原文，源自官方文档 ingestion/realtime-write。中文翻译正在分批进行中。

RTE Write (Upsert)

Enable the real time engine on a table and absorb small, frequent writes cheaply, each one committed in milliseconds.

Applies to: Embedded ✅, Enterprise ✅

The real time engine puts an LSM-style write tier in front of a table, turning many small writes into a few large commits. This page covers enabling it and writing to a table; for querying an RTE table, see RTE reads.

table.enable_rte(spec) is the same call on both connections, but what it does underneath differs:

• Embedded runs Lance's MemWAL ShardWriter directly in your process -- the WAL buffer, memtables and SSTables all live wherever your table does, and there is no separate serving layer to scale or configure.
• Enterprise forwards the call to the server, which runs it on a dedicated, horizontally scalable WAL-node tier instead of your table's query-serving fleet. See Distributed RTE Write for how that tier shards a table's writes and the throughput limits that come with it.

Getting started

1. Create a table with a primary key

RTE deduplicates writes by primary key. Create the table with a non-nullable key column, then declare it with set_unenforced_primary_key. The key is a single int32, int64, string, or binary column, and cannot be changed once set.

The table below holds frames captioned and embedded as they stream in from a live camera feed, such as a robot recording episodes into a dataset shaped like LeRobotDataset's frames table — frame_id is the primary key so a re-captioned frame upserts in place:

from datetime import timedelta

import numpy as np
import pyarrow as pa
from lancedb import RTEWriteSpec
from lancedb.index import FTS, BTree, IvfPq

# Embedded: db = lancedb.connect("./my_db")
db = lancedb.connect("db://my_db", api_key="...", host_override="https://...")

dim = 768
schema = pa.schema([
    pa.field("frame_id", pa.int64(), nullable=False),
    pa.field("video_id", pa.utf8(), nullable=False),
    pa.field("embedding", pa.list_(pa.float32(), dim)),
    pa.field("caption", pa.utf8()),
])
frames = db.create_table("frames", schema=schema)
frames.set_unenforced_primary_key("frame_id")

import * as arrow from "apache-arrow";

const dim = 768;
const schema = new arrow.Schema([
  new arrow.Field("frame_id", new arrow.Int64(), false),
  new arrow.Field("video_id", new arrow.Utf8(), false),
  new arrow.Field(
    "embedding",
    new arrow.FixedSizeList(dim, new arrow.Field("item", new arrow.Float32(), true)),
    true,
  ),
  new arrow.Field("caption", new arrow.Utf8(), true),
]);

const frames = await db.createEmptyTable("frames", schema);
await frames.setUnenforcedPrimaryKey("frame_id");

use std::sync::Arc;

use arrow_schema::{DataType, Field, Schema};

let dim = 768;
let schema = Arc::new(Schema::new(vec![
    Field::new("frame_id", DataType::Int64, false),
    Field::new("video_id", DataType::Utf8, false),
    Field::new(
        "embedding",
        DataType::FixedSizeList(Arc::new(Field::new("item", DataType::Float32, true)), dim),
        true,
    ),
    Field::new("caption", DataType::Utf8, true),
]));

let frames = db.create_empty_table("frames", schema).execute().await?;
frames.set_unenforced_primary_key(["frame_id"]).await?;

Or in SQL:

ALTER TABLE frames SET UNENFORCED PRIMARY KEY (frame_id)

2. Create indexes

By default, RTE maintains every supported index on the table, including indexes created after it is enabled. A scalar index on the primary key is required for partial updates and deletes.

frames.create_index("embedding", config=IvfPq())
frames.create_index("caption", config=FTS())
frames.create_index("frame_id", config=BTree())
frames.wait_for_index(["embedding_idx", "caption_idx", "frame_id_idx"], timeout=timedelta(minutes=5))

3. Enable RTE

frames.enable_rte(RTEWriteSpec.unsharded())

await frames.enableRTE({ specType: "unsharded" });

use lancedb::table::RTEWriteSpec;

frames.enable_rte(RTEWriteSpec::unsharded()).await?;

Or in SQL:

ALTER TABLE frames SET RTE

4. Write

注意

On an RTE table, an upsert through merge_insert is the only way to write rows. Use the form below exactly: when_matched_update_all() with no filter, plus when_not_matched_insert_all(). Other merge_insert forms, such as conditional updates or deleting rows missing from the source, are rejected.

Writes go through merge_insert on the primary key, and are visible to queries immediately — see RTE reads for querying them back:

new_frames = pa.table({
    "frame_id": pa.array([2000, 2001], pa.int64()),
    "video_id": pa.array(["clip-9", "clip-9"]),
    "embedding": pa.FixedSizeListArray.from_arrays(
        pa.array(np.random.rand(2 * dim).astype("float32")), dim),
    "caption": pa.array(["a gripper closing on a cup", "a cup set down on the table"]),
}, schema=schema)

frames.merge_insert("frame_id") \
    .when_matched_update_all() \
    .when_not_matched_insert_all() \
    .execute(new_frames)

Partial updates. Send the primary key plus only the columns you want to change; omitted columns keep their current values. This requires a scalar index on the primary key. A re-captioning job, for example, only touches caption:

patch = pa.table(
    {"frame_id": [2000, 2001], "caption": ["a two-finger gripper closing on a cup", "the cup resting upright"]},
    schema=pa.schema([schema.field("frame_id"), schema.field("caption")]),
)
frames.merge_insert("frame_id").when_matched_update_all().when_not_matched_insert_all().execute(patch)

Deletes. The predicate must bind the primary key to literal values. Ranges, OR, and filters on other columns are rejected. Deletes require a scalar index on the primary key.

frames.delete("frame_id IN (2000, 2001)")

Disabling RTE

Stop writes and wait for in-flight writes to finish, then flush pending writes into the base table and disable. checkpoint_rte() only drains writes that existed when it started, so anything written during the checkpoint can be left behind. disable_rte() is rejected while the write-ahead log still holds rows, so run checkpoint_rte() first.

frames.checkpoint_rte()
frames.disable_rte()

Write throughput

RTE acknowledges a write as soon as it is durable in the write-ahead log, so ingest is no longer limited by how fast the table can commit. On Embedded, durability means your local disk or object store, so throughput is bounded by your own machine's resources.

An Enterprise table can be split into shards that write independently, with each shard's durable-append throughput bounded by its WAL node's own capacity -- measured, not an enforced cap. See Distributed RTE Write for the numbers.

Next steps

• RTE reads — query an RTE table
• Distributed RTE Write — how Enterprise shards and scales this
• How the real time engine works
• Merge insert
• ALTER TABLE ... SET RTE