polsinello commented on code in PR #27927:
URL: https://github.com/apache/flink/pull/27927#discussion_r3125033440
##########
flink-python/pyflink/fn_execution/embedded/converters.py:
##########
@@ -162,12 +199,291 @@ def __init__(self, field_converter: DataConverter):
super(ArrayDataConverter, self).__init__(field_converter)
def to_internal(self, value) -> IN:
- return tuple(super(ArrayDataConverter, self).to_internal(value))
+ if value is None:
+ return None
+
+ return list(super(ArrayDataConverter, self).to_internal(value))
Review Comment:
The asymmetry is intentional and driven by pemja's Python↔Java type-coercion
table:
- `to_internal` (Java → Python UDF arg) returns `list` to match
process-mode's Beam-based path — UDFs see mutable sequences in both modes, so
code like `arr.append(x) / arr[i] = v` doesn't
break only in thread mode. Before this change, thread mode returned a
`tuple`, which created a subtle UDF-visible divergence between modes.
- `to_external` (Python return → Java) stays `tuple` because pemja coerces
`tuple` → `Object[]` (what the downstream Java ArrayDataConverter expects),
whereas `list` → `ArrayList`, which would
break `ArrayDataConverter.toExternalImpl`'s typed-array element loop.
I've added a class-level docstring on `ArrayDataConverter` documenting the
asymmetry.
##########
flink-python/pyflink/fn_execution/embedded/converters.py:
##########
@@ -162,12 +199,291 @@ def __init__(self, field_converter: DataConverter):
super(ArrayDataConverter, self).__init__(field_converter)
def to_internal(self, value) -> IN:
- return tuple(super(ArrayDataConverter, self).to_internal(value))
+ if value is None:
+ return None
+
+ return list(super(ArrayDataConverter, self).to_internal(value))
def to_external(self, value) -> OUT:
+ if value is None:
+ return None
+
return tuple(super(ArrayDataConverter, self).to_external(value))
+class LocalDateTimeConverter(DataConverter):
+ """Converter for TIMESTAMP(precision) columns in embedded (thread) mode.
+
+ On the input side (Java -> Python UDF argument) the value can arrive as
+ either (a) a native Python ``datetime.datetime`` — pemja's automatic
+ Java->Python conversion handles top-level ``java.sql.Timestamp`` /
+ ``LocalDateTime`` method arguments — or (b) a ``pemja.PyJObject``
+ wrapping a ``java.time.LocalDateTime``, which is what happens for values
+ nested inside arrays, rows, or map values (pemja does not recursively
+ auto-convert Object[] elements).
+
+ On the output side (Python return -> Java) the converter explicitly
+ constructs a ``java.time.LocalDateTime`` so the downstream Java converter
+ chain — which uses ``DataFormatConverters.LocalDateTimeConverter`` as the
+ default for TIMESTAMP fields in nested rows — receives the type it
+ expects instead of a pemja-defaulted ``java.sql.Timestamp``.
+
+ Precision: Python ``datetime`` is microsecond-precise, so values with
+ TIMESTAMP(p) where p <= 6 round-trip without loss. For TIMESTAMP(9) the
+ sub-microsecond nanoseconds are silently truncated.
+ """
+
+ def to_internal(self, value) -> IN:
+ if value is None:
+ return None
+ # Fast path: pemja already produced a Python datetime (happens for
+ # top-level method arguments when pemja auto-converts the bare Java
+ # value crossing the JNI boundary).
+ #
+ # We check via ``type(value) is datetime`` rather than ``isinstance``
+ # because pemja's C extension causes CPython's ``isinstance`` to raise
+ # "SystemError: <built-in function isinstance> returned NULL without
+ # setting an error" when invoked on a ``pemja.PyJObject``.
+ # ``type(value)`` uses Py_TYPE at the C level and is unaffected.
+ if type(value) is _dt.datetime:
+ return value
+ # Nested case: value is a pemja.PyJObject wrapping a Java
+ # java.time.LocalDateTime. Extract the fields via Java getters.
+ return _dt.datetime(
+ value.getYear(),
+ value.getMonthValue(),
+ value.getDayOfMonth(),
+ value.getHour(),
+ value.getMinute(),
+ value.getSecond(),
+ value.getNano() // 1000,
+ )
+
+ def to_external(self, value) -> OUT:
+ if value is None:
+ return None
+ return JLocalDateTime.of(
+ value.year,
+ value.month,
+ value.day,
+ value.hour,
+ value.minute,
+ value.second,
+ value.microsecond * 1000,
+ )
+
+
+class LocalDateConverter(DataConverter):
+ """Converter for DATE columns in embedded (thread) mode.
+
+ Mirrors LocalDateTimeConverter's pattern: the input value may be a native
+ ``datetime.date`` (top-level method arg auto-converted by pemja) or a
+ ``pemja.PyJObject`` wrapping a ``java.time.LocalDate`` (nested positions).
+ The output is always an explicit Java ``LocalDate`` to match the external
+ type the downstream Java converter chain expects.
+ """
+
+ def to_internal(self, value) -> IN:
+ if value is None:
+ return None
+ if type(value) is _dt.date:
+ return value
+ return _dt.date(
+ value.getYear(),
+ value.getMonthValue(),
+ value.getDayOfMonth(),
+ )
+
+ def to_external(self, value) -> OUT:
+ if value is None:
+ return None
+ return JLocalDate.of(value.year, value.month, value.day)
+
+
+class LocalTimeConverter(DataConverter):
+ """Converter for TIME(precision) columns in embedded (thread) mode.
+
+ Mirrors LocalDateTimeConverter's pattern against ``java.time.LocalTime``.
+ Python ``datetime.time`` is microsecond-precise so TIME(9) loses
+ sub-microsecond nanoseconds silently.
+ """
+
+ def to_internal(self, value) -> IN:
+ if value is None:
+ return None
+ if type(value) is _dt.time:
+ return value
+ return _dt.time(
+ value.getHour(),
+ value.getMinute(),
+ value.getSecond(),
+ value.getNano() // 1000,
+ )
+
+ def to_external(self, value) -> OUT:
+ if value is None:
+ return None
+ return JLocalTime.of(
+ value.hour,
+ value.minute,
+ value.second,
+ value.microsecond * 1000,
+ )
+
+
+class InstantConverter(DataConverter):
+ """Converter for TIMESTAMP_LTZ / TIMESTAMP WITH LOCAL TIME ZONE columns.
+
+ Java's ``java.time.Instant`` is absolute UTC. Python's ``datetime`` may be
+ naive or tz-aware. Process mode returns naive ``datetime`` (interpreted as
+ UTC); we match that shape to keep thread mode compatible.
+
+ On the input side we extract epoch-second and nano from the Java Instant
+ and construct a naive UTC ``datetime``.
+
+ On the output side we accept both naive (treated as UTC, matching process
+ mode) and tz-aware datetimes and convert to a Java ``Instant`` via
+ ``Instant.ofEpochSecond(seconds, nanos)``.
+ """
+
+ def to_internal(self, value) -> IN:
+ if value is None:
+ return None
+ if type(value) is _dt.datetime:
+ return value
+ # pemja.PyJObject wrapping java.time.Instant
+ epoch_s = value.getEpochSecond()
+ nano = value.getNano()
+ # Construct a naive UTC datetime. Python datetime is microsecond
+ # precise, so sub-microsecond nanoseconds are truncated.
+ return _dt.datetime(1970, 1, 1) + _dt.timedelta(
+ seconds=epoch_s, microseconds=nano // 1000
+ )
+
+ def to_external(self, value) -> OUT:
+ if value is None:
+ return None
+ # Normalise to an epoch-second + nanos. For naive datetimes we assume
+ # UTC, matching process-mode behaviour.
+ if value.tzinfo is None:
+ delta = value - _dt.datetime(1970, 1, 1)
+ else:
+ delta = value - _dt.datetime(1970, 1, 1, tzinfo=_dt.timezone.utc)
+ epoch_s = int(delta.total_seconds())
Review Comment:
good catch, fixed by computing via integer microseconds with divmod, which
floor-divides correctly for negative values:
```py
total_us = delta // _dt.timedelta(microseconds=1)
epoch_s, us_rem = divmod(total_us, 1_000_000)
return JInstant.ofEpochSecond(epoch_s, us_rem * 1000)
```
##########
flink-python/src/main/java/org/apache/flink/table/runtime/typeutils/PythonTypeUtils.java:
##########
@@ -832,25 +993,22 @@ public DataConverter visit(BinaryType binaryType) {
@Override
public DataConverter visit(DateType dateType) {
- return new
IdentityDataConverter<>(DataFormatConverters.DateConverter.INSTANCE);
+ return new DateLocalDateDataConverter();
}
@Override
public DataConverter visit(TimeType timeType) {
- return TimeDataConverter.INSTANCE;
Review Comment:
done
##########
flink-python/src/main/java/org/apache/flink/table/runtime/typeutils/PythonTypeUtils.java:
##########
@@ -493,6 +494,153 @@ public final OUT toExternal(RowData row, int column) {
abstract OUT toExternalImpl(INTER value);
}
+ /**
+ * Converts between pemja's auto-converted {@link java.sql.Timestamp} and
{@link
+ * java.time.LocalDateTime}.
+ *
+ * <p>Pemja maps Python {@code datetime} to {@link java.sql.Timestamp}.
The outer {@link
+ * DataFormatConverters.RowConverter} expects {@link
java.time.LocalDateTime} (the default
+ * bridge class for TIMESTAMP columns). This converter bridges the
mismatch.
+ */
+ public static final class TimestampLocalDateTimeDataConverter
+ extends DataConverter<TimestampData, java.time.LocalDateTime,
Object> {
+
+ private final int precision;
Review Comment:
Same as above — dropped the redundant local `precision` field; the inner
`DataFormatConverters.InstantConverter` already stores its own copy.
##########
flink-python/src/main/java/org/apache/flink/table/runtime/typeutils/PythonTypeUtils.java:
##########
@@ -493,6 +494,153 @@ public final OUT toExternal(RowData row, int column) {
abstract OUT toExternalImpl(INTER value);
}
+ /**
+ * Converts between pemja's auto-converted {@link java.sql.Timestamp} and
{@link
+ * java.time.LocalDateTime}.
+ *
+ * <p>Pemja maps Python {@code datetime} to {@link java.sql.Timestamp}.
The outer {@link
+ * DataFormatConverters.RowConverter} expects {@link
java.time.LocalDateTime} (the default
+ * bridge class for TIMESTAMP columns). This converter bridges the
mismatch.
+ */
+ public static final class TimestampLocalDateTimeDataConverter
+ extends DataConverter<TimestampData, java.time.LocalDateTime,
Object> {
+
+ private final int precision;
+
+ TimestampLocalDateTimeDataConverter(int precision) {
+ super(new DataFormatConverters.LocalDateTimeConverter(precision));
+ this.precision = precision;
+ }
+
+ @Override
+ java.time.LocalDateTime toInternalImpl(Object value) {
+ if (value == null) {
+ return null;
+ }
+ if (value instanceof java.time.LocalDateTime) {
+ return (java.time.LocalDateTime) value;
+ }
+ if (value instanceof java.sql.Timestamp) {
+ return ((java.sql.Timestamp) value).toLocalDateTime();
+ }
+ throw new IllegalArgumentException(
+ "Expected LocalDateTime or Timestamp but got: "
+ + value.getClass().getName());
+ }
+
+ @Override
+ Object toExternalImpl(java.time.LocalDateTime value) {
+ if (value == null) {
+ return null;
+ }
+ return java.sql.Timestamp.valueOf(value);
+ }
+ }
+
+ /**
+ * Converts between pemja's auto-converted {@link java.sql.Timestamp} and
{@link
+ * java.time.Instant} for TIMESTAMP_WITH_LOCAL_TIME_ZONE columns.
+ */
+ public static final class TimestampInstantDataConverter
+ extends DataConverter<TimestampData, java.time.Instant, Object> {
+
+ private final int precision;
Review Comment:
Dropped the field + the `this.precision = precision` assignment; kept the
constructor parameter since it still flows to the inner converter.
##########
flink-python/src/main/java/org/apache/flink/table/runtime/typeutils/PythonTypeUtils.java:
##########
@@ -832,25 +993,22 @@ public DataConverter visit(BinaryType binaryType) {
@Override
public DataConverter visit(DateType dateType) {
- return new
IdentityDataConverter<>(DataFormatConverters.DateConverter.INSTANCE);
+ return new DateLocalDateDataConverter();
Review Comment:
Done — switched to `DateLocalDateDataConverter.INSTANCE` at the
`visit(DateType)` call site; the `INSTANCE` was already defined.
##########
flink-python/src/main/java/org/apache/flink/streaming/api/utils/PythonTypeUtils.java:
##########
@@ -896,89 +894,208 @@ public Long toExternal(Byte value) {
* Python Long will be converted to Long in PemJa, so we need
ShortDataConverter to convert Java
* Long to internal Short.
*/
- public static final class ShortDataConverter extends DataConverter<Short,
Long> {
+ public static final class ShortDataConverter extends DataConverter<Short,
Number> {
private static final long serialVersionUID = 1L;
public static final ShortDataConverter INSTANCE = new
ShortDataConverter();
@Override
- public Short toInternal(Long value) {
+ public Short toInternal(Number value) {
if (value == null) {
return null;
}
-
return value.shortValue();
}
@Override
- public Long toExternal(Short value) {
+ public Number toExternal(Short value) {
if (value == null) {
return null;
}
-
return value.longValue();
}
}
/**
- * Python Long will be converted to Long in PemJa, so we need
IntDataConverter to convert Java
- * Long to internal Integer.
+ * Accepts both Long (from pemja int→Long) and Double (from pemja
float→Double)
+ * via the Number supertype, avoiding bridge-method ClassCastException.
*/
- public static final class IntDataConverter extends DataConverter<Integer,
Long> {
+ public static final class IntDataConverter extends DataConverter<Integer,
Number> {
private static final long serialVersionUID = 1L;
public static final IntDataConverter INSTANCE = new IntDataConverter();
@Override
- public Integer toInternal(Long value) {
+ public Integer toInternal(Number value) {
if (value == null) {
return null;
}
-
return value.intValue();
}
@Override
- public Long toExternal(Integer value) {
+ public Number toExternal(Integer value) {
if (value == null) {
return null;
}
-
return value.longValue();
}
}
/**
- * Python Float will be converted to Double in PemJa, so we need
FloatDataConverter to convert
- * Java Double to internal Float.
+ * Accepts both Double (from pemja float→Double) and Long (from pemja
int→Long)
Review Comment:
You're right — the docstrings were copy-paste that didn't reflect the
widened Number generic, which made the four narrowing converters look
interchangeable. Rewrote so each class now leads with its narrow target type.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
