Ë
    kQêjBp ã                   óØ  — d Z ddlZddlZddlZddlZddlZddlZddlZddlZddl	Z	ddl
mZ ddlmZ ddlmZ ddlmZ ddlmZmZmZmZmZmZmZmZ ddlZddlmZ ddl Z ddl!m"Z"m#Z#m$Z$ d	d
l%m&Z&m'Z'm(Z(m)Z)m*Z*m+Z+m,Z,  ejZ                  e.«      Z/dZ0dZ1dZ2 G d„ de«      Z3 G d„ de«      Z4 G d„ de«      Z5 G d„ d«      Z6 G d„ de#«      Z7 G d„ d«      Z8 G d„ de#«      Z9 G d„ de"«      Z:y)aÝ  Elastic streaming dataloader for PyTorch.

Provides StreamingDataset, a PyTorch IterableDataset that guarantees:

- **Elastic determinism**: for a fixed (num_splits, shuffle_seed, epoch) the set
  of samples that forms each global training step is identical regardless of
  world_size or num_workers.
- **Resumability**: state_dict / load_state_dict capture per-split consumption
  counts so training can resume from an exact mid-epoch position even when the
  distributed topology changes between runs.

Transform failures on bad rows (e.g. nulls or NaNs from incomplete data) can
be tolerated with ``on_transform_error="skip"``; see the parameter
documentation on StreamingDataset for how this interacts with the guarantees
above.
é    N)Údeque)ÚThreadPoolExecutor)Údeepcopy)ÚRawArray)ÚAnyÚCallableÚcastÚIteratorÚLiteralÚ
NamedTupleÚOptionalÚUnion)Ú
DataLoaderÚIterableDatasetÚget_worker_infoé   )ÚPermutationÚPermutationBuilderÚ
TransformsÚpermutation_builderÚ_drop_base_versionÚ_table_from_pickle_stateÚ_table_to_pickle_statei£† é@   é   c                   ó"   — e Zd ZU eed<   ded<   y)Ú_WorkerSampleÚdataÚStreamingDatasetÚdatasetN)Ú__name__Ú
__module__Ú__qualname__r   Ú__annotations__© ó    úV/home/oem/CyberBrain26-unified/.venv/lib/python3.12/site-packages/lancedb/streaming.pyr   r   >   s   … Ø
ƒIØÔr&   r   c                   ó"   — e Zd ZU eed<   eed<   y)Ú_WorkerBatchr   ÚstateN)r!   r"   r#   r   r$   Údictr%   r&   r'   r)   r)   C   s   … Ø
ƒIØ„Kr&   r)   c                   ó"   — e Zd ZU eed<   eed<   y)Ú_ConsumerIteratorLeaseÚowner_tokenÚowner_threadN)r!   r"   r#   Úintr$   r%   r&   r'   r-   r-   H   s   … ØÓØÔr&   r-   c                   ó"   — e Zd ZdZdefd„Zd„ Zy)Ú_CheckpointCollatez9Attach the worker's post-fetch state to a collated batch.Ú
collate_fnc                 ó   — || _         y ©N)Ú_collate_fn)Úselfr3   s     r'   Ú__init__z_CheckpointCollate.__init__P   s
   € Ø%ˆÕr&   c                 óš  — 	 t        |t        «      rX|st        | j                  |«      i «      S |}| j                  |D �cg c]  }|j                  ‘Œ c}«      }|d   j
                  }n'| j                  |j                  «      }|j
                  }t        ||j                  «       «      S c c}w # t        $ r}t        d«      |‚d }~ww xY w)Néÿÿÿÿz8collate_fn raised StopIteration before returning a batch)	Ú
