yzeng1618 commented on code in PR #10048:
URL: https://github.com/apache/seatunnel/pull/10048#discussion_r2562596698
##########
seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/psql/PostgresJdbcRowConverter.java:
##########
@@ -316,4 +348,118 @@ public String microsecondsToIntervalFormatVal(String
intervalVal) {
if (seconds > 0) sb.append(seconds).append(" seconds");
return sb.toString().trim();
}
+
+ private OffsetDateTime getPostgresOffsetDateTime(ResultSet rs, int
columnIndex)
+ throws SQLException {
+ // Read the value once to avoid drivers returning null on subsequent
reads
+ final Object obj = rs.getObject(columnIndex);
+
+ if (obj == null) {
+ return null;
+ }
+
+ // Direct types
+ if (obj instanceof OffsetDateTime) {
+ return (OffsetDateTime) obj;
+ }
+ if (obj instanceof Timestamp) {
+ return ((Timestamp) obj).toInstant().atOffset(ZoneOffset.UTC);
+ }
+ if (obj instanceof java.time.ZonedDateTime) {
+ return ((java.time.ZonedDateTime) obj).toOffsetDateTime();
+ }
+ if (obj instanceof java.util.Date) {
+ return ((java.util.Date) obj).toInstant().atOffset(ZoneOffset.UTC);
+ }
+
+ // Remaining PostgreSQL-specific or driver types: fall back to string
representation
+ return parseTimestampFromObjectString(obj);
+ }
+
+ private OffsetDateTime parseTimestampIfPresent(String str) throws
SQLException {
+ if (str == null) {
+ return null;
+ }
+ return parsePostgresTimestampTz(str);
+ }
+
+ private OffsetDateTime parsePostgresTimestampTz(String str) throws
SQLException {
+ String normalized = normalizeIsoTimestamp(str);
+ if (normalized == null) {
+ return null;
+ }
Review Comment:
The code has been streamlined.
##########
seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/psql/PostgresJdbcRowConverter.java:
##########
@@ -316,4 +348,118 @@ public String microsecondsToIntervalFormatVal(String
intervalVal) {
if (seconds > 0) sb.append(seconds).append(" seconds");
return sb.toString().trim();
}
+
+ private OffsetDateTime getPostgresOffsetDateTime(ResultSet rs, int
columnIndex)
+ throws SQLException {
+ // Read the value once to avoid drivers returning null on subsequent
reads
+ final Object obj = rs.getObject(columnIndex);
+
+ if (obj == null) {
+ return null;
+ }
+
+ // Direct types
+ if (obj instanceof OffsetDateTime) {
+ return (OffsetDateTime) obj;
+ }
+ if (obj instanceof Timestamp) {
+ return ((Timestamp) obj).toInstant().atOffset(ZoneOffset.UTC);
+ }
+ if (obj instanceof java.time.ZonedDateTime) {
+ return ((java.time.ZonedDateTime) obj).toOffsetDateTime();
+ }
+ if (obj instanceof java.util.Date) {
+ return ((java.util.Date) obj).toInstant().atOffset(ZoneOffset.UTC);
+ }
+
+ // Remaining PostgreSQL-specific or driver types: fall back to string
representation
+ return parseTimestampFromObjectString(obj);
+ }
+
+ private OffsetDateTime parseTimestampIfPresent(String str) throws
SQLException {
+ if (str == null) {
+ return null;
+ }
+ return parsePostgresTimestampTz(str);
+ }
+
+ private OffsetDateTime parsePostgresTimestampTz(String str) throws
SQLException {
+ String normalized = normalizeIsoTimestamp(str);
+ if (normalized == null) {
+ return null;
+ }
+
+ try {
+ return OffsetDateTime.parse(normalized);
+ } catch (Exception primary) {
+ log.debug("Failed to parse PostgreSQL timestamptz as ISO-8601:
{}", str, primary);
+ try {
+ String withoutOffset =
+
normalized.replaceFirst("([+-]\\d{2}:?\\d{2}|\\s+UTC|[zZ])$", "");
+ String fallback = withoutOffset.replace('T', ' ').trim();
+ Timestamp ts = Timestamp.valueOf(fallback);
+ return ts.toInstant().atOffset(ZoneOffset.UTC);
+ } catch (Exception secondary) {
+ log.debug(
+ "Failed to parse PostgreSQL timestamptz as UTC
timestamp: {}",
+ str,
+ secondary);
+ throw new SQLException(
+ "Failed to parse PostgreSQL timestamptz string: " +
str, secondary);
+ }
+ }
+ }
+
+ @Nullable private OffsetDateTime parseTimestampFromObjectString(Object
obj) throws SQLException {
+ final String str;
+ try {
+ str = String.valueOf(obj);
+ } catch (Throwable e) {
+ log.debug(
+ "Failed to get PostgreSQL timestamp object string
representation from class: {}",
+ obj.getClass().getName(),
+ e);
+ return null;
+ }
+ return parseTimestampIfPresent(str);
+ }
+
+ private String normalizeIsoTimestamp(String value) {
+ if (value == null) {
+ return null;
+ }
+ // PostgreSQL timestamptz format examples:
+ // "2023-12-25 10:30:45.123456+08:00"
+ // "2023-12-25 10:30:45+08"
+ // "2023-12-25 10:30:45.123456 UTC"
+ String normalized = value.trim();
+ if (normalized.isEmpty()) {
+ return null;
+ }
Review Comment:
The code has been streamlined.
--
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]
