Skip to content

Enable batched read pushdown (SupportsBatchedReadRandomAccessDataSource) on grain.python.ArrayRecordDataSource. - #1412

Open
copybara-service[bot] wants to merge 1 commit into
mainfrom
test_993324537
Open

copybara-service[bot] wants to merge 1 commit into
mainfrom
test_993324537

Conversation

@copybara-service

@copybara-service copybara-service Bot commented Oct 4, 2026 •

Copy link
Copy Markdown

Enable batched read pushdown (SupportsBatchedReadRandomAccessDataSource) on grain.python.ArrayRecordDataSource.

When SourceMapDataset wraps a data source, it checks isinstance(source, SupportsBatchedReadRandomAccessDataSource) and, if true, overrides _getitems to forward batched index lookups (source._getitems(indices)) when BatchMapDataset reads a batch of records.

Previously:

  1. SupportsBatchedReadRandomAccessDataSource was not decorated with @typing.runtime_checkable, and grain.python.ArrayRecordDataSource did not implement _getitems.
  2. As a result, MapDataset.source(ArrayRecordDataSource(...)).batch(B) fell back to per-element __getitem__ calls in SourceMapDataset._getitems, borrowing and returning a shard reader on each individual record rather than holding the pooled reader across the batch.

This change:

  • Decorates SupportsBatchedReadRandomAccessDataSource with @typing.runtime_checkable.
  • Implements _getitems on grain.python.ArrayRecordDataSource (delegating to super().__getitems__ in array_record.ArrayRecordDataSource) so batched reads group record indices by shard and hold the borrowed per-shard reader across the entire batch.
  • Adds unit tests verifying isinstance(source, SupportsBatchedReadRandomAccessDataSource) and source._getitems(...) on grain.python.ArrayRecordDataSource.

@copybara-service
copybara-service Bot force-pushed the test_993324537 branch 2 times, most recently from bbd0b32 to 51adec8 Compare October 5, 2026 04:06
…rce`) on `grain.python.ArrayRecordDataSource`.

When `SourceMapDataset` wraps a data source, it checks `isinstance(source, SupportsBatchedReadRandomAccessDataSource)` and, if true, overrides `_getitems` to forward batched index lookups (`source._getitems(indices)`) when `BatchMapDataset` reads a batch of records.

Previously:
1. `SupportsBatchedReadRandomAccessDataSource` was not decorated with `@typing.runtime_checkable`, and `grain.python.ArrayRecordDataSource` did not implement `_getitems`.
2. As a result, `MapDataset.source(ArrayRecordDataSource(...)).batch(B)` fell back to per-element `__getitem__` calls in `SourceMapDataset._getitems`, borrowing and returning a shard reader on each individual record rather than holding the pooled reader across the batch.

This change:
- Decorates `SupportsBatchedReadRandomAccessDataSource` with `@typing.runtime_checkable`.
- Implements `_getitems` on `grain.python.ArrayRecordDataSource` (delegating to `super().__getitems__` in `array_record.ArrayRecordDataSource`) so batched reads group record indices by shard and hold the borrowed per-shard reader across the entire batch.
- Adds unit tests verifying `isinstance(source, SupportsBatchedReadRandomAccessDataSource)` and `source._getitems(...)` on `grain.python.ArrayRecordDataSource`.

PiperOrigin-RevId: 993324537

This branch has not been deployed

No deployments
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant