apache/beam · error · ValueError

Items obtained by reading the source %r for primary and…

Error message

Items obtained by reading the source %r for primary and residual ranges %s and %s did not produce the expected list of values.

What it means

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.

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.

Example fix

// before
split_position = self.start_position + int(0.5 * (self.stop - self.start)) + 1  # off-by-one gap
// after
split_position = self.start_position + int(0.5 * (self.stop - self.start))
# primary=[start, split_position), residual=[split_position, stop)

Defensive patterns

Strategy: validation

Validate before calling

primary = read_from_source(source, *primary_range)
residual = read_from_source(source, *residual_range)
assert primary + residual == expected_items, 'split is not a partition of the items'

Try / catch

try:
    assert_split_at_fraction_behavior(source, outcome, fraction, k)
except ValueError as e:
    if 'did not produce the expected list of values' in str(e):
        logging.error('Split lost/duplicated items: %s', e)

Prevention

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

When it happens

Trigger: 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.

Common situations: 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.

Understand the failure class

Background: "This is a bug, please report it": internal invariant violations, unreachable panics, and SNH errors explained — this error's family across 47 libraries.

Related errors

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

Appendix: source

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

    # For unsuccessful splits, residual_range should be None.
    assert not residual_range

  residual_items = (
      read_from_source(source, *residual_range) if split_successful else [])

  total_items = primary_items + residual_items

  if current_items != primary_items:
    raise ValueError(
        'Current source %r and a source created using the '
        'range of the primary source %r determined '
        'by performing dynamic work rebalancing at fraction '
        '%r produced different values. Expected '
        'these sources to produce the same list of values.' %
        (source, _range_to_str(*primary_range), split_fraction))

  if expected_items != total_items:
    raise ValueError(
        'Items obtained by reading the source %r for primary '
        'and residual ranges %s and %s did not produce the '
        'expected list of values.' %
        (source, _range_to_str(*primary_range), _range_to_str(*residual_range)))

  result = (len(primary_items), len(residual_items) if split_successful else -1)
  return result


def assert_split_at_fraction_succeeds_and_consistent(
    source, num_items_to_read_before_split, split_fraction):
  """Verifies some consistency properties of dynamic work rebalancing.

  Equivalent to the following pseudocode:::

    original_range_tracker = source.getRangeTracker(None, None)
    original_reader = source.read(original_range_tracker)
    items_before_split = read N items from original_reader

View on GitHub (pinned to 12126d8942)