more tests

Author: jerrinot

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

@@ -9,26 +9,28 @@
 import java.util.List;
 import java.util.Set;
 
+/**
+ * Allow to add regular fields before adding symbols.
+ * <p>
+ * Internally it buffers symbols and fields and sends them on calling <code>atNow()</code> or <code>at()</code>.
+ *
+ */
 final class BufferingSender implements Sender {
-    private final Sender sender;
-
-    private List<CharSequence> timestampNames = new ArrayList<>(4);
-    private LongList timestampValues = new LongList(4);
-
-    private List<CharSequence> longNames = new ArrayList<>(4);
-    private LongList longValues = new LongList(4);
-
-    private List<CharSequence> doubleNames = new ArrayList<>(4);
-    private DoubleList doubleValues = new DoubleList();
+    private static final int DEFAULT_CAPACITY = 4;
 
-    private List<CharSequence> boolNames = new ArrayList<>(4);
-    private BoolList boolValues = new BoolList(4);
-
-    private List<CharSequence> stringNames = new ArrayList<>(4);
-    private List<CharSequence> stringValues = new ArrayList<>(4);
-
-    private List<CharSequence> symbolColumnNames = new ArrayList<>(4);
-    private List<CharSequence> symbolColumnValues = new ArrayList<>(4);
+    private final Sender sender;
+    private final List<CharSequence> timestampNames = new ArrayList<>(DEFAULT_CAPACITY);
+    private final LongList timestampValues = new LongList(DEFAULT_CAPACITY);
+    private final List<CharSequence> longNames = new ArrayList<>(DEFAULT_CAPACITY);
+    private final LongList longValues = new LongList(DEFAULT_CAPACITY);
+    private final List<CharSequence> doubleNames = new ArrayList<>(DEFAULT_CAPACITY);
+    private final DoubleList doubleValues = new DoubleList(DEFAULT_CAPACITY);
+    private final List<CharSequence> boolNames = new ArrayList<>(DEFAULT_CAPACITY);
+    private final BoolList boolValues = new BoolList(DEFAULT_CAPACITY);
+    private final List<CharSequence> stringNames = new ArrayList<>(DEFAULT_CAPACITY);
+    private final List<CharSequence> stringValues = new ArrayList<>(DEFAULT_CAPACITY);
+    private final List<CharSequence> symbolColumnNames = new ArrayList<>(DEFAULT_CAPACITY);
+    private final List<CharSequence> symbolColumnValues = new ArrayList<>(DEFAULT_CAPACITY);
 
 
     private Set<CharSequence> symbolColumns = new HashSet<>();

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

@@ -1,30 +1,30 @@
 package io.questdb.kafka;
 
-public class DoubleList {
+final class DoubleList {
     private double[] buffer;
     private int pos = 0;
 
-    public DoubleList() {
-        buffer = new double[4];
+    DoubleList(int initialCapacity) {
+        buffer = new double[initialCapacity];
     }
 
-    public void add(double value) {
+    void add(double value) {
         ensureCapacity0(pos + 1);
         buffer[pos++] = value;
     }
 
-    public double get(int index) {
+    double get(int index) {
         if (index >= pos) {
             throw new ArrayIndexOutOfBoundsException(index);
         }
         return buffer[index];
     }
 
-    public void clear() {
+    void clear() {
         pos = 0;
     }
 
-    private void ensureCapacity0(int i) {
+    void ensureCapacity0(int i) {
         if (i > buffer.length) {
             double[] newBuffer = new double[buffer.length << 1];
             System.arraycopy(buffer, 0, newBuffer, 0, pos);

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

@@ -139,6 +139,87 @@ public void testSymbol() {
                 "select firstname,lastname,age from " + topicName);
     }
 
+    @Test
+    public void testSymbol_withAllOtherILPTypes() {
+        connect.kafka().createTopic(topicName, 1);
+        Map<String, String> props = baseConnectorProps(topicName);
+        props.put(QuestDBSinkConnectorConfig.SYMBOL_COLUMNS_CONFIG, "firstname");
+        connect.configureConnector(CONNECTOR_NAME, props);
+        assertConnectorTaskRunningEventually();
+        Schema schema = SchemaBuilder.struct().name("com.example.Person")
+                .field("firstname", Schema.STRING_SCHEMA)
+                .field("lastname", Schema.STRING_SCHEMA)
+                .field("age", Schema.INT8_SCHEMA)
+                .field("vegan", Schema.BOOLEAN_SCHEMA)
+                .field("height", Schema.FLOAT64_SCHEMA)
+                .field("birth", Timestamp.SCHEMA)
+                .build();
+
+        java.util.Date birth = new Calendar.Builder()
+                .setTimeZone(TimeZone.getTimeZone("UTC"))
+                .setDate(2022, 9, 23) // note: month is 0-based
+                .setTimeOfDay(13, 53, 59, 123)
+                .build().getTime();
+        Struct p1 = new Struct(schema)
+                .put("firstname", "John")
+                .put("lastname", "Doe")
+                .put("age", (byte) 42)
+                .put("vegan", true)
+                .put("height", 1.80)
+                .put("birth", birth);
+
+        birth = new Calendar.Builder()
+                .setTimeZone(TimeZone.getTimeZone("UTC"))
+                .setDate(2021, 9, 23) // note: month is 0-based
+                .setTimeOfDay(13, 53, 59, 123)
+                .build().getTime();
+        Struct p2 = new Struct(schema)
+                .put("firstname", "Jane")
+                .put("lastname", "Doe")
+                .put("age", (byte) 41)
+                .put("vegan", false)
+                .put("height", 1.60)
+                .put("birth", birth);
+
+        connect.kafka().produce(topicName, "p1", new String(converter.fromConnectData(topicName, schema, p1)));
+        connect.kafka().produce(topicName, "p2", new String(converter.fromConnectData(topicName, schema, p2)));
+
+        QuestDBUtils.assertSqlEventually(questDBContainer, "\"firstname\",\"lastname\",\"age\",\"vegan\",\"height\",\"birth\"\r\n"
+                        + "\"John\",\"Doe\",42,true,1.8,\"2022-10-23T13:53:59.123000Z\"\r\n"
+                        + "\"Jane\",\"Doe\",41,false,1.6,\"2021-10-23T13:53:59.123000Z\"\r\n",
+                "select firstname,lastname,age,vegan,height,birth from " + topicName);
+    }
+
+    @Test
+    public void testUpfrontTable_withSymbols() {
+        connect.kafka().createTopic(topicName, 1);
+        Map<String, String> props = baseConnectorProps(topicName);
+        props.put(QuestDBSinkConnectorConfig.SYMBOL_COLUMNS_CONFIG, "firstname,lastname");
+        connect.configureConnector(CONNECTOR_NAME, props);
+        assertConnectorTaskRunningEventually();
+        Schema schema = SchemaBuilder.struct().name("com.example.Person")
+                .field("firstname", Schema.STRING_SCHEMA)
+                .field("lastname", Schema.STRING_SCHEMA)
+                .field("age", Schema.INT8_SCHEMA)
+                .build();
+
+        Struct struct = new Struct(schema)
+                .put("firstname", "John")
+                .put("lastname", "Doe")
+                .put("age", (byte) 42);
+
+        QuestDBUtils.assertSql(questDBContainer,
+                "{\"ddl\":\"OK\"}\n",
+                "create table " + topicName + " (firstname symbol, lastname symbol, age int)",
+                QuestDBUtils.Endpoint.EXEC);
+
+        connect.kafka().produce(topicName, "key", new String(converter.fromConnectData(topicName, schema, struct)));
+
+        QuestDBUtils.assertSqlEventually(questDBContainer, "\"firstname\",\"lastname\",\"age\",\"key\"\r\n"
+                        + "\"John\",\"Doe\",42,\"key\"\r\n",
+                "select * from " + topicName);
+    }
+
     @Test
     public void testUpfrontTable() {
         connect.kafka().createTopic(topicName, 1);