isinstanceÚlistr)   r6   r   r    ÚStopIterationÚRuntimeErrorÚ_checkpoint_snapshot)r7   ÚsamplesÚworker_samplesÚsampler   r    Úexcs          r'   Ú__call__z_CheckpointCollate.__call__S   sÀ   € ð	Ü˜'¤4Ô(ÙÜ'¨×(8Ñ(8¸Ó(AÀ2ÓFÐFØ!(�Ø×'Ñ'À>Ö(R¸¨¯«Ò(RÓS�Ø(¨Ñ,×4Ñ4‘à×'Ñ'¨¯©Ó5�Ø!Ÿ/™/�ô
 ˜D '×">Ñ">Ó"@ÓAÐAùò )Søô
 ò 	ÜØJóàðûð	ús.   ‚,B0 ¯B0 Á B+Á>B0 Â+B0 Â0	C
Â9CÃC
N)r!   r"   r#   Ú__doc__r   r8   rD   r%   r&   r'   r2   r2   M   s   „ ÙCð& 8ó &óBr&   r2   c                   ó0   ‡ — e Zd ZdZdˆ fd„Zd„ Zd„ Zˆ xZS )Ú_StreamingDatasetAdapterz?Yield private sample wrappers for :class:`StreamingDataLoader`.c                 ó0   •— t         ‰| �  «        || _        y r5   )Úsuperr8   r    )r7   r    Ú	__class__s     €r'   r8   z!_StreamingDatasetAdapter.__init__h   s   ø€ Ü‰ÑÔØˆ�r&   c              #   ó~   K  — | j                   j                  d¬«      D ]  }t        || j                   «      –— Œ y ­w)NT©Úconsumer_checkpoint_transport)r    Ú_iterr   )r7   rB   s     r'   Ú__iter__z!_StreamingDatasetAdapter.__iter__l   s8   è ø€ Ø—l‘l×(Ñ(ÀtÐ(ÓLò 	6ˆFÜ ¨¯©Ó5Ó5ñ	6ùs   ‚;=c                 ój   — | j                   j                  d«      }|€t        |«      ‚t        ||«      S )Nr    )Ú__dict__ÚgetÚAttributeErrorÚgetattr)r7   Únamer    s      r'   Ú__getattr__z$_StreamingDatasetAdapter.__getattr__p   s3   € Ø—-‘-×#Ñ# IÓ.ˆØˆ?Ü  Ó&Ð&Ü�w Ó%Ð%r&   )r    r   )r!   r"   r#   rE   r8   rO   rV   Ú__classcell__©rJ   s   @r'   rG   rG   e   s   ø„ ÙIõò6ö&r&   rG   c                   óF   — e Zd Zdddedefd„Zd„ Zd„ Zdd	„Zd
„ Z	d„ Z
d„ Zy)Ú_ConsumerCommitIteratorr    r   r.   Úrequire_uniformc                óX   — || _         || _        || _        || _        d| _        d| _        y )NF)Ú	_iteratorÚ_datasetÚ_owner_tokenÚ_require_uniformÚ	_releasedÚ	_terminal)r7   Úiteratorr    r.   r[   s        r'   r8   z _ConsumerCommitIterator.__init__x   s0   € ð "ˆŒØˆŒØ'ˆÔØ /ˆÔØˆŒØˆ�r&   c                 ó   — | S r5   r%   ©r7   s    r'   rO   z _ConsumerCommitIterator.__iter__‡   s   € Øˆr&   c                 ó  — | j                   rt        ‚	 t        | j                  «      }	 t        |t        «      st        d«      ‚| j                  j                  |j                  | j                  ¬«       |j                  S # t        $ r d| _         | j	                  «        ‚ t
        $ r$}| j                  j                  d|› �«       ‚ d }~ww xY w# t
        $ r$}| j                  j                  d|› �«       ‚ d }~ww xY w)NTz2a DataLoader batch failed before it was returned: z>StreamingDataLoader did not receive worker checkpoint metadata)r[   )rb   r=   Únextr]   Ú_releaseÚBaseExceptionr^   Ú_invalidate_checkpointr;   r)   r>   Ú_commit_worker_stater*   r`   r   )r7   ÚbatchrC   s      r'   Ú__next__z _ConsumerCommitIterator.__next__Š   sõ   € Ø�>Š>ÜÐð
	Ü˜Ÿ™Ó(ˆEð	Ü˜e¤\Ô2Ü"ØTóð ð �M‰M×.Ñ.Ø—‘¨T×-BÑ-Bð /ô ð —:‘:Ðøô# ò 	Ø!ˆDŒNØ�M‰MŒOØÜò 	Ø�M‰M×0Ñ0ØDÀSÀEÐJôð ûð		ûô ò 	Ø�M‰M×0Ñ0ØDÀSÀEÐJôð ûð		ús/   ”B ªAC Â)CÂ+C
Ã
CÃ	C?ÃC:Ã:C?Nc                 ó¾   — | j                   j                  dd«      ry d| _        | j                   j                  d«      }|�|j                  | j                  «       y y )Nra   Tr^   )rQ   rR   ra   Ú_release_consumer_iteratorr_   )r7   r    s     r'   rh   z _ConsumerCommitIterator._release§   sS   € Ø�=‰=×Ñ˜[¨$Ô/ØØˆŒØ—-‘-×#Ñ# JÓ/ˆØÐØ×.Ñ.¨t×/@Ñ/@ÕAð r&   c                 ó¨  — | j                   j                  dd«      ry d| _        | j                   j                  d«      }t        |dd «      }	 |� |«        n;t        |dd «      }t        |dd «      }t        |dd «      }|€t	        d«      ‚ |«        | j                  «        y # t        $ r$}| j                  j                  d	|› �«       ‚ d }~ww xY w)
Nra   Tr]   Ú_shutdown_workersÚ_dataset_fetcherÚdataset_iterÚclosez6StreamingDataLoader could not close its inner iteratorz5a DataLoader iterator could not be shut down safely: )	rQ   rR   rb   rT   r>   rh   ri   r^   rj   )r7   rc   ÚshutdownÚfetcherÚdataset_iteratorrt   rC   s          r'   rq   z)_ConsumerCommitIterator._shutdown_workers¯   sÚ   € Ø�=‰=×Ñ˜[¨$Ô/ØØˆŒØ—=‘=×$Ñ$ [Ó1ˆÜ˜8Ð%8¸$Ó?ˆð	ØÐ#Ù•
ä! (Ð,>ÀÓE�Ü#*¨7°NÀDÓ#IÐ ÜÐ 0°'¸4Ó@�Ø�=Ü&ØPóð ñ ”ð �M‰M�Oøô ò 	Ø�M‰M×0Ñ0ØGÈÀuÐMôð ûð		ús   ÁAB$ Â$	CÂ-CÃCc                 óD   — 	 | j                  «        y # t        $ r Y y w xY wr5   )rq   ri   re   s    r'   Ú__del__z_ConsumerCommitIterator.__del__É   s%   € ð	Ø×"Ñ"Õ$øÜò 	Ùð	ús   ‚ “	žc                 ó.   — t        | j                  |«      S r5   )rT   r]   )r7   rU   s     r'   rV   z#_ConsumerCommitIterator.__getattr__Ï   s   € Ü�t—~‘~ tÓ,Ð,r&   ©ÚreturnN)r!   r"   r#   r0   Úboolr8   rO   rm   rh   rq   ry   rV   r%   r&   r'   rZ   rZ   w   sD   „ ðð $ðð
 ðð óòòó:Bòò4ó-r&   rZ   c            .       óŠ  ‡ — e Zd ZdZdddddddeedddddddddddddddœdee   d	ed
ee	e
ge
f      dee   dedededededeee      dee   dee   dee	   dee   dee   dee   dee   deeeed   f      deee	egef   f   dee   dee	egef      dee   f,ˆ fd„Zd efd!„Zd ee   fd"„Zd eeeef      fd#„Zd$d%œd&ed eeeef      fd'„Zd&ed eeeef      fd(„Zed efd)„«       Zed efd*„«       Zed efd+„«       Zed efd,„«       Zed efd-„«       Zed efd.„«       Z ed efd/„«       Z!ed efd0„«       Z"d1„ Z#d2„ Z$d efd3„Z%ed efd4„«       Z&d efd5„Z'd6ed dfd7„Z(ed efd8„«       Z)ed ee   fd9„«       Z*ed ee   fd:„«       Z+d efd;„Z,d<ed dfd=„Z-d>ee.   d dfd?„Z/d@edAed dfdB„Z0d@ed dfdC„Z1e2dDee   d efdE„«       Z3ˆ xZ4S )Fr   uë,  An elastic, resumable PyTorch IterableDataset backed by a LanceDB table.

    The table is partitioned into ``num_splits`` fixed splits using a
    deterministic random shuffle controlled by ``shuffle_seed`` and ``epoch``.
    Each rank is assigned a contiguous block of splits, and within a rank each
    DataLoader worker is assigned a contiguous sub-block.  Samples are yielded
    by round-robining over the assigned splits, one sample per split per cycle.

    Internally ``__iter__`` runs a two-stage pipeline:

    - **Stage 1 (I/O)**: one thread pool with ``num_splits * io_queue_depth``
      workers fetches raw ``RecordBatch`` objects from LanceDB in parallel
      across all splits and places them in a per-split raw-batch queue.
    - **Stage 2 (transform)**: a second thread pool with
      ``transform_parallelism`` workers picks up raw batches, applies the
      transform, and places the results in a per-split cooked-row queue.  By
      default, the number of workers is determined by ``os.cpu_count()``.

    The main thread round-robins over the cooked queues, yielding one row per
    split per cycle.

    Parameters
    ----------
    table:
        LanceDB table to stream from.
    num_splits:
        Number of fixed splits to partition the table into.  Must be divisible
        by ``world_size``.  When used with DataLoader workers it must also be
        divisible by ``world_size * num_workers``.  Defaults to ``world_size``.
        If the row count (after any ``filter``) is not evenly divisible by
        ``num_splits``, the surplus rows â€” at most ``num_splits - 1`` per epoch
        â€” are silently dropped to keep all splits the same length.
    shuffle:
        Whether to randomly assign rows to splits.  When ``True`` (the
        default) rows are shuffled using ``shuffle_seed`` and ``epoch``.
        When ``False`` rows are divided into splits sequentially in storage
        order, which can be useful for deterministic debugging or evaluation.
        Ignored when ``split_strategy`` is provided.
    split_strategy:
        Escape hatch for any ``PermutationBuilder`` split strategy this class
        doesn't otherwise expose a dedicated parameter for -- ``split_hash``,
        ``split_calculated``, ``bootstrap_sample`` (uniform sampling with
        replacement), or any strategy added later.  A callable that receives a
        fresh ``PermutationBuilder`` for ``table`` and returns a configured one;
        it is called and ``.execute()``'d directly, taking over permutation
        construction entirely.  When provided, ``shuffle``, ``shuffle_seed``,
        ``epoch``, and ``shuffle_clump_size`` are ignored -- the callable is
        solely responsible for how rows are split and ordered.  For example,
        ``split_strategy=lambda b: b.bootstrap_sample(10_000, fixed=num_splits,
        seed=42)`` draws 10,000 samples with replacement, divided evenly across
        ``num_splits`` splits.
    shuffle_seed:
        Base seed for the random permutation.  Combined with ``epoch`` so
        each epoch produces a different ordering.  Pass ``None`` to generate
        a random seed at construction time.
    epoch:
        Current training epoch.  Combined with ``shuffle_seed`` so that each
        epoch produces a different sample ordering.
    rank:
        This process's rank in the distributed training group.
    world_size:
        Total number of processes in the distributed training group.
    read_batch_size:
        Number of rows fetched from each split in a single ``take_offsets``
        call.  Larger values amortise per-request overhead (critical on object
        storage) at the cost of higher memory usage per split buffer.  Defaults
        to ``DEFAULT_READ_BATCH_SIZE`` (64).
    io_queue_depth:
        Number of fetched batches to keep ahead of the transform stage per
        split: this bounds the combined count of in-flight fetches and
        already-fetched batches still waiting for a transform slot.  Higher
        values overlap storage latency with transform and training compute
        at the cost of more memory and threads, but I/O stops running ahead
        once this many batches are buffered, applying backpressure when the
        transform stage (or ``transform_queue_depth``) is the bottleneck.
        Must be greater than zero.  Defaults to ``DEFAULT_PREFETCH_BATCHES``
        (4).
    columns:
        Optional list of column names to read.  When set, only those columns
        are fetched from storage; all others are omitted.  ``None`` (the
        default) reads every column.
    shuffle_clump_size:
        When set, rows are shuffled in contiguous groups of this size rather
        than individually.  Larger clumps improve I/O locality (important on
        object storage) at the cost of reduced randomness.  ``None`` (the
        default) shuffles rows individually.
    filter:
        Optional SQL filter expression (e.g. ``"label = 'dog'"``).  Only rows
        that satisfy the predicate are included in the permutation.  The filter
        is applied during permutation construction so split sizes reflect the
        filtered row count.
    transform:
        Optional callable applied to each ``pyarrow.RecordBatch`` before rows
        are yielded.  Receives one batch at a time and must return an iterable
        whose length equals the number of rows in the batch.  When ``None``
        (the default) rows are returned as plain Python dicts.
    transform_parallelism:
        Maximum number of transforms to run concurrently.  Must be greater
        than zero.  When ``None`` (the default), uses ``os.cpu_count()`` or 1
        when the CPU count is unavailable.
    pack_sequences:
        Sequence-packing mode: token lists from consecutive documents are
        joined with ``eos_id`` and sliced into blocks of this many tokens.
        Each item is then a dict of two ``(pack_sequences,)`` LongTensors â€”
        ``input_ids`` and ``doc_ids`` (per-position document index within
        the block, for block-diagonal masks or position-id resets).
        * Packing happens independently per owned split and preserves per-split
        resume state.
        * When a split cannot fill a real block for a cycle but an owned sibling
        still can, or when only a short tail remains at epoch end, the short buffer
        is padded to ``pack_sequences`` with ``pad_id`` so every local cycle emits
        one block per owned split.
        * ``eos_id``, ``pad_id``, and ``columns`` naming a single integer-list
        column are required; incompatible with ``transform``.
    eos_id:
        Separator token id between packed documents.  Required with
        ``pack_sequences``, ignored otherwise.
    pad_id:
        Padding token id used to complete blocks when a split runs out of
        real tokens mid-cycle or at epoch end.  Required with
        ``pack_sequences``, ignored otherwise. It must be reserved for padding:
        padding positions retain the preceding document's ``doc_id`` (or zero
        in an all-padding block), so callers must mask them separately using
        ``input_ids == pad_id``.
    blocks_per_epoch:
        Total number of packed blocks emitted globally per epoch. Required with
        ``pack_sequences``. An integer must be divisible by ``num_splits``.
        Every logical split emits exactly ``blocks_per_epoch / num_splits``
        blocks: exhausted splits emit padding, while tokens beyond the budget
        are left out of the epoch. This fixed per-split budget keeps packed
        iteration and checkpoints independent of rank topology.
        Pass ``"auto"`` to estimate a corpus-level budget from a bounded sample
        of token lists. The estimate may be inaccurate.
    on_transform_error:
        What to do when the transform raises an exception:

        - ``"raise"`` (the default): the exception propagates and iteration
          aborts.
        - ``"skip"``: the failing rows are dropped and iteration continues.
        - ``"warn"``: like ``"skip"``, but a warning is logged for each
          failing batch.
        - a callable ``handler(exc) -> bool``: called with the exception;
          return ``True`` to skip the failing rows or ``False`` to re-raise.
          Useful to skip only expected error types (compatible with
          ``webdataset.handlers`` style handlers).

        When a batch fails, the transform is re-invoked on each single-row
        slice of the batch so that only the rows that actually fail are
        dropped.  Transforms should therefore be deterministic and accept
        batches of any size (including one row).  Skipped rows are counted in
        ``rows_skipped``.

        Skipping weakens the elastic-determinism guarantee at the end of the
        epoch: splits that lose more rows than others run dry earlier, and
        each rank's iterator ends at the last cycle where every split *it
        owns* still has a row.  Because bad rows are not distributed evenly
        across splits, this means one rank's iterator can yield noticeably
        fewer or more steps than another rank's *in the same run* â€” there is
        no cross-rank coordination that stops every rank at the same global
        step.  This is generally safe for asynchronous or single-rank use,
        but synchronous distributed training (e.g. ranks that call
        ``all_reduce`` every step) can hang or deadlock if one rank's
        iterator is exhausted while others are still stepping; callers doing
        synchronous multi-rank training with ``on_transform_error != "raise"``
        are responsible for their own cross-rank stopping mechanism (e.g.
        broadcasting a stop signal on ``StopIteration``).  The final few
        global steps can also differ across topologies (bounded by the skew
        in bad-row counts across splits).  The sequence of samples yielded
        from each split remains deterministic.  Mid-epoch
        checkpoints remain exact provided the transform fails
        deterministically; in multi-rank training each rank must save its
        own ``state_dict`` and the states must be combined with
        ``merge_state_dicts`` before resuming on a different topology.
        Prefer the ``filter`` parameter when bad rows can be expressed as a
        SQL predicate (e.g. ``"col IS NOT NULL"``) â€” filtering happens before
        splits are built, so every guarantee is fully preserved.
    transform_queue_depth:
        Number of transform-result batches to buffer per split in the
        post-transform queue before backpressure is applied to the transform
        stage.  When the combined count of in-flight transform futures and
        already-buffered rows for a split reaches
        ``transform_queue_depth * read_batch_size``, no new transforms are
        submitted for that split until the consumer catches up.  Useful for
        capping peak memory when the consumer (e.g. a GPU training step) is
        slower than the transform stage.  Must be greater than zero.
        ``None`` (the default) imposes no limit.
    worker_info_override:
        If set, used in place of ``torch.utils.data.get_worker_info()`` to
        determine the DataLoader worker assignment.  Intended for unit tests
        that need to simulate multiple workers without spawning real processes.
        If both this and the real worker info are non-None a warning is logged
        and the override takes precedence.
    NTr   r   Úraise)Ú
num_splitsÚshuffleÚsplit_strategyÚshuffle_seedÚepochÚrankÚ
world_sizeÚread_batch_sizeÚio_queue_depthÚcolumnsÚshuffle_clump_sizeÚfilterÚ	transformÚtransform_parallelismÚpack_sequencesÚeos_idÚpad_idÚblocks_per_epochÚon_transform_errorÚtransform_queue_depthÚconnection_factoryÚworker_info_overrideÚprefetch_batchesr€   r�   r‚   rƒ   r„   r…   r†   r‡   rˆ   r‰   rŠ   r‹   rŒ   r�   rŽ   r�   r�   r‘   Úautor’   r“   r”   r–   c          	      óð  •— t         ‰!| �  «        |�t        j                  d«       |}
|€|}|€t	        j
                  d«      }||z  dk7  rt        d|› d|› d�«      ‚|
dk  rt        d«      ‚|�|dk  rt        d«      ‚|��Ë|dk  rt        d	«      ‚|€t        d
«      ‚|€t        d«      ‚|€t        d«      ‚|dk7  rUt        |t        «      rt        |t        «      rt        d«      ‚|dk  rt        d«      ‚||z  dk7  rt        d|› d|› d�«      ‚|�t        d«      ‚|�t        |«      dk7  rt        d«      ‚|j                  j                  |d   «      }t        j                  j                  |j                   «      spt        j                  j#                  |j                   «      sGt        j                  j%                  |j                   «      st        d|d   › d|j                   › �«      ‚t        j                  j'                  |j                   j(                  «      s5t        d|d   › d|j                   j(                  › �«      ‚|�t        d«      ‚|dvrt+        |«      st        d|›�«      ‚|�|dk  rt        d«      ‚|| _        || _        || _        || _        || _        || _        || _        |	| _        |
| _        || _        || _         || _!        || _"        || _#        || _$        || _%        || _&        || _'        || _(        || _)        || _*        || _+        dg|z  | _,        i | _-        dg|z  | _.        d | _/        d | _0        d | _1        d | _2        d | _3        ti        tj        jl                  d«      | _7        ti        tj        jl                  d«      | _8        d | _9        d| _:        tw        jx                  «       | _=        d| _>        d | _?        d| _@        d| _A        d| _B        d| _C        d| _D        i | _E        i | _F        t�        |«      }|�|j‘                  |«      }|�� ||«      j“                  «       | _J        | j”                  j—                  «       j™                  d «      j›                  «       }t�        tŸ        |j                  d!«      j¡                  «       |j                  d"«      j¡                  «       «      «      }t£        t¥        |«      «      }t£        |«      |k7  r!t        d#t§        |«      › d$|› d%|dz
  › d&�«      ‚t        t£        |j©                  «       «      «      dkD  rjt        d'|› d(�«      ‚|r4||tª        z  z   }|j­                  |||¬)«      j“                  «       | _J        n%|j¯                  |¬*«      j“                  «       | _J        |d u| _X        d | _Y        | jN                  dk(  r| jµ                  «       | _'        ||z  }||z  } t·        t¥        | | |z   «      «      | _\        y )+Nzbprefetch_batches is deprecated and will be removed in a future version; use io_queue_depth insteadl        r   znum_splits (z#) must be divisible by world_size (ú)z%io_queue_depth must be greater than 0z,transform_parallelism must be greater than 0z%pack_sequences must be greater than 0z-eos_id is required when pack_sequences is setz-pad_id is required when pack_sequences is setz7blocks_per_epoch is required when pack_sequences is setr—   z5blocks_per_epoch must be a positive integer or 'auto'z'blocks_per_epoch must be greater than 0zblocks_per_epoch (z#) must be divisible by num_splits (z0transform cannot be combined with pack_sequencesr   zRpack_sequences requires columns to name exactly one list-typed column of token idsz3pack_sequences requires a list-typed token column; z
 has type z<pack_sequences requires a token column with integer values; z has value type z(blocks_per_epoch requires pack_sequences)r   ÚskipÚwarnzGon_transform_error must be 'raise', 'skip', 'warn', or a callable, got z,transform_queue_depth must be greater than 0é   Fg        Úsplit_idÚvaluesÚcountsz"split_strategy produced split ids z, but num_splits=z requires exactly {0, ..., z{}. Pass num_splits matching the strategy's own split count (e.g. `fixed=num_splits` on bootstrap_sample/split_random/etc.).z,split_strategy produced unequal split sizes zø (split id -> row count). Every split must have the same row count. Use `fixed=num_splits` rather than `ratios`/`counts`/weighted `split_hash`, which can produce unequal sizes, or split unevenly-sized groups into separate StreamingDatasets instead.)ÚfixedÚseedÚ
clump_size)r    )]rI   r8   ÚloggerÚwarningÚrandomÚ	randrangeÚ
ValueErrorr;   r0   r}   ÚlenÚschemaÚfieldÚpaÚtypesÚis_listÚtypeÚis_large_listÚis_fixed_size_listÚ
is_integerÚ
value_typeÚcallableÚ_tableÚ_num_splitsÚ_shuffleÚ_shuffle_seedÚ_epochÚ_rankÚ_world_sizeÚ_read_batch_sizeÚ_io_queue_depthÚ_columnsÚ_shuffle_clump_sizeÚ_filterÚ
_transformÚ_transform_parallelismÚ_pack_sequencesÚ_eos_idÚ_pad_idÚ_blocks_per_epochÚ_on_transform_errorÚ_transform_queue_depthÚ_connection_factoryÚ_worker_info_overrideÚ_pack_consumedÚ_pack_buffersÚ_pack_blocks_emittedÚ_raw_batches_refÚ_cooked_refÚ_fetch_head_refÚ_split_sizes_refÚ_local_consumed_refr   ÚctypesÚc_int64Ú_worker_statsÚ_untracked_worker_iterationÚ_checkpoint_invalid_reasonÚ%_consumer_checkpoint_requires_uniformÚ	threadingÚLockÚ_consumer_iterator_lockÚ_consumer_iterator_generationÚ_consumer_iterator_leaseÚ_bytes_loadedÚ_fetch_timeÚ_transform_timeÚ_rows_skippedÚ_resume_offsetÚ_resume_samplesÚ_resume_positionsr   r‹   ÚexecuteÚ_perm_tableÚto_arrowÚcolumnÚvalue_countsr+   ÚzipÚ	to_pylistÚsetÚrangeÚsortedrž   Ú_EPOCH_PRIMEÚsplit_randomÚsplit_sequentialÚ_uses_split_strategyÚ_perm_fingerprint_cacheÚ_estimate_blocks_per_epochr<   Ú_rank_splits)"r7   Útabler€   r�   r‚   rƒ   r„   r…   r†   r‡   rˆ   r‰   rŠ   r‹   rŒ   r�   rŽ   r�   r�   r‘   r’   r“   r”   r•   r–   rª   ÚbuilderÚsplit_value_countsÚcounts_by_splitÚexpected_splitsÚ	perm_seedÚsplits_per_rankÚ
rank_startrJ   s"                                    €r'   r8   zStreamingDataset.__init__–  s�  ø€ ô> 	‰ÑÔØÐ'Ü�N‰Nð6ôð .ˆNØÐØ#ˆJØÐÜ!×+Ñ+¨EÓ2ˆLØ˜
Ñ" aÒ'ÜØ˜z˜lð +Ø)˜l¨!ð-óð ð ˜QÒÜÐDÓEÐEØ Ð,Ð1FÈ!Ò1KÜÐKÓLÐLØÑ%Ø Ò"Ü Ð!HÓIÐIØˆ~Ü Ð!PÓQÐQØˆ~Ü Ð!PÓQÐQØÐ'Ü ØMóð ð   6Ò)Ü!Ð"2´CÔ8¼JØ$¤dô=ô %ØOóð ð $ qÒ(Ü$Ð%NÓOÐOØ# jÑ0°AÒ5Ü$Ø,Ð-=Ð,>ð ?'Ø'1 l°!ð5óð ð Ð$Ü Ð!SÓTÐTØˆ¤# g£,°!Ò"3Ü ð5óð ð —L‘L×&Ñ& w¨q¡zÓ2ˆEä—‘× Ñ  §¡Ô,Ü—8‘8×)Ñ)¨%¯*©*Ô5Ü—8‘8×.Ñ.¨u¯z©zÔ:ä ØIØ˜q‘z�l *¨U¯Z©Z¨Lð:óð ô —8‘8×&Ñ& u§z¡z×'<Ñ'<Ô=Ü ØRØ˜q‘z�lÐ"2°5·:±:×3HÑ3HÐ2IðKóð ð Ð)ÜÐGÓHÐHØÐ%>Ñ>ÄxØôH
ô ð!Ø!3Ð 6ð8óð ð !Ð,Ð1FÈ!Ò1KÜÐKÓLÐLàˆŒØ%ˆÔØˆŒØ)ˆÔØˆŒØˆŒ
Ø%ˆÔØ /ˆÔØ-ˆÔØˆŒØ#5ˆÔ ØˆŒØ#ˆŒØ&;ˆÔ#Ø-ˆÔØˆŒØˆŒØ!1ˆÔØ#5ˆÔ Ø&;ˆÔ#Ø#5ˆÔ Ø%9ˆÔ"ð +,¨¨zÑ)9ˆÔØ>@ˆÔØ01¨s°ZÑ/?ˆÔ!ð
 8<ˆÔØ26ˆÔØ48ˆÔØ59ˆÔØ8<ˆÔ ô (0´·±ÀÓ'BˆÔô 6>¼f¿n¹nÈaÓ5PˆÔ(ð :>ˆÔ'Ø5:ˆÔ2Ü'0§~¡~Ó'7ˆÔ$Ø-.ˆÔ*ØJNˆÔ%ð #$ˆÔà"%ˆÔØ&)ˆÔà"#ˆÔð
 $%ˆÔð 02ˆÔð 24ˆÔô & eÓ,ˆØÐØ—n‘n VÓ,ˆGØÑ%ñ  .¨gÓ6×>Ñ>Ó@ˆDÔð × Ñ ×)Ñ)Ó+×2Ñ2°:Ó>×KÑKÓMð ô #ÜØ&×,Ñ,¨XÓ6×@Ñ@ÓBØ&×,Ñ,¨XÓ6×@Ñ@ÓBóóˆOô "¤%¨
Ó"3Ó4ˆOÜ�?Ó# Ò6Ü Ø8¼ÀÓ9PÐ8Qð R&Ø&0 \ð 2!Ø!+¨a¡Ð 0ð 1;ð;óð ô  ”3�×-Ñ-Ó/Ó0Ó1°AÒ5Ü ØBØ&Ð'ð (1ð1óð ñ Ø$ u¬|Ñ';Ñ;ˆIØ&×3Ñ3Ø  yÐ=Oð  4ó  ç‰g‹ið Õð  '×7Ñ7¸jÐ7ÓI×QÑQÓSˆDÔð %3¸$Ð$>ˆÔ!ð 7;ˆÔ$à×!Ñ! VÒ+Ø%)×%DÑ%DÓ%FˆDÔ"ð %¨
Ñ2ˆØ˜OÑ+ˆ
Ü'+Ü�*˜j¨?Ñ:Ó;ó(
ˆÕr&   r|   c                 ó´  — | j                   �| j                  st        d«      ‚| j                   }| j                  d   }t        dd| j                  z  «      }d}d}d}t        j                  | j                  «      }t        j                  d«       t        | j                  «      D �]y  }t        j                  | j                  | j                  |¬«      }	|	j                  |g«      }	|	j!                  t"        j$                  «      }	|	j&                  }
|
dk(  rt)        d«      ‚t+        |
t        dt+        |
dz   d	z  |«      «      «      }t-        |j/                  t        |
«      |«      «      }t        d| j0                  «      }t        d||«      D ]ƒ  }|	j3                  ||||z    «      }t5        j6                  |j9                  d«      «      }|j:                  rt)        d
«      ‚|t=        t5        j>                  |«      jA                  «       «      z  }Œ… ||z  }||
z  }�Œ| ||z   |z  |z  }||z  }t        | j                  ||| j                  z  z
  «      S )zAEstimate a fixed packed-block budget from a bounded token sample.z7packing must be configured before estimating its budgetr   r   i † zmblocks_per_epoch='auto' uses an approximate token-count sample; pass an explicit value for exact epoch sizing©Úsplitz8blocks_per_epoch='auto' cannot estimate an empty datasetéc   éd   z0pack_sequences does not support null token lists)!rÂ   r½   r>   Úmaxrµ   r¥   ÚRandomr·   Úwarningsr›   rì   r   Úfrom_tablesr´   rå   Úselect_columnsÚwith_transformr   Úarrow2arrowÚnum_rowsr§   Úminrí   rB   r»   Ú__getitems__ÚpcÚlist_value_lengthrç   Ú
null_countr0   ÚsumÚas_py)r7   Úpack_lenÚtoken_columnÚsample_cap_per_splitÚsampled_tokensÚtotal_sampledÚ
total_rowsÚrngrÿ   ÚpermutationÚ
split_rowsÚsample_rowsÚsample_offsetsÚsample_batch_sizeÚstartrl   ÚlengthsÚestimated_tokensÚblockss                      r'   ró   z+StreamingDataset._estimate_blocks_per_epoch¨  s\  € ð ×ÑÐ'¨t¯}ª}ÜØIóð ð ×'Ñ'ˆØ—}‘} QÑ'ˆÜ" 1 g°×1AÑ1AÑ&AÓBÐØˆØˆØˆ
Ü�m‰m˜D×.Ñ.Ó/ˆä�‰ð<ô	
ô
 ˜4×+Ñ+Ó,ó 	%ˆEÜ%×1Ñ1Ø—‘˜T×-Ñ-°UôˆKð &×4Ñ4°l°^ÓDˆKØ%×4Ñ4´Z×5KÑ5KÓLˆKØ$×-Ñ-ˆJØ˜QŠÜ ØNóð ô ØÜ�A”s˜J¨™O°Ñ3Ð5IÓJÓKóˆKô $ C§J¡J¬u°ZÓ/@À+Ó$NÓOˆNÜ # A t×'<Ñ'<Ó =ÐÜ˜q +Ð/@ÓAò ?�Ø#×0Ñ0Ø" 5¨5Ð3DÑ+DÐEó�ô ×.Ñ.¨u¯|©|¸A«Ó?�Ø×%Ò%Ü$Ð%WÓXÐXØ¤#¤b§f¡f¨W£o×&;Ñ&;Ó&=Ó">Ñ>‘ð?ð ˜[Ñ(ˆMØ˜*Ñ$ŠJð=	%ðF ˜mÑ+¨zÑ9¸]ÑJð 	ð " XÑ-ˆÜØ×ÑØ�V˜d×.Ñ.Ñ.Ñ.ó
ð 	
r&   c                 ó\  — t        «       }| j                  �$|�t        j                  d«       | j                  }n|}|€| j                  S |j
                  }|j                  }t        | j                  «      }||z  dk7  rt        d|› d|› d�«      ‚||z  }||z  }| j                  |||z    S )z?Return the split indices this instance should read in __iter__.zÖworker_info_override is set but get_worker_info() also returned a non-None value; ignoring the real torch worker info and using the override instead.  This may lead to duplicated or incorrect data from the dataset.r   zNumber of rank splits (z$) must be divisible by num_workers (r™   )	r   rÉ   r£   r¤   rô   Únum_workersÚidr¨   r§   )r7   Útorch_worker_infoÚworker_infor"  Ú	worker_idÚn_rank_splitsÚsplits_per_workerr  s           r'   Ú_resolve_my_splitsz#StreamingDataset._resolve_my_splitsé  sÙ   € ä+Ó-ÐØ×%Ñ%Ð1Ø Ð,Ü—‘ð(ôð ×4Ñ4‰Kà+ˆKàÐØ×$Ñ$Ð$à&×2Ñ2ˆØ$Ÿ™ˆ	Ü˜D×-Ñ-Ó.ˆØ˜;Ñ&¨!Ò+ÜØ)¨-¨ð 9 Ø +˜}¨Að/óð ð *¨[Ñ8ÐØÐ-Ñ-ˆØ× Ñ  ¨Ð1BÑ)BÐCÐCr&   c                 ó"   — | j                  «       S r5   )rN   re   s    r'   rO   zStreamingDataset.__iter__  s   € Ø�z‰z‹|Ðr&   FrL   rM   c             #   óN  K  — d }| j                   }|r| j                  st        d«      ‚	 | j                  «       }	 | j                  |¬«      E d {  –—†  |�| j                  |«       y y # t        $ r | j                  |«       ‚ w xY w7 Œ8# |�| j                  |«       w w xY w­w)NzSStreamingDataLoader worker transport requires an active parent iterator reservationrL   )rÜ   Ú_consumer_iterator_activer>   Ú_acquire_consumer_iteratorri   Ú/_release_consumer_iterator_after_failed_acquireÚ_iter_ownedro   )r7   rM   r.   Úprevious_leases       r'   rN   zStreamingDataset._iter
  sÌ   è ø€ ð ˆØ×6Ñ6ˆÙ(Ø×1Ò1Ü"ð2óð ð
Ø"×=Ñ=Ó?�ð	=Ø×'Ñ'Ø.Kð (ó ÷ ð ð Ð&Ø×/Ñ/°Õ<ð 'øô !ò Ø×DÑDÀ^ÔTØðúðùð Ð&Ø×/Ñ/°Õ<ð 'üs?   ‚(B%«A+ ¼B ÁB
ÁB ÁB%Á+BÂB%Â
B ÂB"Â"B%c          
   #   ó´  ‡ ‡‡‡‡‡‡‡‡‡ ‡!‡"‡#‡$‡%‡&‡'‡(‡)‡*‡+‡,‡-‡.‡/‡0‡1‡2‡3‡4‡5‡6‡7‡8‡9‡:‡;‡<‡=‡>‡?‡@‡A‡BK  — ‰ j                   �t        d«      ‚t        «       d uŠ>‰>r|sd‰ j                  d<   ‰ j	                  «       Š4‰4sy g Š;g Š.g Š-‰4D �]  }t        j                  ‰ j                  ‰ j                  |¬«      }‰ j                  �|j                  ‰ j                  «      }|j                  t        j                  «      }‰ j                  j                  |‰ j                   «      }‰ j"                  �‰ j$                  |   n‰ j&                  j                  ||«      }|dkD  r|j)                  |«      }‰.j+                  |«       ‰-j+                  |«       ‰;j+                  |«       �Œ t-        ‰;«      Š5‰;D �cg c]  }|j.                  ‘Œ c}Š?dg‰5z  Š2t1        ‰-«      Š<‰ j2                  Š'‰ j4                  Š1‰ j6                  �‰ j6                  nt9        j:                  «       xs d}‰ j"                  �+dt<        j>                  dt0        t0        t@              fd„}|Š+n(‰ jB                  �‰ jB                  nt        jD                  Š+‰ jF                  �‰ jF                  ‰'z  nd Š3dg‰5z  Š*tI        ‰5«      D �cg c]  }tK        «       ‘Œ c}Š/tI        ‰5«      D �cg c]  }tK        «       ‘Œ c}Š=tI        ‰5«      D �cg c]  }tK        «       ‘Œ c}Š@tI        ‰5«      D �cg c]  }tK        «       ‘Œ c}Š(tM        jN                  |«      ŠBˆ fd„Šd	t@        dd fˆˆ'ˆ*ˆ-ˆ/ˆ0ˆ;ˆ?fd
„Š!d	t@        dd fˆ!ˆ*ˆ/ˆ1ˆ=ˆ?fd„Šd	t@        dd fˆ/ˆ=fd„Š‰ jP                  Š6dtR        dtT        fˆ6fd„Š dt0        dt@        dd fd„Šˆˆ ˆ+ˆ6ˆ fd„Š#ˆˆ ˆ#ˆ+fd„Š"ˆ"ˆ ˆBfd„Š%d	t@        dd fˆ%ˆ'ˆ(ˆ3ˆ=ˆ@ˆAˆBfd„Š$d	t@        dd fˆ(ˆ@fd„Šd	t@        dd fˆˆˆˆ$fd„Šd	t@        dd fˆˆ%ˆ(ˆ/ˆ=ˆ@ˆAˆBfd„ŠddœdtT        dd fˆ(ˆ*ˆ2ˆ5ˆ=ˆ ˆ?fd„Š&tW        t@        ‰ j"                  «      Š:tW        t@        ‰ jX                  «      Š)tW        t@        ‰ jZ                  «      }	‰ j"                  �'tW        t@        ‰ j\                  «      ‰ j^                  z  nd}
t1        ‰ j$                  «      Š9ta        ‰ jb                  «      Š8t1        ‰ jd                  «      Š7d	t@        dtf        th        t0        t@           f   fˆ4ˆ8fd„Šd	t@        dd fˆˆˆˆ(ˆ)ˆ2ˆ4ˆ9ˆ:f	d„}d	t@        dtf        th        tj        f   fˆˆ:fd„}d(ˆ7ˆ8ˆ9ˆ fd „}tm        ‰5‰1z  ¬!«      5 Š0tm        |¬!«      5 ŠA‰=‰ _         ‰(‰ _7        ‰*‰ _8        ‰?‰ _9        ‰2‰ _:        	 tI        ‰5«      D ]
  } ‰|«       Œ d	t@        fˆˆ(ˆ.ˆ2ˆ4ˆ<ˆ fd"„}d(ˆ&ˆ.ˆ2ˆ5ˆ>ˆ fd#„}‰ j"                  ��‰7‰4d      Š,tw        ˆ,ˆ7fd$„‰4dd  D «       «      rty        d%«      ‚‰7‰4d      |
k  r tI        ‰5«      D ]
  } ||«       Œ tI        ‰5«      D ]n  } ‰|«      d&   }t-        |«      ‰:k  r!|j{                  |	g‰:t-        |«      z
  z  «        ||«      }‰7‰4|   xx   dz  cc<    |«        |‰5dz
  k(  r ‰&«        |–— Œp ‰7‰4d      |
k  rŒ 	  ‰&d'¬«       d ‰ _         d ‰ _7        d ‰ _8        d ‰ _9        d ‰ _:        d d d «       d d d «       y t-        t}        ‰.«      «      dkD  rìt        ‰.«      }tI        ‰5«      D �cg c]  }‰.|   |k  r‰.|   ‰4|   |f‘Œ }}t�        j‚                  |«       |r¡t�        j„                  |«      \  }}} ‰|«       ‰(|   s>	  ‰&d'¬«       d ‰ _         d ‰ _7        d ‰ _8        d ‰ _9        d ‰ _:        d d d «       d d d «       y  ||«      }|dz   |k  rt�        j†                  ||dz   ‰4|   |f«        |«        |–— |rŒ¡	 d}tI        ‰5«      D ]  } ‰|«       ‰(|   rŒd'} n |rn,tI        ‰5«      D ]  } ||«      }|‰5dz
  k(  r |«        |–— Œ ŒR	  ‰&d'¬«       d ‰ _         d ‰ _7        d ‰ _8        d ‰ _9        d ‰ _:        	 d d d «       d d d «       y c c}w c c}w c c}w c c}w c c}w c c}w #  ‰&d'¬«       d ‰ _         d ‰ _7        d ‰ _8        d ‰ _9        d ‰ _:        w xY w# 1 sw Y   ŒaxY w# 1 sw Y   y xY w­w))NúqStreamingDataset does not support concurrent iteration. Only one active iterator per dataset instance is allowed.r   r   rþ   rl   r|   c                 óÞ   — | j                  d«      }|j                  s|j                  «       j                  rt        d«      ‚t	        t
        t
        t              |j                  «       «      S )Nr   z:pack_sequences does not support null token lists or values)rç   r  Úflattenr§   r	   r<   r0   rê   )rl   r  s     r'   Úarrow_tokensz2StreamingDataset._iter_owned.<locals>.arrow_tokens_  sY   € Ø$Ÿ|™|¨A›�Ø×*Ò*¨l×.BÑ.BÓ.D×.OÒ.OÜ$ØTóð ô œD¤¤c¡™O¨\×-CÑ-CÓ-EÓFÐFr&   c                 óä   •— t        j                  «       }| j                  |«      }‰xj                  |j                  z  c_        ‰xj
                  t        j                  «       |z
  z  c_        |S r5   )ÚtimeÚperf_counterr  rÝ   ÚnbytesrÞ   )ÚpermÚindicesÚt0rl   r7   s       €r'   Ú_io_callz.StreamingDataset._iter_owned.<locals>._io_call„  sZ   ø€ Ü×"Ñ"Ó$ˆBØ×%Ñ% gÓ.ˆEØ×Ò %§,¡,Ñ.ÕØ×Ò¤× 1Ñ 1Ó 3°bÑ 8Ñ8ÕØˆLr&   Úic                 óü   •— ‰|    ‰	|    z
  }|dk  ry t        ‰|«      }‰	|    }‰	| xx   |z  cc<   ‰|    }t        t        |||z   «      «      }‰
|    |z   }‰|    j                  |‰j	                  ‰||«      f«       y )Nr   )r
  r<   rì   ÚappendÚsubmit)r>  Ú	remainingÚfetchr  Úperm_ir;  Ú	abs_startr=  Ú
batch_sizeÚ
fetch_headÚinitial_positionsÚ
io_pendingÚio_poolÚpermutationsÚsplit_sizess          €€€€€€€€r'   Ú
_submit_ioz0StreamingDataset._iter_owned.<locals>._submit_io‹  s“   ø€ Ø# A™¨°A©Ñ6ˆIØ˜AŠ~ØÜ˜
 IÓ.ˆEØ˜q‘MˆEØ�q‹M˜UÑ"‹MØ! !‘_ˆFÜœ5 ¨°©Ó6Ó7ˆGØ)¨!Ñ,¨uÑ4ˆIØ�q‰M× Ñ  )¨W¯^©^¸HÀfÈgÓ-VÐ!WÕXr&   c                 óÊ   •— t        ‰|    «      t        ‰|    «      z   ‰k  rB‰|    ‰|    k  r6 ‰| «       t        ‰|    «      t        ‰|    «      z   ‰k  r‰|    ‰|    k  rŒ4y y y y r5   ©r¨   )r>  rM  rG  rI  rˆ   Úraw_batchesrL  s    €€€€€€r'   Ú_fill_ioz.StreamingDataset._iter_owned.<locals>._fill_io—  s{   ø€ ô �J˜q‘MÓ"¤S¨°Q©Ó%8Ñ8¸>ÒIØ˜q‘M K°¡NÒ2á˜1”ô �J˜q‘MÓ"¤S¨°Q©Ó%8Ñ8¸>ÒIØ˜q‘M K°¡NÔ2ð JØ2ð Jr&   c                 óú   •— ‰|    ru‰|    d   d   j                  «       r[‰|    j                  «       \  }}‰|    j                  ||j                  «       f«       ‰|    r‰|    d   d   j                  «       rŒYyyyy)z;Move completed I/O futures into raw_batches non-blockingly.r   r   N)ÚdoneÚpopleftr@  Úresult)r>  rE  ÚfutrI  rP  s      €€r'   Ú	_drain_ioz/StreamingDataset._iter_owned.<locals>._drain_io¦  s   ø€ à˜Q’- J¨q¡M°!Ñ$4°QÑ$7×$<Ñ$<Ô$>Ø!+¨A¡×!6Ñ!6Ó!8‘�	˜3Ø˜A‘×%Ñ% y°#·*±*³,Ð&?Ô@ð ˜Q’- J¨q¡M°!Ñ$4°QÑ$7×$<Ñ$<Ö$>�-Ð$>�-r&   rC   c                 óJ   •— ‰dk(  ryt        ‰«      rt         ‰| «      «      S y)Nr   FT)r³   r}   )rC   Úon_errors    €r'   Ú_should_skipz2StreamingDataset._iter_owned.<locals>._should_skip°  s(   ø€ Ø˜7Ò"ØÜ˜Ô!Ü™H S›MÓ*Ð*Ør&   Úrowsr	  c                 óV   — t        | «      |k7  rt        dt        | «      › d|› d�«      ‚y )Nztransform returned z rows for a batch of z�; transforms must return exactly one output row per input row.  To drop bad rows, raise inside the transform and pass on_transform_error='skip'.)r¨   r§   )r[  r	  s     r'   Ú_check_row_countz6StreamingDataset._iter_owned.<locals>._check_row_count·  s?   € Ü�4‹y˜HÒ$Ü Ø)¬#¨d«)¨Ð4IØ�jð !DðDóð ð %r&   c           	      ó¤  •— g }d}d}t        |j                  «      D ]F  }	 t         ‰|j                  |d«      «      «      } ‰	|d«       |j                  | |z   |d   f«       ŒH ‰xj                  |z  c_        |r+‰dk(  r&t        j                  d||j                  |�|n|«       |S # t        $ r} ‰
|«      s‚ |dz  }|€|}Y d}~Œ­d}~ww xY w)z=Re-run the transform on single-row slices, dropping failures.r   Nr   r›   z>Skipped %d of %d rows whose transform failed (first error: %r))	rì   r	  r<   ÚsliceÚ	Exceptionr@  rà   r£   r¤   )rE  rl   Ú	batch_excÚoutÚskippedÚ	first_excÚjr[  rC   r]  rZ  Úfinal_transformrY  r7   s            €€€€€r'   Ú_transform_isolatedz9StreamingDataset._iter_owned.<locals>._transform_isolatedÀ  sé   ø€ àˆCØˆGØˆIÜ˜5Ÿ>™>Ó*ò 5�ðÜ¡°·±¸A¸qÓ0AÓ BÓC�Dñ !  qÔ)Ø—
‘
˜I¨™M¨4°©7Ð3Õ4ð5ð ×Ò 'Ñ)ÕÙ˜8 vÒ-Ü—‘ØTØØ—N‘NØ!*Ð!6‘I¸Iô	ð ˆJøô# !ò Ù'¨Ô,ØØ˜q‘L�GØ Ð(Ø$'˜	Üûðús   ¡!B*Â*	CÂ3C
Ã
Cc                 óò   •— 	 t         ‰	|«      «      } ‰||j                  «       t        |«      D ��cg c]  \  }}| |z   |f‘Œ c}}S # t        $ r} ‰|«      s‚  ‰| ||«      cY d}~S d}~ww xY wc c}}w )z5Apply the transform, returning [(abs_pos, row), ...].N)r<   r`  r	  Ú	enumerate)
rE  rl   r[  rC   re  Úrowr]  rZ  rg  rf  s
         €€€€r'   Ú_transform_batchz6StreamingDataset._iter_owned.<locals>._transform_batchÛ  s}   ø€ ðBÜ™O¨EÓ2Ó3�ñ
 ˜T 5§>¡>Ô2Ü7@À³×G©V¨Q°�Y ‘] CÒ(ÓGÐGøô ò BÙ# CÔ(ØÙ*¨9°e¸SÓAÕAûðBüó
 Hs"   ƒA
 µA3Á
	A0ÁA+Á%A0Á+A0c                 óâ   •— 	 t        j                  «       } ‰| |«      }‰xj                  t        j                  «       |z
  z  c_        |‰j                  «        S # ‰j                  «        w xY wr5   )r7  r8  rß   Úrelease)rE  rl   r<  rU  rk  r7   Útx_semaphores       €€€r'   Ú_tx_call_guardedz6StreamingDataset._iter_owned.<locals>._tx_call_guardedæ  s]   ø€ ð'Ü×&Ñ&Ó(�Ù)¨)°UÓ;�Ø×$Ò$¬×(9Ñ(9Ó(;¸bÑ(@Ñ@Õ$Øà×$Ñ$Õ&ø�×$Ñ$Õ&ús   ƒAA ÁA.c                 ó  •— ‰|    r€‰�)t        ‰|    «      t        ‰	|    «      ‰z  z   }|‰z   ‰kD  ry‰j                  d¬«      sy‰|    j                  «       \  }}‰	|    j                  ‰
