{"record":{"id":"073416562d50391b","repo":"apache/beam","slug":"reference-source-must-produce-the-same-number-of-records-as","errorCode":null,"errorMessage":"Reference source must produce the same number of records as the list of sources. Number of records were %d and %d instead.","messagePattern":"Reference source must produce the same number of records as the list of sources\\. Number of records were (.+?) and (.+?) instead\\.","errorType":"validation","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/io/source_test_utils.py","lineNumber":171,"sourceCode":"    if not (isinstance(source_info, tuple) and len(source_info) == 3 and\n            isinstance(source_info[0], iobase.BoundedSource)):\n      raise ValueError(\n          'source_info must a three tuple where first'\n          'item of the tuple gives a '\n          'iobase.BoundedSource. Received: %r' % source_info)\n    if (type(reference_source_info[0].default_output_coder())\n        != type(source_info[0].default_output_coder())):\n      raise ValueError(\n          'Reference source %r and the source %r must use the same coder. '\n          'They are using %r and %r respectively instead.' % (\n              reference_source_info[0],\n              source_info[0],\n              type(reference_source_info[0].default_output_coder()),\n              type(source_info[0].default_output_coder())))\n    source_records.extend(read_from_source(*source_info))\n\n  if len(reference_records) != len(source_records):\n    raise ValueError(\n        'Reference source must produce the same number of records as the '\n        'list of sources. Number of records were %d and %d instead.' %\n        (len(reference_records), len(source_records)))\n\n  if equal_to(reference_records)(source_records):\n    raise ValueError(\n        'Reference source and provided list of sources must produce the '\n        'same set of records.')\n\n\ndef assert_reentrant_reads_succeed(source_info):\n  \"\"\"Tests if a given source can be read in a reentrant manner.\n\n  Assume that given source produces the set of values ``{v1, v2, v3, ... vn}``.\n  For ``i`` in range ``[1, n-1]`` this method performs a reentrant read after\n  reading ``i`` elements and verifies that both the original and reentrant read\n  produce the expected set of values.\n","sourceCodeStart":153,"sourceCodeEnd":189,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/io/source_test_utils.py#L153-L189","documentation":"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.","triggerScenarios":"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.","commonSituations":"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).","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."],"exampleFix":"# before: range boundary excludes the last record\nfor record in records[start_position:stop_position - 1]:\n  yield record\n# after\nfor record in records[start_position:stop_position]:\n  yield record","handlingStrategy":"validation","validationCode":"ref_records = list(read_from_source(*ref_info))\nsrc_records = [r for si in sources_info for r in read_from_source(*si)]\nassert len(ref_records) == len(src_records), (len(ref_records), len(src_records))","typeGuard":"def is_source_info(x):\n    return isinstance(x, tuple) and len(x) == 3 and isinstance(x[0], iobase.BoundedSource)","tryCatchPattern":"try:\n    source_test_utils.assert_sources_equal_reference_source(ref_info, sources_info)\nexcept ValueError as e:\n    logger.error(\"source equivalence failed: %s\", e)\n    raise","preventionTips":["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"],"tags":["apache-beam","bounded-source","record-count-mismatch","io-testing"],"backgroundTag":"unexpected-response-shape","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T11:17:12.474Z"}