Files
lancedb/nodejs/src/permutation.rs
T
Weston Pace c6db80dd0b feat: add an elastic dataloader as an iterable dataset (#3509)
# Elastic Streaming Dataloader

## Motivation

Training large models on LanceDB tables today requires loading the
entire dataset
into memory or writing bespoke batching logic. This PR introduces
`StreamingDataset`, a PyTorch `IterableDataset` that streams directly
from a
LanceDB table with two hard guarantees that are difficult to achieve
together:
**elastic determinism** and **resumability**.

## Goals

### Elastic determinism

The dataset partitions the table into a fixed number of *splits*
(controlled by
`num_splits`, `shuffle_seed`, and `epoch`). Samples are yielded by
round-robining
over splits one sample per split per cycle. Because the split structure
is fixed,
the set of samples that makes up each global training step is identical
regardless
of `world_size` or `num_workers`. You can scale your cluster up or down
between
runs and the model sees the same data in the same order — no
re-sharding, no
gradient variance from topology changes.

### Resumability

`state_dict()` / `load_state_dict()` capture how many samples each split
has
consumed. Because all splits are the same size and the round-robin
design keeps
them in lockstep, the state reduces to a single scalar
(`samples_consumed_per_split`)
that is topology-independent. A checkpoint saved with 8 GPUs can resume
correctly
on 4 GPUs or 16 GPUs without any adjustment.

### PyTorch `IterableDataset` / streaming

`StreamingDataset` implements the standard PyTorch `IterableDataset`
interface, so
it drops into any existing `DataLoader` pipeline without modification.
Data is
fetched lazily from Lance in chunks — only the rows needed for the
current batch are
ever in memory.

Compared to the map dataset this takes more work from pytorch and puts
it into the dataset itself (e.g. shuffling, filtering, etc.). We do this
because we cannot achieve things like elastic determinism or
prefiltering otherwise.

### Multi-worker support

DataLoader workers are automatically assigned contiguous sub-blocks of
splits (the
rank's splits are divided evenly across workers). Each worker is
independent:
no shared state, no inter-process coordination. The only constraint is
that
`num_splits` must be divisible by `world_size * num_workers`.

That being said, multi-worker is highly discouraged as it relies on
multiprocessing which is inefficient. Still, we want to support it.

### Filters as prefilters

Filters are applied at *permutation-build time* via
`PermutationBuilder.filter()`,
not re-evaluated on every fetch. The filtered row IDs are stored in the
permutation
table so that subsequent reads see only the matching rows. This allows
us to avoid loading rows that don't match the filter (which is the
default pytorch behavior)

### Prefetching

Two parameters control the I/O pipeline:

- `read_batch_size` (default 64) — number of rows fetched per
`take_offsets` call.
Larger values amortise per-request overhead, which is critical on object
storage
  where a single round-trip can cost ~100 ms.
- `prefetch_batches` (default 4) — number of batches prefetched in
parallel per
split via a `ThreadPoolExecutor`. While the model processes the current
batch,
the next several batches are already in flight, hiding storage latency
behind
  compute.

If set correctly then you can get good performance even with
num_workers=0 (unless you are bottlenecked on transform).

### Transform parallelism

The underlying `Permutation` API supports a `with_transform()` callback
for
decoding, augmentation, and format conversion. Unfortunately, this is
not parallelized. Pytorch typically parallelizes this with num_workers
which is multiprocessing which is highly inefficient. For simple
transforms we should be able to utilize multithreading and Rust based
UDFs. For complex python UDFs we could have a dedicated multiprocessing
pipeline for just the transform. Or we could just utilize
multithreading. In both cases we would exclude the I/O stage from the
multiprocessing because that ends up being very memory hungry and
inefficient.

---------

Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-07-06 05:50:45 -07:00

255 lines
7.4 KiB
Rust

// SPDX-License-Identifier: Apache-2.0
// SPDX-FileCopyrightText: Copyright The LanceDB Authors
use std::sync::{Arc, Mutex};
use crate::{error::NapiErrorExt, table::Table};
use lancedb::dataloader::{
permutation::builder::{PermutationBuilder as LancePermutationBuilder, ShuffleStrategy},
permutation::split::{SplitSizes, SplitStrategy},
};
use napi_derive::napi;
#[napi(object)]
pub struct SplitRandomOptions {
pub ratios: Option<Vec<f64>>,
pub counts: Option<Vec<i64>>,
pub fixed: Option<i64>,
pub seed: Option<i64>,
pub clump_size: Option<i64>,
pub split_names: Option<Vec<String>>,
}
#[napi(object)]
pub struct SplitHashOptions {
pub columns: Vec<String>,
pub split_weights: Vec<i64>,
pub discard_weight: Option<i64>,
pub split_names: Option<Vec<String>>,
}
#[napi(object)]
pub struct SplitSequentialOptions {
pub ratios: Option<Vec<f64>>,
pub counts: Option<Vec<i64>>,
pub fixed: Option<i64>,
pub split_names: Option<Vec<String>>,
}
#[napi(object)]
pub struct SplitCalculatedOptions {
pub calculation: String,
pub split_names: Option<Vec<String>>,
}
#[napi(object)]
pub struct ShuffleOptions {
pub seed: Option<i64>,
pub clump_size: Option<i64>,
}
pub struct PermutationBuilderState {
pub builder: Option<LancePermutationBuilder>,
}
#[napi]
pub struct PermutationBuilder {
state: Arc<Mutex<PermutationBuilderState>>,
}
impl PermutationBuilder {
pub fn new(builder: LancePermutationBuilder) -> Self {
Self {
state: Arc::new(Mutex::new(PermutationBuilderState {
builder: Some(builder),
})),
}
}
}
impl PermutationBuilder {
fn modify(
&self,
func: impl FnOnce(LancePermutationBuilder) -> LancePermutationBuilder,
) -> napi::Result<Self> {
let mut state = self.state.lock().unwrap();
let builder = state
.builder
.take()
.ok_or_else(|| napi::Error::from_reason("Builder already consumed"))?;
state.builder = Some(func(builder));
Ok(Self {
state: self.state.clone(),
})
}
}
#[napi]
impl PermutationBuilder {
#[napi]
pub fn persist(
&self,
connection: &crate::connection::Connection,
table_name: String,
) -> napi::Result<Self> {
let database = connection.database()?;
self.modify(|builder| builder.persist(database, table_name))
}
/// Configure random splits
#[napi]
pub fn split_random(&self, options: SplitRandomOptions) -> napi::Result<Self> {
// Check that exactly one split type is provided
let split_args_count = [
options.ratios.is_some(),
options.counts.is_some(),
options.fixed.is_some(),
]
.iter()
.filter(|&&x| x)
.count();
if split_args_count != 1 {
return Err(napi::Error::from_reason(
"Exactly one of 'ratios', 'counts', or 'fixed' must be provided",
));
}
let sizes = if let Some(ratios) = options.ratios {
SplitSizes::Percentages(ratios)
} else if let Some(counts) = options.counts {
SplitSizes::Counts(counts.into_iter().map(|c| c as u64).collect())
} else if let Some(fixed) = options.fixed {
SplitSizes::Fixed(fixed as u64)
} else {
unreachable!("One of the split arguments must be provided");
};
let seed = options.seed.map(|s| s as u64);
let clump_size = options.clump_size.map(|c| c as u64);
self.modify(|builder| {
builder.with_split_strategy(
SplitStrategy::Random {
seed,
sizes,
clump_size,
},
options.split_names.clone(),
)
})
}
/// Configure hash-based splits
#[napi]
pub fn split_hash(&self, options: SplitHashOptions) -> napi::Result<Self> {
let split_weights = options
.split_weights
.into_iter()
.map(|w| w as u64)
.collect();
let discard_weight = options.discard_weight.unwrap_or(0) as u64;
self.modify(move |builder| {
builder.with_split_strategy(
SplitStrategy::Hash {
columns: options.columns,
split_weights,
discard_weight,
},
options.split_names,
)
})
}
/// Configure sequential splits
#[napi]
pub fn split_sequential(&self, options: SplitSequentialOptions) -> napi::Result<Self> {
// Check that exactly one split type is provided
let split_args_count = [
options.ratios.is_some(),
options.counts.is_some(),
options.fixed.is_some(),
]
.iter()
.filter(|&&x| x)
.count();
if split_args_count != 1 {
return Err(napi::Error::from_reason(
"Exactly one of 'ratios', 'counts', or 'fixed' must be provided",
));
}
let sizes = if let Some(ratios) = options.ratios {
SplitSizes::Percentages(ratios)
} else if let Some(counts) = options.counts {
SplitSizes::Counts(counts.into_iter().map(|c| c as u64).collect())
} else if let Some(fixed) = options.fixed {
SplitSizes::Fixed(fixed as u64)
} else {
unreachable!("One of the split arguments must be provided");
};
self.modify(move |builder| {
builder.with_split_strategy(SplitStrategy::Sequential { sizes }, options.split_names)
})
}
/// Configure calculated splits
#[napi]
pub fn split_calculated(&self, options: SplitCalculatedOptions) -> napi::Result<Self> {
self.modify(move |builder| {
builder.with_split_strategy(
SplitStrategy::Calculated {
calculation: options.calculation,
},
options.split_names,
)
})
}
/// Configure shuffling
#[napi]
pub fn shuffle(&self, options: ShuffleOptions) -> napi::Result<Self> {
let seed = options.seed.map(|s| s as u64);
let clump_size = options.clump_size.map(|c| c as u64);
self.modify(|builder| {
builder.with_shuffle_strategy(ShuffleStrategy::Random { seed, clump_size })
})
}
/// Configure filtering
#[napi]
pub fn filter(&self, filter: String) -> napi::Result<Self> {
self.modify(|builder| builder.with_filter(filter))
}
/// Execute the permutation builder and create the table
#[napi]
pub async fn execute(&self) -> napi::Result<Table> {
let builder = {
let mut state = self.state.lock().unwrap();
state
.builder
.take()
.ok_or_else(|| napi::Error::from_reason("Builder already consumed"))?
};
let table = builder.build().await.default_error()?;
Ok(Table::new(table))
}
}
/// Create a permutation builder for the given table
#[napi]
pub fn permutation_builder(table: &crate::table::Table) -> napi::Result<PermutationBuilder> {
use lancedb::dataloader::permutation::builder::PermutationBuilder as LancePermutationBuilder;
let inner_table = table.inner_ref()?.clone();
let inner_builder = LancePermutationBuilder::new(inner_table);
Ok(PermutationBuilder::new(inner_builder))
}