Carl-Zhou-CN commented on code in PR #10122:
URL: https://github.com/apache/seatunnel/pull/10122#discussion_r2646801544
##########
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:
It seems that the entire rewrite is not necessary. Can only the TIMESTAMP_TZ
part be rewritten?
##########
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:
I don't think it's suitable to be placed here. You can push the Settings to
config
##########
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:
Simplifying the judgment, it seems too complicated now. Wouldn't it be
better to have a default value
--
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]
