apache/beam · error · ValueError

Reference source %r and the source %r must use the same…

Error message

Reference source %r and the source %r must use the same coder. They are using %r and %r respectively instead.

What it means

assert_sources_equal_reference_source compares records read from a reference BoundedSource against those read from candidate sources, which requires identical encoding. Beam raises ValueError when the reference source's default_output_coder type differs from a candidate source's, because byte-level record comparison would be meaningless across coders.

Solutions

1. Make the tested source's default_output_coder() return the same coder type as the reference source
2. Align the element types produced by both sources
3. Wrap elements in a common coder (e.g. use FastPrimitivesCoder in both) if types differ legitimately
4. Update the test's reference source after any intentional coder change

Example fix

// before
class MySource(BoundedSource):
  def default_output_coder(self):
    return ProtoCoder(MyProto)
// after
class MySource(BoundedSource):
  def default_output_coder(self):
    return FastPrimitivesCoder()  # matches reference source

from apache_beam.io import iobase
assert type(ref[0].default_output_coder()) == type(src[0].default_output_coder())

def coders_match(ref_source, src):
    return type(ref_source.default_output_coder()) == type(src.default_output_coder())

try:
    source_test_utils.assert_sources_equal_reference_source(ref, sources)
except ValueError as e:
    logger.error('Coder mismatch between sources: %s', e)

When it happens

Trigger: Comparing a source whose default_output_coder() returns, e.g., a ProtoCoder against one returning a FastPrimitivesCoder, or comparing sources of different element types (str vs bytes).

Common situations: Testing a new source against a reference implementation written with a different coder, or changing element types in one source but not its test reference.

Understand the failure class

Background: "is not a compatible type" / "cannot merge" errors: when a value's type doesn't match what the library requires — this error's family across 65 libraries.

Related errors

• At most one of --create_test and --fix_tests may be…
• Coder for the GroupByKey operation
• coder is not of type Coder
• Encountered unhashable element
• Error decoding input stream with coder

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

Appendix: source

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

    raise ValueError(
        'reference_source_info must a three-tuple where first'
        'item of the tuple gives a '
        'iobase.BoundedSource. Received: %r' % reference_source_info)
  reference_records = read_from_source(*reference_source_info)

  source_records = []
  for source_info in sources_info:
    assert isinstance(source_info, tuple)
    assert len(source_info) == 3
    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.')

View on GitHub (pinned to 12126d8942)