Create a varint32 coder and used it for RowCoder

[Abacn]

Fix #34089

This fixes negative int32 field passed from java to python in xlang transform

Please add a meaningful description for your change here

---

Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:

• Mention the appropriate issue in your description (for example: addresses #123), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, comment fixes #<ISSUE NUMBER> instead.
• Update CHANGES.md with noteworthy changes.
• If this contribution is large, please file an Apache Individual Contributor License Agreement.

See the Contributor Guide for more tips on how to make review process smoother.

To check the build health, please visit https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md

GitHub Actions Tests Status (on master branch)

See CI.md for more information about GitHub Actions CI or the workflows README to see a list of phrases to trigger workflows.

[Abacn]

Python3.9 test passed but 3.2 test failed, due to

E           FileNotFoundError: [Errno 2] No such file or directory: '/home/runner/.apache_beam/cache/jars/mysql-connector-java-8.0.28.jar.tmp' -> '/home/runner/.apache_beam/cache/jars/mysql-connector-java-8.0.28.jar'

Likely a race condition that two rename run at the same time when tests run in parallel:

os.rename(cached_jar + '.tmp', cached_jar)

[Abacn]

R: @liferoad

[github-actions]

Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control. If you'd like to restart, comment assign set of reviewers

[Abacn]

R: @kennknowles also

[liferoad]

can we add this to CHANGES.md?

[Abacn]

Thanks, will do as a follow up (otherwise need to rebase this PR onto latest master to be able to add to 2.65.0 section)

[Abacn]

opened #34408

[Abacn]

This appears to break

test_avro_schema_to_beam_schema_with_nullable_atomic_fields (apache_beam.io.avroio_test.TestFastAvro)

E File "apache_beam/coders/coder_impl.py", line 983, in apache_beam.coders.coder_impl.VarInt32CoderImpl.encode_to_stream
E OverflowError: value too large to convert to int32_t [while running 'Map()-ptransform-56']

[Amar3tto]

This appears to break

test_avro_schema_to_beam_schema_with_nullable_atomic_fields (apache_beam.io.avroio_test.TestFastAvro)

E File "apache_beam/coders/coder_impl.py", line 983, in apache_beam.coders.coder_impl.VarInt32CoderImpl.encode_to_stream E OverflowError: value too large to convert to int32_t [while running 'Map()-ptransform-56']

Also breaks these XVR jobs: #31418, #30601, #30602, #30593, #30517

[Abacn]

Yeah they all fails on same test. Let me investigate now.

From first glance it fails on encoding, which means the row passing in already have large integers

[Abacn]

hmmm test passed locally:

apache_beam/io/avroio_test.py::TestFastAvro::test_avro_schema_to_beam_schema_with_nullable_atomic_fields PASSED [100%]

================================================================================ 1 passed in 13.05s ================================================================================

[Abacn]

It turns out that the test already had an integar overflow in Row passed in:

print out the pcollection pass going to pass to SqlTransform:

beam/sdks/python/apache_beam/io/avroio_test.py

Line 193 in 867d70c

| beam.Map(avro_dict_to_beam_row(avro_schema, beam_schema))

Row(name='Henry', favorite_number=3, favorite_color='green')
Row(name='Toby', favorite_number=7, favorite_color='brown')
Row(name='Gordon', favorite_number=4, favorite_color='blue')
Row(name='Emily', favorite_number=4294967295, favorite_color='Red')
Row(name='Percy', favorite_number=6, favorite_color='Green')
Row(name='Bruce', favorite_number=None, favorite_color=None)

this happens to both before and after this change (#34354)

more suprisingly, when read back, it converts back to -1:

{'name': 'Thomas', 'favorite_number': 1, 'favorite_color': 'blue'}
{'name': 'Henry', 'favorite_number': 3, 'favorite_color': 'green'}
{'name': 'Toby', 'favorite_number': 7, 'favorite_color': 'brown'}
{'name': 'Gordon', 'favorite_number': 4, 'favorite_color': 'blue'}
{'name': 'Emily', 'favorite_number': -1, 'favorite_color': 'Red'}
{'name': 'Percy', 'favorite_number': 6, 'favorite_color': 'Green'}
{'name': 'Bruce', 'favorite_number': None, 'favorite_color': None}