Merge pull request #7 from questdb/feat/timestamp_parsing

jerrinot · web-flow · commit 4961939ae39f · 2023-08-07T10:12:27.000+02:00
feat: native support for parsing string timestamps
diff --git a/connector/pom.xml b/connector/pom.xml
@@ -120,7 +120,7 @@
                             <goal>kafka-connect</goal>
                         </goals>
                         <configuration>
-                            <title>Kafka Connect QuestDB</title>
+                            <title>QuestDB Kafka Connector</title>
                             <documentationUrl>https://questdb.io/docs/third-party-tools/kafka/questdb-kafka/</documentationUrl>
                             <description>
                                 QuestDB Sink connector for Apache Kafka. QuestDB is a columnar time-series database with
diff --git a/connector/src/main/java/io/questdb/kafka/QuestDBSinkConnectorConfig.java b/connector/src/main/java/io/questdb/kafka/QuestDBSinkConnectorConfig.java
@@ -33,8 +33,11 @@ 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_STRING_FIELDS = "timestamp.string.fields";
+    private static final String TIMESTAMP_STRING_FIELDS_DOC = "Comma-separated list of string fields that should be parsed as timestamps.";
+
     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";
+    private static final String TIMESTAMP_UNITS_DOC = "Units of designated 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";
@@ -60,6 +63,11 @@ public final class QuestDBSinkConnectorConfig extends AbstractConfig {
     public static final String MAX_RETRIES = "max.retries";
     private static final String MAX_RETRIES_DOC = "The maximum number of times to retry on errors before failing the task";
 
+    public static final String TIMESTAMP_FORMAT = "timestamp.string.format";
+    private static final String TIMESTAMP_FORMAT_DOC = "Timestamp format. Used when parsing timestamp string fields";
+
+    private static final String DEFAULT_TIMESTAMP_FORMAT = "yyyy-MM-dd'T'HH:mm:ss.SSSSSS'Z'";
+
     public QuestDBSinkConnectorConfig(ConfigDef config, Map<String, String> parsedConfig) {
         super(config, parsedConfig);
     }
@@ -84,7 +92,9 @@ public static ConfigDef conf() {
                 .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, null, -1, ConfigDef.Width.NONE, TIMESTAMP_UNITS_CONFIG, Collections.emptyList(), TimestampUnitsRecommender.INSTANCE)
                 .define(RETRY_BACKOFF_MS, Type.LONG, 3_000, Importance.LOW, RETRY_BACKOFF_MS_DOC)
-                .define(MAX_RETRIES, Type.INT, 10, Importance.LOW, MAX_RETRIES_DOC);
+                .define(MAX_RETRIES, Type.INT, 10, Importance.LOW, MAX_RETRIES_DOC)
+                .define(TIMESTAMP_FORMAT, Type.STRING, DEFAULT_TIMESTAMP_FORMAT, TimestampFormatValidator.INSTANCE, Importance.MEDIUM, TIMESTAMP_FORMAT_DOC)
+                .define(TIMESTAMP_STRING_FIELDS, Type.STRING, null, Importance.MEDIUM, TIMESTAMP_STRING_FIELDS_DOC);
     }
 
     public String getHost() {
@@ -95,6 +105,14 @@ public String getTable() {
         return getString(TABLE_CONFIG);
     }
 
+    public String getTimestampFormat() {
+        return getString(TIMESTAMP_FORMAT);
+    }
+
+    public String getTimestampStringFields() {
+        return getString(TIMESTAMP_STRING_FIELDS);
+    }
+
     public String getKeyPrefix() {
         return getString(KEY_PREFIX_CONFIG);
     }
@@ -174,6 +192,16 @@ public boolean visible(String name, Map<String, Object> parsedConfig) {
         }
     }
 
