Merge pull request #5 from questdb/feat/timestamp-unit-detection

jerrinot · web-flow · commit efeab770b59a · 2022-11-26T14:52:40.000+01:00
[feat] timestamp unit configuration
diff --git a/connector/src/main/java/io/questdb/kafka/QuestDBSinkConnectorConfig.java b/connector/src/main/java/io/questdb/kafka/QuestDBSinkConnectorConfig.java
@@ -4,8 +4,10 @@
 import org.apache.kafka.common.config.ConfigDef;
 import org.apache.kafka.common.config.ConfigDef.Importance;
 import org.apache.kafka.common.config.ConfigDef.Type;
+import org.apache.kafka.connect.errors.ConnectException;
 
 import java.util.Map;
+import java.util.concurrent.TimeUnit;
 
 public final class QuestDBSinkConnectorConfig extends AbstractConfig {
     public static final String HOST_CONFIG = "host";
@@ -26,6 +28,9 @@ public final class QuestDBSinkConnectorConfig extends AbstractConfig {
     public static final String DESIGNATED_TIMESTAMP_COLUMN_NAME_CONFIG = "timestamp.field.name";
     private static final String DESIGNATED_TIMESTAMP_COLUMN_NAME_DOC = "Designated timestamp field name";
 
+    public static final String TIMESTAMP_UNITS_CONFIG = "timestamp.units";
+    private static final String TIMESTAMP_UNITS_DOC = "Units of timestamp field. Possible values: auto, millis, micros, nanos";
+
     public static final String INCLUDE_KEY_CONFIG = "include.key";
     private static final String INCLUDE_KEY_DOC = "Include key in the table";
 
@@ -61,7 +66,8 @@ public static ConfigDef conf() {
                 .define(SYMBOL_COLUMNS_CONFIG, Type.STRING, null, Importance.MEDIUM, SYMBOL_COLUMNS_DOC)
                 .define(USERNAME, Type.STRING, "admin", Importance.MEDIUM, USERNAME_DOC)
                 .define(TOKEN, Type.PASSWORD, null, Importance.MEDIUM, TOKEN_DOC)
-                .define(TLS, Type.BOOLEAN, false, Importance.MEDIUM, TLS_DOC);
+                .define(TLS, Type.BOOLEAN, false, Importance.MEDIUM, TLS_DOC)
+                .define(TIMESTAMP_UNITS_CONFIG, Type.STRING, "auto", ConfigDef.ValidString.in("auto", "millis", "micros", "nanos"), Importance.LOW, TIMESTAMP_UNITS_DOC);
     }
 
     public String getHost() {
@@ -107,4 +113,20 @@ public String getToken() {
     public boolean isTls() {
         return getBoolean(TLS);
     }
+
+    public TimeUnit getTimestampUnitsOrNull() {
+        String configured = getString(TIMESTAMP_UNITS_CONFIG);
+        switch (configured) {
+            case "millis":
+                return TimeUnit.MILLISECONDS;
+            case "micros":
+                return TimeUnit.MICROSECONDS;
+            case "nanos":
+                return TimeUnit.NANOSECONDS;
+            case "auto":
+                return null;
+            default:
+                throw new ConnectException("Unknown timestamp units mode: " + configured + ". Possible values: auto, millis, micros, nanos");
+        }
+    }
 }
diff --git a/connector/src/main/java/io/questdb/kafka/QuestDBSinkTask.java b/connector/src/main/java/io/questdb/kafka/QuestDBSinkTask.java
@@ -32,6 +32,7 @@ public final class QuestDBSinkTask extends SinkTask {
     private QuestDBSinkConnectorConfig config;
     private String timestampColumnName;
     private long timestampColumnValue = Long.MIN_VALUE;
+    private TimeUnit timestampUnits;
 
     @Override
     public String version() {
@@ -43,6 +44,7 @@ public void start(Map<String, String> map) {
         this.config = new QuestDBSinkConnectorConfig(map);
         this.sender = createSender();
         this.timestampColumnName = config.getDesignatedTimestampColumnName();
+        this.timestampUnits = config.getTimestampUnitsOrNull();
     }
 
     private Sender createSender() {
@@ -153,7 +155,7 @@ private void handleObject(String name, Schema schema, Object value, String fallb
         writePhysicalTypeWithoutSchema(name, value, fallbackName);
     }
 
-    private static long resolveDesignatedTimestampColumnValue(Object value, Schema schema) {
+    private long resolveDesignatedTimestampColumnValue(Object value, Schema schema) {
         if (value instanceof java.util.Date) {
             return TimeUnit.MILLISECONDS.toNanos(((java.util.Date) value).getTime());
         }
@@ -162,15 +164,11 @@ private static long resolveDesignatedTimestampColumnValue(Object value, Schema s
         }
         long longValue = (Long) value;
         TimeUnit inputUnit;
-        if (schema == null || schema.name() == null) {
-            // no schema, assuming millis since epoch
-            inputUnit = TimeUnit.MILLISECONDS;
-        } else if ("io.debezium.time.MicroTimestamp".equals(schema.name())) {
+        if (schema == null || !"io.debezium.time.MicroTimestamp".equals(schema.name())) {
+            inputUnit = TimestampHelper.getTimestampUnits(timestampUnits, longValue);
+        } else {
             // special case: Debezium micros since epoch
             inputUnit = TimeUnit.MICROSECONDS;
-        } else {
-            // no idea what's that, let's assume it's again millis since epoch and hope for the best
-            inputUnit = TimeUnit.MILLISECONDS;
         }
         return inputUnit.toNanos(longValue);
     }
diff --git a/connector/src/main/java/io/questdb/kafka/TimestampHelper.java b/connector/src/main/java/io/questdb/kafka/TimestampHelper.java
@@ -0,0 +1,26 @@
+package io.questdb.kafka;
+
+import java.util.concurrent.TimeUnit;
+
+public final class TimestampHelper {
+    private TimestampHelper() {
+    }
+
+    static TimeUnit guessTimestampUnits(long timestamp) {
+        if (timestamp < 10000000000000L) { // 11/20/2286, 5:46:40 PM in millis and 4/26/1970, 5:46:40 PM in micros
+            return TimeUnit.MILLISECONDS;
+        } else if (timestamp < 10000000000000000L) {
+            return TimeUnit.MICROSECONDS;
+        } else {
+            return TimeUnit.NANOSECONDS;
+        }
+    }
+
+    static TimeUnit getTimestampUnits(TimeUnit hint, long timestamp) {
+        if (hint == null) {
+            return guessTimestampUnits(timestamp);
+        } else {
+            return hint;
+        }
+    }
+}
diff --git a/connector/src/test/java/io/questdb/kafka/QuestDBSinkConnectorEmbeddedTest.java b/connector/src/test/java/io/questdb/kafka/QuestDBSinkConnectorEmbeddedTest.java
@@ -27,14 +27,14 @@
 import org.testcontainers.junit.jupiter.Container;
 import org.testcontainers.junit.jupiter.Testcontainers;
 
-import java.io.IOException;
 import java.math.BigDecimal;
 import java.util.Calendar;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.TimeZone;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import static java.util.Collections.singletonMap;
@@ -221,6 +221,93 @@ public void testUpfrontTable_withSymbols() {
     }
 
     @Test
+    public void testTimestampUnitResolution_auto() {
+        connect.kafka().createTopic(topicName, 1);
+        Map<String, String> props = baseConnectorProps(topicName);
+        props.put("value.converter.schemas.enable", "false");
+        props.put(QuestDBSinkConnectorConfig.DESIGNATED_TIMESTAMP_COLUMN_NAME_CONFIG, "birth");
+        connect.configureConnector(CONNECTOR_NAME, props);
+        assertConnectorTaskRunningEventually();
+
+        java.util.Date birth = new Calendar.Builder()
+                .setTimeZone(TimeZone.getTimeZone("UTC"))
+                .setDate(2022, 9, 23) // note: month is 0-based
+                .setTimeOfDay(13, 53, 59, 123)
+                .build().getTime();
+
+        long birthInMillis = birth.getTime();
+        long birthInMicros = birthInMillis * 1000;
+        long birthInNanos = birthInMicros * 1000;
+
+
+        connect.kafka().produce(topicName, "foo", "{\"firstname\":\"John\",\"lastname\":\"Doe\",\"birth\":" + birthInMillis + "}");
+        connect.kafka().produce(topicName, "bar", "{\"firstname\":\"Jane\",\"lastname\":\"Doe\",\"birth\":" + birthInMicros + "}");
+        connect.kafka().produce(topicName, "baz", "{\"firstname\":\"Jack\",\"lastname\":\"Doe\",\"birth\":" + birthInNanos + "}");
+
+        QuestDBUtils.assertSqlEventually(questDBContainer, "\"firstname\",\"lastname\",\"timestamp\"\r\n"
+                        + "\"John\",\"Doe\",\"2022-10-23T13:53:59.123000Z\"\r\n"
+                        + "\"Jane\",\"Doe\",\"2022-10-23T13:53:59.123000Z\"\r\n"
+                        + "\"Jack\",\"Doe\",\"2022-10-23T13:53:59.123000Z\"\r\n",
+                "select firstname,lastname,timestamp from " + topicName);
+    }
+
+    @Test
+    public void testTimestampUnitResolution_millis() {
+        testTimestampUnitResolution0("millis");
+    }
+
+    @Test
+    public void testTimestampUnitResolution_micros() {
+        testTimestampUnitResolution0("micros");
+    }
+
+    @Test
+    public void testTimestampUnitResolution_nanos() {
+        testTimestampUnitResolution0("nanos");
+    }
+
+    private void testTimestampUnitResolution0(String mode) {
+        TimeUnit unit;
+        switch (mode) {
+            case "nanos":
+                unit = TimeUnit.NANOSECONDS;
+                break;
+            case "micros":
+                unit = TimeUnit.MICROSECONDS;
+                break;
+            case "millis":
+                unit = TimeUnit.MILLISECONDS;
+                break;
+            default:
+                throw new IllegalArgumentException("Unknown mode: " + mode);
+        }
+        connect.kafka().createTopic(topicName, 1);
+        Map<String, String> props = baseConnectorProps(topicName);
+        props.put("value.converter.schemas.enable", "false");
+        props.put(QuestDBSinkConnectorConfig.DESIGNATED_TIMESTAMP_COLUMN_NAME_CONFIG, "birth");
+        props.put(QuestDBSinkConnectorConfig.TIMESTAMP_UNITS_CONFIG, mode);
+        connect.configureConnector(CONNECTOR_NAME, props);
+        assertConnectorTaskRunningEventually();
+
+        long birthMillis = new Calendar.Builder()
+                .setTimeZone(TimeZone.getTimeZone("UTC"))
+                .setDate(2206, 10, 20) // note: month is 0-based
+                .setTimeOfDay(17, 46, 39, 999)
+                .build().getTime().getTime();
+
+        long birthTarget = unit.convert(birthMillis, MILLISECONDS);
+
+        connect.kafka().produce(topicName, "foo", "{\"firstname\":\"John\",\"lastname\":\"Doe\",\"birth\":0}");
+        connect.kafka().produce(topicName, "bar", "{\"firstname\":\"Jane\",\"lastname\":\"Doe\",\"birth\":" + birthTarget + "}");
+
+        QuestDBUtils.assertSqlEventually(questDBContainer, "\"firstname\",\"lastname\",\"timestamp\"\r\n"
+                        + "\"John\",\"Doe\",\"1970-01-01T00:00:00.000000Z\"\r\n"
+                        + "\"Jane\",\"Doe\",\"2206-11-20T17:46:39.999000Z\"\r\n",
+                "select firstname,lastname,timestamp from " + topicName);
+    }
+
+
+                                              @Test
     public void testUpfrontTable() {
         connect.kafka().createTopic(topicName, 1);
         Map<String, String> props = baseConnectorProps(topicName);
diff --git a/connector/src/test/java/io/questdb/kafka/TimestampHelperTest.java b/connector/src/test/java/io/questdb/kafka/TimestampHelperTest.java
@@ -0,0 +1,52 @@
+package io.questdb.kafka;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Calendar;
+import java.util.Date;
+import java.util.TimeZone;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+public class TimestampHelperTest {
+
+    @Test
+    public void testBoundaries_auto() {
+        // 4/26/1970, 5:46:40 PM
+        Date lowerBound = new Calendar.Builder()
+                .setTimeZone(TimeZone.getTimeZone("UTC"))
+                .setDate(1970, 3, 26) // note: month is 0-based
+                .setTimeOfDay(17, 46, 40, 1)
+                .build().getTime();
+
+        // 11/20/2286, 5:46:40 PM
+        Date upperBound = new Calendar.Builder()
+                .setTimeZone(TimeZone.getTimeZone("UTC"))
+                .setDate(2286, 10, 20) // note: month is 0-based
+                .setTimeOfDay(17, 46, 39, 999)
+                .build().getTime();
+
+        assertEquals(TimeUnit.MILLISECONDS, TimestampHelper.getTimestampUnits(null, lowerBound.getTime()));
+        assertEquals(TimeUnit.MILLISECONDS, TimestampHelper.getTimestampUnits(null, upperBound.getTime()));
+
+        assertEquals(TimeUnit.MICROSECONDS, TimestampHelper.getTimestampUnits(null, lowerBound.getTime() * 1000));
+        assertEquals(TimeUnit.MICROSECONDS, TimestampHelper.getTimestampUnits(null, upperBound.getTime() * 1000));
+
+        assertEquals(TimeUnit.NANOSECONDS, TimestampHelper.getTimestampUnits(null, lowerBound.getTime() * 1000000));
+        assertEquals(TimeUnit.NANOSECONDS, TimestampHelper.getTimestampUnits(null, Long.MAX_VALUE)); //upper bound in nanos does not fit in long
+    }
+
+    @Test
+    public void testBoundaries_explicit() {
+        assertEquals(TimeUnit.MILLISECONDS, TimestampHelper.getTimestampUnits(TimeUnit.MILLISECONDS, 0));
+        assertEquals(TimeUnit.MILLISECONDS, TimestampHelper.getTimestampUnits(TimeUnit.MILLISECONDS, Long.MAX_VALUE));
+
+        assertEquals(TimeUnit.MICROSECONDS, TimestampHelper.getTimestampUnits(TimeUnit.MICROSECONDS, 0));
+        assertEquals(TimeUnit.MICROSECONDS, TimestampHelper.getTimestampUnits(TimeUnit.MICROSECONDS, Long.MAX_VALUE));
+
+        assertEquals(TimeUnit.NANOSECONDS, TimestampHelper.getTimestampUnits(TimeUnit.NANOSECONDS, 0));
+        assertEquals(TimeUnit.NANOSECONDS, TimestampHelper.getTimestampUnits(TimeUnit.NANOSECONDS, Long.MAX_VALUE));
+    }
+
+}
diff --git a/kafka-questdb-connector-samples/stocks/pom.xml b/kafka-questdb-connector-samples/stocks/pom.xml
@@ -79,6 +79,7 @@
             <plugin>
                 <groupId>org.springframework.boot</groupId>
                 <artifactId>spring-boot-maven-plugin</artifactId>
+                <version>${spring.boot.version}</version>
                 <configuration>
                     <mainClass>io.questdb.kafka.samples.StocksApp</mainClass>
                 </configuration>
diff --git a/readme.md b/readme.md
@@ -44,6 +44,7 @@ The connector supports following Options:
 | value.prefix           | STRING  | from_value                                                  | N/A                | Prefix for value fields                                    |
 | skip.unsupported.types | BOOLEAN | false                                                       | false              | Skip unsupported types                                     |
 | timestamp.field.name   | STRING  | pickup_time                                                 | N/A                | Designated timestamp field name                            |
+| timestamp.units        | STRING  | micros                                                      | auto               | Designated timestamp field units                           |
 | include.key            | BOOLEAN | false                                                       | true               | Include message key in target table                        |
 | symbols                | STRING  | instrument,stock                                            | N/A                | Comma separated list of columns that should be symbol type |
 | username               | STRING  | user1                                                       | admin              | User name for QuestDB. Used only when token is non-empty   |
@@ -76,7 +77,7 @@ The connector will create a table with the following columns:
 | John                        | Doe                        | 30                  | Main Street                      | New York                       |
 
 ## Designated Timestamps
-The connector supports designated timestamps. If the message contains a field with a timestamp, the connector can use it as a timestamp for the row. The field name must be configured using `timestamp.field.name` option. The field must either a simple number or a timestamp. When it's a simple number, the connector will interpret it as a Unix timestamp in milliseconds.
+The connector supports designated timestamps. If a message contains a field with a timestamp, the connector can use it as a timestamp for the row. The field name must be configured using `timestamp.field.name` option. The field must either a plain integer or being labelled as a timestamp in a message schema. When it's a plain integer, the connector will autodetect its units. This works for timestamps after 4/26/1970, 5:46:40 PM. The units can be also configured explicitly using `timestamp.units` option. Supported configuration values are `nanos`, `micros`, `millis` and `auto`. 
 
 ## QuestDB Symbol Type
 QuestDB supports a special type called [Symbol](https://questdb.io/docs/concept/symbol/). This connector never creates a column with a type `SYMBOL`. Instead, it creates a column with a type `STRING`. If you want to use `SYMBOL` type, you can pre-create a table in QuestDB and use it as a target table.
