You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by zh...@apache.org on 2022/08/25 08:30:37 UTC
[rocketmq-connect] branch master updated: [ISSUE #202]Debezium MySQL Source Connector datatime type column transform problem (#281)
This is an automated email from the ASF dual-hosted git repository.
zhoubo pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/rocketmq-connect.git
The following commit(s) were added to refs/heads/master by this push:
new aad06e6 [ISSUE #202]Debezium MySQL Source Connector datatime type column transform problem (#281)
aad06e6 is described below
commit aad06e61bba0889ccbcbc3d7871434b7860fedd9
Author: xiaoyi <su...@163.com>
AuthorDate: Thu Aug 25 16:30:32 2022 +0800
[ISSUE #202]Debezium MySQL Source Connector datatime type column transform problem (#281)
* Debezium MySQL Source Connector datatime type column transform problem #202
* reformat code
---
.../connect/jdbc/common/DebeziumTimeTypes.java | 80 +++++++++++++++++++---
.../jdbc/dialect/impl/GenericDatabaseDialect.java | 37 +++-------
2 files changed, 83 insertions(+), 34 deletions(-)
diff --git a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/common/DebeziumTimeTypes.java b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/common/DebeziumTimeTypes.java
index 81268cd..a2f04b1 100644
--- a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/common/DebeziumTimeTypes.java
+++ b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/common/DebeziumTimeTypes.java
@@ -17,8 +17,18 @@
package org.apache.rocketmq.connect.jdbc.common;
+import io.debezium.time.Date;
+import io.debezium.time.ZonedTimestamp;
+import io.openmessaging.connector.api.data.Schema;
+import org.apache.rocketmq.connect.jdbc.util.DateTimeUtils;
+
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
import java.time.LocalDate;
+import java.time.LocalDateTime;
import java.time.ZoneOffset;
+import java.time.format.DateTimeFormatter;
+import java.util.TimeZone;
/**
@@ -39,18 +49,72 @@ public class DebeziumTimeTypes {
public static final String ZONED_TIME = "io.debezium.time.ZonedTime";
public static final String ZONED_TIMESTAMP = "io.debezium.time.ZonedTimestamp";
- public static Object toMillsTimestamp(String schemaName, Object value){
- if(schemaName == null){
+
+ /**
+ * maybe bind debezium logical
+ *
+ * @param statement
+ * @param index
+ * @param schema
+ * @param value
+ * @param timeZone
+ * @return
+ * @throws SQLException
+ */
+ public static boolean maybeBindDebeziumLogical(
+ PreparedStatement statement,
+ int index,
+ Schema schema,
+ Object value,
+ TimeZone timeZone
+ ) throws SQLException {
+ if (schema.getName() != null) {
+ switch (schema.getName()) {
+ case Date.SCHEMA_NAME:
+ statement.setDate(index,
+ new java.sql.Date(
+ (long) DebeziumTimeTypes.toMillsTimestamp(Date.SCHEMA_NAME, value)
+ ),
+ DateTimeUtils.getTimeZoneCalendar(timeZone)
+ );
+ return true;
+ case io.debezium.time.Timestamp.SCHEMA_NAME:
+
+ statement.setTimestamp(index,
+ new java.sql.Timestamp((long) DebeziumTimeTypes.toMillsTimestamp(io.debezium.time.Timestamp.SCHEMA_NAME, value)),
+ DateTimeUtils.getTimeZoneCalendar(timeZone)
+ );
+ return true;
+ case ZonedTimestamp.SCHEMA_NAME:
+
+ statement.setTimestamp(index,
+ new java.sql.Timestamp((long) toMillsTimestamp(ZonedTimestamp.SCHEMA_NAME, value)),
+ DateTimeUtils.getTimeZoneCalendar(timeZone)
+ );
+ return true;
+ default:
+ return false;
+ }
+ }
+ return false;
+ }
+
+
+ private static Object toMillsTimestamp(String schemaName, Object value) {
+ if (schemaName == null) {
return value;
}
- switch (schemaName){
- case DATE:
- return LocalDate.ofEpochDay((long)value)
- .atStartOfDay(ZoneOffset.ofHours(8))
+ switch (schemaName) {
+ case ZonedTimestamp.SCHEMA_NAME:
+ DateTimeFormatter formatter = ZonedTimestamp.FORMATTER;
+ LocalDateTime localDateTime = LocalDateTime.parse(value.toString(), formatter);
+ return localDateTime.toInstant(ZoneOffset.ofHours(0)).toEpochMilli();
+
+ case Date.SCHEMA_NAME:
+ return LocalDate.ofEpochDay((long) value)
+ .atStartOfDay(ZoneOffset.ofHours(0))
.toInstant()
.toEpochMilli();
- case TIMESTAMP:
- return value;
default:
return value;
}
diff --git a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/dialect/impl/GenericDatabaseDialect.java b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/dialect/impl/GenericDatabaseDialect.java
index 0c26be3..247febb 100644
--- a/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/dialect/impl/GenericDatabaseDialect.java
+++ b/connectors/rocketmq-connect-jdbc/src/main/java/org/apache/rocketmq/connect/jdbc/dialect/impl/GenericDatabaseDialect.java
@@ -16,7 +16,6 @@
*/
package org.apache.rocketmq.connect.jdbc.dialect.impl;
-import io.debezium.time.ZonedTimestamp;
import io.openmessaging.connector.api.data.FieldType;
import io.openmessaging.connector.api.data.Schema;
import io.openmessaging.connector.api.data.SchemaBuilder;
@@ -70,9 +69,7 @@ import java.sql.SQLXML;
import java.sql.Statement;
import java.sql.Timestamp;
import java.sql.Types;
-import java.time.LocalDateTime;
import java.time.ZoneOffset;
-import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Calendar;
@@ -1559,6 +1556,9 @@ public class GenericDatabaseDialect implements DatabaseDialect {
}
} else {
boolean bound = maybeBindLogical(statement, index, schema, value);
+ if (!bound) {
+ bound = maybeBindDebeziumLogical(statement, index, schema, value);
+ }
if (!bound) {
bound = maybeBindPrimitive(statement, index, schema, value);
}
@@ -1615,29 +1615,6 @@ public class GenericDatabaseDialect implements DatabaseDialect {
DateTimeUtils.getTimeZoneCalendar(timeZone)
);
return true;
-
- case DebeziumTimeTypes.DATE:
- statement.setDate(index,
- new java.sql.Date((long)DebeziumTimeTypes.toMillsTimestamp(DebeziumTimeTypes.DATE, value)),
- DateTimeUtils.getTimeZoneCalendar(timeZone)
- );
- return true;
- case DebeziumTimeTypes.TIMESTAMP:
- statement.setTimestamp(index,
- new java.sql.Timestamp((long)DebeziumTimeTypes.toMillsTimestamp(DebeziumTimeTypes.TIMESTAMP, value)),
- DateTimeUtils.getTimeZoneCalendar(timeZone)
- );
- return true;
- case ZonedTimestamp.SCHEMA_NAME:
- DateTimeFormatter formatter = ZonedTimestamp.FORMATTER;
- LocalDateTime localDateTime=LocalDateTime.parse(value.toString(),formatter);
- Long format = localDateTime.toInstant(ZoneOffset.ofHours(8)).toEpochMilli();
- statement.setTimestamp(index,
- new java.sql.Timestamp(format),
- DateTimeUtils.getTimeZoneCalendar(timeZone)
- );
- return true;
-
default:
return false;
}
@@ -1697,6 +1674,14 @@ public class GenericDatabaseDialect implements DatabaseDialect {
return null;
}
+ protected boolean maybeBindDebeziumLogical(
+ PreparedStatement statement,
+ int index,
+ Schema schema,
+ Object value
+ ) throws SQLException {
+ return DebeziumTimeTypes.maybeBindDebeziumLogical(statement, index, schema, value, timeZone);
+ }
protected boolean maybeBindPrimitive(
PreparedStatement statement,
int index,