Support for QuestDB arrays (#33)

What This Enables

This PR allows you to stream multi-dimensional array data from Kafka to QuestDB.

Previously, any message containing arrays would cause the connector to crash with Unsupported type errors. Now you can store sensor readings, machine learning features, financial time series, and other array-based data.
Real-World Use Cases Now Supported

IoT & Sensor Data

{
  "device_id": "sensor_001",
  "temperature_readings": [20.1, 20.3, 20.5, 20.7],
  "location": "warehouse_a"
}

Machine Learning Features

{
  "model_id": "classifier_v2",
  "feature_matrix": [[0.1, 0.8], [0.3, 0.6], [0.9, 0.2]],
  "prediction_confidence": 0.94
}

Financial Market Data

{
  "symbol": "AAPL",
  "ohlc_5min": [[150.1, 151.2, 149.8, 150.9], [150.9, 152.1, 150.3, 151.5]],
  "volume": [12500, 13200]
}

Your array data flows through to QuestDB and gets stored as native arrays, queryable with QuestDB's array functions.
Data Format Support
With Kafka Connect Schema

// Your existing schema definitions work
Schema arraySchema = SchemaBuilder.array(Schema.FLOAT64_SCHEMA).build();

Schema-Free JSON (Most Common)

// Just send your JSON - arrays are auto-detected
{"sensor_data": [23.1, 24.5, 22.8]}`

Current Limitations

Supported

    1D, 2D, 3D arrays (current QuestDB version)
    Numeric data (integers and floats)
    Both schema and schema-free Kafka messages
    Nested arrays in complex JSON structures

Not Yet Supported

    String arrays - ["apple", "banana"]
    Mixed-type arrays - [1, "text", 3.14]
    Null elements - [1.0, null, 3.0]]
    Empty Arrays - this is a temporary limitation, empty arrays will be supported in future versions

Future Support

    Up to 32 dimensions coming in next QuestDB version
    Performance optimizations

Author: jerrinot

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

@@ -1,8 +1,11 @@
 package io.questdb.kafka;
 
 import io.questdb.client.Sender;
+import io.questdb.cutlass.line.array.DoubleArray;
+import io.questdb.cutlass.line.array.LongArray;
 import io.questdb.std.BoolList;
 import io.questdb.std.LongList;
+import io.questdb.std.bytes.DirectByteSlice;
 
 import java.time.Instant;
 import java.time.temporal.ChronoUnit;
@@ -34,6 +37,12 @@ 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);
+    private final List<CharSequence> doubleArray2DNames = new ArrayList<>(DEFAULT_CAPACITY);
+    private final List<double[][]> doubleArray2DValues = new ArrayList<>(DEFAULT_CAPACITY);
+    private final List<CharSequence> doubleArray3DNames = new ArrayList<>(DEFAULT_CAPACITY);
+    private final List<double[][][]> doubleArray3DValues = new ArrayList<>(DEFAULT_CAPACITY);
 
     BufferingSender(Sender sender, String symbolColumns) {
         this.sender = sender;
@@ -97,6 +106,11 @@ public Sender boolColumn(CharSequence name, boolean value) {
         return this;
     }
 
+    @Override
+    public DirectByteSlice bufferView() {
+        throw new UnsupportedOperationException("not implemented");
+    }
+
     @Override
     public void cancelRow() {
         symbolColumnNames.clear();
@@ -111,6 +125,12 @@ public void cancelRow() {
         boolValues.clear();
         timestampNames.clear();
         timestampValues.clear();
+        doubleArrayNames.clear();
+        doubleArrayValues.clear();
+        doubleArray2DNames.clear();
+        doubleArray2DValues.clear();
+        doubleArray3DNames.clear();
+        doubleArray3DValues.clear();
 
         sender.cancelRow();
     }
@@ -193,6 +213,30 @@ 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();
+
+        for (int i = 0, n = doubleArray2DNames.size(); i < n; i++) {
+            CharSequence fieldName = doubleArray2DNames.get(i);
+            double[][] fieldValue = doubleArray2DValues.get(i);
+            sender.doubleArray(fieldName, fieldValue);
+        }
+        doubleArray2DNames.clear();
+        doubleArray2DValues.clear();
+
+        for (int i = 0, n = doubleArray3DNames.size(); i < n; i++) {
+            CharSequence fieldName = doubleArray3DNames.get(i);
+            double[][][] fieldValue = doubleArray3DValues.get(i);
+            sender.doubleArray(fieldName, fieldValue);
+        }
+        doubleArray3DNames.clear();
+        doubleArray3DValues.clear();
     }
 
     private static long unitToMicros(long value, ChronoUnit unit) {
@@ -230,4 +274,50 @@ public void flush() {
     public void close() {
         sender.close();
     }
+
+    @Override
+    public Sender doubleArray(CharSequence charSequence, double[] doubles) {
+        doubleArrayNames.add(charSequence);
+        doubleArrayValues.add(doubles);
+        return this;
+    }
+
+    @Override
+    public Sender doubleArray(CharSequence charSequence, double[][] doubles) {
+        doubleArray2DNames.add(charSequence);
+        doubleArray2DValues.add(doubles);
+        return this;
+    }
+
+    @Override
+    public Sender doubleArray(CharSequence charSequence, double[][][] doubles) {
+        doubleArray3DNames.add(charSequence);
+        doubleArray3DValues.add(doubles);
+        return this;
+    }
+
+    @Override
+    public Sender doubleArray(CharSequence charSequence, DoubleArray doubleArray) {
+        throw new UnsupportedOperationException("not implemented");
+    }
+
+    @Override
+    public Sender longArray(CharSequence charSequence, long[] longs) {
+        throw new UnsupportedOperationException("not implemented");
+    }
+
+    @Override
+    public Sender longArray(CharSequence charSequence, long[][] longs) {
+        throw new UnsupportedOperationException("not implemented");
+    }
+
+    @Override
+    public Sender longArray(CharSequence charSequence, long[][][] longs) {
+        throw new UnsupportedOperationException("not implemented");
+    }
+
+    @Override
+    public Sender longArray(CharSequence charSequence, LongArray longArray) {
+        throw new UnsupportedOperationException("not implemented");
+    }
 }