apache/beam · error · NotImplementedError

BigQuery storage source must be split before being read

Error message

BigQuery storage source must be split before being read

What it means

The BigQuery storage API source (ReadFromBigQuery with method=STORAGE_API_READ) likewise raises NotImplementedError from read(): the top-level source is only a placeholder that must be split into _CustomBigQueryStorageStreamSource objects (one per read-session stream) before reading. Direct read() on the un-split storage source is invalid.

Solutions

Call source.split() and read only the returned stream sub-sources.
Use the beam.io.ReadFromBigQuery(method=THE_STORAGE_API) transform so the framework performs split/read.
Fix custom runner code to follow the BoundedSource split-then-read protocol.

Example fix

// before
src = beam.io.bigquery._CustomBigQueryStorageSource(...)
rows = src.read(src.default_range_tracker())  # NotImplementedError
// after
for sub in src.split(float('inf'), None):
    rows = sub.read(sub.default_range_tracker())

Defensive patterns

Strategy: try-catch

Validate before calling

splits = source.split(float('inf'), None)
assert splits, 'storage source must produce stream sub-sources before read()'

Type guard

def is_stream_sub_source(src):
    return isinstance(src, _CustomBigQueryStorageStreamSource)

Try / catch

try:
    rows = source.read(rt)
except NotImplementedError:
    rows = [s.read(s.default_range_tracker()) for s in source.split(float('inf'), None)]

Prevention

Use ReadFromBigQuery(method=THE_STORAGE_API) instead of raw sources
Always split() a BoundedSource before read()
Test custom runners against the split-then-read contract

When it happens

Trigger: Calling read(range_tracker) directly on the source produced for BigQuery storage API reads instead of first calling split(), e.g. in custom runner code or tests exercising the storage API path.

Common situations: Testing the storage API read path manually, or a custom/direct runner that bypasses BoundedSource.split before read.

Related errors

AI-assisted analysis of apache/beam@12126d8942 (2026-09-13). Data as JSON: /api/errors/08a82d4003951db6. Report an issue: GitHub.

Appendix: source

Thrown at sdks/python/apache_beam/io/gcp/bigquery.py:1310

    for source in self.split_result:
      yield SourceBundle(
          weight=1.0, source=source, start_position=None, stop_position=None)

  def get_range_tracker(self, start_position, stop_position):
    class NonePositionRangeTracker(RangeTracker):
      """A RangeTracker that always returns positions as None. Prevents the
      BigQuery Storage source from being read() before being split()."""
      def start_position(self):
        return None

      def stop_position(self):
        return None

    return NonePositionRangeTracker()

  def read(self, range_tracker):
    raise NotImplementedError(
        'BigQuery storage source must be split before being read')


class _CustomBigQueryStorageStreamSource(BoundedSource):
  """A source representing a single stream in a read session."""

  # Runner will act on this counter on scaling event, if supported
  THROTTLE_COUNTER = Metrics.counter(__name__, 'cumulativeThrottlingSeconds')

  def __init__(
      self,
      read_stream_name: str,
      use_native_datetime: Optional[bool] = True,
      timeout: Optional[float] = None):
    self.read_stream_name = read_stream_name
    self.use_native_datetime = use_native_datetime
    self.timeout = timeout

View on GitHub (pinned to 12126d8942)