j	                  ‰||«      «       ‰|    rŒyy)z>Submit transforms for raw_batches[i] up to available capacity.NF)Úblocking)r¨   ÚacquirerT  r@  rA  )r>  Úin_pipelinerE  rl   ro  rF  ÚcookedÚmax_cooked_rowsrP  Ú
tx_pendingÚtx_poolrn  s       €€€€€€€€r'   Ú_try_submit_txz4StreamingDataset._iter_owned.<locals>._try_submit_txï  s—   ø€ à˜a’.ð #Ð.Ü"% f¨Q¡i£.´3°zÀ!±}Ó3EÈ
Ñ3RÑ"R�KØ" ZÑ/°/ÒAØØ#×+Ñ+°UÐ+Ô;ØØ#.¨q¡>×#9Ñ#9Ó#;Ñ �	˜5Ø˜1‘×$Ñ$ W§^¡^Ð4DÀiÐQVÓ%WÔXð ˜a•.r&   c                 óà   •— ‰|    rh‰|    d   j                  «       rQ‰|    j                  ‰|    j                  «       j                  «       «       ‰|    r‰|    d   j                  «       rŒOyyyy)z<Move completed transform futures into cooked non-blockingly.r   N)rS  ÚextendrT  rU  )r>  rt  rv  s    €€r'   Ú	_drain_txz/StreamingDataset._iter_owned.<locals>._drain_tx  sj   ø€ à˜Q’- J¨q¡M°!Ñ$4×$9Ñ$9Ô$;Ø�q‘	× Ñ  ¨A¡×!6Ñ!6Ó!8×!?Ñ!?Ó!AÔBð ˜Q’- J¨q¡M°!Ñ$4×$9Ñ$9Ö$;�-Ð$;�-r&   c                 óF   •—  ‰| «        ‰| «        ‰| «        ‰| «       y)z'Non-blocking pipeline pump for split i.Nr%   )r>  rW  r{  rQ  rx  s    €€€€r'   Ú_advancez.StreamingDataset._iter_owned.<locals>._advance  s    ø€ á�aŒLÙ�aŒLÙ˜1ÔÙ�Q�Kr&   c                 óä  •—  ‰| «       ‰|    sâ‰	|    r<‰|    j                  ‰	|    j                  «       j                  «       «        ‰| «       nš‰|    rL‰j                  «        ‰|    j                  «       \  }}‰	|    j	                  ‰
j                  ‰||«      «       nI‰|    rC‰|    j                  «       \  }}‰|    j	                  ||j                  «       f«        ‰| «       ny‰|    sŒáyy)z=Ensure cooked[i] has at least one row, blocking if necessary.N)rz  rT  rU  rr  r@  rA  )r>  rE  rl   rV  r}  ro  rt  rI  rP  rv  rw  rn  s       €€€€€€€€r'   Ú_ensure_cookedz4StreamingDataset._iter_owned.<locals>._ensure_cooked  så   ø€ á�QŒKØ˜Q’iØ˜a’=à˜1‘I×$Ñ$ Z°¡]×%:Ñ%:Ó%<×%CÑ%CÓ%EÔFÙ˜Q•KØ  ’^ð !×(Ñ(Ô*Ø'2°1¡~×'=Ñ'=Ó'?Ñ$�I˜uØ˜q‘M×(Ñ(ØŸ™Ð'7¸ÀEÓJõð   ’]à%/°¡]×%:Ñ%:Ó%<‘N�I˜sØ ‘N×)Ñ)¨9°c·j±j³lÐ*CÔDÙ˜Q•Kàð' ˜Q•ir&   F)Úidler€  c                 óˆ  •— ‰j                   }t        ˆˆfd„t        ‰«      D «       «      |d<   | rdnt        d„ ‰D «       «      |d<   | rdnt        d„ ‰D «       «      |d<   t        ‰«      |d<   ‰j                  |d<   t	        ‰j
                  d	z  «      |d
