dataset
js/data/dataset.ts
fino:data/dataset - deterministic datasets and pull-driven data loading.
Dataset owns finite, deterministic random-access values.
IterableDataset is the single lazy composition path for finite and
streaming inputs, and DataLoader adds batching plus collation without
introducing a second scheduler. Every transform is an AsyncIterable, so
consumers control demand: the pipeline reads one item only when the next
item or batch is requested. Early return and cancellation close upstream
iterators.
Randomness is explicit. Buffered shuffle derives its stream from seed,
epoch, and workerId, which makes repeated evaluation and realm-worker
execution reproducible. Loader iterators expose serializable checkpoints,
and DataLoader.restore() resumes at the next unseen batch by replaying the
deterministic source without re-running collators for skipped batches.
Built-in adapters cover CSV, JSON Lines, Arrow IPC, Parquet, SQLite, HTTP,
and hub-style HTTP repositories. Tabular binary sources yield Arrow
RecordBatch objects; row sources can use arrowCollator at the loader
boundary.
import { DataLoader, csvDataset, arrowCollator } from 'fino:data/dataset';
const rows = csvDataset('id,text\n1,hello\n2,world\n', {
header: true,
cast: true,
});
const batches = new DataLoader(rows, {
batchSize: 128,
shuffle: { bufferSize: 2048 },
seed: 7,
collate: arrowCollator,
});
for await (const batch of batches.forEpoch(0)) {
console.log(batch.numRows);
}Types
type MaybePromise<T> = T | Promise<T>
Value returned directly or through a promise by transform and collator callbacks.
type DatasetSource<T> = Iterable<T> | AsyncIterable<T>
Synchronous or asynchronous iterable accepted by dataset composition APIs.
type DatasetFactory<T> = (context: DatasetIterationContext) => DatasetSource<T>
Factory used by IterableDataset.
A factory should create a fresh iterable for each traversal. It receives the normalized epoch, worker, seed, and cancellation context.
type CollateFunction<T, B> = (values: readonly T[], context: CollateContext) => MaybePromise<B>
Convert one item group into the value yielded by DataLoader.
type DataLoaderWorkerFunction<T, B> = (
values: readonly T[],
context: DataLoaderWorkerContext,
) => MaybePromise<B | SharedBatchWrite>
Default export shape for a DataLoader realm-worker module.
type ByteSource =
| string
| Uint8Array
| ArrayBuffer
| Iterable<Uint8Array>
| AsyncIterable<Uint8Array>
Text, bytes, or chunked bytes accepted by row-oriented source adapters.
Strings are interpreted as content, not filesystem paths. Use a
DiskFileSystem reader as the async iterable for bounded file input.
type ArrowDatasetSource =
| RecordBatch
| Table
| Uint8Array
| ArrayBuffer
| Iterable<Uint8Array>
| AsyncIterable<Uint8Array>
Arrow values accepted by arrowDataset.
type HttpDatasetFormat = 'json' | 'jsonl' | 'csv' | 'arrow' | 'parquet'
Wire formats decoded by httpDataset and hubDataset.
Interfaces
interface DatasetIterationOptions {
Reproducibility and cancellation inputs for one dataset traversal.
epoch and workerId default to zero. seed also defaults to zero unless
an operation, such as shuffle, supplies its own seed. A canceled signal
rejects the next pull and closes the active upstream iterator.
Properties
epoch?: number
Epoch number mixed into seeded transforms. Defaults to 0.
workerId?: number
Worker number mixed into seeded transforms. Defaults to 0.
seed?: number
Base seed used when a transform does not provide one. Defaults to 0.
signal?: AbortSignal
Optional cancellation signal checked before every upstream pull.
interface DatasetIterationContext {
Fully normalized context received by dataset factories.
Numeric values are always present and validated; only signal is optional.
Properties
epoch: number
Epoch number for this traversal.
workerId: number
Worker number for this traversal.
seed: number
Base random seed for this traversal.
signal?: AbortSignal
Optional cancellation signal.
interface DatasetTransformContext extends DatasetIterationContext {
Context passed to map and filter callbacks.
index is the zero-based position in that traversal, before the current
transform removes or changes any item.
Properties
index: number
Zero-based source position for the current traversal.
interface ShuffleOptions {
Options for bounded-memory streaming shuffle.
Properties
bufferSize: number
Maximum number of items retained for random selection.
Must be a positive integer. Larger values approach a full random permutation while using proportionally more memory.
seed?: number
Seed for this shuffle. Defaults to the traversal's seed.
interface SplitOptions {
Options for deterministic dataset partitioning.
Properties
seed?: number
Seed controlling membership. Defaults to 0.
interface BatchOptions {
Options for grouping adjacent items.
Properties
dropLast?: boolean
Omit the final group when it contains fewer than size items.
interface CollateContext {
Context passed to a DataLoader collator.
Properties
batchIndex: number
Zero-based batch position within this loader traversal.
epoch: number
Epoch selected for this traversal.
workerId: number
Worker id selected for this traversal.
signal?: AbortSignal
Cancellation signal, when supplied.
interface DataLoaderState {
Serializable position for a deterministic DataLoader traversal.
The checkpoint records the next batch to yield, not prefetched work. Restore replays and skips source batches, so the source and transforms must be replayable and deterministic for the recorded seed context.
Properties
version: 1
Checkpoint format version.
epoch: number
Epoch used by the traversal.
workerId: number
Dataset worker partition used by the traversal.
seed: number
Seed used by the traversal.
batchesYielded: number
Number of batches already yielded to the consumer.
loader: {
batchSize: number;
dropLast: boolean;
shuffle: { bufferSize: number; seed: number } | null;
}
Configuration that affects deterministic source grouping and order.
interface DataLoaderIterator<B> extends AsyncIterableIterator<B> {
DataLoader iterator with a durable checkpoint for its next unseen batch.
Methods
state(): DataLoaderState
Snapshot the committed traversal position as plain JSON-compatible data.
interface DataLoaderWorkerContext extends Omit<CollateContext, 'signal'> {
Metadata passed to a realm-worker collator.
Realm workers do not receive an AbortSignal because cancellation
terminates the active Realm. When shared-memory collation is enabled,
shared identifies the writable slab for this source batch.
Properties
interface DataLoaderWorkerOptions {
Reactor-pooled collator module used by DataLoader.
Each submitted batch gets a fresh movable Realm isolate scheduled by Fino's
existing reactor pool. size bounds concurrent calls; it does not create a
second thread scheduler.
Properties
entry: string
Module whose default export implements DataLoaderWorkerFunction.
size?: number
Maximum concurrent realm calls. Defaults to 1.
realm?: Omit<
RealmOptions,
'entry' | 'process' | 'remote' | 'watch' | 'repl' | 'input' | 'output'
>
Realm loader/import configuration applied to every worker call.
interface DataLoaderOptions<T, B = T[]> {
DataLoader construction options.
Properties
batchSize?: number
Number of source items per yielded value. Defaults to 1.
dropLast?: boolean
Omit the final incomplete group. Defaults to false.
shuffle?: ShuffleOptions
Enable deterministic buffered shuffle.
The buffer bound is explicit so memory use cannot grow to the full source by accident.
seed?: number
Loader seed mixed with traversal epoch and worker id. Defaults to 0.
collate?: CollateFunction<T, B>
Convert an item group into the yielded batch value. Defaults to a new array.
prefetch?: number
Maximum number of batches processed ahead of the consumer.
Defaults to the realm worker size when worker is present and 1
otherwise. Results still yield in source order.
worker?: DataLoaderWorkerOptions
Run collation in fresh movable Realm isolates on the existing reactor pool.
The worker module replaces collate; specifying both is an error.
interface JsonlDatasetOptions {
JSON Lines parsing options.
Properties
reviver?: (this: unknown, key: string, value: unknown) => unknown
Optional JSON.parse reviver.
skipEmptyLines?: boolean
Ignore blank or whitespace-only lines. Defaults to true.
interface HttpDatasetOptions<F extends HttpDatasetFormat = HttpDatasetFormat> {
HTTP-backed dataset options.
Properties
format: F
Response wire format.
headers?: HeadersInit
Request headers merged into request.
csv?: CsvParseOptions
CSV dialect and row-shape options for format: 'csv'.
jsonl?: JsonlDatasetOptions
JSON Lines options for format: 'jsonl'.
request?: RequestInit
Other request settings. Traversal cancellation overrides its signal.
fetch?: (request: Request) => Promise<Response>
Injectable Fetch-compatible implementation for testing or custom transports.
interface HubDatasetOptions< F extends HttpDatasetFormat = HttpDatasetFormat, > extends HttpDatasetOptions<F> {
Options for a revision-addressed hub repository.
Properties
baseUrl?: string | URL
Hub root. Defaults to the Hugging Face-compatible dataset endpoint.
revision?: string
Repository revision. Defaults to "main".
token?: string
Optional bearer token added unless authorization is already set.
Classes
class IterableDataset<T> implements AsyncIterable<T> {
Lazy, repeatable dataset backed by a synchronous or asynchronous factory.
Transform methods return another IterableDataset and do no work until it
is iterated. Use iterate(options) when an epoch, worker id, seed, or
cancellation signal matters; ordinary for await uses all-zero defaults.
import { IterableDataset } from 'fino:data/dataset';
const values = IterableDataset.from([1, 2, 3, 4])
.filter((value) => value % 2 === 0)
.map((value) => value * 10);
console.log(await Array.fromAsync(values)); // [20, 40]Constructors
constructor(factory: DatasetFactory<T>)
Create a lazy dataset from a factory.
The factory runs once for each traversal and should return a fresh iterable when repeatability is required.
Static Methods
static from<T>(source: DatasetSource<T>): IterableDataset<T>
Wrap an existing iterable.
Arrays, sets, and normal iterable containers can be traversed repeatedly. A self-returning one-shot iterator remains one-shot; use the constructor with a factory when the source must be reopened.
Methods
iterate(options: DatasetIterationOptions = {}): AsyncIterableIterator<T>
Start a traversal with explicit reproducibility and cancellation inputs.
map<U>(
mapper: (value: T, context: DatasetTransformContext) => MaybePromise<U>,
): IterableDataset<U>
Lazily transform every item.
The mapper may be synchronous or asynchronous. Results preserve source order.
filter(
predicate: (value: T, context: DatasetTransformContext) => MaybePromise<boolean>,
): IterableDataset<T>
Lazily retain items accepted by predicate.
The predicate may be synchronous or asynchronous. Its index refers to the input position, not the number of accepted rows.
shuffle(options: ShuffleOptions): IterableDataset<T>
Shuffle with bounded memory.
The first bufferSize items fill a reservoir. Each later item replaces a
randomly selected slot whose previous value is emitted; the remaining
slots are drained randomly at the end. The same seed, epoch, worker id,
and input yield the same order.
batch(size: number, options: BatchOptions = {}): IterableDataset<T[]>
Group adjacent items.
At most size source values are retained. The final partial batch is
emitted unless dropLast is true.
take(count: number): IterableDataset<T>
Yield at most count items and then close the source iterator.
split(weights: readonly number[], options: SplitOptions = {}): IterableDataset<T>[]
Partition a stream deterministically according to positive weights.
Membership is a stable hash of source position and seed; it does not
change with epoch or worker id. Each returned dataset traverses the source
independently, so use a replayable source when consuming several splits.
interleave<U>(other: DatasetSource<U>): IterableDataset<T | U>
Round-robin this dataset with another source.
When one source finishes, the remaining source continues alone. Both iterators are closed if the consumer stops early.
class Dataset<T> extends IterableDataset<T> {
Finite dataset with deterministic asynchronous random access.
Dataset.from snapshots its input array, so later mutations of the caller's
array cannot change indexing or iteration. get is always asynchronous to
keep consumer code compatible with future random-access storage sources.
import { Dataset } from 'fino:data/dataset';
const rows = Dataset.from([{ id: 1 }, { id: 2 }]);
console.log(await rows.get(1)); // { id: 2 }Readonly Properties
readonly length: number
Number of indexed values.
Static Methods
static from<T>(values: readonly T[]): Dataset<T>
Snapshot an array-like sequence as a deterministic indexed dataset.
Methods
async get(index: number): Promise<T>
Read one position.
Throws RangeError for indices outside [0, length).
toArray(): T[]
Copy all indexed values into a plain array.
class DataLoader<T, B = T[]> implements AsyncIterable<B> {
Pull-driven batching and collation over an IterableDataset.
Local collation remains one-batch-at-a-time by default. prefetch opts into
a bounded ordered window, and worker runs those calls in movable Realm
isolates owned by the runtime's reactor scheduler. A shared-memory ring can
provide strongly retained shared host slabs for zero-copy descriptors and
explicit H2D handoff backpressure.
import { DataLoader, Dataset } from 'fino:data/dataset';
const loader = new DataLoader(Dataset.from([1, 2, 3]), {
batchSize: 2,
collate: (values) => new Uint32Array(values),
});
for await (const values of loader) console.log(values);Constructors
constructor(source: IterableDataset<T> | DatasetSource<T>, options: DataLoaderOptions<T, B> = {})
Create a reusable loader over a dataset or arbitrary iterable.
Methods
iterate(options: DatasetIterationOptions = {}): DataLoaderIterator<B>
Traverse and collate with explicit epoch, worker, and cancellation inputs.
restore(
state: DataLoaderState,
options: Pick<DatasetIterationOptions, 'signal'> = {},
): DataLoaderIterator<B>
Resume a serialized traversal checkpoint.
Restore rejects checkpoints from incompatible batching/shuffle configuration. It replays the source to the saved batch boundary without invoking local or realm collators for skipped batches.
forEpoch(epoch: number, options: Omit<DatasetIterationOptions, 'epoch'> = {}): IterableDataset<B>
Return an iterable view for one epoch.
Calling this repeatedly with the same epoch reproduces the same shuffle and batch order.
Functions
function arrowCollator(rows: readonly Record<string, unknown>[]): RecordBatch
Collate plain object rows into one Arrow RecordBatch.
Column order follows first appearance across the rows; missing values become
null. Type inference uses RecordBatch.from, so provide a custom collator
when an explicit schema is required or a column contains only nulls.
function csvDataset(
source: ByteSource,
options?: CsvParseOptions & { header?: false; columns?: undefined },
): IterableDataset<string[]>
function csvDataset(
source: ByteSource,
options?: CsvParseOptions & { header?: false; columns?: undefined },
): IterableDataset<string[]>
function csvDataset(
source: ByteSource,
options: CsvParseOptions & { header: true },
): IterableDataset<Record<string, unknown>>
function csvDataset(
source: ByteSource,
options: CsvParseOptions & { columns: string[] },
): IterableDataset<Record<string, unknown>>
function csvDataset(
source: ByteSource,
options?: CsvParseOptions,
): IterableDataset<Record<string, unknown> | string[]>Stream CSV content as positional arrays or header-keyed records.
Parsing delegates to fino:format/csv, including dialect, header, casting,
and column-count behavior. Chunked sources retain only the current logical
row.
function jsonlDataset<T = unknown>(
source: ByteSource,
options: JsonlDatasetOptions = {},
): IterableDataset<T>
Stream newline-delimited JSON values across arbitrary UTF-8 chunk boundaries.
Each non-empty line is parsed independently. A final line does not require a trailing newline.
function arrowDataset(source: ArrowDatasetSource): IterableDataset<RecordBatch>
Expose Arrow tables, batches, or IPC input as a record-batch stream.
Existing RecordBatch and Table objects are yielded without copying.
IPC byte inputs use RecordBatchReader; its current streaming helper
buffers bytes before decode, matching the underlying reader contract.
function parquetDataset(source: ByteSource): IterableDataset<RecordBatch>
Decode a complete or chunked Parquet file as Arrow record batches.
Parquet footer discovery requires the complete file, so chunked sources are
bounded by the file size rather than by one row group. Each decoded row
group is then yielded as its own RecordBatch.
function sqliteDataset(
database: Database,
query: string,
parameters: readonly SqlValue[] = [],
): IterableDataset<Record<string, SqlValue>>
Stream rows from a SQLite query.
The statement is prepared lazily for every traversal, resets on early
return through Statement.iterate, and is finalized when traversal ends.
function httpDataset<T = unknown>(
input: string | URL | Request,
options: HttpDatasetOptions<'json' | 'jsonl'>,
): IterableDataset<T>
function httpDataset<T = unknown>(
input: string | URL | Request,
options: HttpDatasetOptions<'json' | 'jsonl'>,
): IterableDataset<T>
function httpDataset(
input: string | URL | Request,
options: HttpDatasetOptions<'csv'>,
): IterableDataset<Record<string, unknown> | string[]>
function httpDataset(
input: string | URL | Request,
options: HttpDatasetOptions<'arrow' | 'parquet'>,
): IterableDataset<RecordBatch>
function httpDataset<T = unknown>(
input: string | URL | Request,
options: HttpDatasetOptions,
): IterableDataset<T>Fetch and decode a dataset lazily.
No request is sent until iteration starts. JSON arrays yield one item per
element; a non-array JSON document yields once. JSONL and CSV consume the
response stream incrementally. Arrow IPC and Parquet follow their underlying
whole-buffer reader constraints and yield RecordBatch values.
function hubDataset<T = unknown>(
repository: string,
path: string,
options: HubDatasetOptions<'json' | 'jsonl'>,
): IterableDataset<T>
function hubDataset<T = unknown>(
repository: string,
path: string,
options: HubDatasetOptions<'json' | 'jsonl'>,
): IterableDataset<T>
function hubDataset(
repository: string,
path: string,
options: HubDatasetOptions<'csv'>,
): IterableDataset<Record<string, unknown> | string[]>
function hubDataset(
repository: string,
path: string,
options: HubDatasetOptions<'arrow' | 'parquet'>,
): IterableDataset<RecordBatch>
function hubDataset<T = unknown>(
repository: string,
path: string,
options: HubDatasetOptions,
): IterableDataset<T>Read a file from a revision-addressed HTTP dataset repository.
The URL shape is
{baseUrl}/{repository}/resolve/{revision}/{path}. Custom baseUrl and
injectable fetch make the adapter usable with self-hosted or
Hugging-Face-compatible hubs without adding a second transport path.