{"record":{"id":"c852d33a7d9c3058","repo":"apache/beam","slug":"items-obtained-by-reading-the-source-r-for-primary-and","errorCode":null,"errorMessage":"Items obtained by reading the source %r for primary and residual ranges %s and %s did not produce the expected list of values.","messagePattern":"Items obtained by reading the source %r for primary and residual ranges (.+?) and (.+?) did not produce the expected list of values\\.","errorType":"validation","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/io/source_test_utils.py","lineNumber":397,"sourceCode":"    # For unsuccessful splits, residual_range should be None.\n    assert not residual_range\n\n  residual_items = (\n      read_from_source(source, *residual_range) if split_successful else [])\n\n  total_items = primary_items + residual_items\n\n  if current_items != primary_items:\n    raise ValueError(\n        'Current source %r and a source created using the '\n        'range of the primary source %r determined '\n        'by performing dynamic work rebalancing at fraction '\n        '%r produced different values. Expected '\n        'these sources to produce the same list of values.' %\n        (source, _range_to_str(*primary_range), split_fraction))\n\n  if expected_items != total_items:\n    raise ValueError(\n        'Items obtained by reading the source %r for primary '\n        'and residual ranges %s and %s did not produce the '\n        'expected list of values.' %\n        (source, _range_to_str(*primary_range), _range_to_str(*residual_range)))\n\n  result = (len(primary_items), len(residual_items) if split_successful else -1)\n  return result\n\n\ndef assert_split_at_fraction_succeeds_and_consistent(\n    source, num_items_to_read_before_split, split_fraction):\n  \"\"\"Verifies some consistency properties of dynamic work rebalancing.\n\n  Equivalent to the following pseudocode:::\n\n    original_range_tracker = source.getRangeTracker(None, None)\n    original_reader = source.read(original_range_tracker)\n    items_before_split = read N items from original_reader","sourceCodeStart":379,"sourceCodeEnd":415,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/io/source_test_utils.py#L379-L415","documentation":"Thrown by _verify_single_split_fraction_result when primary items plus residual items, read from the two ranges produced by split_at_fraction, do not equal the expected full list of items. A correct split must partition the source's items with no loss or duplication.","triggerScenarios":"Calling assert_split_at_fraction_behavior/binary/concurrent on a source whose primary+residual ranges skip or duplicate records — e.g. try_split returns ranges whose union is not [start, stop), or read_from_source on a range omits boundary records.","commonSituations":"Custom RangeTracker.off-by-one in try_split (primary stop != residual start); source reader that buffers and drops items when the range changes; residual range computed from the wrong stop_position_before_split.","solutions":["Ensure try_split returns (split_position, None-style) ranges where primary=[start, split) and residual=[split, stop) with no gap or overlap.","Check that reading a range includes the record starting exactly at its start position and excludes records at/after stop.","Fix the reader to honor RangeTracker position updates exactly, including the item at split_position belonging to the residual range.","Print the sorted primary+residual item lists vs expected to find which records are missing or duplicated."],"exampleFix":"// before\nsplit_position = self.start_position + int(0.5 * (self.stop - self.start)) + 1  # off-by-one gap\n// after\nsplit_position = self.start_position + int(0.5 * (self.stop - self.start))\n# primary=[start, split_position), residual=[split_position, stop)","handlingStrategy":"validation","validationCode":"primary = read_from_source(source, *primary_range)\nresidual = read_from_source(source, *residual_range)\nassert primary + residual == expected_items, 'split is not a partition of the items'","typeGuard":null,"tryCatchPattern":"try:\n    assert_split_at_fraction_behavior(source, outcome, fraction, k)\nexcept ValueError as e:\n    if 'did not produce the expected list of values' in str(e):\n        logging.error('Split lost/duplicated items: %s', e)","preventionTips":["Ensure try_split ranges satisfy residual_start == primary_stop","Honor RangeTracker position transitions exactly in the reader, including boundary records","Diff primary+residual against the full item list when debugging custom sources"],"tags":["python","apache-beam","iobase","split-at-fraction","data-loss"],"backgroundTag":"internal-invariant-violation","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-20T03:17:13.778Z"}