<   t	        ‰j                  d	z  «      |d<   ‰j                  |d<   y)z:Refresh pipeline statistics visible to the parent process.c              3   ó4   •K  — | ]  }‰|   ‰|   z
  –— Œ y ­wr5   r%   )Ú.0re  rG  rL  s     €€r'   ú	<genexpr>zFStreamingDataset._iter_owned.<locals>._update_stats.<locals>.<genexpr>*  s   øè ø€ ÒI¸1˜ A™¨°A©Õ6ÑIùó   ƒr   c              3   óH   K  — | ]  }|D ]  \  }}|j                   –— Œ Œ y ­wr5   ©r	  ©rƒ  ÚqÚ_rl   s       r'   r„  zFStreamingDataset._iter_owned.<locals>._update_stats.<locals>.<genexpr>.  s$   è ø€ ÒO¨AÈQÒOÁÀÀE˜Ÿ�ÐO˜ÑOùó   ‚ "r   c              3   ó2   K  — | ]  }t        |«      –— Œ y ­wr5   rO  ©rƒ  r‰  s     r'   r„  zFStreamingDataset._iter_owned.<locals>._update_stats.<locals>.<genexpr>0  s   è ø€ Ò&>°!¤s¨1§vÑ&>ùó   ‚é   é   r   é@B é   é   é   N)rÔ   r  rì   rÝ   r0   rÞ   rß   rà   )	r€  Úwsrt  rG  Úlocal_consumedÚnrP  r7   rL  s	     €€€€€€€r'   Ú_update_statsz3StreamingDataset._iter_owned.<locals>._update_stats'  s¿   ø€ à×#Ñ#ˆBÜÔIÄÀaÃÔIÓIˆBˆq‰Eñ ñ äÑO°ÔOÓOð ˆq‰Eñ
 ‘A¤3Ñ&>°vÔ&>Ó#>ˆBˆq‰EÜ˜Ó'ˆBˆq‰EØ×&Ñ&ˆBˆq‰EÜ˜×(Ñ(¨9Ñ4Ó5ˆBˆq‰EÜ˜×,Ñ,¨yÑ8Ó9ˆBˆq‰EØ×&Ñ&ˆBˆqŠEr&   c                 ó4   •— ‰j                  ‰|    g g dœ«      S )N©ÚtokensÚstarts)Ú
