apache/beam · error · NotImplementedError

BigQuery source must be split before being read

Error message

BigQuery source must be split before being read

What it means

BoundedSource.read() on BigQuerySourceBase raises NotImplementedError because a BigQuery source is a logical source that must first be split into concrete sub-sources (e.g. per-stream storage API sources) before any data can be read. The Beam framework is expected to call split() and then read() on each returned sub-source; calling read() directly on the un-split source is a programming error.

Source

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

    for path in self.export_result.paths:
      source = self._create_source(path, self.export_result.coder)
      yield SourceBundle(
          weight=1.0, source=source, start_position=None, stop_position=None)

  def get_range_tracker(self, start_position, stop_position):
    class CustomBigQuerySourceRangeTracker(RangeTracker):
      """A RangeTracker that always returns positions as None."""
      def start_position(self):
        return None

      def stop_position(self):
        return None

    return CustomBigQuerySourceRangeTracker()

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

  @check_accessible(['query'])
  def _setup_temporary_dataset(self, bq):
    if self.temp_dataset:
      # Temp dataset was provided by the user so we can just return.
      return
    location = bq.get_query_location(
        self._get_project(), self.query.get(), self.use_legacy_sql)
    bq.create_temporary_dataset(
        self._get_project(), location, kms_key=self.kms_key)

  @check_accessible(['query'])
  def _execute_query(self, bq):
    query_job_name = bigquery_tools.generate_bq_job_name(
        self._job_name,
        self._source_uuid,
        bigquery_tools.BigQueryJobTypes.QUERY,
        '%s_%s' % (int(time.time()), random.randint(0, 1000)))

View on GitHub (pinned to 12126d8942)

Solutions

  1. Call source.split(desired_num_splits, pipeline_options) first and call read() only on the returned sub-sources.
  2. Use the high-level beam.io.ReadFromBigQuery() transform instead of consuming the source directly, which handles splitting and reading.
  3. If implementing a custom runner, implement the BoundedSource protocol: split() then read() each split with its RangeTracker.

Example fix

// before
source = beam.io.BigQuerySource('project:dataset.table')
rows = source.read(source.default_range_tracker())  # NotImplementedError
// after
source = beam.io.BigQuerySource('project:dataset.table')
for split in source.split(1, None):
    rows = split.read(split.default_range_tracker())
# or simply:
_ = p | beam.io.ReadFromBigQuery(query='SELECT ...')
Defensive patterns

Strategy: try-catch

Validate before calling

if isinstance(source, BoundedSource) and source is the top-level BigQuery source:
    splits = source.split(desired_num_splits, pipeline_options)
    # only call read() on each split

Type guard

def is_split_source(src):
    return isinstance(src, iobase.BoundedSource) and not isinstance(src, bigquery.BigQuerySourceBase)

Try / catch

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

Prevention

When it happens

Trigger: Calling read(range_tracker) directly on the BigQuery source returned by beam.io.BigQuerySource (or gcsio-backed query source) instead of splitting it first, typically in custom source code, tests, or a custom runner that bypasses the standard BoundedSource protocol.

Common situations: Custom I/O experimentation, unit tests that instantiate BigQuerySource and try to read() it, or a runner/DoFn harness that does not implement the split-then-read BoundedSource contract.

Related errors

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