[BEAM-11037] Python DirectRunner InputStream Issues (RuntimeError: VarLong too long)

• Type: Bug
• Status: Open
• Priority: P3
• Resolution: Unresolved
• Affects Version/s: 2.24.0
• Component/s: runner-direct, sdk-py-core
• Environment: Windows 10 64bit, Python 3.7.9
• Assignee: Unassigned
• Reporter: Michael Stauffer
• Created: 07/Oct/20 16:06
• Updated: 04/Jun/22 18:21

Description

I have a Beam workflow that runs fine on Dataflow. However, when I run it on my local direct runner (Windows 10, Python SDK), I run into some strange behaviour if I have a negative number in my PCollection (see below for an example).

Minimal code example to reproduce the exception:

from apache_beam.options.pipeline_options import PipelineOptions
from apache_beam.pipeline import Pipeline
import apache_beam

if __name__ == '__main__':
    pipeline_options = PipelineOptions()

    with Pipeline(options=pipeline_options) as p:
        lines = (
                p
                | apache_beam.Create([
                    {"Test": -1}
                ]))

        lines | apache_beam.io.WriteToText('./lines')

Exception:

  File "apache_beam\coders\coder_impl.py", line 222, in apache_beam.coders.coder_impl.StreamCoderImpl.decode
  File "apache_beam\coders\coder_impl.py", line 224, in apache_beam.coders.coder_impl.StreamCoderImpl.decode
  File "apache_beam\coders\coder_impl.py", line 470, in apache_beam.coders.coder_impl.FastPrimitivesCoderImpl.decode_from_stream
  File "apache_beam\coders\coder_impl.py", line 450, in apache_beam.coders.coder_impl.FastPrimitivesCoderImpl.decode_from_stream
  File "apache_beam\coders\stream.pyx", line 193, in apache_beam.coders.stream.InputStream.read_var_int64
RuntimeError: VarLong too long. [while running 'Create/Map(decode)']

 Any help is highly appreciated.