flink
flink copied to clipboard
apache
[FLINK-38418][python] Support pickling Instants when collecting TIMESTAMP_LTZ types in PyFlink
What is the purpose of the change
Currently, when trying to collect a TIMESTAMP_LTZ value in PyFlink, you get a pickling error:
py4j.protocol.Py4JJavaError: An error occurred while calling z:org.apache.flink.api.common.python.PythonBridgeUtils.getPickledBytesFromRow.
: org.apache.flink.api.python.shaded.net.razorvine.pickle.PickleException: couldn't introspect javabean: java.lang.IllegalArgumentException: wrong number of arguments
This is because there is no case to handle pickling LocalZonedTimestampType Instant values in the pickler, which just defaults to trying to pickle the Instant object which fails. This PR adds a clause to handle this (by creating a Timestamp value from the Instant, similar to how Timestamps are pickled), and adds a handler on the PyFlink side to convert this value to a datetime.
Brief change log
- Adds case to handle
LocalZonedTimestampTypetypes in the pickling/unpickling logic during collection.
Verifying this change
I've added an extra case in the Table API environment type tests to include a test for TIMESTAMP_LTZ types insert/collection tests, which before this change fails with the above pickling error.
Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): (no)
- The public API, i.e., is any changed class annotated with
@Public(Evolving): (no) - The serializers: (no)
- The runtime per-record code paths (performance sensitive): (no)
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (no)
- The S3 file system connector: (no)
Documentation
- Does this pull request introduce a new feature? (no)
- If yes, how is the feature documented? (not applicable)
