yzeng1618 commented on code in PR #10122:
URL: https://github.com/apache/seatunnel/pull/10122#discussion_r2652014651
##########
seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/connection/SimpleJdbcConnectionProvider.java:
##########
@@ -108,8 +108,19 @@ public Connection getOrEstablishConnection() throws
SQLException, ClassNotFoundE
if (jdbcConfig.getPassword().isPresent()) {
info.setProperty("password", jdbcConfig.getPassword().get());
}
+ // Apply server time zone for MySQL if user configured it and did not
override
+ // serverTimezone
+ String serverTimeZone = jdbcConfig.getServerTimeZone();
+ String url = jdbcConfig.getUrl();
+ if (serverTimeZone != null
+ && !serverTimeZone.isEmpty()
+ && url != null
+ && url.startsWith("jdbc:mysql:")
+ && !jdbcConfig.getProperties().containsKey("serverTimezone")) {
+ info.setProperty("serverTimezone", serverTimeZone);
+ }
info.putAll(jdbcConfig.getProperties());
- connection = driver.connect(jdbcConfig.getUrl(), info);
+ connection = driver.connect(url, info);
Review Comment:
Optimized and revised
##########
seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/mysql/MysqlJdbcRowConverter.java:
##########
@@ -17,16 +17,119 @@
package org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.mysql;
+import org.apache.seatunnel.api.table.catalog.TableSchema;
+import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
+import org.apache.seatunnel.api.table.type.SeaTunnelRow;
+import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
+import org.apache.seatunnel.common.exception.CommonError;
import
org.apache.seatunnel.connectors.seatunnel.jdbc.internal.converter.AbstractJdbcRowConverter;
import
org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.DatabaseIdentifier;
+import org.apache.seatunnel.connectors.seatunnel.jdbc.utils.JdbcFieldTypeUtils;
+import java.sql.Date;
import java.sql.PreparedStatement;
+import java.sql.ResultSet;
import java.sql.SQLException;
+import java.sql.Timestamp;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.LocalTime;
+import java.time.ZoneId;
+import java.util.Optional;
public class MysqlJdbcRowConverter extends AbstractJdbcRowConverter {
+ private final ZoneId serverTimeZone;
+
+ public MysqlJdbcRowConverter() {
+ this(null);
+ }
+
+ public MysqlJdbcRowConverter(ZoneId serverTimeZone) {
+ this.serverTimeZone = serverTimeZone == null ? ZoneId.systemDefault()
: serverTimeZone;
+ }
+
+ @Override
+ public SeaTunnelRow toInternal(ResultSet rs, TableSchema tableSchema)
throws SQLException {
Review Comment:
Optimized and revised
##########
seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/mysql/MySqlDialectFactory.java:
##########
@@ -51,4 +57,30 @@ public JdbcDialect create(@Nonnull String compatibleMode,
String fieldIde) {
}
return new MysqlDialect(fieldIde);
}
+
+ @Override
+ public JdbcDialect create(
+ @Nonnull String compatibleMode,
+ String fieldIde,
+ JdbcConnectionConfig jdbcConnectionConfig) {
+ if (DatabaseIdentifier.STARROCKS.equalsIgnoreCase(compatibleMode)) {
+ return new StarRocksDialect(fieldIde);
+ }
+ ZoneId zoneId = null;
+ boolean timestampWithTimeZone = false;
+ if (jdbcConnectionConfig != null
+ && jdbcConnectionConfig.getServerTimeZone() != null
+ && !jdbcConnectionConfig.getServerTimeZone().isEmpty()) {
Review Comment:
Optimized and revised
--
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]
