Fix a logical type issue about JdbcDateType and JdbcTimeType

[shunping]

fixes #35244
fixes #28151

The date type in Java is encoded with the same logic as Timestamp in Python.

When we construct the schema proto of a python date field, it first resolves its logical type to JdbcDateType. From there, it finds its representation type.

Originally, the representation type of JdbcDateType is Timestamp. Because Timestamp is not an atomic type, it tries to find what is the corresponding logical type of Timestamp. However, there are two logical types defining Timestamp as their language type, MillisInstant and MicrosInstant. The latter one overrides the registration of the former. Therefore, we resolve the representation type of JdbcDataType as MicrosInstant, which is a logical type that can be expanded to a representation type of a named tuple of int64.

This causes the conflict on the coders. The data that comes out from the JDBCIO transform is encoded with Timestamp-like coder (in Java), but the downstream step (in Python) thinks the values are named tuples.

[github-actions]

Assigning reviewers:

R: @tvalentyn for label python.

Note: If you would like to opt out of this review, comment assign to next reviewer.

Available commands:

• stop reviewer notifications - opt out of the automated review tooling
• remind me after tests pass - tag the comment author after tests pass
• waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).

[Abacn]

Can you confirm the tests now work without this line?

beam/sdks/python/apache_beam/io/external/xlang_jdbcio_it_test.py

Line 260 in 7f09169

LogicalType.register_logical_type(MillisInstant)

[shunping]

xlang_jdbcio_it_test.py

Can you confirm the tests now work without this line?

beam/sdks/python/apache_beam/io/external/xlang_jdbcio_it_test.py

Line 260 in 7f09169

LogicalType.register_logical_type(MillisInstant)

I didn't run this test, but I have some local ones that has a similar logic (like the one in #35244). They failed prior to my change with the errors, and now they run successfully.

[Abacn]

To run this test, just remove this line in the test, make any change to https://github.com/apache/beam/blob/master/.github/trigger_files/beam_PostCommit_Python.json and push a commit to this PR

[shunping]

remind me after tests pass

[github-actions]

Ok - I'll remind @shunping after tests pass

[shunping]

I checked and verified that the failure is not related to the change here.

[shunping]

@Abacn, could you please take another look? Thanks!

[Abacn]

Thank you!

[claudevdm]

I think this PR is causing failures in the xlang direct tests https://github.com/apache/beam/actions/workflows/beam_PostCommit_Python_Xlang_Gcp_Direct.yml?page=2

@shunping

[shunping]

I think this PR is causing failures in the xlang direct tests https://github.com/apache/beam/actions/workflows/beam_PostCommit_Python_Xlang_Gcp_Direct.yml?page=2

@shunping

Do you see the error messages? I checked but they are not very useful.

We can try to revert this PR and run xlang postcommit and see if it passes.

[claudevdm]

Here is one

ValueError: Error decoding input stream with coder WindowedValueCoder[window_coder=GlobalWindowCoder, value_coder=RowCoder]
self = <apache_beam.runners.worker.bundle_processor.DataInputOperation object at 0x7917f5f45d60>
encoded_windowed_values = b'\x7f\xdf;dZ\x1c\xac\t\x00\x00\x00\x01\x0fQ\x0b\x00\xfd\xff\xff\xff\x0f\xfd\xff\xff\xff\xff\xff\xff\xff\xff\x01?\xb9\...\x05Test9\x05Test9\x05Test9\x80\x00\x01\x97{"\xe0C\x02\x02\x037\x80\x00\x00@t\x17p\x00\x80\x00\x00\x00\x01\xf6\xc3\x11'

    def process_encoded(self, encoded_windowed_values: bytes) -> None:
      input_stream = coder_impl.create_InputStream(encoded_windowed_values)
      while input_stream.size() > 0:
        with self.splitting_lock:
          if self.index == self.stop - 1:
            return
          self.index += 1
        try:
