better handling of debezium temporal types

Author: jerrinot

connector/src/main/java/io/questdb/kafka/QuestDBSinkTask.java

@@ -1,6 +1,7 @@
 package io.questdb.kafka;
 
 import io.questdb.client.Sender;
+import io.questdb.std.datetime.microtime.Timestamps;
 import org.apache.kafka.clients.consumer.OffsetAndMetadata;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.connect.data.Date;
@@ -234,6 +235,11 @@ private boolean tryWriteLogicalType(String name, Schema schema, Object value) {
                 long l = (Long) value;
                 sender.timestampColumn(name, l);
                 return true;
+            case "io.debezium.time.Date":
+                int i = (Integer) value;
+                long micros = Timestamps.addDays(0, i);
+                sender.timestampColumn(name, micros);
+                return true;
             case Timestamp.LOGICAL_NAME:
             case Date.LOGICAL_NAME:
                 java.util.Date d = (java.util.Date) value;

integration-tests/debezium/src/test/java/kafka/DebeziumIT.java

@@ -240,6 +240,48 @@ public void testEventTime() throws SQLException {
         }
     }
 
+    @Test
+    public void testEventTimeMicros() throws SQLException {
+        String questTableName = "test_event_time_micros";
+        try (Connection connection = getConnection(postgresContainer);
+             Statement statement = connection.createStatement()) {
+
+            statement.execute("create schema " + PG_SCHEMA_NAME);
+            statement.execute("create table " + PG_SCHEMA_NAME + "." + PG_TABLE_NAME + " (id int8 not null, title varchar(255), created_at timestamp(6), primary key (id))");
+            statement.execute("insert into "  + PG_SCHEMA_NAME + "." + PG_TABLE_NAME + " values (1, 'Learn CDC', '2021-01-02T01:02:03.123456Z')");
+
+            startDebeziumConnector();
+            ConnectorConfiguration questSink = newQuestSinkBaseConfig(questTableName);
+            questSink.with(QuestDBSinkConnectorConfig.DESIGNATED_TIMESTAMP_COLUMN_NAME_CONFIG, "created_at");
+            debeziumContainer.registerConnector(QUESTDB_CONNECTOR_NAME, questSink);
+
+            QuestDBUtils.assertSqlEventually(questDBContainer, "\"id\",\"title\",\"timestamp\"\r\n"
+                            + "1,\"Learn CDC\",\"2021-01-02T01:02:03.123456Z\"\r\n",
+                    "select id, title, timestamp from " + questTableName);
+        }
+    }
+
+    @Test
+    public void testEventTimeNanos() throws SQLException {
+        String questTableName = "test_event_time_nanos";
+        try (Connection connection = getConnection(postgresContainer);
+             Statement statement = connection.createStatement()) {
+
+            statement.execute("create schema " + PG_SCHEMA_NAME);
+            statement.execute("create table " + PG_SCHEMA_NAME + "." + PG_TABLE_NAME + " (id int8 not null, title varchar(255), created_at timestamp(9), primary key (id))");
+            statement.execute("insert into "  + PG_SCHEMA_NAME + "." + PG_TABLE_NAME + " values (1, 'Learn CDC', '2021-01-02T01:02:03.123456789Z')");
+
+            startDebeziumConnector();
+            ConnectorConfiguration questSink = newQuestSinkBaseConfig(questTableName);
+            questSink.with(QuestDBSinkConnectorConfig.DESIGNATED_TIMESTAMP_COLUMN_NAME_CONFIG, "created_at");
+            debeziumContainer.registerConnector(QUESTDB_CONNECTOR_NAME, questSink);
+
+            QuestDBUtils.assertSqlEventually(questDBContainer, "\"id\",\"title\",\"timestamp\"\r\n"
+                            + "1,\"Learn CDC\",\"2021-01-02T01:02:03.123457Z\"\r\n",
+                    "select id, title, timestamp from " + questTableName);
+        }
+    }
+
     @Test
     public void testNonDesignatedTimestamp() throws SQLException {
         String questTableName = "test_non_designated_timestamp";
@@ -260,6 +302,26 @@ public void testNonDesignatedTimestamp() throws SQLException {
         }
     }
 
+    @Test
+    public void testDate() throws SQLException {
+        String questTableName = "test_date";
+        try (Connection connection = getConnection(postgresContainer);
+             Statement statement = connection.createStatement()) {
+
+            statement.execute("create schema " + PG_SCHEMA_NAME);
+            statement.execute("create table " + PG_SCHEMA_NAME + "." + PG_TABLE_NAME + " (id int8 not null, title varchar(255), created_at date, primary key (id))");
+            statement.execute("insert into "  + PG_SCHEMA_NAME + "." + PG_TABLE_NAME + " values (1, 'Learn CDC', '2021-01-02')");
+
+            startDebeziumConnector();
+            ConnectorConfiguration questSink = newQuestSinkBaseConfig(questTableName);
+            debeziumContainer.registerConnector(QUESTDB_CONNECTOR_NAME, questSink);
+
+            QuestDBUtils.assertSqlEventually(questDBContainer, "\"id\",\"title\",\"created_at\"\r\n"
+                            + "1,\"Learn CDC\",\"2021-01-02T00:00:00.000000Z\"\r\n",
+                    "select id, title, created_at from " + questTableName);
+        }
+    }
+
     private static void startDebeziumConnector() {
         ConnectorConfiguration connector = ConnectorConfiguration
                 .forJdbcContainer(postgresContainer)