Skip to content

[FLINK-35180][python] Fix embedded (thread-mode) type converters#27927

Draft
polsinello wants to merge 1 commit intoapache:release-1.19from
polsinello:FLINK-35180-fix-thread-mode-type-converters
Draft

[FLINK-35180][python] Fix embedded (thread-mode) type converters#27927
polsinello wants to merge 1 commit intoapache:release-1.19from
polsinello:FLINK-35180-fix-thread-mode-type-converters

Conversation

@polsinello
Copy link
Copy Markdown

What is the purpose of the change

In embedded (thread) mode, Pemja's JNI bridge auto-converts Python datetime/date/time objects to java.sql.Timestamp/Date/Time, but Flink's ExternalSerializer and DataFormatConverters.RowConverter expect the modern java.time.* bridge classes (LocalDateTime, LocalDate, LocalTime, Instant). This mismatch causes ClassCastException at serialization boundaries for any UDF that returns or receives temporal types. Process mode is unaffected because its Beam-based runnerOutputTypeSerializer resolves to the correct java.time.* serializers via LegacyTypeInfoDataTypeConverter.

This patch fixes several related type-conversion bugs in the embedded Python execution path so that all Flink-supported types work correctly in thread mode.

Brief change log

  • Add bridge-aware DataConverter implementations for all temporal types (TIMESTAMP, DATE, TIME, TIMESTAMP_LTZ) on both the Table API (PythonTypeUtils in flink-table-runtime) and DataStream (PythonTypeUtils in flink-streaming-java) paths
  • Replace IdentityDataConverter for temporal types where it silently passed java.sql.* objects through without conversion
  • Fix ROW/TUPLE buffer reuse in ArrayDataConverter that caused silent data corruption in ARRAY<ROW<...>> results
  • Widen numeric DataConverter generics from Long/Double to Number to avoid bridge-method ClassCastException when Pemja returns Integer/Float for small values
  • Cache the original Java ExternalTypeInfo in the Python ExternalTypeInfo wrapper to prevent lossy DECIMAL precision round-trips through legacy TypeInfo conversion

Verifying this change

This change is already covered by existing tests — the existing PyFlink Table API and DataStream test suites exercise all affected type paths.

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 (fixes type conversion before values reach the serializer)
  • The runtime per-record code paths (performance sensitive): yes — the embedded Python UDF execution path now performs explicit type conversions for temporal/numeric types instead of relying on identity pass-through. The overhead is minimal (one java.time.* factory call per value).
  • 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

…temporal, numeric, and nested types

Pemja auto-converts Python datetime/date/time to java.sql.Timestamp/Date/Time,
but Flink's ExternalSerializer and DataFormatConverters.RowConverter expect the
modern java.time.* bridge classes (LocalDateTime, LocalDate, LocalTime, Instant).
In thread mode this mismatch causes ClassCastException at serialization boundaries.
Process mode is unaffected because its Beam-based runnerOutputTypeSerializer
resolves to the correct java.time.* serializers via LegacyTypeInfoDataTypeConverter.

This patch adds bridge-aware DataConverters for all temporal types on both the
Table API and DataStream paths, replaces IdentityDataConverter where it silently
passed java.sql.* through, fixes row/tuple buffer reuse that caused silent data
corruption in ARRAY<ROW<>>, widens numeric DataConverter generics from Long/Double
to Number to avoid bridge-method ClassCastException, and caches the original Java
ExternalTypeInfo to prevent lossy DECIMAL precision round-trips through legacy
TypeInfo conversion.
@flinkbot
Copy link
Copy Markdown
Collaborator

flinkbot commented Apr 14, 2026

CI report:

Bot commands The @flinkbot bot supports the following commands:
  • @flinkbot run azure re-run the last Azure build

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants