Add a config option to force double types when schema is unavailable

Author: jerrinot

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

@@ -37,6 +37,9 @@ public final class QuestDBSinkConnectorConfig extends AbstractConfig {
     public static final String SYMBOL_COLUMNS_CONFIG = "symbols";
     private static final String SYMBOL_COLUMNS_DOC = "Comma separated list of columns that should be symbol type";
 
+    public static final String DOUBLE_COLUMNS_CONFIG = "doubles";
+    private static final String DOUBLE_COLUMNS_DOC = "Comma separated list of columns that should be double type";
+
     public static final String USERNAME = "username";
     private static final String USERNAME_DOC = "Username for QuestDB ILP authentication";
 
@@ -64,6 +67,7 @@ public static ConfigDef conf() {
                 .define(DESIGNATED_TIMESTAMP_COLUMN_NAME_CONFIG, Type.STRING, null, Importance.MEDIUM, DESIGNATED_TIMESTAMP_COLUMN_NAME_DOC)
                 .define(INCLUDE_KEY_CONFIG, Type.BOOLEAN, true, Importance.MEDIUM, INCLUDE_KEY_DOC)
                 .define(SYMBOL_COLUMNS_CONFIG, Type.STRING, null, Importance.MEDIUM, SYMBOL_COLUMNS_DOC)
+                .define(DOUBLE_COLUMNS_CONFIG, Type.STRING, null, Importance.MEDIUM, DOUBLE_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)
@@ -102,6 +106,10 @@ public String getSymbolColumns() {
         return getString(SYMBOL_COLUMNS_CONFIG);
     }
 
+    public String getDoubleColumns() {
+        return getString(DOUBLE_COLUMNS_CONFIG);
+    }
+
     public String getUsername() {
         return getString(USERNAME);
     }

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

@@ -18,8 +18,11 @@
 import org.slf4j.LoggerFactory;
 
 import java.util.Collection;
+import java.util.Collections;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.TimeUnit;
 
 public final class QuestDBSinkTask extends SinkTask {
@@ -33,6 +36,7 @@ public final class QuestDBSinkTask extends SinkTask {
     private String timestampColumnName;
     private long timestampColumnValue = Long.MIN_VALUE;
     private TimeUnit timestampUnits;
+    private Set<CharSequence> doubleColumns;
 
     @Override
     public String version() {
@@ -42,6 +46,15 @@ public String version() {
     @Override
     public void start(Map<String, String> map) {
         this.config = new QuestDBSinkConnectorConfig(map);
+        String doubleColumnsConfig = config.getDoubleColumns();
+        if (doubleColumnsConfig == null) {
+            doubleColumns = Collections.emptySet();
+        } else {
+            doubleColumns = new HashSet<>();
+            for (String symbolColumn : doubleColumnsConfig.split(",")) {
+                doubleColumns.add(symbolColumn.trim());
+            }
+        }
         this.sender = createSender();
         this.timestampColumnName = config.getDesignatedTimestampColumnName();
         this.timestampUnits = config.getTimestampUnitsOrNull();
@@ -181,9 +194,19 @@ private void writePhysicalTypeWithoutSchema(String name, Object value, String fa
         if (value instanceof String) {
             sender.stringColumn(actualName, (String) value);
         } else if (value instanceof Long) {
-            sender.longColumn(actualName, (Long) value);
+            Long longValue = (Long) value;
+            if (doubleColumns.contains(actualName)) {
+                sender.doubleColumn(actualName, longValue.doubleValue());
+            } else {
+                sender.longColumn(actualName, longValue);
+            }
         } else if (value instanceof Integer) {
-            sender.longColumn(actualName, (Integer) value);
+            Integer intValue = (Integer) value;
+            if (doubleColumns.contains(actualName)) {
+                sender.doubleColumn(actualName, intValue.doubleValue());
+            } else {
+                sender.longColumn(actualName, intValue);
+            }
         } else if (value instanceof Boolean) {
             sender.boolColumn(actualName, (Boolean) value);
         } else if (value instanceof Double) {

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

@@ -452,6 +452,24 @@ public void testJsonNoSchema() {
                 "select firstname,lastname,age from " + topicName);
     }
 
+    @Test
+    public void testJsonNoSchema_mixedFlotingAndIntTypes() {
+        connect.kafka().createTopic(topicName, 1);
+        Map<String, String> props = baseConnectorProps(topicName);
+        props.put("value.converter.schemas.enable", "false");
+        props.put(QuestDBSinkConnectorConfig.DOUBLE_COLUMNS_CONFIG, "age");
+        connect.configureConnector(CONNECTOR_NAME, props);
+        assertConnectorTaskRunningEventually();
+        connect.kafka().produce(topicName, "key", "{\"firstname\":\"John\",\"lastname\":\"Doe\",\"age\":42}");
+        connect.kafka().produce(topicName, "key", "{\"firstname\":\"Jane\",\"lastname\":\"Doe\",\"age\":42.5}");
+
+        QuestDBUtils.assertSqlEventually(questDBContainer, "\"firstname\",\"lastname\",\"age\"\r\n"
+                        + "\"John\",\"Doe\",42.0\r\n"
+                        + "\"Jane\",\"Doe\",42.5\r\n",
+                "select firstname,lastname,age from " + topicName);
+    }
+
+
     @Test
     public void testJsonNoSchema_ArrayNotSupported() {
         connect.kafka().createTopic(topicName, 1);

readme.md

@@ -47,6 +47,7 @@ The connector supports following Options:
 | 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 |
+| doubles                | STRING  | volume,price                                                | N/A                | Comma separated list of columns that should be double type |
 | username               | STRING  | user1                                                       | admin              | User name for QuestDB. Used only when token is non-empty   |
 | token                  | STRING  | <sub>QgHCOyq35D5HocCMrUGJinEsjEscJlCp7FZQETH21Bw</sub>      | N/A                | Token for QuestDB authentication                           |
 | tls                    | BOOLEAN | true                                                        | false              | Use TLS for QuestDB connection                             |