What causes "Expected size >= 0 but received %s." in apache/beam?

Fix restriction_size() to clamp at 0 (e.g. max(0, end - start))

apache/beam · error · ValueError

Expected size >= 0 but received

Error message

Expected size >= 0 but received %s.

What it means

During SDF process invocation with a deferred (truncated) result, Beam computes the restriction's size via restriction_size() and requires it to be non-negative, because size feeds element weighting for splitting and progress. A negative size means the user's RestrictionProvider returned an invalid size, so Beam raises immediately.

Solutions

Fix restriction_size() to clamp at 0 (e.g. max(0, end - start))
Validate the restriction invariants (start <= end) in the RestrictionProvider before returning sizes
Check the RestrictionTracker's try_split/current_restriction logic for producing inverted restrictions

Example fix

# before
def restriction_size(self, element, restriction):
    return restriction.end - restriction.start
# after
def restriction_size(self, element, restriction):
    return max(0, restriction.end - restriction.start)

Defensive patterns

Strategy: validation

Validate before calling

def validate_restriction_provider(provider, element, restriction):
    size = provider.restriction_size(element, restriction)
    if size < 0:
        raise ValueError('restriction_size returned %s; clamp to >= 0' % size)
    return size

Type guard

def has_valid_size(provider, element, restriction):
    return provider.restriction_size(element, restriction) >= 0

Try / catch

try:
    process_element(element)
except ValueError as e:
    if 'Expected size >= 0' in str(e):
        logging.error('RestrictionProvider produced negative size: %s', e)
    raise

Prevention

Implement restriction_size as max(0, end - start)
Assert restriction.start <= restriction.stop in your RestrictionTracker
Test restriction_size with boundary and degenerate restrictions

When it happens

Trigger: A custom RestrictionProvider's restriction_size() returns a negative number while processing deferred/continuation results from a splittable DoFn.

Common situations: Custom restriction size functions computing ranges (end-start) that can go negative on malformed restrictions; off-by-one or reversed start/end in a custom RestrictionTracker; division/negation bugs in size estimation.

Understand the failure class

Background: "value must be between 0 and 1" / "out of range" / "must not be negative" errors: fixing range-validation failures across open-source libraries — this error's family across 42 libraries.

Related errors

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

Appendix: source

Thrown at sdks/python/apache_beam/runners/common.py:1072

    if additional_kwargs:
      kwargs_for_process.update(additional_kwargs)

    self.output_handler.handle_process_outputs(
        windowed_value,
        self.process_method(*args_for_process, **kwargs_for_process),
        self.threadsafe_watermark_estimator)

    if self.is_splittable:
      assert self.threadsafe_restriction_tracker is not None
      self.threadsafe_restriction_tracker.check_done()
      deferred_status = self.threadsafe_restriction_tracker.deferred_status()
      if deferred_status:
        deferred_restriction, deferred_timestamp = deferred_status
        element = windowed_value.value
        size = self.signature.get_restriction_provider().restriction_size(
            element, deferred_restriction)
        if size < 0:
          raise ValueError('Expected size >= 0 but received %s.' % size)
        current_watermark = (
            self.threadsafe_watermark_estimator.current_watermark())
        estimator_state = (
            self.threadsafe_watermark_estimator.get_estimator_state())
        residual_value = ((element, (deferred_restriction, estimator_state)),
                          size)
        return SplitResultResidual(
            residual_value=windowed_value.with_value(residual_value),
            current_watermark=current_watermark,
            deferred_timestamp=deferred_timestamp)
    return None

  def _invoke_process_batch_per_window(
      self,
      windowed_batch: WindowedBatch,
      additional_args,
      additional_kwargs,
  ):

View on GitHub (pinned to 12126d8942)