dianfu commented on code in PR #27927:
URL: https://github.com/apache/flink/pull/27927#discussion_r3107958671
##########
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:
The description seems confusing for me. What's the difference between
IntDataConverter and FloatDataConverter?
##########
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:
TimeDataConverter is not used any more. We can remove it in this PR.
##########
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:
Review comments from AI:
```
converters.py line ~375. int() truncates towards zero, which gives
incorrect epoch_s/nano pairs for negative timestamps (before 1970):
epoch_s = int(delta.total_seconds())
nano = (
delta.microseconds + (delta.total_seconds() - epoch_s) * 1_000_000
) * 1000
Example: for a datetime representing -99.3s from epoch:
- delta.total_seconds() = -99.3
- epoch_s = int(-99.3) = -99 (truncates towards zero, should be -100)
- nano computes to 400,000,000 instead of 700,000,000
- Result: Instant.ofEpochSecond(-99, 400000000) = -98.6s (wrong, should be
-99.3s)
Fix: use int(delta.total_seconds() // 1) or math.floor(), or better yet,
compute via integer microseconds:
total_us = int(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/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:
What's the purpose of this change? (tuple -> list)
##########
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:
Could be removed since it's not used
##########
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:
Could use `DateLocalDateDataConverter.INSTANCE`?
##########
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:
Could be removed since it's not used
--
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]
