support for symbols

Author: jerrinot

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

@@ -37,6 +37,8 @@ final class BufferingSender implements Sender {
     private final List<CharSequence> symbolColumnNames = new ArrayList<>(DEFAULT_CAPACITY);
     private final List<CharSequence> symbolColumnValues = new ArrayList<>(DEFAULT_CAPACITY);
     private final Set<CharSequence> symbolColumns = new HashSet<>();
+    private final List<CharSequence> doubleArrayNames = new ArrayList<>(DEFAULT_CAPACITY);
+    private final List<double[]> doubleArrayValues = new ArrayList<>(DEFAULT_CAPACITY);
 
     BufferingSender(Sender sender, String symbolColumns) {
         this.sender = sender;
@@ -201,6 +203,14 @@ private void transferFields() {
         }
         timestampNames.clear();
         timestampValues.clear();
+
+        for (int i = 0, n = doubleArrayNames.size(); i < n; i++) {
+            CharSequence fieldName = doubleArrayNames.get(i);
+            double[] fieldValue = doubleArrayValues.get(i);
+            sender.doubleArray(fieldName, fieldValue);
+        }
+        doubleArrayNames.clear();
+        doubleArrayValues.clear();
     }
 
     private static long unitToMicros(long value, ChronoUnit unit) {
@@ -241,7 +251,9 @@ public void close() {
 
     @Override
     public Sender doubleArray(CharSequence charSequence, double[] doubles) {
-        throw new UnsupportedOperationException("not implemented");
+        doubleArrayNames.add(charSequence);
+        doubleArrayValues.add(doubles);
+        return this;
     }
 
     @Override

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

@@ -577,6 +577,7 @@ private void handleArray(String name, Object value, Schema schema) {
             }
             sender.doubleArray(name, doubleArray);
         } else if (elementType == Schema.Type.ARRAY) {
+            // todo: handle multidimensional arrays
             onUnsupportedType(name, "Multidimensional ARRAY");
         } else {
             onUnsupportedType(name, "ARRAY<" + elementType + ">");

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

@@ -2067,6 +2067,7 @@ public void testFloat64ArraySupport(boolean useHttp) {
         connect.kafka().createTopic(topicName, 1);
         Map<String, String> props = ConnectTestUtils.baseConnectorProps(questDBContainer, topicName, useHttp);
         props.put(VALUE_CONVERTER_CLASS_CONFIG, JsonConverter.class.getName());
+        props.put(QuestDBSinkConnectorConfig.SYMBOL_COLUMNS_CONFIG, "devices");
         connect.configureConnector(ConnectTestUtils.CONNECTOR_NAME, props);
         ConnectTestUtils.assertConnectorTaskRunningEventually(connect);