setdefault)r>  Ú	my_splitsÚpack_bufferss    €€r'   Ú_pack_bufferz2StreamingDataset._iter_owned.<locals>._pack_bufferD  s    ø€ Ø×*Ñ*¨9°Q©<ÀBÐRTÑ9UÓVÐVr&   c                 óv  •	—  ‰| «      }t        |d   «      ‰k  rŸ ‰| «       ‰|    sy|d   j                  t        |d   «      «       ‰|    j                  «       \  }}|d   j                  |«       |d   j                  ‰«       |dz   ‰‰
|    <   ‰	| xx   dz  cc<    ‰| «       t        |d   «      ‰k  rŒžyy)z8Fill split i's buffer to one block or exhaust the split.r›  Nrœ  r   )r¨   r@  rT  rz  )r>  ÚbufÚposr›  r}  r  r   rt  r�   r–  rž  Úpack_consumedr  s       €€€€€€€€€r'   Ú_fill_blockz1StreamingDataset._iter_owned.<locals>._fill_blockG  sÀ   ø€ á˜q“/ˆCÜ�c˜(‘mÓ$ xÒ/Ù˜qÔ!Ø˜a’yØØ�H‘×$Ñ$¤S¨¨X©Ó%7Ô8Ø$ Q™i×/Ñ/Ó1‘��VØ�H‘×$Ñ$ VÔ,Ø�H‘×$Ñ$ VÔ,Ø.1°A©g�˜i¨™lÑ+Ø˜qÓ! QÑ&Ó!Ù˜”ô �c˜(‘mÓ$ xÕ/r&   c                 óŒ  •—  ‰| «      }|d   |d   }}t        j                  ‰	t         j                  ¬«      }|D �cg c]  }d|cxk  r‰	k  sŒn n|‘Œ }}d||<   |j                  d¬«       t        j                  |d ‰	 t         j                  ¬«      |dœ}|d ‰	…= |D �cg c]  }|‰	k\  sŒ	|‰	z
  ‘Œ c}|d<   |S c c}w c c}w )Nr›  rœ  )Údtyper   r   )Údim)Ú	input_idsÚdoc_ids)ÚtorchÚzerosÚint64Úcumsum_Útensor)
r>  r¢  r›  rœ  rª  ÚsÚ
doc_startsÚblockr   r  s
           €€r'   Ú_emit_blockz1StreamingDataset._iter_owned.<locals>._emit_blockV  sÊ   ø€ Ù˜q“/ˆCØ  ™]¨C°©M�FˆFô —k‘k (´%·+±+Ô>ˆGØ%+Ö@ ¨q°1Ô/?°xÖ/?š!Ð@ˆJÐ@Ø"#ˆG�JÑØ�O‰O ˆOÔ"ä"Ÿ\™\¨&°°(Ð*;Ä5Ç;Á;ÔOØ"ñˆEð �y˜�yÐ!à39ÖK¨a¸QÀ(»]˜Q ›\ÒKˆC�‰MØˆLùò Aùò Ls   ½B<ÁB<Â"
CÂ-Cc            
      óÚ   •— t        ‰«      ‰_        ‰j                  «       D � �ci c]#  \  } }| t        |d   «      t        |d   «      dœ“Œ% c}} ‰_        t        ‰«      ‰_        y c c}} w )Nr›  rœ  rš  )r<   rÊ   ÚitemsrË   rÌ   )rÿ   ÚbufferÚpack_blocks_emittedrŸ  r¤  r7   s     €€€€r'   Ú_commit_pack_statez8StreamingDataset._iter_owned.<locals>._commit_pack_stateh  su   ø€ Ü"& }Ó"5ˆDÔð &2×%7Ñ%7Ó%9÷"ñ
 "�E˜6ð	 Ü" 6¨(Ñ#3Ó4Ü" 6¨(Ñ#3Ó4ññ ó"ˆDÔô )-Ð-@Ó(AˆDÕ%ùó"s   ¥(A')Úmax_workersc                 óÌ   •— ‰|    j                  «       \  }} ‰| «       ‰| xx   dz  cc<   |dz   ‰	| <   ‰|    }‰|    ‰|    z   ‰
j                  |<   ‰	|    ‰
j                  |<   |S )Nr   )rT  râ   rã   )r>  r£  rj  Ú	split_idxr}  rt  Úinitial_samplesr–  rž  Úpos_consumedr7   s       €€€€€€€r'   Ú
_yield_rowz0StreamingDataset._iter_owned.<locals>._yield_row�  s†   ø€ Ø#)¨!¡9×#4Ñ#4Ó#6™˜˜Sñ ! œØ& qÓ)¨QÑ.Ó)Ø*-°©'˜ Q™Ø$-¨a¡L˜	à+¨AÑ.°ÀÑ1BÑBð ×,Ñ,¨YÑ7ð =IÈ¹O˜×.Ñ.¨yÑ9Ø"˜
r&   c                  ó^   •— ‰s#t        ˆˆfd„t        ‰«      D «       «      ‰_         ‰ «        y )Nc              3   ó4   •K  — | ]  }‰|   ‰|   z   –— Œ y ­wr5   r%   )rƒ  re  r¼  r–  s     €€r'   r„  zOStreamingDataset._iter_owned.<locals>._update_progress_stats.<locals>.<genexpr>‘  s%   øè ø€ ò 6ØKL °Ñ 2°^ÀAÑ5FÕ Fñ6ùr…  )r
  rì   rá   )r˜  r¼  r–  r—  Úreal_workerr7   s   €€€€€€r'   Ú_update_progress_statsz<StreamingDataset._iter_owned.<locals>._update_progress_stats�  s-   ø€ Ù*Ü25ô 6ÜPUÐVWÓPXô6ó 3˜DÔ/ñ &�r&   c              3   ó.   •K  — | ]  }‰|   ‰k7  –— Œ y ­wr5   r%   )rƒ  rÿ   Úfirst_countr·  s     €€r'   r„  z/StreamingDataset._iter_owned.<locals>.<genexpr>˜  s#   øè ø€ ò à %ð 0°Ñ6¸+ÕEñùs   ƒz Packed checkpoint is not aligned across the splits owned by this iterator; merge every rank state with merge_state_dicts before resuming on a different topologyr›  Tr{   )DrÍ   r>   r   rÕ   r)  r   r  r´   rå   r½   r  r  r   r  râ   rR   rá   rÂ   rÊ   rã   Ú	with_skipr@  r¨   r	  r<   r»   r¼   rÁ   ÚosÚ	cpu_countr«   ÚRecordBatchr0   rÀ   Úarrow2pythonrÇ   rì   r   rØ   Ú	SemaphorerÆ   r`  r}   r	   rÃ   rÄ   rÅ   rµ   r   rË   rÌ   r+   Ústrr   r   rÎ   rÏ   rÐ   rÑ   Úanyr§   rz  rë   r  ÚheapqÚheapifyÚheappopÚheappush)Cr7   rM   r»  r:  Úsample_countÚ	start_posÚtransform_workersr5  rŠ  r�   Úblocks_per_splitr¥  r³  r¸  r>  r¾  rÂ  r›  r²  Úcatch_up_toÚpendingÚconsumedrj  Ú	exhaustedr}  r]  rW  r{  r  rQ  r=  r   rZ  rM  rk  rg  rx  ro  r˜  rF  rt  r�   rG  rf  rÄ  rH  r¼  rI  rJ  rˆ   r–  ru  rž  r—  rY  r·  rŸ  r¤  r  rK  r½  rP  rÁ  rL  rv  rw  rn  sC   `                       @@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@r'   r/  zStreamingDataset._iter_owned#  s9	  ÿÿÿÿÿûè ø€ ð × Ñ Ð,ÜðLóð ô &Ó'¨tÐ3ˆÙÑ<Ø23ˆD×,Ñ,¨QÑ/à×+Ñ+Ó-ˆ	ÙØð +-ˆØ%'ˆØ')ÐØ"ó 	&ˆIÜ×*Ñ*Ø—‘˜T×-Ñ-°YôˆDð �}‰}Ð(Ø×*Ñ*¨4¯=©=Ó9�Ø×&Ñ&¤z×'=Ñ'=Ó>ˆDØ×/Ñ/×3Ñ3°I¸t×?RÑ?RÓSˆLð
 ×'Ñ'Ð3ð ×#Ñ# IÒ.à×+Ñ+×/Ñ/°	¸<ÓHð ð
 ˜1Š}Ø—~‘~ iÓ0�Ø×"Ñ" <Ô0Ø×$Ñ$ YÔ/Ø×Ñ Ö%ð'	&ô* �ÓˆØ1=Ö>¨�t—}“}Ò>ˆØ˜˜q™ˆô Ð-Ó.ˆà×*Ñ*ˆ
Ø×-Ñ-ˆð ×*Ñ*Ð6ð ×'Ò'ä—,‘,“.Ò% Að 	ð ×ÑÐ+ðG¤B§N¡Nð G´t¼DÄ¹I±ó Gð +‰Oð —?‘?Ð.ð —’ä×,Ñ,ð ð ×*Ñ*Ð6ð ×'Ñ'¨*Ò4àð 	ð �S˜1‘Wˆ
Ü',¨Q£xÖ0 !”e•gÒ0ˆ
Ü(-¨a«Ö1 1”u•wÒ1ˆÜ',¨Q£xÖ0 !”e•gÒ0ˆ
Ü#(¨£8Ö,˜a”%•'Ò,ˆô !×*Ñ*Ð+<Ó=ˆô	ð
	Yœ#ð 
	Y $÷ 
	Yô 
	Yð	œð 	 ÷ 	ò 	ð	Aœð 	A ö 	Að ×+Ñ+ˆð	œið 	¬Dõ 	ð	¤4ð 	´3ð 	¸4ó 	÷	ð 	÷6		Hö	'ð	Yœcð 	Y d÷ 	Yô 	Yð$	Cœð 	C ö 	Cð	œð 	 ÷ 	ð 	ð	œcð 	 d÷ 	ô 	ð0 +0ò 	'¤4ð 	'°D÷ 	'ó 	'ô" œ˜T×1Ñ1Ó2ˆÜ”c˜4Ÿ<™<Ó(ˆÜ”c˜4Ÿ<™<Ó(ˆð ×#Ñ#Ð/ô ”�d×,Ñ,Ó-°×1AÑ1AÒAàð 	ô
 ˜T×0Ñ0Ó1ˆÜ × 2Ñ 2Ó3ˆÜ" 4×#<Ñ#<Ó=Ðð	WœCð 	W¤D¬¬d´3©i¨Ñ$8ö 	Wð	œ3ð 	 4÷ 	õ 	ð	œ3ð 	¤4¬¬S¨¡>ö 	÷$		Bð 		Bô  ¨A°Ñ,>Ô?ð B	4À7Ü#Ð0AÔBð A4ÀgØ(3�Ô%Ø#)�Ô Ø'1�Ô$Ø(3�Ô%Ø+9�Ô(ðz4Ü" 1›Xò $˜Ù  �ð$ð#¤c÷ #ó #÷(ò (ð ×+Ñ+Ñ7Ø&9¸)ÀA¹,Ñ&G˜Üô à)2°1°2¨ôô ô #-ð!5ó#ð ð 2°)¸A±,Ñ?ÐBRÒRô &+¨1£Xò / Ù +¨A¥ð/ô &+¨1£Xò , Ù)5°a«¸Ñ)B Ü#& v£;°Ò#9Ø$*§M¡M°6°(¸hÌÈVËÑ>TÑ2UÔ$VÙ(3°A« Ø 3°I¸a±LÓ AÀQÑ FÓ Añ !3Ô 4Ø#$¨¨A©¢:Ù$1¤OØ&+£ð,ð 2°)¸A±,Ñ?ÐBRÓRð& ñx " tÕ,Ø,0�DÔ)Ø'+�DÔ$Ø+/�DÔ(Ø,0�DÔ)Ø/3�DÔ,÷CA4÷B	4ð B	4ôN œ3˜Ó/Ó0°1Ò4Ü&)¨/Ó&:˜ô &+¨1£Xö#à !Ø.¨qÑ1°KÒ?ð -¨QÑ/°¸1±¸qÒAð#˜ð #ô
 Ÿ™ gÔ.Ù%Ü-2¯]©]¸7Ó-C™N˜H a¨Ù*¨1Ô-Ø#)¨!¢9Ø &ñT " tÕ,Ø,0�DÔ)Ø'+�DÔ$Ø+/�DÔ(Ø,0�DÔ)Ø/3�DÔ,÷CA4÷B	4ð B	4ñh #-¨Q£-˜CØ'¨!™|¨kÒ9Ü %§¡¨w¸ÀA¹ÀyÐQRÁ|ÐUVÐ8WÔ XÙ2Ô4Ø"%šIò &ð ð %*˜	Ü!& q£ò &˜AÙ*¨1Ô-Ø#)¨!£9Ø,0 	Ù %ð	&ñ
 %Ø!ä!& q£ò 
&˜AÙ",¨Q£-˜Cð  ! A¨¡EšzÙ 6Ô 8à"%›Ið
&ð' ð" "ñ$ " tÕ,Ø,0�DÔ)Ø'+�DÔ$Ø+/�DÔ(Ø,0�DÔ)Ø/3�DÕ,÷CA4÷B	4ð B	4ùòO	 ?ùòX 1ùÚ1ùÚ0ùÚ,ùòB
#øñh " tÕ,Ø,0�DÔ)Ø'+�DÔ$Ø+/�DÔ(Ø,0�DÔ)Ø/3�DÕ,ú÷CA4ð A4ú÷B	4ð B	4üsÛ   ®F cÆ.a1ÇC+cÊ,a6Ê=cËa;ËcË.b Ë?cÌbÌ H4cÕcÕ!$c ÖDbÚ,c ÛcÛ	cÛ/bÜb
Ü%?bÝ%,c ÞcÞ	cÞ"<bßbß>4bà3,c á cá("câ
bâ.b=â=c ã c		ãcãcãcc                 ób   — | j                   �| j                  S t        | j                  d   «      S )a^  Cumulative bytes of raw Arrow buffer data fetched from storage.

        Measured on the ``RecordBatch`` before any transform is applied, so
        the value reflects actual I/O rather than the size of transformed
        output.  Accumulates across multiple iterations of the same dataset
        instance and is never reset automatically.
        r   )rÍ   rÝ   r0   rÔ   re   s    r'   Úbytes_loadedzStreamingDataset.bytes_loadedù  ó2   € ð × Ñ Ð,Ø×%Ñ%Ð%Ü�4×%Ñ% aÑ(Ó)Ð)r&   c                 óV   — | j                   �| j                  S | j                  d   dz  S )zòCumulative seconds spent waiting for data from LanceDB.

        Measured per batch in the Stage 1 I/O threads as the total elapsed
        time of the ``take_offsets`` call.  Accumulates across all splits and
        all iterations.
        r’  r‘  )rÍ   rÞ   rÔ   re   s    r'   Ú
