use QuestDB timestamp parser

So we don't allocate with every timestamp like psychopaths :)

Author: jerrinot

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

@@ -38,7 +38,7 @@ public final class QuestDBSinkConnectorConfig extends AbstractConfig {
     private static final String TIMESTAMP_STRING_FIELDS_DOC = "Comma separated list of string fields that should be parsed as timestamp.";
 
     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";
@@ -64,8 +64,8 @@ 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 = "yyyy-MM-dd'T'HH:mm:ss.SSSSSS'Z'";
-    private static final String TIMESTAMP_FORMAT_DOC = "Default timestamp format";
+    public static final String TIMESTAMP_FORMAT = "timestamp.string.format";
+    private static final String TIMESTAMP_FORMAT_DOC = "Default timestamp format. Used when parsing timestamp string fields";
 
     private static final String DEFAULT_TIMESTAMP_FORMAT = "yyyy-MM-dd'T'HH:mm:ss.SSSSSS'Z'";
 
@@ -200,11 +200,6 @@ public void ensureValid(String name, Object value) {
             if (!(value instanceof String)) {
                 throw new ConfigException(name, value, "Timestamp format must be a string");
             }
-            try {
-                DateTimeFormatter.ofPattern((String) value);
-            } catch (IllegalArgumentException e) {
-                throw new ConfigException(name, value, "Timestamp format is not a valid DateTimeFormatter pattern. Error='" + e.getMessage() + "'");
-            }
         }
     }
 

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;
@@ -19,9 +22,6 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.time.format.DateTimeFormatter;
-import java.time.temporal.ChronoField;
-import java.time.temporal.TemporalAccessor;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashSet;
@@ -45,7 +45,7 @@ public final class QuestDBSinkTask extends SinkTask {
     private Set<String> stringTimestampColumns;
     private int remainingRetries;
     private long batchesSinceLastError = 0;
-    private DateTimeFormatter timestampParser;
+    private DateFormat dataFormat;
 
     @Override
     public String version() {
@@ -61,7 +61,7 @@ public void start(Map<String, String> map) {
             for (String symbolColumn : timestampStringFields.split(",")) {
                 stringTimestampColumns.add(symbolColumn.trim());
             }
-            this.timestampParser = DateTimeFormatter.ofPattern(config.getDefaultTimestampFormat());
+            dataFormat = TimestampParserCompiler.compilePattern(config.getDefaultTimestampFormat());
         } else {
             stringTimestampColumns = Collections.emptySet();
         }
@@ -251,7 +251,7 @@ private long resolveDesignatedTimestampColumnValue(Object value, Schema schema)
         }
         if (value instanceof String) {
             log.debug("Timestamp column value is a string");
-            return parseTimestamp((String) value, TimeUnit.NANOSECONDS);
+            return parseToMicros((String) value) * 1000;
         }
         if (!(value instanceof Long)) {
             throw new ConnectException("Unsupported timestamp column type: " + value.getClass());
@@ -277,7 +277,7 @@ private void writePhysicalTypeWithoutSchema(String name, Object value, String fa
         if (value instanceof String) {
             String stringVal = (String) value;
             if (stringTimestampColumns.contains(actualName)) {
-                long timestamp = parseTimestamp(stringVal, TimeUnit.MICROSECONDS);
+                long timestamp = parseToMicros(stringVal);
                 sender.timestampColumn(actualName, timestamp);
             } else {
                 sender.stringColumn(actualName, stringVal);
@@ -310,11 +310,12 @@ private void writePhysicalTypeWithoutSchema(String name, Object value, String fa
         }
     }
 
-    private long parseTimestamp(String timestamp, TimeUnit unit) {
-        TemporalAccessor accessor = timestampParser.parse(timestamp);
-        long seconds = accessor.getLong(ChronoField.INSTANT_SECONDS);
-        int nanos = accessor.get(ChronoField.NANO_OF_SECOND);
-        return unit.convert(seconds, TimeUnit.SECONDS) + unit.convert(nanos, TimeUnit.NANOSECONDS);
+    private long parseToMicros(String timestamp) {
+        try {
+            return dataFormat.parse(timestamp, DateFormatUtils.enLocale);
+        } catch (NumericException e) {
+            throw new ConnectException("Cannot parse timestamp: " + timestamp, e);
+        }
     }
 
     private static String sanitizeName(String name) {
@@ -349,7 +350,7 @@ private boolean tryWritePhysicalTypeFromSchema(String name, Schema schema, Objec
             case STRING:
                 String s = (String) value;
                 if (stringTimestampColumns.contains(primitiveTypesName)) {
-                    long timestamp = parseTimestamp(s, TimeUnit.MICROSECONDS);
+                    long timestamp = parseToMicros(s);
                     sender.timestampColumn(sanitizedName, timestamp);
                 } else {
                     sender.stringColumn(sanitizedName, s);

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

@@ -0,0 +1,31 @@
+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
+    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
+            if (cache == null) {
+                cache = new CharSequenceObjHashMap<>();
+            }
+            DateFormat format = cache.get(timestampPattern);
+            if (format != null) {
+                return format;
+            }
+            format = compiler.compile(timestampPattern);
+            cache.put(timestampPattern, format);
+            return format;
+        }
+    }
+}

connector/src/test/java/io/questdb/kafka/QuestDBSinkConnectorEmbeddedTest.java

@@ -747,7 +747,7 @@ public void testParsingStringTimestamp() {
         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.SSSSSS z");
+        props.put(QuestDBSinkConnectorConfig.TIMESTAMP_FORMAT, "yyyy-MM-dd HH:mm:ss.U+ z");
         props.put(QuestDBSinkConnectorConfig.TIMESTAMP_STRING_FIELDS, "born,death");
 
         connect.configureConnector(CONNECTOR_NAME, props);