diff --git a/docs/web/docs.json b/docs/web/docs.json index a8fb44007..9c9468a3b 100644 --- a/docs/web/docs.json +++ b/docs/web/docs.json @@ -327,6 +327,7 @@ "management/namespace", "management/table", "tables/branching", + "tables/realtime-engine", "management/materialized-view", "management/view" ] diff --git a/docs/web/feature-engineering/materialized-views.mdx b/docs/web/feature-engineering/materialized-views.mdx index ffe4fc959..1552c3638 100644 --- a/docs/web/feature-engineering/materialized-views.mdx +++ b/docs/web/feature-engineering/materialized-views.mdx @@ -52,7 +52,7 @@ In Python SDK 0.40.0, create the source with above does. Stable row IDs let the view track source rows across updates and compaction. Set this option when creating the table; it cannot be added later. -A source with a MemWAL write spec or retained, uncompacted write-ahead-log +A source with the [realtime engine](/tables/realtime-engine) enabled or retained, uncompacted write-ahead-log rows is not accepted by the local materialized-view implementation. ## Refresh behavior {#incremental} diff --git a/docs/web/maintenance/background-jobs.mdx b/docs/web/maintenance/background-jobs.mdx index dcfa31321..1a710fe79 100644 --- a/docs/web/maintenance/background-jobs.mdx +++ b/docs/web/maintenance/background-jobs.mdx @@ -27,13 +27,13 @@ also start cache-warming jobs through the admin API. ## Write-ahead log maintenance {#wal} -For a table with a [write-ahead log](/sql/ddl/table#properties) enabled, two jobs keep the +For a table with the [realtime engine](/tables/realtime-engine) enabled, two jobs keep the log from growing without bound: | Job type | What it does | | --- | --- | -| `sstable_compaction` | Folds a table shard's SSTables into the base table by primary key. Distinct from `compact`, which merges base-table fragments. The two are unrelated. | -| `wal_index_catchup` | Extends a base-table index over the rows `sstable_compaction` just folded in, and records how far the index has caught up. Nothing else advances that record. | +| `memwal_compaction` | Folds a table shard's SSTables into the base table by primary key. Distinct from `compact`, which merges base-table fragments. The two are unrelated. | +| `memwal_catchup_` | Extends a base-table index over the rows `memwal_compaction` just folded in, and records how far the index has caught up. Nothing else advances that record. The suffix names the index type, for example `memwal_catchup_btree`, `memwal_catchup_fts`, or `memwal_catchup_ivfpq`. | ## Index maintenance {#index} diff --git a/docs/web/sql/ddl/table.mdx b/docs/web/sql/ddl/table.mdx index ecb522fe6..792568220 100644 --- a/docs/web/sql/ddl/table.mdx +++ b/docs/web/sql/ddl/table.mdx @@ -102,18 +102,20 @@ duplicates impossible. It tells the planner what to assume. The column must be declared `NOT NULL`. ```sql -ALTER TABLE t SET MEMWAL -ALTER TABLE t SET MEMWAL WITH (sharding = 'bucket', column = 'id', buckets = 4) -ALTER TABLE t SET MEMWAL WITH (maintained_indexes = '') -ALTER TABLE t UNSET MEMWAL +ALTER TABLE t SET RTE +ALTER TABLE t SET RTE WITH (sharding = 'bucket', column = 'id', buckets = 4) +ALTER TABLE t SET RTE WITH (maintained_indexes = '') +ALTER TABLE t UNSET RTE ``` -A write-ahead log makes small frequent writes cheap by batching them out of the +The [realtime engine](/tables/realtime-engine) puts a write-ahead log in front +of the table, making small frequent writes cheap by batching them out of the commit path. `sharding` and `buckets` spread the log so concurrent writers do not contend on one place; `bucket` sharding names its `column`, which must be the table's unenforced primary key, and inserts into the table then need a scalar index on that key. `maintained_indexes` chooses which indexes are kept -current against WAL rows. A table with a computed column cannot take a +current against WAL rows; omitted, every supported index is kept current, +including ones created later. A table with a computed column cannot take a write-ahead log. See [Background jobs](/maintenance/background-jobs#wal) for how the log gets folded back into the table. diff --git a/docs/web/tables/realtime-engine.mdx b/docs/web/tables/realtime-engine.mdx new file mode 100644 index 000000000..88738791c --- /dev/null +++ b/docs/web/tables/realtime-engine.mdx @@ -0,0 +1,229 @@ +--- +title: "Realtime engine (RTE)" +sidebarTitle: "Realtime Engine" +description: "Absorb small, frequent writes cheaply and make them queryable within milliseconds." +icon: "bolt" +keywords: ["rte", "realtime engine", "wal", "memwal", "lsm", "write-ahead log", "streaming writes", "upsert"] +--- + +Enterprise + +## What is the realtime engine? {#what-is-rte} + +Every write to a LanceDB table is normally a commit to the underlying Lance +dataset. That is ideal for bulk loads, but a stream of small upserts becomes a +stream of small commits, and each write is only visible once its commit lands. + +The realtime engine (RTE) puts an LSM-style write tier in front of the table. +It turns many small writes into a few large commits, and makes each write +**searchable within milliseconds** of being acknowledged. It is enabled per +table, and nothing about how you query that table changes. + +## How it works {#how-it-works} + +- **Writes.** A `merge_insert` or `delete` lands in an in-memory **memtable** + and is acknowledged once it is durable in the write-ahead log on object + storage. +- **Flush and compaction.** Memtables freeze into **SSTables**, which background + jobs compact into the base table, catching its indexes up as they go. See + [Background jobs](/maintenance/background-jobs#wal). +- **Reads.** Every query runs against the base table and the fresh tier + (memtables and SSTables) in parallel. Fresh rows supersede base rows with the + same primary key, and the two are merged into one result. +- **Indexes stay fresh.** Vector, full-text, BTree, bitmap, and most other + scalar indexes are maintained on the memtable, so new rows are searchable + through the index as soon as they are written. + +## Getting started {#getting-started} + +### 1. Create a table with a primary key {#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. + +```python +from datetime import timedelta + +import lancedb +import numpy as np +import pyarrow as pa +from lancedb import RTEWriteSpec +from lancedb.index import FTS, BTree, IvfPq + +db = lancedb.connect("db://my_db", api_key="...", host_override="https://...") + +n, dim = 1024, 128 +schema = pa.schema([ + pa.field("id", pa.int64(), nullable=False), + pa.field("vector", pa.list_(pa.float32(), dim)), + pa.field("text", pa.utf8()), +]) +initial_data = pa.table({ + "id": pa.array(range(n), pa.int64()), + "vector": pa.FixedSizeListArray.from_arrays( + pa.array(np.random.rand(n * dim).astype("float32")), dim), + "text": pa.array([f"doc {i} alpha bravo" for i in range(n)]), +}, schema=schema) +table = db.create_table("docs", initial_data, schema=schema) +table.set_unenforced_primary_key("id") +``` + +Or in SQL: + +```sql +ALTER TABLE docs SET UNENFORCED PRIMARY KEY (id) +``` + +### 2. Create indexes {#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. + +```python +table.create_index("vector", config=IvfPq()) +table.create_index("text", config=FTS()) +table.create_index("id", config=BTree()) +table.wait_for_index(["vector_idx", "text_idx", "id_idx"], timeout=timedelta(minutes=5)) +``` + +### 3. Enable RTE {#enable} + +```python +table.enable_rte(RTEWriteSpec.unsharded()) +``` + +Or in SQL: + +```sql +ALTER TABLE docs SET RTE +``` + +### 4. Write and query {#write-and-query} + + +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: + +```python +new_rows = pa.table({ + "id": pa.array([2000, 2001], pa.int64()), + "vector": pa.FixedSizeListArray.from_arrays( + pa.array(np.random.rand(2 * dim).astype("float32")), dim), + "text": pa.array(["fresh charlie delta", "fresh echo foxtrot"]), +}, schema=schema) +query_vector = np.random.rand(dim).astype("float32") + +table.merge_insert("id") \ + .when_matched_update_all() \ + .when_not_matched_insert_all() \ + .execute(new_rows) + +table.search(query_vector).limit(10).to_list() +table.search("charlie").limit(10).to_list() +``` + +**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. + +```python +patch = pa.table( + {"id": [0, 1], "text": ["doc 0 revised", "doc 1 revised"]}, + schema=pa.schema([schema.field("id"), schema.field("text")]), +) +table.merge_insert("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. + +```python +table.delete("id IN (2000, 2001)") +``` + +### Disabling RTE {#disable} + +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. + +```python +table.checkpoint_rte() +table.disable_rte() +``` + +## Performance {#performance} + +### Writes {#write-performance} + +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. + +**Compared with writing directly to the table.** On tables ranging from +hundreds of thousands to tens of millions of rows, RTE delivered 4–12× the +write throughput and 24–52× lower p99 write latency, depending on the mix of +inserts, updates, and deletes. + +**Scaling with shards.** An RTE table can be split into +[shards](/sql/ddl/table#properties) that write independently, and a single +table's ingest grows with its shard count. In our testing on AWS, with every +index on the table (vector, full-text, and BTree) kept current, a single table +reached these peak rates: + +| Shards | Peak ingest | +| --: | --: | +| 1 | 90 MB/s | +| 2 | 170 MB/s | +| 4 | 340 MB/s | +| 8 | 625 MB/s | +| 16 | 690 MB/s | +| 32 | 885 MB/s | + +Adding shards scales a table's ingest nearly linearly. + +### Reads {#read-performance} + +Reads on an RTE table stay close to those on a fully compacted table, even +with the fresh tier at its fullest before compaction. + +The table below compares a 5M-row table with a 128-dimension IVF-PQ index plus +BTree and FTS indexes, with and without RTE. The RTE table holds one +unflushed memtable and three SSTables, each from a 100k-row write wave, half of +which are updates to existing rows. Latencies are p50 in milliseconds. + +| Query | Without RTE | With RTE | +| --- | --: | --: | +| Primary-key lookup | 62 | 63 | +| Vector search | 68 | 71 | +| Vector search with prefilter | 90 | 91 | +| Full-text term | 63 | 64 | +| Full-text phrase | 68 | 82 | +| Hybrid search (RRF) | 68 | 83 | +| SQL point lookup | 189 | 187 | + +Most queries are within a few milliseconds of the base table. Hybrid and phrase +queries add roughly 15 ms to merge the two tiers. + + +Support for all table operations on RTE tables is coming soon. Time travel, +branching, `count_rows`, row IDs (`with_row_id`), and `TABLESAMPLE` are not yet +available. + + +## Next steps {#next-steps} + +- [Merge insert](/ingestion/merge-insert) +- [Background jobs](/maintenance/background-jobs#wal) +- [`ALTER TABLE ... SET RTE`](/sql/ddl/table#properties) diff --git a/python/python/lancedb/_lancedb.pyi b/python/python/lancedb/_lancedb.pyi index 988482a97..5a613b210 100644 --- a/python/python/lancedb/_lancedb.pyi +++ b/python/python/lancedb/_lancedb.pyi @@ -850,9 +850,9 @@ class RTEWriteSpec: @staticmethod def unsharded() -> "RTEWriteSpec": ... def with_maintained_indexes(self, indexes: Optional[List[str]]) -> "RTEWriteSpec": - """Set which indexes the RTE keeps up to date. None resolves every - index on the table at install, failing if one cannot be maintained; - a list is verbatim, empty means none.""" + """Set which indexes the RTE keeps up to date. None is every index on + the table, now and later, skipping any it cannot maintain; a list is + verbatim, empty means none.""" ... def with_writer_config_defaults(self, defaults: Dict[str, str]) -> "RTEWriteSpec": """Return a copy of this spec recording the given default diff --git a/python/python/lancedb/table.py b/python/python/lancedb/table.py index 48d5582d1..60404b73c 100644 --- a/python/python/lancedb/table.py +++ b/python/python/lancedb/table.py @@ -5564,12 +5564,12 @@ class AsyncTable: via [`set_unenforced_primary_key`]; bucket sharding additionally requires it to be the single column being bucketed. - By default the RTE maintains every index on the table, resolved - here — a snapshot, so an index created afterwards needs the spec unset - and set again. This fails if one cannot be maintained; name the set - with ``with_maintained_indexes`` to install anyway. That pins an exact - set (a still-building index is rejected, not omitted); ``[]`` maintains - none. + By default the RTE maintains every index on the table, re-read as the + table changes, so an index created later is maintained too; one of a + kind the RTE cannot mirror is skipped. Name the set with + ``with_maintained_indexes`` to pin it instead: each name must already + exist and be maintainable (a still-building index is rejected, not + omitted); ``[]`` maintains none. Parameters ----------