apache/beam · error · ValueError

Reference source must produce the same number of records as

Error message

Reference source must produce the same number of records as the list of sources. Number of records were %d and %d instead.

What it means

Raised by assert_sources_equal_reference_source in Apache Beam's source_test_utils when the reference BoundedSource and the list of sources under test produce a different total number of records. The utility reads the reference source fully, then reads each source in sources_info and concatenates their records; if the counts differ, the sources are not equivalent and the check fails before any element-by-element comparison. This guards custom I/O connectors against producing or dropping records across splits.

Source

Thrown at sdks/python/apache_beam/io/source_test_utils.py:171

    if not (isinstance(source_info, tuple) and len(source_info) == 3 and
            isinstance(source_info[0], iobase.BoundedSource)):
      raise ValueError(
          'source_info must a three tuple where first'
          'item of the tuple gives a '
          'iobase.BoundedSource. Received: %r' % source_info)
    if (type(reference_source_info[0].default_output_coder())
        != type(source_info[0].default_output_coder())):
      raise ValueError(
          'Reference source %r and the source %r must use the same coder. '
          'They are using %r and %r respectively instead.' % (
              reference_source_info[0],
              source_info[0],
              type(reference_source_info[0].default_output_coder()),
              type(source_info[0].default_output_coder())))
    source_records.extend(read_from_source(*source_info))

  if len(reference_records) != len(source_records):
    raise ValueError(
        'Reference source must produce the same number of records as the '
        'list of sources. Number of records were %d and %d instead.' %
        (len(reference_records), len(source_records)))

  if equal_to(reference_records)(source_records):
    raise ValueError(
        'Reference source and provided list of sources must produce the '
        'same set of records.')


def assert_reentrant_reads_succeed(source_info):
  """Tests if a given source can be read in a reentrant manner.

  Assume that given source produces the set of values ``{v1, v2, v3, ... vn}``.
  For ``i`` in range ``[1, n-1]`` this method performs a reentrant read after
  reading ``i`` elements and verifies that both the original and reentrant read
  produce the expected set of values.

View on GitHub (pinned to 12126d8942)

Solutions

Fix the custom source's read()/RangeTracker logic so each position range [start_position, stop_position) covers its records exactly once with no gaps or overlaps.
Verify the start/stop positions passed for each source_info tuple actually match the ranges produced by source.split().
Check default_output_coder consistency so records aren't silently skipped/mangled during decoding.
Print len(reference_records) and the per-source record counts (the error message gives the two totals) to locate which split is wrong.

Example fix

# before: range boundary excludes the last record
for record in records[start_position:stop_position - 1]:
  yield record
# after
for record in records[start_position:stop_position]:
  yield record

Defensive patterns

Strategy: validation

Validate before calling

ref_records = list(read_from_source(*ref_info))
src_records = [r for si in sources_info for r in read_from_source(*si)]
assert len(ref_records) == len(src_records), (len(ref_records), len(src_records))

Type guard

def is_source_info(x):
    return isinstance(x, tuple) and len(x) == 3 and isinstance(x[0], iobase.BoundedSource)

Try / catch

try:
    source_test_utils.assert_sources_equal_reference_source(ref_info, sources_info)
except ValueError as e:
    logger.error("source equivalence failed: %s", e)
    raise

Prevention

Unit-test each split range independently before running the equivalence harness
Always use half-open [start, stop) position semantics in read()
Compute expected record counts per split and assert them

When it happens

Trigger: Calling assert_sources_equal_reference_source((ref_source, start, stop), [(source, start, stop), ...]) where len(read_from_source(reference)) != sum of records read from each source in sources_info.

Common situations: Testing a custom BoundedSource whose range splitting drops or duplicates records (off-by-one in read() range boundaries, wrong stop-position handling, or a dynamic split that loses the last record).

Related errors

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