dianfu commented on code in PR #27999:
URL: https://github.com/apache/flink/pull/27999#discussion_r3141619568
##########
flink-python/src/main/java/org/apache/flink/table/runtime/typeutils/PythonTypeUtils.java:
##########
@@ -493,6 +494,147 @@ 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> {
+
+ TimestampLocalDateTimeDataConverter(int precision) {
+ super(new DataFormatConverters.LocalDateTimeConverter(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> {
+
+ TimestampInstantDataConverter(int precision) {
+ super(new DataFormatConverters.InstantConverter(precision));
+ }
+
+ @Override
+ java.time.Instant toInternalImpl(Object value) {
+ if (value == null) {
+ return null;
+ }
+ if (value instanceof java.time.Instant) {
+ return (java.time.Instant) value;
+ }
+ if (value instanceof java.sql.Timestamp) {
+ return ((java.sql.Timestamp) value).toInstant();
+ }
+ throw new IllegalArgumentException(
+ "Expected Instant or Timestamp but got: "
+ + value.getClass().getName());
+ }
+
+ @Override
+ Object toExternalImpl(java.time.Instant value) {
+ if (value == null) {
+ return null;
+ }
+ return java.sql.Timestamp.from(value);
+ }
+ }
+
+ /** Converts between pemja's auto-converted {@link java.sql.Date} and
{@link java.time.LocalDate}. */
+ public static final class DateLocalDateDataConverter
+ extends DataConverter<Integer, java.time.LocalDate, Object> {
+
+ static final DateLocalDateDataConverter INSTANCE = new
DateLocalDateDataConverter();
+
+ DateLocalDateDataConverter() {
+ super(DataFormatConverters.LocalDateConverter.INSTANCE);
+ }
+
+ @Override
+ java.time.LocalDate toInternalImpl(Object value) {
+ if (value == null) {
+ return null;
+ }
+ if (value instanceof java.time.LocalDate) {
+ return (java.time.LocalDate) value;
+ }
+ if (value instanceof java.sql.Date) {
+ return ((java.sql.Date) value).toLocalDate();
+ }
+ throw new IllegalArgumentException(
+ "Expected LocalDate or Date but got: " +
value.getClass().getName());
+ }
+
+ @Override
+ Object toExternalImpl(java.time.LocalDate value) {
+ if (value == null) {
+ return null;
+ }
+ return java.sql.Date.valueOf(value);
+ }
+ }
+
+ /** Converts between pemja's auto-converted {@link java.sql.Time} and
{@link java.time.LocalTime}. */
+ public static final class TimeLocalTimeDataConverter
+ extends DataConverter<Integer, java.time.LocalTime, Object> {
+
+ static final TimeLocalTimeDataConverter INSTANCE = new
TimeLocalTimeDataConverter();
+
+ TimeLocalTimeDataConverter() {
+ super(DataFormatConverters.LocalTimeConverter.INSTANCE);
+ }
+
+ @Override
+ java.time.LocalTime toInternalImpl(Object value) {
+ if (value == null) {
+ return null;
+ }
+ if (value instanceof java.time.LocalTime) {
+ return (java.time.LocalTime) value;
+ }
+ if (value instanceof java.sql.Time) {
+ return ((java.sql.Time) value).toLocalTime();
+ }
+ throw new IllegalArgumentException(
+ "Expected LocalTime or Time but got: " +
value.getClass().getName());
+ }
+
+ @Override
+ Object toExternalImpl(java.time.LocalTime value) {
+ return value;
Review Comment:
Should we convert it to java.sql.Time?
--
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]