fetch_timezStreamingDataset.fetch_time  s2   € ð × Ñ Ð,Ø×#Ñ#Ð#Ø×!Ñ! !Ñ$ yÑ0Ð0r&   c                 óV   — | j                   �| j                  S | j                  d   dz  S )a:  Cumulative seconds spent applying the transform.

        Measured per batch in the Stage 2 transform threads as the elapsed
        time inside the transform callable (or the default ``arrow2python``
        conversion when no transform is set).  Accumulates across all splits
        and all iterations.
        r“  r‘  )rÍ   rß   rÔ   re   s    r'   Útransform_timezStreamingDataset.transform_time  s2   € ð × Ñ Ð,Ø×'Ñ'Ð'Ø×!Ñ! !Ñ$ yÑ0Ð0r&   c                 ó‚   — | j                   �t        d„ | j                   D «       «      S t        | j                  d   «      S )a  Number of raw rows waiting for a transform thread across all splits.

        A persistently non-zero value means Stage 2 (transform) is the
        bottleneck: I/O is completing faster than transforms can consume
        batches.  Returns 0 when not iterating.
        c              3   óH   K  — | ]  }|D ]  \  }}|j                   –— Œ Œ y ­wr5   r‡  rˆ  s       r'   r„  z3StreamingDataset.raw_queue_depth.<locals>.<genexpr>(  s%   è ø€ ÒW¨!ÐUVÒWÉÈÈE�u—~•~ÐW�~ÑWùr‹  r   )rÍ   r  r0   rÔ   re   s    r'   Úraw_queue_depthz StreamingDataset.raw_queue_depth  s=   € ð × Ñ Ð,ÜÑW¨t×/DÑ/DÔWÓWÐWÜ�4×%Ñ% aÑ(Ó)Ð)r&   c                 ó‚   — | j                   �t        d„ | j                   D «       «      S t        | j                  d   «      S )u  Number of rows transformed and ready to yield across all splits.

        Counts rows whose transform has completed and are sitting in memory
        waiting for the main thread â€” rows that can be handed off with no
        I/O or CPU wait.  Returns 0 when not iterating.
        c              3   ó2   K  — | ]  }t        |«      –— Œ y ­wr5   rO  r�  s     r'   r„  z8StreamingDataset.prefetch_queue_depth.<locals>.<genexpr>4  s   è ø€ Ò8 !”s˜1—vÑ8ùrŽ  r�  )rÎ   r  r0   rÔ   re   s    r'   Úprefetch_queue_depthz%StreamingDataset.prefetch_queue_depth+  s=   € ð ×ÑÐ'ÜÑ8 t×'7Ñ'7Ô8Ó8Ð8Ü�4×%Ñ% aÑ(Ó)Ð)r&   c                 óª   — | j                   �0t        d„ t        | j                  | j                   «      D «       «      S t	        | j
                  d   «      S )a  Number of rows not yet submitted to the I/O stage across all splits.

        Decreases as the I/O stage submits fetch requests.  When this reaches
        zero all data has been requested from storage (though it may not have
        arrived yet).  Returns 0 when not iterating.
        c              3   ó,   K  — | ]  \  }}||z
  –— Œ y ­wr5   r%   )rƒ  ÚsizeÚheads      r'   r„  z2StreamingDataset.unscanned_rows.<locals>.<genexpr>@  s    è ø€ ò á�D˜$ð �t•ñùs   ‚r   )rÏ   r  ré   rÐ   r0   rÔ   re   s    r'   Úunscanned_rowszStreamingDataset.unscanned_rows7  sU   € ð ×ÑÐ+Üñ ä"% d×&;Ñ&;¸T×=QÑ=QÓ"Rôó ð ô �4×%Ñ% aÑ(Ó)Ð)r&   c                 ób   — | j                   �| j                  S t        | j                  d   «      S )aK  Number of rows dropped because their transform raised an exception.

        Only ever non-zero when ``on_transform_error`` is set to ``"skip"``,
        ``"warn"``, or a callable that returned ``True``.  Accumulates across
        multiple iterations of the same dataset instance and is never reset
        automatically.
        r”  )rÍ   rà   r0   rÔ   re   s    r'   Úrows_skippedzStreamingDataset.rows_skippedF  rÛ  r&   c                 ót   — | j                   �t        | j                   «      S t        | j                  d   «      S )z¦Number of rows already yielded to the caller across all splits.

        Monotonically increases throughout iteration.  Returns 0 when not
        iterating.
        r�  )rÑ   r  r0   rÔ   re   s    r'   Úconsumed_rowszStreamingDataset.consumed_rowsS  s7   € ð ×#Ñ#Ð/Ü�t×/Ñ/Ó0Ð0Ü�4×%Ñ% aÑ(Ó)Ð)r&   c                 óH  — | j                   j                  «       }| j                  j                  |d<   | j                  �d|d<   nt        | j                  «      |d<   | j                  j                  | j                  j                  «       f|d<   dD ]  }d||<   Œ	 d|d<   |S )a   Support pickling for multi-worker DataLoader (forkserver / spawn).

        The live LanceDB table object contains non-picklable connection state
        (sockets, Rust-backed PyO3 objects).  If a ``connection_factory`` was
        supplied only the table name is serialised; the factory is called in
        the worker to reopen the connection without embedding any credentials.
        Without a factory the table's own picklable reopen state is captured
        via ``_table_to_pickle_state`` (mirrors the ``Permutation`` approach).
        Ú_table_nameNr´   rå   )rÍ   rÎ   rÏ   rÐ   rÑ   rÚ   )rQ   Úcopyr´   rU   rÈ   r   rå   ræ   )r7   r*   Úkeys      r'   Ú__getstate__zStreamingDataset.__getstate__^  s®   € ð —‘×"Ñ"Ó$ˆà#Ÿ{™{×/Ñ/ˆˆmÑØ×#Ñ#Ð/Ø"ˆE�(ŠOä4°T·[±[ÓAˆE�(‰Oð ×Ñ×!Ñ!Ø×Ñ×%Ñ%Ó'ð 
ˆˆmÑð
ò 	ˆCð ˆE�#ŠJð	ð ,0ˆÐ'Ñ(Øˆr&   c                 óª  — ddl m} |j                  d«      }|j                  d«      }|j                  d«      \  }}| j                  j	                  |«       t        j                  «       | _        | j                  �| j                  |«      | _	        n#t        |«      | _	        |d   dk(  rt        |«      } |d	«      j                  ||«      | _        y)
z:Reconnect to LanceDB after unpickling in a worker process.r   )Úconnectrð  r´   rå   NÚkindÚmemoryz	memory://)Ú rõ  ÚpoprQ   ÚupdaterØ   rÙ   rÚ   rÈ   r´   r   r   Úcreate_tablerå   )r7   r*   Ú_connectÚ
table_nameÚtable_stateÚ	perm_nameÚ	perm_datas          r'   Ú__setstate__zStreamingDataset.__setstate__€  s°   € å)à—Y‘Y˜}Ó-ˆ
Ø—i‘i Ó)ˆØ$Ÿy™y¨Ó7Ñˆ	�9Ø�‰×Ñ˜UÔ#Ü'0§~¡~Ó'7ˆÔ$Ø×#Ñ#Ð/Ø×2Ñ2°:Ó>ˆD�Kä2°;Ó?ˆDŒKØ˜6Ñ" hÒ.ä.¨yÓ9�	Ù# KÓ0×=Ñ=¸iÈÓSˆÕr&   c                 ó  — | j                   d   rt        «       €t        d«      ‚| j                  �t        d| j                  › d�«      ‚| j	                  «       }| j
                  �C| j                  D �cg c]
  }|d   |   ‘Œ }}t        t        |«      «      dkD  rt        d«      ‚|S | j                  rC|d   }| j                  D �cg c]  }||   ‘Œ	 }}t        t        |«      «      dkD  rt        d	«      ‚|S c c}w c c}w )
aÄ  Snapshot the dataset's consumption state.

        When using DataLoader workers, construct a
        [StreamingDataLoader][lancedb.streaming.StreamingDataLoader].  It
        commits worker state only when a prefetched batch is returned to the
        trainer.  A standard multi-process ``DataLoader`` cannot expose that
        boundary, so calling this method after one has started raises
        ``RuntimeError`` instead of returning stale producer state.

        In row mode, the returned dict is topology-independent at global step
        boundaries. ``positions_consumed_per_split`` records how far each
        split's permutation has advanced, which can differ from the sample
        count when ``on_transform_error`` skips rows. ``StreamingDataLoader``
        combines worker state in its parent process. Combine state dicts from
        every rank with
        [merge_state_dicts][lancedb.streaming.StreamingDataset.merge_state_dicts]
        before resuming on a different topology.

        Packed state includes partial token buffers and emitted block counts
        for every logical split. When packing is sharded, merge every rank
        state with ``merge_state_dicts`` before loading it.
        r   z¬StreamingDataset cannot checkpoint a standard DataLoader with num_workers > 0 because prefetched worker progress is not consumer-committed. Use StreamingDataLoader instead.z2StreamingDataset checkpointing is invalid because zH. Load the last valid checkpoint into a fresh dataset before continuing.Úblocks_emitted_per_splitr   zÒPacked StreamingDataset checkpointing is only safe at a complete logical step boundary, when every split assigned to this rank has emitted the same block count. Consume more batches before calling state_dict().Úsamples_consumed_per_splitzæStreamingDataLoader checkpointing with multiple workers is only safe at a complete logical step boundary, when every split assigned to this rank has the same consumed-sample count. Consume more batches before calling state_dict().)
rÕ   r   r>   rÖ   r?   rÂ   rô   r¨   rë   r×   )r7   r*   rÿ   Úrank_blocksr@   Úrank_sampless         r'   Ú
state_dictzStreamingDataset.state_dict’  sA  € ð. ×+Ñ+¨AÒ.´?Ó3DÐ3LÜðGóð ð
 ×*Ñ*Ð6ÜØDØ×2Ñ2Ð3ð 4EðEóð ð
 ×)Ñ)Ó+ˆØ×ÑÐ+àFJ×FWÑFWöØ=B�Ð0Ñ1°%Ó8ðˆKð ô ”3�{Ó#Ó$ qÒ(Ü"ð;óð ð  ˆð ×7Ò7ØÐ8Ñ9ˆGØ8<×8IÑ8IÖJ¨u˜G E›NÐJˆLÐJÜ”3�|Ó$Ó%¨Ò)Ü"ðOóð ð ˆùò)ùò Ks   Á4C:ÃC?c                 óÆ  — | j                   €Ê| j                  j                  «       }t        j                  «       }|j                  t        |j                  «      j                  «       «       dD ]P  }|j                  |«      j                  «       }|j                  |j                  d¬«      j                  «       «       ŒR |j                  «       | _         | j                   S )aÚ  Cheap identity check for the built permutation, computed once and cached
        (``_perm_table`` never changes after construction).

        Only meaningful -- and only recorded in a checkpoint -- for the
        ``split_strategy`` path (see ``_checkpoint_snapshot``): ``num_splits``/
        ``shuffle_seed`` alone cannot detect a `load_state_dict` resuming saved
        per-split progress against a *different* permutation than the one that
        produced it, since a ``split_strategy`` callable ignores ``shuffle_seed``
        entirely -- e.g. a ``bootstrap_sample`` with no seed (or any changed
        callable) can rebuild a completely different permutation on the next run
        while ``shuffle_seed`` keeps its same default value and the existing check
        passes anyway. ``split_random``/``split_sequential`` are already fully
        pinned by ``shuffle_seed``/``num_splits`` and must *not* additionally be
        fingerprinted: this hashes physical row ids, which are not stable across
        compaction by default, so checking it on those paths would reject an
        otherwise-valid resume after the table gets compacted between save and
        load even though the seeded permutation itself is unchanged.
        )Úrow_idr�   F)Úzero_copy_only)rò   rå   ræ   ÚhashlibÚsha256rú  rË  r	  Úencoderç   Úcombine_chunksÚto_numpyÚtobytesÚ	hexdigest)r7   Ú	arrow_tblÚhasherrU   rç   s        r'   Ú_perm_fingerprintz"StreamingDataset._perm_fingerprintÍ  s¹   € ð( ×'Ñ'Ð/Ø×(Ñ(×1Ñ1Ó3ˆIÜ—^‘^Ó%ˆFØ�M‰Mœ#˜i×0Ñ0Ó1×8Ñ8Ó:Ô;Ø.ò O�Ø"×)Ñ)¨$Ó/×>Ñ>Ó@�Ø—‘˜fŸo™o¸U˜oÓC×KÑKÓMÕNðOð ,2×+;Ñ+;Ó+=ˆDÔ(Ø×+Ñ+Ð+r&   c                 óè  — | j                   rd| j                  ini }| j                  �‘| j                  | j                  dœ|¥| j
                  | j                  | j                  | j                  | j                  t        | j                  «      t        | j                  «      t        | j                  «      dœ¥S t        | j                  «      D �cg c](  }| j                  j!                  || j"                  «      ‘Œ* }}t        | j                  «      D �cg c]!  }| j$                  j!                  |||   «      ‘Œ# }}| j                  | j                  dœ|¥| j
                  ||dœ¥S c c}w c c}w )NÚperm_fingerprint)rƒ   r€   )r„   rŽ   r�   r�   r‘   r  r  rŸ  )r„   r  Úpositions_consumed_per_split)rñ   r  rÂ   r·   rµ   r¸   rÃ   rÄ   rÅ   r<   rÊ   rÌ   r   rË   rì   râ   rR   rá   rã   )r7   r  rÿ   r@   Ú	positionss        r'   r?   z%StreamingDataset._checkpoint_snapshotë  s{  € ð ×)Ò)ð   ×!7Ñ!7Ñ8àð 	ð
 ×ÑÐ+à $× 2Ñ 2Ø"×.Ñ.ñð #ðð Ÿ™Ø"&×"6Ñ"6ØŸ,™,ØŸ,™,Ø$(×$:Ñ$:Ü.2°4×3FÑ3FÓ.GÜ,0°×1JÑ1JÓ,KÜ (¨×);Ñ);Ó <òð ô ˜t×/Ñ/Ó0ö
