Repository navigation
Writing
Write a stream of nd record batches, then read a projection back as a stream:
use futures::{StreamExt, stream};
use lance_nd::{ColumnOptions, Compression, WriteOptions, scan, write};
let options = WriteOptions {
column_defaults: Some(
ColumnOptions::default().with_compression(Compression::Zstd { level: Some(3) }),
),
..WriteOptions::default()
};
// `batches` is an iterator of nd_arrow_array::NdRecordBatch.
let rows = stream::iter(batches.map(Ok));
let table = write("data/arrays.lance", rows, &options).await?;
// Read two arrays of one row. Each item is one NdRecordBatch.
let mut rows = scan(&table)
.project(&["temperature", "x"])
.filter("__global_attrs__.id = 'a'")
.try_into_stream()
.await?;
while let Some(batch) = rows.next().await {
let batch = batch?;
println!("{:?}", batch.target());
}The main functions:
| Function | Purpose |
|---|---|
open(uri) |
Open a table. Fails with InvalidTable for a table without the lance-nd format key of this release. |
open_with(uri, &storage_options) |
Open a table on an object store, see Object stores. |
write(uri, rows, &options) |
Create, append to or overwrite a table from a stream of nd record batches, in one commit. |
append(&mut table, rows, &options) |
Append a stream of nd record batches to the latest version of a table, in one commit. |
NdWriter::create and NdWriter::append
|
Write nd record batches in several calls, then commit all of them in one version with finish. write_batch takes rows that are already encoded. |
scan(&table) |
Read nd record batches as a stream, with project, filter, filter_expr, lance_filter and limit. |
variables(&table) |
List the array columns of the table. |
column_defaults and set_column_defaults
|
Read and change the encoding defaults for new columns. |
delete and delete_expr
|
Delete the rows that match an exact predicate. |
compact and cleanup_old_versions
|
Merge small fragments, split large fragments, remove deleted rows and free disk space. |
versions, checkout and restore
|
List the versions of a table, read an earlier version, or make an earlier version the latest in a new commit. |
lance-nd logs with tracing. Each write,
commit, delete, compaction, widen and restore has a span with the table
uri. Install a subscriber in your application to see them, for example
tracing_subscriber::fmt::init().
project takes whole variables, such as temperature, not nested fields
such as temperature.attrs.units. Each row keeps all of its global
attributes. An empty projection reads only the global attributes, one row
without arrays for each table row. A filter can use nested fields, see
Filters.
Rows can be large, so a scan reads Lance batches of at most 256 rows or about
32 MiB, and it decodes them in parallel. batch_size, batch_size_bytes,
batch_readahead and decode_parallelism change these limits.
Error tells the cause: TableNotFound, TableExists, InvalidTable,
SchemaMismatch, or CommitConflict with retryable when another commit
changed the table first. After a failed fragment write, each call of the
NdWriter returns WriterFailed with the first error.
A write that fails a check, for example a type change, is rejected, and the
NdWriter stays usable. finish commits the other rows. A failed fragment
write makes the writer fail, and finish then commits nothing.
An append NdWriter adds new variables and attribute keys to the table
before it writes the first fragment with them, so readers can see new empty
columns before finish. finish then commits the new rows only. Lance
merges this commit with concurrent appends, deletes and compactions.
NdWriter buffers rows and writes them as Lance fragments of about 128 MiB
on disk, with up to 4 fragment writes at the same time. Lance compresses the
data, so the writer learns the ratio of disk to memory bytes from the
fragments it wrote. The first fragments use the size in memory, and the
buffer holds at most 8 times the target in memory. A row is never split.
Each commit writes the rest of the buffer as a smaller fragment, so use
compact for exact sizes. WriteOptions::target_fragment_bytes and
WriteOptions::write_parallelism change these limits. The writer must run
on a Tokio runtime.
Scans and writes allocate many large buffers. On Windows, the system allocator makes them much slower. Set a faster global allocator in your application, for example mimalloc:
#[global_allocator]
static GLOBAL: mimalloc::MiMalloc = mimalloc::MiMalloc;WriteOptions::casting decides what a write does with a variable or an
attribute that has another type than the table. The rules follow the
casting argument of numpy:
Casting |
A different type |
|---|---|
No (default) |
Rejects the write. |
Safe |
Casts the new values when no value can change, for example Float32 to Float64 or UInt16 to Int32. |
SameKind |
Also casts within a kind, or to a later kind in the order bool, unsigned, signed, float, for example Float64 to Float32. An integer that does not fit fails the write. |
Promote |
As numpy type promotion. Casts as Safe when the new values fit the table type. Otherwise widens the variable in the table to the smallest type that holds both, for example Int32 and Float32 to Float64, then casts the new values. |
With No, Safe and SameKind, the table type wins: a cast changes the
new rows, never the table. So with Safe, a Float64 variable cannot go
into a Float32 table. The
cast comes before the stats, so the stats get the table type too. An
attribute with a number in one row and an array in another row is never
cast. Promote widens only variable values; attributes follow Safe.
widen(&mut table, "sst", &DataType::Float64) changes the value type of a
variable in the table, and the type of its stats. The new type must hold
each old value exactly, as Safe allows. Lance 11 cannot cast a nested
field in place, so lance-nd writes the variable and its stats column again
under new field ids, in one Lance commit. The other variables stay as they
are.
A writer that started before a widen and commits after it holds rows with
the old field ids. Its commit fails with Error::SchemaMismatch and adds no
rows. Write the rows again. A widen that a concurrent write preempts starts
again, up to 10 times. Under Promote, the writers call widen
themselves. An NdWriter then first commits the rows that it wrote before.
Each column can have its own Lance encoding: compression, compression level, byte stream split, run-length encoding and structural encoding. The table stores default options, and each new column gets them. You can also set options for named columns.