Skip to content

Support for QuestDB arrays #33

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 9 commits into from
Aug 4, 2025
Merged

Support for QuestDB arrays #33

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
ea07d9d
Update QuestDB client
jerrinot Jul 28, 2025
08f432a
buffered client stubs
jerrinot Jul 28, 2025
a60ec9f
adjust for new cancelRow() behaviour
jerrinot Jul 28, 2025
d16cd05
1D arrays supported
jerrinot Jul 29, 2025
82da49a
support for symbols
jerrinot Jul 29, 2025
e719d7e
adjust auth test for conf strings
jerrinot Jul 29, 2025
9c0a99a
support for 2D and 3D arrays
jerrinot Aug 4, 2025
491c40c
support for 2D and 3D array in BufferedSender
jerrinot Aug 4, 2025
1533491
regression fixed
jerrinot Aug 4, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
90 changes: 90 additions & 0 deletions connector/src/main/java/io/questdb/kafka/BufferingSender.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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;
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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();
Expand All @@ -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();
}
Expand Down Expand Up @@ -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) {
Expand Down Expand Up @@ -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");
}
}
Loading