àð × Ñ ×$Ñ$ U¨D×,?Ñ,?Õ@ð
ˆð 
ô ˜t×/Ñ/Ó0ö
àð ×"Ñ"×&Ñ& u¨g°e©nÕ=ð
ˆ	ð 
ð
 !×.Ñ.Ø×*Ñ*ñ
ð ð
ð —[‘[Ø*1Ø,5ò
ð 	
ùò
ùò
s   Ã-E*Ä&E/Úreasonc                 ó,   — | j                   €|| _         y y r5   )rÖ   )r7   r  s     r'   rj   z'StreamingDataset._invalidate_checkpoint  s   € Ø×*Ñ*Ð2Ø.4ˆDÕ+ð 3r&   c                 ó   — | j                   d uS r5   )rÜ   re   s    r'   r,  z*StreamingDataset._consumer_iterator_active  s   € à×,Ñ,°DÐ8Ð8r&   c                 ó:   — | j                   }|�|j                  S d S r5   )rÜ   r.   ©r7   Úleases     r'   Ú_consumer_iterator_ownerz)StreamingDataset._consumer_iterator_owner  s#   € à×-Ñ-ˆØ$)Ð$5ˆu× Ñ Ð?¸4Ð?r&   c                 ó:   — | j                   }|�|j                  S d S r5   )rÜ   r/   r  s     r'   Ú_consumer_iterator_owner_threadz0StreamingDataset._consumer_iterator_owner_thread!  s#   € à×-Ñ-ˆØ%*Ð%6ˆu×!Ñ!Ð@¸DÐ@r&   c                 ó
  — | j                   5  | j                  s| j                  �t        d«      ‚t	        j
                  «       }| j                  dz   }t        ||«      }|| _        || _        |cddd«       S # 1 sw Y   yxY w)z>Reserve this parent dataset for one checkpoint-aware iterator.Nr2  r   )	rÚ   r,  rÍ   r>   rØ   Ú	get_identrÛ   r-   rÜ   )r7   r/   r.   r  s       r'   r-  z+StreamingDataset._acquire_consumer_iterator&  s†   € à×)Ñ)ñ 	Ø×-Ò-°×1FÑ1FÐ1RÜ"ðPóð ô %×.Ñ.Ó0ˆLØ×<Ñ<¸qÑ@ˆKÜ*¨;¸ÓEˆEØ1<ˆDÔ.Ø,1ˆDÔ)Ø÷	÷ 	ò 	ús   �A"A9Á9Br.   c                 óŽ   — | j                   5  | j                  }|�|j                  |k(  rd | _        d d d «       y # 1 sw Y   y xY wr5   )rÚ   rÜ   r.   )r7   r.   r  s      r'   ro   z+StreamingDataset._release_consumer_iterator5  sF   € Ø×)Ñ)ñ 	5Ø×1Ñ1ˆEØÐ  U×%6Ñ%6¸+Ò%EØ04�Ô-÷	5÷ 	5ñ 	5ús	   �%;»Ar0  c                 óÒ   — t        j                  «       j                  }| j                  5  | j                  }|�||ur|j
                  |k(  rd| _        ddd«       y# 1 sw Y   yxY w)zGClean up when an interrupted acquire set a lease but did not return it.N)rØ   Úcurrent_threadÚidentrÚ   rÜ   r/   )r7   r0  r/   r  s       r'   r.  z@StreamingDataset._release_consumer_iterator_after_failed_acquire;  se   € ô !×/Ñ/Ó1×7Ñ7ˆØ×)Ñ)ñ 	5Ø×1Ñ1ˆEàÐ!Ø Ñ/Ø×&Ñ&¨,Ò6à04�Ô-÷	5÷ 	5ñ 	5ús   «)AÁA&r*   r[   c          	      ó¤  ‡ — d‰ j                   fd‰ j                  fd‰ j                  ffD ];  \  }}|j                  |«      |k7  sŒt	        |› d|j                  |«      › d|› �«      ‚ d|v }|‰ j
                  duk7  rt	        d«      ‚|�r�d	D ]G  }t        ‰ d
|› �«      }|j                  |«      |k7  sŒ't	        |› d|j                  |«      › d|› �«      ‚ |d   }|d   }t        |«      ‰ j                  k7  st        |«      ‰ j                  k7  rt	        d«      ‚|d   }t        t        ||«      «      D ]Ù  \  }	\  }
}t        |«      t        |
«      f}‰ j                  |	   ‰ j                  |	   f}||kD  sŒC|d   ‰ j                  |	<   |d   ‰ j                  |	<   |j                  |	|j                  t        |	«      «      «      }|€‰ j                  j                  |	d«       Œ°t!        |d   «      t!        |d   «      dœ‰ j                  |	<   ŒÛ ‰ xj"                  |z  c_        y|d   }|j                  d|«      }t        |«      D ]M  \  }	}
‰ j$                  j                  |	‰ j&                  «      }t)        |t        |
«      «      ‰ j$                  |	<   ŒO t        |«      D ]g  \  }	}‰ j*                  j                  |	‰ j$                  j                  |	‰ j&                  «      «      }t)        |t        |«      «      ‰ j*                  |	<   Œi t-        ˆ fd„t/        ‰ j                  «      D «       «      ‰ _        ‰ xj"                  |z  c_        y)z:Merge one trainer-consumed worker batch into parent state.rƒ   r€   r„   z  mismatch in worker checkpoint: ú != rŸ  Nz1worker checkpoint mode does not match the dataset©rŽ   r�   r�   r‘   rŠ  r  r  z9packed worker checkpoint must contain one entry per splitr   r   r›  rœ  rš  r  c              3   ój   •K  — | ]*  }‰j                   j                  |‰j                  «      –— Œ, y ­wr5   )râ   rR   rá   )rƒ  rÿ   r7   s     €r'   r„  z8StreamingDataset._commit_worker_state.<locals>.<genexpr>…  s2   øè ø€ ò "
àð × Ñ ×$Ñ$ U¨D×,?Ñ,?×@ñ"
ùs   ƒ03)r·   rµ   r¸   rR   r§   rÂ   rT   r¨   ri  ré   r0   rÌ   rÊ   rË  rË   rù  r<   r×   râ   rá   r  rã   r
  rì   )r7   r*   r[   rò  ÚexpectedÚpackedr@   ÚemittedÚbuffersrÿ   Úcountr   ÚincomingÚcurrentr¶  r  Úpositions   `                r'   rk   z%StreamingDataset._commit_worker_stateI  sƒ  ø€ ð ˜T×/Ñ/Ð0Ø˜4×+Ñ+Ð,Ø�d—k‘kÐ"ð
ò 		‰MˆC�ð
 �y‰y˜‹~ Ó)Ü Ø�eÐ;Ø—y‘y “~Ð& d¨8¨*ð6óð ð		ð   5Ð(ˆØ�d×*Ñ*°$Ð6Ò7ÜÐPÓQÐQÚØQò �Ü" 4¨1¨S¨E¨Ó3�Ø—9‘9˜S“> XÓ-Ü$Ø˜%Ð?Ø Ÿ9™9 S›>Ð*¨$¨x¨jð:óð ðð Ð8Ñ9ˆGØÐ6Ñ7ˆGÜ�7‹|˜t×/Ñ/Ò/´3°w³<À4×CSÑCSÒ3SÜ ØOóð ð ˜NÑ+ˆGÜ*3´C¸ÀÓ4IÓ*Jò Ñ&�‘˜˜vÜ ›K¬¨U«Ð4�à×-Ñ-¨eÑ4Ø×'Ñ'¨Ñ.ð�ð ˜gÓ%Ø7?À±{�D×-Ñ-¨eÑ4Ø19¸!±�D×'Ñ'¨Ñ.Ø$Ÿ[™[¨°·±¼CÀ»JÓ0GÓH�FØ�~Ø×*Ñ*×.Ñ.¨u°dÕ;ô '+¨6°(Ñ+;Ó&<Ü&*¨6°(Ñ+;Ó&<ñ5˜×*Ñ*¨5Ò1ðð" ×6Ò6¸/ÑIÕ6ØàÐ4Ñ5ˆØ—I‘IÐ<¸gÓFˆ	Ü% gÓ.ò 	C‰LˆE�5Ø×*Ñ*×.Ñ.¨u°d×6IÑ6IÓJˆGÜ*-¨g´s¸5³zÓ*BˆD× Ñ  Ò'ð	Cô  )¨Ó3ò 	H‰OˆE�8Ø×,Ñ,×0Ñ0Ø�t×+Ñ+×/Ñ/°°t×7JÑ7JÓKóˆGô -0°¼¸X»Ó,GˆD×"Ñ" 5Ò)ð		Hô
 "ó "
ä˜t×/Ñ/Ó0ô"
ó 
ˆÔð 	×2Ò2°oÑEÖ2r&   c           
      ó®  — |d   | j                   k7  rt        d|d   › d| j                   › �«      ‚|d   | j                  k7  rt        d|d   › d| j                  › �«      ‚|j                  d«      }|�|| j                  k7  rt        d«      ‚d	| _        d
|v s| j                  �ßdD ]G  }t        | d|› �«      }|j                  |«      |k7  sŒ't        |› d|j                  |«      › d|› �«      ‚ |d   D �cg c]  }t        |«      ‘Œ c}| _	        |d   D �cg c]  }t        |«      ‘Œ c}| _
        |d
   j                  «       D ��ci c],  \  }}t        |«      t        |d   «      t        |d   «      dœ“Œ. c}}| _        y|d   }t        |t        «      rB|rt        |«      nd| _        t#        |«      D �	�
ci c]  \  }	}
|	t        |
«      “Œ c}
}	| _        nt        |«      | _        i | _        |j                  d«      }|€i | _        yt#        |«      D �	�ci c]  \  }	}|	t        |«      “Œ c}}	| _        yc c}w c c}w c c}}w c c}
}	w c c}}	w )a  Resume from a previously snapshotted state.

        Raises ``ValueError`` if ``num_splits`` or ``shuffle_seed`` differ
        from the checkpoint, since a different split structure or shuffle order
        makes mid-epoch resumption meaningless. Also raises if the checkpoint
        was built with ``split_strategy`` and records a permutation fingerprint
        that does not match the current dataset's permutation -- this catches a
        changed or non-deterministically-reseeded ``split_strategy`` that
        ``num_splits``/``shuffle_seed`` alone cannot detect, since ``shuffle_seed``
        is unused on that path. Checkpoints from ``split_random``/
        ``split_sequential`` never carry this fingerprint (row ids are not stable
        across compaction, so it would reject an otherwise-valid resume), and
        continue to rely on ``shuffle_seed``/``num_splits`` alone as before.
        Packed checkpoints pin ``pack_sequences``, ``eos_id``, ``pad_id``,
        ``blocks_per_epoch``, and ``epoch``.
        r€   z$num_splits mismatch: checkpoint has z, current dataset has rƒ   z&shuffle_seed mismatch: checkpoint has r  Na/  permutation mismatch: this checkpoint was saved against a different permutation (e.g. a split_strategy that resamples differently between runs, such as bootstrap_sample with no seed, or a changed callable) than the current dataset built. Resuming would apply saved per-split positions to the wrong rows.FrŸ  )rŽ   r�   r�   r‘   r„   rŠ  z mismatch: checkpoint has r  r  r›  rœ  rš  r   r  )rµ   r§   r·   rR   r  r×   rÂ   rT   r0   rÊ   rÌ   rµ  r<   rË   r;   r
  rá   ri  râ   rã   )r7   r*   Úcheckpoint_fingerprintrò  ÚoursÚcÚgÚbr×  rÿ   r0  r  r£  s                r'   Úload_state_dictz StreamingDataset.load_state_dict‹  s¨  € ð" �Ñ $×"2Ñ"2Ò2ÜØ6°u¸\Ñ7JÐ6Kð L'Ø'+×'7Ñ'7Ð&8ð:óð ð �Ñ  D×$6Ñ$6Ò6ÜØ8¸¸~Ñ9NÐ8Oð P'Ø'+×'9Ñ'9Ð&:ð<óð ð "'§¡Ð+=Ó!>Ðà"Ð.Ø&¨$×*@Ñ*@Ò@äðóð ð 6;ˆÔ2à˜UÑ" d×&:Ñ&:Ð&Fðò �ô ˜t q¨¨ YÓ/�Ø—9‘9˜S“> TÓ)Ü$Ø˜%Ð9¸%¿)¹)ÀC».Ð9Ið J/Ø/3¨fð6óð ðð 49Ð9UÑ3VÖ"W¨a¤3 q¥6Ò"WˆDÔà %Ð&@Ñ Aö)Ø”�A•ò)ˆDÔ%ð
 " .Ñ1×7Ñ7Ó9÷"á�A�qô �A“¤4¨¨(©Ó#4ÄÀQÀxÁ[Ó@QÑRÑRó"ˆDÔð àÐ5Ñ6ˆÜ�h¤Ô%Ù3;¤# h¤-ÀˆDÔä6?ÀÓ6I÷$Ù&2 e¨U�”s˜5“zÑ!ó$ˆDÕ ô #& h£-ˆDÔØ#%ˆDÔ ð —I‘IÐ<Ó=ˆ	ØÐØ%'ˆDÕ"ô 3<¸IÓ2F÷&Ù$. E¨3�”s˜3“x‘ó&ˆDÕ"ùò5 #Xùò)ùó"ùó$ùó&s   Ã8H;ÄI Å	1IÆ;IÈIÚstatesc           
      ó–  — | st        d«      ‚| d   }d|v }g d¢}d|v r|j                  d«       |r|j                  g d¢«       | dd D ]D  }d|v |k7  rt        d	«      ‚|D ]+  }||vrŒ||   ||   k7  sŒt        |› d
||   › d||   › �«      ‚ ŒF |�r.|d   }| D ](  }dD ]!  }t        ||   «      |k7  sŒt        |› d�«      ‚ Œ* g }g }i }	t	        |«      D ]ª  }
| d   }|d   |
   |d   |
   f}| dd D ]  }|d   |
   |d   |
   f}||kD  sŒ|}|}Œ |j                  |d   |
   «       |j                  |d   |
   «       |d   j                  |
|d   j                  t        |
«      «      «      }|€Œ�t        |«      |	|
<   Œ¬ t        t        |«      «      dkD  rt        d«      ‚t        |«      }||d<   ||d<   |	|d<   |S t        |«      }t        d„ | D «       Ž D �cg c]  }t        |«      ‘Œ c}|d<   | D �cg c]  }|j                  d|d   «      ‘Œ }}t        |Ž D �cg c]  }t        |«      ‘Œ c}|d<   |S c c}w c c}w c c}w )u£	  Merge state dicts saved by different ranks into one exact state.

        In row mode, each rank records exact consumer-committed progress for
        its own splits and lower bounds for the rest, so elementwise maxima
        recover both sample counts and permutation positions. In packed mode,
        the state that emitted the most blocks for each logical split supplies
        that split's permutation position and partial token buffer. Packed
        states must cover every rank at the same global step.

        Raises ``ValueError`` if the states are empty, were not produced by
        the same run, or do not represent the same global step.

        The merge is always all-to-all and topology-agnostic: collect the
        ``state_dict()`` from every rank of the *previous* run into
        one list, merge that whole list, and hand the identical merged result
        to every rank of the *next* run â€” regardless of whether the
        topology grew, shrank, or stayed the same. There is no pairwise or
        subset merging step, because each split's exact state is only known to
        whichever iterator owned that split.

        For example, checkpointing 8 ranks and resuming on 4 (the same
        pattern applies when growing, e.g. 4 ranks resuming on 8)::

            states = [ds.state_dict() for ds in previous_run_datasets]  # 8
            merged = StreamingDataset.merge_state_dicts(states)
            for ds in resumed_datasets:  # now only 4 ranks
                ds.load_state_dict(merged)  # same dict on every rank

        The rank count on either side never affects the merge itself, since
        ``merge_state_dicts`` only cares about the list of states it is
        given.  Each split's position is recovered by elementwise maximum;
        here rank 0 owned split 0 (and skipped two rows there) while rank 1
        owned split 1 (and skipped one row):

        >>> rank0 = {
        ...     "shuffle_seed": 0, "num_splits": 2, "epoch": 0,
        ...     "samples_consumed_per_split": [3, 3],
        ...     "positions_consumed_per_split": [5, 3],
        ... }
        >>> rank1 = {
        ...     "shuffle_seed": 0, "num_splits": 2, "epoch": 0,
        ...     "samples_consumed_per_split": [3, 3],
        ...     "positions_consumed_per_split": [3, 4],
        ... }
        >>> merged = StreamingDataset.merge_state_dicts([rank0, rank1])
        >>> merged["positions_consumed_per_split"]
        [5, 4]
        z2merge_state_dicts requires at least one state dictr   rŸ  )rƒ   r€   r„   r  r*  r   Nz,cannot merge packed and unpacked state dictsz mismatch across state dicts: r)  r€   )r  r  z) must contain one entry per logical splitr  r  zWpacked state dicts were not captured at the same global step or do not cover every rankc              3   ó&   K  — | ]	  }|d    –— Œ y­w)r  Nr%   )rƒ  r*   s     r'   r„  z5StreamingDataset.merge_state_dicts.<locals>.<genexpr>b  s   è ø€ ÒJ¸%�%Ð4Õ5ÑJùs   ‚r  )r§   r@  rz  r¨   rì   rR   rË  r   rë   r+   ré   r  )r;  Úfirstr-  Úconfig_keysr*   rò  r€   Úmerged_consumedÚmerged_emittedÚmerged_buffersrÿ   ÚownerÚowner_progressÚprogressr¶  ÚmergedÚ	per_splitÚall_positionss                     r'   Úmerge_state_dictsz"StreamingDataset.merge_state_dictsá  sR  € ñd ÜÐQÓRÐRØ�q‘	ˆØ 5Ð(ˆÚ=ˆØ Ñ&ð ×ÑÐ1Ô2ÙØ×ÑÚJôð ˜A˜B�Zò 
	ˆEØ %Ð'¨FÒ2Ü Ð!OÓPÐPØ"ò �Ø˜eÑ#ØØ˜‘:  s¡Ó+Ü$Ø˜%Ð=Ø  ™:˜, d¨5°©:¨,ð8óð ñ	ð
	ò Ø˜|Ñ,ˆJØò �ðò �Cô ˜5 ™:“¨*Ó4Ü(Ø"˜eÐ#LÐMóð ñðð !ˆOØˆNØˆNÜ˜zÓ*ò =�Ø˜q™	�àÐ4Ñ5°eÑ<ØÐ6Ñ7¸Ñ>ð"�ð $ A B˜Zò 2�EàÐ8Ñ9¸%Ñ@ØÐ:Ñ;¸EÑBð �Hð   .Ó0Ø %˜Ø)1™ð2ð  ×&Ñ& uÐ-IÑ'JÈ5Ñ'QÔRØ×%Ñ% eÐ,FÑ&GÈÑ&NÔOØ˜~Ñ.×2Ñ2Ø˜5 Ñ0×4Ñ4´S¸³ZÓ@ó�ð Ñ%Ü,4°VÓ,<�N 5Ò)ð)=ô, ”3�~Ó&Ó'¨!Ò+Ü ð6óð ô
 ˜%“[ˆFØ3BˆFÐ/Ñ0Ø1?ˆFÐ-Ñ.Ø%3ˆF�>Ñ"ØˆMä�e“ˆô !ÙJÀ6ÔJðö0
àô �	�Nò0
ˆÐ+Ñ,ð  ö	
ð ð �I‰IØ.°Ð6RÑ0Sõð
ˆð 
ô -0°Ð,?ö2
Ø(ŒC�	�Nò2
ˆÐ-Ñ.ð ˆùò0
ùò
ùò2
s   ÇH<Ç:IÈ"I)5r!   r"   r#   rE   ÚDEFAULT_READ_BATCH_SIZEÚDEFAULT_PREFETCH_BATCHESr   r0   r}   r   r   r<   rË  r   r   r`  r   r8   ró   r)  r
   r+   rO   rN   r/  ÚpropertyrÚ  ÚfloatrÝ  rß  râ  rå  rê  rì  rî  ró  r  r  r  r?   rj   r,  r  r!  r-  ro   r-   r.  rk   r:  ÚstaticmethodrI  rW   rX   s   @r'   r   r   Ó   s€  ø„ ñ@ðL %)Øð Ø&'ØØØØ6Ø6Ø'+Ø,0Ø $Ø(,Ø/3Ø(,Ø $Ø $ØBFØFMØ/3Ø=AØ!à*.ò;P
