Skip to content

Commit e805863

Browse files
committed
bugfixes
1 parent 1e16ba4 commit e805863

11 files changed

Lines changed: 301 additions & 22 deletions

File tree

core/src/main/java/io/questdb/client/Sender.java

Lines changed: 15 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1823,10 +1823,11 @@ private LineSenderBuilder fromConfig(CharSequence configurationString) {
18231823
}
18241824
} else {
18251825
if (Chars.equalsIgnoreCase("off", sink)) {
1826-
throw new LineSenderException("WebSocket transport must have auto_flush_bytes enabled");
1826+
autoFlushBytes(0);
1827+
} else {
1828+
int autoFlushBytes = parseIntValue(sink, "auto_flush_bytes");
1829+
autoFlushBytes(autoFlushBytes);
18271830
}
1828-
int autoFlushBytes = parseIntValue(sink, "auto_flush_bytes");
1829-
autoFlushBytes(autoFlushBytes);
18301831
}
18311832
autoFlushBytesSet = true;
18321833
} else if (Chars.equals("auto_flush", sink)) {
@@ -2162,6 +2163,17 @@ public LineSenderBuilder authToken(String token) {
21622163
LineSenderBuilder.this.shouldDestroyPrivKey = true;
21632164
return LineSenderBuilder.this;
21642165
}
2166+
2167+
/**
2168+
* Authenticate by using a {@link PrivateKey} directly.
2169+
*
2170+
* @param privateKey authentication private key
2171+
* @return an instance of LineSenderBuilder for further configuration
2172+
*/
2173+
public LineSenderBuilder privateKey(PrivateKey privateKey) {
2174+
LineSenderBuilder.this.privateKey = privateKey;
2175+
return LineSenderBuilder.this;
2176+
}
21652177
}
21662178
}
21672179
}

core/src/main/java/io/questdb/client/cutlass/http/client/HttpClient.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -606,7 +606,7 @@ private void connect(CharSequence host, int port) {
606606
throw new HttpClientException("could not allocate a file descriptor").errno(nf.errno());
607607
}
608608
if (nf.setTcpNoDelay(fd, true) < 0) {
609-
LOG.info("could not turn off Nagle's algorithm [fd={}, errno={}", nf.errno(), fd);
609+
LOG.info("could not turn off Nagle's algorithm [fd={}, errno={}]", fd, nf.errno());
610610
}
611611
socket.of(fd);
612612

core/src/main/java/io/questdb/client/cutlass/http/client/WebSocketClient.java