+    private static class TimestampFormatValidator implements ConfigDef.Validator {
+        private static final TimestampFormatValidator INSTANCE = new TimestampFormatValidator();
+        @Override
+        public void ensureValid(String name, Object value) {
+            if (!(value instanceof String)) {
+                throw new ConfigException(name, value, "Timestamp format must be a string");
+            }
+        }
+    }
+
     private static class TablenameValidator implements ConfigDef.Validator {
         private static final TablenameValidator INSTANCE = new TablenameValidator();
         @Override
diff --git a/connector/src/main/java/io/questdb/kafka/QuestDBSinkTask.java b/connector/src/main/java/io/questdb/kafka/QuestDBSinkTask.java
@@ -2,7 +2,10 @@
 
 import io.questdb.client.Sender;
 import io.questdb.cutlass.line.LineSenderException;
+import io.questdb.std.NumericException;
+import io.questdb.std.datetime.DateFormat;
 import io.questdb.std.datetime.microtime.Timestamps;
+import io.questdb.std.datetime.millitime.DateFormatUtils;
 import org.apache.kafka.clients.consumer.OffsetAndMetadata;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.connect.data.Date;
@@ -39,8 +42,10 @@ public final class QuestDBSinkTask extends SinkTask {
     private long timestampColumnValue = Long.MIN_VALUE;
     private TimeUnit timestampUnits;
     private Set<CharSequence> doubleColumns;
+    private Set<String> stringTimestampColumns;
     private int remainingRetries;
     private long batchesSinceLastError = 0;
+    private DateFormat dataFormat;
 
     @Override
     public String version() {
@@ -50,6 +55,17 @@ public String version() {
     @Override
     public void start(Map<String, String> map) {
         this.config = new QuestDBSinkConnectorConfig(map);
+        String timestampStringFields = config.getTimestampStringFields();
+        if (timestampStringFields != null) {
+            stringTimestampColumns = new HashSet<>();
+            for (String symbolColumn : timestampStringFields.split(",")) {
+                stringTimestampColumns.add(symbolColumn.trim());
+            }
+            dataFormat = TimestampParserCompiler.compilePattern(config.getTimestampFormat());
+        } else {
+            stringTimestampColumns = Collections.emptySet();
+        }
+
         String doubleColumnsConfig = config.getDoubleColumns();
         if (doubleColumnsConfig == null) {
             doubleColumns = Collections.emptySet();
@@ -233,6 +249,10 @@ private long resolveDesignatedTimestampColumnValue(Object value, Schema schema)
             log.debug("Timestamp column value is a java.util.Date");
             return TimeUnit.MILLISECONDS.toNanos(((java.util.Date) value).getTime());
         }
+        if (value instanceof String) {
+            log.debug("Timestamp column value is a string");
+            return parseToMicros((String) value) * 1000;
+        }
         if (!(value instanceof Long)) {
             throw new ConnectException("Unsupported timestamp column type: " + value.getClass());
         }
@@ -255,7 +275,13 @@ private void writePhysicalTypeWithoutSchema(String name, Object value, String fa
         }
         String actualName = name.isEmpty() ? fallbackName : sanitizeName(name);
         if (value instanceof String) {
-            sender.stringColumn(actualName, (String) value);
+            String stringVal = (String) value;
+            if (stringTimestampColumns.contains(actualName)) {
+                long timestamp = parseToMicros(stringVal);
+                sender.timestampColumn(actualName, timestamp);
+            } else {
+                sender.stringColumn(actualName, stringVal);
+            }
         } else if (value instanceof Long) {
             Long longValue = (Long) value;
             if (doubleColumns.contains(actualName)) {
@@ -284,6 +310,16 @@ private void writePhysicalTypeWithoutSchema(String name, Object value, String fa
         }
     }
 
+    private long parseToMicros(String timestamp) {
+        try {
+            return dataFormat.parse(timestamp, DateFormatUtils.enLocale);
+        } catch (NumericException e) {
+            throw new ConnectException("Cannot parse timestamp: " + timestamp + " with the configured format '" + config.getTimestampFormat() +"' use '"
+                    + QuestDBSinkConnectorConfig.TIMESTAMP_FORMAT + "' to configure the right timestamp format. " +
+                    "See https://questdb.io/docs/reference/function/date-time/#date-and-timestamp-format for timestamp parser documentation. ", e);
+        }
+    }
+
     private static String sanitizeName(String name) {
         // todo: proper implementation
         return name.replace('.', '_');
@@ -315,7 +351,12 @@ private boolean tryWritePhysicalTypeFromSchema(String name, Schema schema, Objec
                 break;
             case STRING:
                 String s = (String) value;
-                sender.stringColumn(sanitizedName, s);
+                if (stringTimestampColumns.contains(primitiveTypesName)) {
+                    long timestamp = parseToMicros(s);
+                    sender.timestampColumn(sanitizedName, timestamp);
+                } else {
+                    sender.stringColumn(sanitizedName, s);
+                }
                 break;
             case STRUCT:
                 handleStruct(name, (Struct) value, schema);
diff --git a/connector/src/main/java/io/questdb/kafka/TimestampParserCompiler.java b/connector/src/main/java/io/questdb/kafka/TimestampParserCompiler.java
@@ -0,0 +1,29 @@
+package io.questdb.kafka;
+
+import io.questdb.std.CharSequenceObjHashMap;
+import io.questdb.std.datetime.DateFormat;
+import io.questdb.std.datetime.microtime.TimestampFormatCompiler;
+
+class TimestampParserCompiler {
+    private static TimestampFormatCompiler compiler;
+    private static final Object MUTEX = new Object();
+    // we assume that there will just a few patterns hence to issue with unbounded cache growth
+    private static CharSequenceObjHashMap<DateFormat> cache;
+
+    public static DateFormat compilePattern(String timestampPattern) {
+        synchronized (MUTEX) {
+            if (compiler == null) {
+                compiler = new TimestampFormatCompiler();
+                // DateFormat instances are thread-safe, so we can cache them and use for multiple workers
+                cache = new CharSequenceObjHashMap<>();
+            }
+            DateFormat format = cache.get(timestampPattern);
+            if (format != null) {
+                return format;
+            }
+            format = compiler.compile(timestampPattern);
+            cache.put(timestampPattern, format);
+            return format;
+        }
+    }
+}
diff --git a/connector/src/test/java/io/questdb/kafka/QuestDBSinkConnectorEmbeddedTest.java b/connector/src/test/java/io/questdb/kafka/QuestDBSinkConnectorEmbeddedTest.java
@@ -458,14 +458,14 @@ private void testTimestampUnitResolution0(String mode) {
     }
 
     @Test
-    public void testTimestampSMT_parseMicroseconds_schemaLess() {
+    public void testTimestampSMT_parseTimestamp_schemaLess() {
         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, "born");
         props.put(QuestDBSinkConnectorConfig.INCLUDE_KEY_CONFIG, "false");
 
-        String timestampFormat = "yyyy-MM-dd HH:mm:ss.SSSSSS z";
+        String timestampFormat = "yyyy-MM-dd HH:mm:ss.SSS z";
         props.put("transforms", "Timestamp-born,Timestamp-death");
         props.put("transforms.Timestamp-born.type", "org.apache.kafka.connect.transforms.TimestampConverter$Value");
         props.put("transforms.Timestamp-born.field", "born");
@@ -485,8 +485,8 @@ public void testTimestampSMT_parseMicroseconds_schemaLess() {
                 "create table " + topicName + " (firstname string, lastname string, death timestamp, born timestamp) timestamp(born)",
                 QuestDBUtils.Endpoint.EXEC);
 
-        String birthTimestamp = "1985-08-02 16:41:55.402095 UTC";
-        String deadTimestamp = "2023-08-02 16:41:55.402095 UTC";
+        String birthTimestamp = "1985-08-02 16:41:55.402 UTC";
+        String deadTimestamp = "2023-08-02 16:41:55.402 UTC";
         connect.kafka().produce(topicName, "foo",
                 "{\"firstname\":\"John\""
                         + ",\"lastname\":\"Doe\""
@@ -495,18 +495,18 @@ public void testTimestampSMT_parseMicroseconds_schemaLess() {
         );
 
         QuestDBUtils.assertSqlEventually(questDBContainer, "\"firstname\",\"lastname\",\"death\",\"born\"\r\n" +
-                        "\"John\",\"Doe\",\"2023-08-02T16:48:37.095000Z\",\"1985-08-02T16:48:37.095000Z\"\r\n",
+                        "\"John\",\"Doe\",\"2023-08-02T16:41:55.402000Z\",\"1985-08-02T16:41:55.402000Z\"\r\n",
                 "select * from " + topicName);
     }
 
     @Test
-    public void testTimestampSMT_parseMicroseconds_withSchema() {
+    public void testTimestampSMT_parseTimestamp_withSchema() {
         connect.kafka().createTopic(topicName, 1);
         Map<String, String> props = baseConnectorProps(topicName);
         props.put(QuestDBSinkConnectorConfig.DESIGNATED_TIMESTAMP_COLUMN_NAME_CONFIG, "born");
         props.put(QuestDBSinkConnectorConfig.INCLUDE_KEY_CONFIG, "false");
 
-        String timestampFormat = "yyyy-MM-dd HH:mm:ss.SSSSSS z";
+        String timestampFormat = "yyyy-MM-dd HH:mm:ss.SSS z";
         props.put("transforms", "Timestamp-born,Timestamp-death");
         props.put("transforms.Timestamp-born.type", "org.apache.kafka.connect.transforms.TimestampConverter$Value");
         props.put("transforms.Timestamp-born.field", "born");
@@ -530,14 +530,14 @@ public void testTimestampSMT_parseMicroseconds_withSchema() {
         Struct struct = new Struct(schema)
                 .put("firstname", "John")
                 .put("lastname", "Doe")
-                .put("born", "1985-08-02 16:41:55.402095 UTC")
-                .put("death", "2023-08-02 16:41:55.402095 UTC");
+                .put("born", "1985-08-02 16:41:55.402 UTC")
+                .put("death", "2023-08-02 16:41:55.402 UTC");
 
 
         connect.kafka().produce(topicName, "key", new String(converter.fromConnectData(topicName, schema, struct)));
 
         QuestDBUtils.assertSqlEventually(questDBContainer, "\"firstname\",\"lastname\",\"death\",\"timestamp\"\r\n" +
-                        "\"John\",\"Doe\",\"2023-08-02T16:48:37.095000Z\",\"1985-08-02T16:48:37.095000Z\"\r\n",
+                        "\"John\",\"Doe\",\"2023-08-02T16:41:55.402000Z\",\"1985-08-02T16:41:55.402000Z\"\r\n",
                 "select * from " + topicName);
     }
 
@@ -740,6 +740,38 @@ public void testPrimitiveKey() {
                 "select firstname, lastname, age, key from " + topicName);
     }
 
+    @Test
+    public void testParsingStringTimestamp() {
+        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, "born");
+        props.put(QuestDBSinkConnectorConfig.INCLUDE_KEY_CONFIG, "false");
+        props.put(QuestDBSinkConnectorConfig.TIMESTAMP_FORMAT, "yyyy-MM-dd HH:mm:ss.SSSUUU z");
+        props.put(QuestDBSinkConnectorConfig.TIMESTAMP_STRING_FIELDS, "born,death");
+
+        connect.configureConnector(CONNECTOR_NAME, props);
+        assertConnectorTaskRunningEventually();
+
+        QuestDBUtils.assertSql(questDBContainer,
+                "{\"ddl\":\"OK\"}\n",
+                "create table " + topicName + " (firstname string, lastname string, death timestamp, born timestamp) timestamp(born)",
+                QuestDBUtils.Endpoint.EXEC);
+
+        String birthTimestamp = "1985-08-02 16:41:55.402095 UTC";
+        String deadTimestamp = "2023-08-02 16:41:55.402095 UTC";
+        connect.kafka().produce(topicName, "foo",
+                "{\"firstname\":\"John\""
+                        + ",\"lastname\":\"Doe\""
+                        + ",\"death\":\"" + deadTimestamp + "\""
+                        + ",\"born\":\"" + birthTimestamp + "\"}"
+        );
+
+        QuestDBUtils.assertSqlEventually(questDBContainer, "\"firstname\",\"lastname\",\"death\",\"born\"\r\n" +
+                        "\"John\",\"Doe\",\"2023-08-02T16:41:55.402095Z\",\"1985-08-02T16:41:55.402095Z\"\r\n",
+                "select * from " + topicName);
+    }
+
     @Test
     public void testCustomPrefixWithPrimitiveKeyAndValues() {
         connect.kafka().createTopic(topicName, 1);