ð ˜S‘Mð	P
ð
 ðP
ð !ØÐ(Ð)Ð+=Ð=Ñ>ñ
ðP
ð ˜s‘mðP
ð ðP
ð ðP
ð ðP
ð ðP
ð ðP
ð ˜$˜s™)Ñ$ðP
ð  % S™Mð!P
ð" ˜‘ð#P
ð$ ˜HÑ%ð%P
ð&  (¨™}ð'P
ð( ! ™ð)P
ð* ˜‘ð+P
ð, ˜‘ð-P
ð. # 5¨¨g°f©oÐ)=Ñ#>Ñ?ð/P
ð0 " # x°°¸TÐ0AÑ'BÐ"BÑCð1P
ð2  (¨™}ð3P
ð4 % X¨s¨e°S¨jÑ%9Ñ:ð5P
ð: # 3™-õ;P
ðd?
¨Có ?
ðBD D¨¡Ió Dð<˜( 4¨¨S¨¡>Ñ2ó ð 8=ò=Ø04ð=à	�$�s˜C�x‘.Ñ	!ó=ð2T4Ø04ðT4à	�$�s˜C�x‘.Ñ	!óT4ðl ð
*˜cò 
*ó ð
*ð ð	1˜Eò 	1ó ð	1ð ð
1 ò 
1ó ð
1ð ð	* ò 	*ó ð	*ð ð	* cò 	*ó ð	*ð ð* ò *ó ð*ð ð
*˜cò 
*ó ð
*ð ð*˜sò *ó ð*ò òDTð$9˜Dó 9ðv ð, 3ò ,ó ð,ð:'
 dó '
ðR5¨Sð 5°Tó 5ð ð9¨4ò 9ó ð9ð ð@¨(°3©-ò @ó ð@ð ðA°¸#±ò Aó ðAð¨Có ð5°cð 5¸dó 5ð5Ø&Ð'=Ñ>ð5à	ó5ð@F¨$ð @FÀDð @FÈTó @FðDT Tð T¨dó Tðl ðL $ t¡*ð L°ò Ló ôLr&   r   c                   ó2   ‡ — e Zd ZdZdefˆ fd„Zˆ fd„Zˆ xZS )ÚStreamingDataLoadera  A PyTorch DataLoader with consumer-committed dataset checkpoints.

    PyTorch workers prefetch batches ahead of the trainer, so worker-local
    producer progress is not a safe checkpoint.  This loader carries a state
    snapshot alongside every internal batch and applies it to the parent
    [StreamingDataset][lancedb.streaming.StreamingDataset] only when that batch
    is returned by ``next()``.
    The trainer receives the same collated batch it would receive from a
    standard ``torch.utils.data.DataLoader``.

    With more than one worker, row-mode ``state_dict()`` is available only at
    complete logical step boundaries, when every split assigned to the rank has
    the same consumed-sample count. Packed checkpoints require equal emitted-block
    counts across the rank's splits for any worker count. ``persistent_workers=True``
    is not supported because prefetched worker copies cannot be restored from
    parent-committed state. If batch collation raises, checkpointing remains
    invalid for that dataset instance; restore the last valid checkpoint into a
    fresh dataset before continuing.
    Only one active iterator may own a dataset at a time, including when worker
    processes are used. Exhausting or explicitly shutting down the iterator
    releases that ownership. ``drop_last=True`` is not supported because worker
    replicas discard incomplete tails independently, which cannot produce a
    topology-independent checkpoint.

    Parameters are the same as ``torch.utils.data.DataLoader`` except that
    ``dataset`` must be a
    [StreamingDataset][lancedb.streaming.StreamingDataset].
    Subclasses that override ``StreamingDataset.__iter__`` are not supported
    because the custom iterator cannot provide the exact per-yield checkpoint
    snapshots required by this loader.

    Examples
    --------
    >>> # dataset = StreamingDataset(table, num_splits=2)
    >>> # loader = StreamingDataLoader(dataset, batch_size=8, num_workers=2)
    >>> # batch = next(iter(loader))
    >>> # checkpoint = dataset.state_dict()
    r    c                 ó¼  •— t        |t        «      st        d«      ‚t        |«      j                  t        j                  urt        d«      ‚|j                  dd«      du rt        d«      ‚|j                  dd«      rt        d«      ‚|| _        t        ‰| �$  t        |«      g|¢­i |¤Ž | j                  rt        d	«      ‚t        | j                  «      | _        y )
Nz/StreamingDataLoader requires a StreamingDatasetz”StreamingDataLoader does not support StreamingDataset subclasses that override __iter__ because they cannot provide exact per-yield checkpoint stateÚin_orderTFzQStreamingDataLoader requires in_order=True for deterministic consumer checkpointsÚpersistent_workersz|StreamingDataLoader does not support persistent_workers=True because worker prefetch state cannot be reset from a checkpointz€StreamingDataLoader does not support drop_last=True because discarded worker tails cannot be checkpointed topology-independently)r;   r   Ú	TypeErrorr®   rO   rR   r§   Ú_streaming_datasetrI   r8   rG   Ú	drop_lastr2   r3   )r7   r    ÚargsÚkwargsrJ   s       €r'   r8   zStreamingDataLoader.__init__™  sâ   ø€ Ü˜'Ô#3Ô4ÜÐMÓNÐNÜ�‹=×!Ñ!Ô)9×)BÑ)BÑBÜð-óð ð
 �:‰:�j $Ó'¨5Ñ0Üð'óð ð �:‰:Ð*¨EÔ2ÜðRóð ð #*ˆÔÜ‰ÑÔ1°'Ó:ÐL¸TÒLÀVÒLØ�>Š>Üð)óð ô
 -¨T¯_©_Ó=ˆ�r&   c                 ó–  •— | j                   }|j                  }d }	 |j                  «       }|j                  «       }|j                  d u}|rC|d   }|j
                  D �cg c]  }||   ‘Œ	 }}t        t        |«      «      dkD  r]t        d«      ‚| j                  dkD  rC|d   }	|j
                  D �cg c]  }|	|   ‘Œ	 }
}t        t        |
«      «      dkD  rt        d«      ‚t        t        ‰| �1  «       ||| j                  dkD  xs |¬«      S c c}w c c}w # t        $ r' |�|j                  |«       ‚ |j                  |«       ‚ w xY w)Nr  r   z StreamingDataLoader cannot start from a partial packed logical step; resume from a checkpoint whose splits assigned to this rank have equal emitted-block countsr  z¬StreamingDataLoader cannot start multiple workers from a partial logical step; resume from a checkpoint whose splits assigned to this rank have equal consumed-sample counts)r.   r[   )rU  rÜ   r-  r?   rÂ   rô   r¨   rë   r>   r"  rZ   rI   rO   ri   ro   r.  )r7   r    r0  r.   r*   r-  r   rÿ   r  r@   r  rJ   s              €r'   rO   zStreamingDataLoader.__iter__¶  sw  ø€ Ø×)Ñ)ˆØ ×9Ñ9ˆØˆð"	Ø!×<Ñ<Ó>ˆKØ×0Ñ0Ó2ˆEØ×,Ñ,°DÐ8ˆFÙØÐ9Ñ:�Ø:A×:NÑ:NÖO°˜v e›}ÐO�ÐOÜ”s˜;Ó'Ó(¨1Ò,Ü&ðPóð ð
 ×!Ñ! AÒ%ØÐ <Ñ=�Ø<C×<PÑ<PÖQ°5 ¨£ÐQ�ÐQÜ”s˜<Ó(Ó)¨AÒ-Ü&ð!óð ô +Ü‘Ñ Ó"ØØ'Ø $× 0Ñ 0°1Ñ 4Ò >¸ô	ð ùò# Pùò  Røô ò 	ØÐ&Ø×2Ñ2°;Ô?ð ð ×GÑGÈÔWØð	ús,   �AD Á DÁ,AD Â2DÂ>AD Ä
D Ä0E)r!   r"   r#   rE   r   r8   rO   rW   rX   s   @r'   rP  rP  q  s    ø„ ñ%ðN>Ð 0õ >÷:&ð &r&   rP  );rE   rÒ   r  rÍ  ÚloggingrÆ  r¥   rØ   r7  r  Úcollectionsr   Úconcurrent.futuresr   rñ  r   Úmultiprocessingr   Útypingr   r   r	   r
   r   r   r   r   Úpyarrowr«   Úpyarrow.computeÚcomputer  r«  Útorch.utils.datar   r   r   r  r   r   r   r   r   r   r   Ú	getLoggerr!   r£   rî   rJ  rK  r   r)   r-   r2   rG   rZ   r   rP  r%   r&   r'   ú<module>rd     sç   ðñó" Û Û Û Û 	Û Û Û Û Ý Ý 1Ý Ý $ß V× VÓ Vã Ý Û ß IÑ I÷÷ ñ ð 
ˆ×	Ñ	˜8Ó	$€ð
 €àÐ ØÐ ô �Jô  ô
�:ô ô
˜Zô ÷
Bñ Bô0&˜ô &÷$Y-ñ Y-ôx[�ô [ô|4k˜*õ kr&   