Lines changed: 9 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -92,7 +92,8 @@ public abstract class WebSocketClient implements QuietCloseable {
9292
private final int maxRecvBufSize;
9393
private final SecureRnd rnd;
9494
private final WebSocketSendBuffer sendBuffer;
95-
private boolean closed;
95+
// volatile: written by user thread in close(), read by I/O thread in checkConnected()/sendFrame()/receiveFrame()
96+
private volatile boolean closed;
9697
private int fragmentBufPos;
9798
private long fragmentBufPtr; // native buffer for accumulating fragment payloads
9899
private int fragmentBufSize;
@@ -368,8 +369,11 @@ public void sendPing(int timeout) {
368369
checkConnected();
369370
controlFrameBuffer.reset();
370371
WebSocketSendBuffer.FrameInfo frame = controlFrameBuffer.writePingFrame();
371-
doSend(controlFrameBuffer.getBufferPtr() + frame.offset, frame.length, timeout);
372-
controlFrameBuffer.reset();
372+
try {
373+
doSend(controlFrameBuffer.getBufferPtr() + frame.offset, frame.length, timeout);
374+
} finally {
375+
controlFrameBuffer.reset();
376+
}
373377
}
374378

375379
/**
@@ -557,7 +561,7 @@ private void appendToFragmentBuffer(long payloadPtr, int payloadLen) {
557561
if (payloadLen == 0) {
558562
return;
559563
}
560-
int required = fragmentBufPos + payloadLen;
564+
long required = (long) fragmentBufPos + payloadLen;
561565
if (required > maxRecvBufSize) {
562566
throw new HttpClientException("WebSocket fragment buffer size exceeded maximum [required=")
563567
.put(required)
@@ -566,7 +570,7 @@ private void appendToFragmentBuffer(long payloadPtr, int payloadLen) {
566570
.put(']');
567571
}
568572
if (fragmentBufPtr == 0) {
569-
fragmentBufSize = Math.max(required, DEFAULT_RECV_BUFFER_SIZE);
573+
fragmentBufSize = (int) Math.max(required, DEFAULT_RECV_BUFFER_SIZE);
570574
fragmentBufPtr = Unsafe.malloc(fragmentBufSize, MemoryTag.NATIVE_DEFAULT);
571575
} else if (required > fragmentBufSize) {
572576
int newSize = (int) Math.min(Math.max((long) fragmentBufSize * 2, required), maxRecvBufSize);

core/src/main/java/io/questdb/client/cutlass/qwp/client/QwpWebSocketEncoder.java

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -41,7 +41,6 @@ public class QwpWebSocketEncoder implements QuietCloseable {
4141
private NativeBufferWriter buffer;
4242
private byte flags;
4343
private int payloadStart;
44-
private byte savedFlags;
4544
private byte version = VERSION_1;
4645

4746
public QwpWebSocketEncoder() {
@@ -67,9 +66,11 @@ public void beginMessage(
6766
buffer.reset();
6867
int deltaStart = confirmedMaxId + 1;
6968
int deltaCount = Math.max(0, batchMaxId - confirmedMaxId);
70-
savedFlags = flags;
71-
flags |= FLAG_DELTA_SYMBOL_DICT;
69+
byte headerFlags = (byte) (flags | FLAG_DELTA_SYMBOL_DICT);
70+
byte origFlags = flags;
71+
flags = headerFlags;
7272
writeHeader(tableCount, 0);
73+
flags = origFlags;
7374
payloadStart = buffer.getPosition();
7475
buffer.putVarint(deltaStart);
7576
buffer.putVarint(deltaCount);
@@ -114,7 +115,6 @@ public int encodeWithDeltaDict(
114115
public int finishMessage() {
115116
int payloadLength = buffer.getPosition() - payloadStart;
116117
buffer.patchInt(8, payloadLength);
117-
flags = savedFlags;
118118
return buffer.getPosition();
119119
}
120120

core/src/main/java/io/questdb/client/cutlass/qwp/protocol/QwpTableBuffer.java

Lines changed: 12 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -37,7 +37,7 @@
3737
import io.questdb.client.std.Decimal256;
3838
import io.questdb.client.std.Decimal64;
3939
import io.questdb.client.std.Decimals;
40-
import io.questdb.client.std.LowerCaseAsciiCharSequenceIntHashMap;
40+
import io.questdb.client.std.LowerCaseCharSequenceIntHashMap;
4141
import io.questdb.client.std.MemoryTag;
4242
import io.questdb.client.std.NumericException;
4343
import io.questdb.client.std.ObjList;
@@ -61,7 +61,7 @@
6161
public class QwpTableBuffer implements QuietCloseable {
6262

6363
private static final int MAX_COLUMN_NAME_LENGTH = 127;
64-
private final LowerCaseAsciiCharSequenceIntHashMap columnNameToIndex;
64+
private final LowerCaseCharSequenceIntHashMap columnNameToIndex;
6565
private final ObjList<ColumnBuffer> columns;
6666
private final QwpWebSocketSender sender;
6767
private final String tableName;
@@ -87,7 +87,7 @@ public QwpTableBuffer(String tableName, QwpWebSocketSender sender) {
8787
this.tableName = tableName;
8888
this.sender = sender;
8989
this.columns = new ObjList<>();
90-
this.columnNameToIndex = new LowerCaseAsciiCharSequenceIntHashMap();
90+
this.columnNameToIndex = new LowerCaseCharSequenceIntHashMap();
9191
this.rowCount = 0;
9292
this.schemaId = -1;
9393
this.columnDefsCacheValid = false;
@@ -1039,7 +1039,7 @@ public void addNull() {
10391039
break;
10401040
case TYPE_STRING:
10411041
case TYPE_VARCHAR:
1042-
stringOffsets.putInt((int) stringData.getAppendOffset());
1042+
stringOffsets.putInt(checkedStringOffset(stringData.getAppendOffset()));
10431043
break;
10441044
case TYPE_SYMBOL:
10451045
dataBuffer.putInt(-1);
@@ -1093,7 +1093,7 @@ public void addString(CharSequence value) {
10931093
if (value != null) {
10941094
stringData.putUtf8(value);
10951095
}
1096-
stringOffsets.putInt((int) stringData.getAppendOffset());
1096+
stringOffsets.putInt(checkedStringOffset(stringData.getAppendOffset()));
10971097
valueCount++;
10981098
}
10991099
size++;
@@ -1737,6 +1737,13 @@ private int getOrAddLocalSymbol(CharSequence value) {
17371737
return idx;
17381738
}
17391739

1740+
private static int checkedStringOffset(long offset) {
1741+
if (offset > Integer.MAX_VALUE) {
1742+
throw new LineSenderException("string column data exceeds 2 GiB per batch, flush more frequently");
1743+
}
1744+
return (int) offset;
1745+
}
1746+
17401747
private void markNull(int index) {
17411748
long longAddr = nullBufPtr + ((long) (index >>> 6)) * 8;
17421749
int bitIndex = index & 63;

core/src/main/java/io/questdb/client/std/AbstractLowerCaseAsciiCharSequenceHashSet.java

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -54,13 +54,13 @@ public void clear() {
5454
}
5555

5656
public int keyIndex(CharSequence key) {
57-
int index = Chars.lowerCaseHashCode(key) & mask;
57+
int index = Chars.lowerCaseAsciiHashCode(key) & mask;
5858

5959
if (keys[index] == noEntryKey) {
6060
return index;
6161
}
6262

63-
if (Chars.equalsIgnoreCase(key, keys[index])) {
63+
if (Chars.equalsLowerCaseAscii(key, keys[index])) {
6464
return -index - 1;
6565
}
6666

@@ -77,7 +77,7 @@ private int probe(CharSequence key, int index) {
7777
if (keys[index] == noEntryKey) {
7878
return index;
7979
}
80-
if (Chars.equalsIgnoreCase(key, keys[index])) {
80+
if (Chars.equalsLowerCaseAscii(key, keys[index])) {
8181
return -index - 1;
8282
}
8383
} while (true);
Lines changed: 84 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,84 @@
1+
/*+*****************************************************************************
2+
* ___ _ ____ ____
3+
* / _ \ _ _ ___ ___| |_| _ \| __ )
4+
* | | | | | | |/ _ \/ __| __| | | | _ \
5+
* | |_| | |_| | __/\__ \ |_| |_| | |_) |
6+
* \__\_\\__,_|\___||___/\__|____/|____/
7+
*
8+
* Copyright (c) 2014-2019 Appsicle
9+
* Copyright (c) 2019-2026 QuestDB
10+
*
11+
* Licensed under the Apache License, Version 2.0 (the "License");
12+
* you may not use this file except in compliance with the License.
13+
* You may obtain a copy of the License at
14+
*
15+
* http://www.apache.org/licenses/LICENSE-2.0
16+
*
17+
* Unless required by applicable law or agreed to in writing, software
18+
* distributed under the License is distributed on an "AS IS" BASIS,
19+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
20+
* See the License for the specific language governing permissions and
21+
* limitations under the License.
22+
*
23+
******************************************************************************/
24+
25+
package io.questdb.client.std;
26+
27+
import java.util.Arrays;
28+
29+
public abstract class AbstractLowerCaseCharSequenceHashSet implements Mutable {
30+
protected static final int MIN_INITIAL_CAPACITY = 16;
31+
protected static final CharSequence noEntryKey = null;
32+
protected final double loadFactor;
33+
protected int capacity;
34+
protected int free;
35+
protected CharSequence[] keys;
36+
protected int mask;
37+
38+
public AbstractLowerCaseCharSequenceHashSet(int initialCapacity, double loadFactor) {
39+
if (loadFactor <= 0d || loadFactor >= 1d) {
40+
throw new IllegalArgumentException("0 < loadFactor < 1");
41+
}
42+
43+
free = this.capacity = Math.max(initialCapacity, MIN_INITIAL_CAPACITY);
44+
this.loadFactor = loadFactor;
45+
keys = new CharSequence[Numbers.ceilPow2((int) (this.capacity / loadFactor))];
46+
mask = keys.length - 1;
47+
}
48+
49+
@Override
50+
public void clear() {
51+
Arrays.fill(keys, noEntryKey);
52+
free = capacity;
53+
}
54+
55+
public int keyIndex(CharSequence key) {
56+
int index = Chars.lowerCaseHashCode(key) & mask;
57+
58+
if (keys[index] == noEntryKey) {
59+
return index;
60+
}
61+
62+
if (Chars.equalsIgnoreCase(key, keys[index])) {
63+
return -index - 1;
64+
}
65+
66+
return probe(key, index);
67+
}
68+
69+
public int size() {
70+
return capacity - free;
71+
}
72+
73+
private int probe(CharSequence key, int index) {
74+
do {
75+
index = (index + 1) & mask;
76+
if (keys[index] == noEntryKey) {
77+
return index;
78+
}
79+
if (Chars.equalsIgnoreCase(key, keys[index])) {
80+
return -index - 1;
81+
}
82+
} while (true);
83+
}
84+
}

core/src/main/java/io/questdb/client/std/Chars.java

Lines changed: 29 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -233,6 +233,19 @@ public static boolean isBlank(CharSequence s) {
233233
return true;
234234
}
235235

236+
public static int lowerCaseAsciiHashCode(CharSequence value) {
237+
int len = value.length();
238+
if (len == 0) {
239+
return 0;
240+
}
241+
242+
int h = 0;
243+
for (int p = 0; p < len; p++) {
244+
h = 31 * h + toLowerCaseAscii(value.charAt(p));
245+
}
246+
return h;
247+
}
248+
236249
public static int lowerCaseHashCode(CharSequence value) {
237250
int len = value.length();
238251
if (len == 0) {
@@ -266,6 +279,22 @@ public static String toLowerCase(@Nullable CharSequence value) {
266279
return b.toString();
267280
}
268281

282+
public static String toLowerCaseAscii(@Nullable CharSequence value) {
283+
if (value == null) {
284+
return null;
285+
}
286+
final int len = value.length();
287+
if (len == 0) {
288+
return "";
289+
}
290+
291+
final Utf16Sink b = Misc.getThreadLocalSink();
292+
for (int i = 0; i < len; i++) {
293+
b.put(toLowerCaseAscii(value.charAt(i)));
294+
}
295+
return b.toString();
296+
}
297+
269298
public static char toLowerCaseAscii(char character) {
270299
return character > 64 && character < 91 ? (char) (character + 32) : character;
271300
}

core/src/main/java/io/questdb/client/std/LowerCaseAsciiCharSequenceIntHashMap.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -65,7 +65,7 @@ public boolean putAt(int index, CharSequence key, int value) {
6565
values[-index - 1] = value;
6666
return false;
6767
}
68-
putAt0(index, Chars.toLowerCase(key), value);
68+
putAt0(index, Chars.toLowerCaseAscii(key), value);
6969
return true;
7070
}
7171

0 commit comments

Comments
 (0)