DBZ-7920 Add support for timestamp infinity values for Db2

debezium · Jul 9, 2024 · a64b2c6 · a64b2c6
1 parent bd739fd
commit a64b2c6
Show file tree

Hide file tree

Showing 4 changed files with 33 additions and 26 deletions.
diff --git a/src/main/java/io/debezium/connector/jdbc/dialect/db2/Db2DatabaseDialect.java b/src/main/java/io/debezium/connector/jdbc/dialect/db2/Db2DatabaseDialect.java
@@ -128,6 +128,16 @@ public String getAlterTableColumnDelimiter() {
         return " ";
     }
 
+    @Override
+    public String getTimestampPositiveInfinityValue() {
+        return "9999-12-31T23:59:59+00:00";
+    }
+
+    @Override
+    public String getTimestampNegativeInfinityValue() {
+        return "0001-01-01T00:00:00+00:00";
+    }
+
     @Override
     public String getUpsertStatement(TableDescriptor table, SinkRecordDescriptor record) {
         final SqlStatementBuilder builder = new SqlStatementBuilder();

diff --git a/src/main/java/io/debezium/connector/jdbc/dialect/db2/ZonedTimestampType.java b/src/main/java/io/debezium/connector/jdbc/dialect/db2/ZonedTimestampType.java
@@ -9,12 +9,7 @@
 import java.time.ZonedDateTime;
 import java.util.List;
 
-import org.apache.kafka.connect.data.Schema;
-import org.apache.kafka.connect.errors.ConnectException;
-
 import io.debezium.connector.jdbc.ValueBindDescriptor;
-import io.debezium.connector.jdbc.dialect.DatabaseDialect;
-import io.debezium.connector.jdbc.type.AbstractTimestampType;
 import io.debezium.connector.jdbc.type.Type;
 import io.debezium.time.ZonedTimestamp;
 
@@ -23,34 +18,28 @@
  *
  * @author Chris Cranford
  */
-public class ZonedTimestampType extends AbstractTimestampType {
+public class ZonedTimestampType extends io.debezium.connector.jdbc.type.debezium.ZonedTimestampType {
 
     public static final ZonedTimestampType INSTANCE = new ZonedTimestampType();
 
-    @Override
-    public String[] getRegistrationKeys() {
-        return new String[]{ ZonedTimestamp.SCHEMA_NAME };
-    }
+    protected List<ValueBindDescriptor> infinityTimestampValue(int index, Object value) {
+        final ZonedDateTime zdt;
 
-    @Override
-    public String getDefaultValueBinding(DatabaseDialect dialect, Schema schema, Object value) {
-        return dialect.getFormattedTimestampWithTimeZone((String) value);
+        if (POSITIVE_INFINITY.equals(value)) {
+            zdt = ZonedDateTime.parse(getDialect().getTimestampPositiveInfinityValue(), ZonedTimestamp.FORMATTER);
+        }
+        else {
+            zdt = ZonedDateTime.parse(getDialect().getTimestampNegativeInfinityValue(), ZonedTimestamp.FORMATTER);
+        }
+
+        return List.of(new ValueBindDescriptor(index, Timestamp.valueOf((zdt.toLocalDateTime()))));
     }
 
     @Override
-    public List<ValueBindDescriptor> bind(int index, Schema schema, Object value) {
-
-        if (value == null) {
-            return List.of(new ValueBindDescriptor(index, null));
-        }
-        if (value instanceof String) {
+    protected List<ValueBindDescriptor> normalTimestampValue(int index, Object value) {
 
-            final ZonedDateTime zdt = ZonedDateTime.parse((String) value, ZonedTimestamp.FORMATTER).withZoneSameInstant(getDatabaseTimeZone().toZoneId());
-
-            return List.of(new ValueBindDescriptor(index, Timestamp.from(zdt.toInstant())));
-        }
+        final ZonedDateTime zdt = ZonedDateTime.parse((String) value, ZonedTimestamp.FORMATTER).withZoneSameInstant(getDatabaseTimeZone().toZoneId());
 
-        throw new ConnectException(String.format("Unexpected %s value '%s' with type '%s'", getClass().getSimpleName(),
-                value, value.getClass().getName()));
+        return List.of(new ValueBindDescriptor(index, Timestamp.from(zdt.toInstant())));
     }
 }
diff --git a/src/test/java/io/debezium/connector/jdbc/e2e/AbstractJdbcSinkPipelineIT.java b/src/test/java/io/debezium/connector/jdbc/e2e/AbstractJdbcSinkPipelineIT.java
@@ -2593,7 +2593,7 @@ public void testTimestampWithTimeZoneDataTypeWithInfinityValue(Source source, Si
     private static @NotNull List<ZonedDateTime> getExpectedZonedDateTimes(Sink sink) {
 
         List<ZonedDateTime> expectedValues = List.of();
-        if (sink.getType().is(SinkType.SQLSERVER)) {
+        if (sink.getType().is(SinkType.SQLSERVER) && sink.getType().is(SinkType.DB2)) {
 
             expectedValues = List.of(ZonedDateTime.of(1, 1, 1, 0, 0, 0, 0, ZoneOffset.UTC),
                     ZonedDateTime.of(9999, 12, 31, 23, 59, 59, 0, ZoneOffset.UTC));

diff --git a/src/test/resources/logback-test.xml b/src/test/resources/logback-test.xml
@@ -28,4 +28,12 @@
         <appender-ref ref="CONSOLE" />
     </logger>
 
+    <!-- To enable Hibernate loggin set lebel to debug -->
+    <logger name="org.hibernate.SQL" level="off">
+        <appender-ref ref="CONSOLE" />
+    </logger>
+    <!-- To enable Hibernate loggin set lebel to trace -->
+    <logger name="org.hibernate.type" level="off">
+        <appender-ref ref="CONSOLE" />
+    </logger>
 </configuration>