apache/beam · error · ValueError

Reference source and provided list of sources must produce…

Error message

Reference source and provided list of sources must produce the same set of records.

What it means

Raised by assert_sources_equal_reference_source when the reference source and the sources under test produce the same number of records but the record sets differ. After count comparison passes, the utility applies the equal_to matcher; if the elements are not identical it raises this error. It means the source under test emits different data (values, order-sensitive mismatches, or corrupt/differently-decoded records) than the reference source.

Solutions

Diff reference_records vs source_records element-wise to find the first mismatching record.
Fix the source's read() so each range yields exactly the records in [start_position, stop_position) with correct values.
Ensure the source uses the same default_output_coder as the reference (the utils already enforce the coder type earlier).
Check for duplicate ranges across the split sources and eliminate overlaps.

Example fix

# before: overlapping splits duplicate a record at the boundary
return records[start_position:stop_position + 1]
# after: half-open range [start, stop)
return records[start_position:stop_position]

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 sorted(map(repr, ref_records)) == sorted(map(repr, src_records))

Type guard

def uses_same_coder(ref, src):
    return type(ref.default_output_coder()) == type(src.default_output_coder())

Try / catch

try:
    source_test_utils.assert_sources_equal_reference_source(ref_info, sources_info)
except ValueError:
    diff = list(set(map(repr, ref_records)) ^ set(map(repr, src_records)))
    logger.error("record mismatch, differing: %s", diff[:10])
    raise

Prevention

Check split ranges for overlaps/duplicates before comparing
Keep the coder identical between reference and test sources
Use deterministic record ordering inside read()

When it happens

Trigger: assert_sources_equal_reference_source succeeds on record counts but equal_to(reference_records)(source_records) detects differing elements between the reference source output and the concatenated output of sources_info.

Common situations: A custom source's split ranges overlap (duplicate records) while totals happen to match, records are decoded with a wrong coder, or nondeterministic ordering/transformation inside read() alters values.

Related errors

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

Appendix: source

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

    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.

  Args:
    source_info (Tuple[~apache_beam.io.iobase.BoundedSource, int, int]):
      a three-tuple that gives the reference
      :class:`~apache_beam.io.iobase.BoundedSource`, position to start reading
      at, and a position to stop reading at.

View on GitHub (pinned to 12126d8942)