>         decoded_value = self.windowed_coder_impl.decode_from_stream(
              input_stream, True)

apache_beam/runners/worker/bundle_processor.py:231: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
apache_beam/coders/coder_impl.py:1549: in decode_from_stream
    value = self._value_coder.decode_from_stream(in_stream, nested)
apache_beam/coders/coder_impl.py:1641: in decode_from_stream
    return self._value_coder.decode(in_stream.read(value_length))
apache_beam/coders/coder_impl.py:239: in decode
    return self.decode_from_stream(create_InputStream(encoded), False)
apache_beam/coders/coder_impl.py:1952: in decode_from_stream
    item = component_coder.decode_from_stream(in_stream, True)
apache_beam/coders/coder_impl.py:2005: in decode_from_stream
    return self.logical_type.to_language_type(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <apache_beam.typehints.schemas.MicrosInstant object at 0x79182e358fa0>
value = BeamSchema_ee15bbf4_32de_4611_a9dc_b1b5c5981f57(seconds=None, micros=None)

    def to_language_type(self, value):
      # type: (MicrosInstantRepresentation) -> Timestamp
>     return Timestamp(seconds=int(value.seconds), micros=int(value.micros))
E     TypeError: int() argument must be a string, a bytes-like object or a number, not 'NoneType'

apache_beam/typehints/schemas.py:934: TypeError

The above exception was the direct cause of the following exception:

a = (<apache_beam.io.external.xlang_jdbcio_it_test.CrossLanguageJdbcIOTest testMethod=test_xlang_jdbc_write_read_0_postgres>,)
kw = {}

    @wraps(func)
    def standalone_func(*a, **kw):
>       return func(*(a + p.args), **p.kwargs, **kw)

../../build/gradleenv/1398941893/lib/python3.9/site-packages/parameterized/parameterized.py:620: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
apache_beam/io/external/xlang_jdbcio_it_test.py:268: in test_xlang_jdbc_write_read
    assert_that(result, equal_to(expected_rows))
apache_beam/pipeline.py:661: in __exit__
    self.result = self.run()
apache_beam/testing/test_pipeline.py:115: in run
    result = super().run(
apache_beam/pipeline.py:635: in run
    return self.runner.run_pipeline(self, self._options)
apache_beam/runners/direct/test_direct_runner.py:42: in run_pipeline
    self.result = super().run_pipeline(pipeline, options)
apache_beam/runners/direct/direct_runner.py:185: in run_pipeline
    return runner.run_pipeline(pipeline, options)
apache_beam/runners/portability/fn_api_runner/fn_runner.py:196: in run_pipeline
    self._latest_run_result = self.run_via_runner_api(
apache_beam/runners/portability/fn_api_runner/fn_runner.py:223: in run_via_runner_api
    return self.run_stages(stage_context, stages)
apache_beam/runners/portability/fn_api_runner/fn_runner.py:470: in run_stages
    bundle_results = self._execute_bundle(
apache_beam/runners/portability/fn_api_runner/fn_runner.py:795: in _execute_bundle
    self._run_bundle(
apache_beam/runners/portability/fn_api_runner/fn_runner.py:1034: in _run_bundle
    result, splits = bundle_manager.process_bundle(
apache_beam/runners/portability/fn_api_runner/fn_runner.py:1360: in process_bundle
    result_future = self._worker_handler.control_conn.push(process_bundle_req)
apache_beam/runners/portability/fn_api_runner/worker_handlers.py:389: in push
    response = self.worker.do_instruction(request)
apache_beam/runners/worker/sdk_worker.py:657: in do_instruction
    return getattr(self, request_type)(
apache_beam/runners/worker/sdk_worker.py:695: in process_bundle
    bundle_processor.process_bundle(instruction_id))
apache_beam/runners/worker/bundle_processor.py:1274: in process_bundle
    input_op_by_transform_id[element.transform_id].process_encoded(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <apache_beam.runners.worker.bundle_processor.DataInputOperation object at 0x7917f5f45d60>
encoded_windowed_values = b'\x7f\xdf;dZ\x1c\xac\t\x00\x00\x00\x01\x0fQ\x0b\x00\xfd\xff\xff\xff\x0f\xfd\xff\xff\xff\xff\xff\xff\xff\xff\x01?\xb9\...\x05Test9\x05Test9\x05Test9\x80\x00\x01\x97{"\xe0C\x02\x02\x037\x80\x00\x00@t\x17p\x00\x80\x00\x00\x00\x01\xf6\xc3\x11'

    def process_encoded(self, encoded_windowed_values: bytes) -> None:
      input_stream = coder_impl.create_InputStream(encoded_windowed_values)
      while input_stream.size() > 0:
        with self.splitting_lock:
          if self.index == self.stop - 1:
            return
          self.index += 1
        try:
          decoded_value = self.windowed_coder_impl.decode_from_stream(
              input_stream, True)
        except Exception as exn:
>         raise ValueError(
              "Error decoding input stream with coder " +
              str(self.windowed_coder)) from exn
E         ValueError: Error decoding input stream with coder WindowedValueCoder[window_coder=GlobalWindowCoder, value_coder=RowCoder]

apache_beam/runners/worker/bundle_processor.py:234: ValueError

[claudevdm]

I think we need to add this back?

[shunping]

Hmmm, removing that line is the main purpose of this PR. Let's revert the PR for now and I will take a look at that failed test later.

[akashorabek]

I created a PR that reverts this one and #35191, which also contributed to the failures of PostCommit Python XLang GCP Direct.

[claudevdm]

The way I understand it is your fix avoids the need for using this to workaround to work withdate (beam:logical_type:javasdk_date:v1) and time (beam:logical_type:javasdk_time:v1) types.

For jdbc timestamp types, java encodes as beam:logical_type:millis_instant:v1, which python interprets as a language type Timestamp, but MicrosInstant is registered to handle timestamps?

[shunping]

So the two logical types MillisInstant (

beam/sdks/python/apache_beam/typehints/schemas.py

Line 867 in 5532af8

class MillisInstant(NoArgumentLogicalType[Timestamp, np.int64]):

) and MicrosInstant (

beam/sdks/python/apache_beam/typehints/schemas.py

Line 911 in 5532af8

class MicrosInstant(NoArgumentLogicalType[Timestamp,

) both register Timestamp as their language type and MillisInstant goes first and then MicrosIntant goes second in the file.

There is a map internally to map a language type to its logical type and it is used when it tries to determine the coder. Without my fix and the workaround, the internal map is going to map Timestamp to MicroInstant.

Then the whole code path when determining the coder of MillisInstant will be MillisInstant -> TimeStamp -> Logical Type of TimeStamp -> MicroInstant and its representation type.

The workaround is to calling the register function again for MillisInstant, so that it overwrite the internal mapping to Timestamp -> Millistant.

My fix was trying to eliminate the need of the workaround.

[claudevdm]

When we read from jdbc, we get a schema with a millis_instant urn e.g.

fields {
  name: "created_at"
  type {
    nullable: true
    logical_type {
      urn: "beam:logical_type:millis_instant:v1"
      representation {
        atomic_type: INT64
      }
    }
  }
  id: 2
  encoding_position: 2
}

Then in named_tuple_from_schema we try map the schema fields to python types

beam/sdks/python/apache_beam/typehints/schemas.py

Line 541 in 0d6c94a

return LogicalType.from_runner_api(

.

[('created_at', typing.Optional[apache_beam.typehints.schemas.Timestamp])]

The coder for this tuple will use MicrosInstant because that was registered last for Timestamp.

I am testing a fix here to use a stub type so that
beam:logical_type:millis_instant:v1 -> Stub Type -> MillisInstant -> Timestamp
#35400