Skip to content

Commit 06b6ef7

Browse files
glasstigerglasstiger
committed
address review comments
1 parent f0a9fa7 commit 06b6ef7

3 files changed

Lines changed: 90 additions & 26 deletions

File tree

‎core/src/main/java/io/questdb/client/cutlass/qwp/client/WebSocketResponse.java‎

Lines changed: 29 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,7 @@
2727
import io.questdb.client.std.LongList;
2828
import io.questdb.client.std.ObjList;
2929
import io.questdb.client.std.Unsafe;
30+
import org.jetbrains.annotations.TestOnly;
3031

3132
import java.nio.charset.StandardCharsets;
3233

@@ -37,15 +38,15 @@
3738
* <pre>
3839
* +--------+----------+------------+--------------------------------------+
3940
* | status | sequence | tableCount | table entries |
40-
* | 1 byte | 8 bytes | 2 bytes | [nameLen(1)+name(N)+seqTxn(8)] * cnt |
41+
* | 1 byte | 8 bytes | 2 bytes | [nameLen(2)+name(N)+seqTxn(8)] * cnt |
4142
* +--------+----------+------------+--------------------------------------+
4243
* </pre>
4344
* <p>
4445
* STATUS_DURABLE_ACK response format:
4546
* <pre>
4647
* +--------+------------+--------------------------------------+
4748
* | status | tableCount | table entries |
48-
* | 1 byte | 2 bytes | [nameLen(1)+name(N)+seqTxn(8)] * cnt |
49+
* | 1 byte | 2 bytes | [nameLen(2)+name(N)+seqTxn(8)] * cnt |
4950
* +--------+------------+--------------------------------------+
5051
* </pre>
5152
* <p>
@@ -92,6 +93,7 @@ public WebSocketResponse() {
9293
/**
9394
* Creates a durable-upload ACK response with a single table entry.
9495
*/
96+
@TestOnly
9597
public static WebSocketResponse durableAck(String tableName, long seqTxn) {
9698
WebSocketResponse response = new WebSocketResponse();
9799
response.status = STATUS_DURABLE_ACK;
@@ -104,6 +106,7 @@ public static WebSocketResponse durableAck(String tableName, long seqTxn) {
104106
/**
105107
* Creates an error response.
106108
*/
109+
@TestOnly
107110
public static WebSocketResponse error(long sequence, byte status, String errorMessage) {
108111
WebSocketResponse response = new WebSocketResponse();
109112
response.status = status;
@@ -150,6 +153,7 @@ public static boolean isStructurallyValid(long ptr, int length) {
150153
/**
151154
* Creates a success response.
152155
*/
156+
@TestOnly
153157
public static WebSocketResponse success(long sequence) {
154158
WebSocketResponse response = new WebSocketResponse();
155159
response.status = STATUS_OK;
@@ -265,23 +269,22 @@ public boolean readFrom(long ptr, int length) {
265269
}
266270

267271
// Error response
268-
if (length < MIN_OK_RESPONSE_SIZE) {
272+
if (length < MIN_ERROR_RESPONSE_SIZE) {
269273
return false;
270274
}
271275
sequence = Unsafe.getUnsafe().getLong(ptr + 1);
272276
int offset = 9;
273-
if (length >= offset + 2) {
274-
int msgLen = Unsafe.getUnsafe().getShort(ptr + offset) & 0xFFFF;
275-
offset += 2;
276-
if (length >= offset + msgLen && msgLen > 0) {
277-
byte[] msgBytes = new byte[msgLen];
278-
for (int i = 0; i < msgLen; i++) {
279-
msgBytes[i] = Unsafe.getUnsafe().getByte(ptr + offset + i);
280-
}
281-
errorMessage = new String(msgBytes, StandardCharsets.UTF_8);
282-
} else {
283-
errorMessage = null;
277+
int msgLen = Unsafe.getUnsafe().getShort(ptr + offset) & 0xFFFF;
278+
offset += 2;
279+
if (length < offset + msgLen) {
280+
return false;
281+
}
282+
if (msgLen > 0) {
283+
byte[] msgBytes = new byte[msgLen];
284+
for (int i = 0; i < msgLen; i++) {
285+
msgBytes[i] = Unsafe.getUnsafe().getByte(ptr + offset + i);
284286
}
287+
errorMessage = new String(msgBytes, StandardCharsets.UTF_8);
285288
} else {
286289
errorMessage = null;
287290
}
@@ -291,6 +294,7 @@ public boolean readFrom(long ptr, int length) {
291294
/**
292295
* Calculates the serialized size of this response.
293296
*/
297+
@TestOnly
294298
public int serializedSize() {
295299
if (status == STATUS_OK) {
296300
return MIN_OK_RESPONSE_SIZE + tableEntriesSize();
@@ -326,6 +330,7 @@ public String toString() {
326330
* @param ptr destination address
327331
* @return number of bytes written
328332
*/
333+
@TestOnly
329334
public int writeTo(long ptr) {
330335
int offset = 0;
331336

@@ -362,11 +367,11 @@ private boolean readTableEntries(long ptr, int remaining) {
362367
int tableCount = Unsafe.getUnsafe().getShort(ptr) & 0xFFFF;
363368
int offset = 2;
364369
for (int i = 0; i < tableCount; i++) {
365-
if (remaining < offset + 1) {
370+
if (remaining < offset + 2) {
366371
return false;
367372
}
368-
int nameLen = Unsafe.getUnsafe().getByte(ptr + offset) & 0xFF;
369-
offset += 1;
373+
int nameLen = Unsafe.getUnsafe().getShort(ptr + offset) & 0xFFFF;
374+
offset += 2;
370375
if (remaining < offset + nameLen + 8) {
371376
return false;
372377
}
@@ -380,13 +385,13 @@ private boolean readTableEntries(long ptr, int remaining) {
380385
tableNames.add(new String(nameBytes, StandardCharsets.UTF_8));
381386
tableSeqTxns.add(seqTxn);
382387
}
383-
return true;
388+
return remaining == offset;
384389
}
385390

386391
private int tableEntriesSize() {
387392
int size = 0;
388393
for (int i = 0, n = tableNames.size(); i < n; i++) {
389-
size += 1 + tableNames.getQuick(i).getBytes(StandardCharsets.UTF_8).length + 8;
394+
size += 2 + tableNames.getQuick(i).getBytes(StandardCharsets.UTF_8).length + 8;
390395
}
391396
return size;
392397
}
@@ -398,11 +403,11 @@ private static boolean validateTableEntries(long ptr, int remaining) {
398403
int tableCount = Unsafe.getUnsafe().getShort(ptr) & 0xFFFF;
399404
int offset = 2;
400405
for (int i = 0; i < tableCount; i++) {
401-
if (remaining < offset + 1) {
406+
if (remaining < offset + 2) {
402407
return false;
403408
}
404-
int nameLen = Unsafe.getUnsafe().getByte(ptr + offset) & 0xFF;
405-
offset += 1;
409+
int nameLen = Unsafe.getUnsafe().getShort(ptr + offset) & 0xFFFF;
410+
offset += 2;
406411
if (remaining < offset + nameLen + 8) {
407412
return false;
408413
}
@@ -418,8 +423,8 @@ private int writeTableEntries(long ptr) {
418423
offset += 2;
419424
for (int i = 0; i < count; i++) {
420425
byte[] nameBytes = tableNames.getQuick(i).getBytes(StandardCharsets.UTF_8);
421-
Unsafe.getUnsafe().putByte(ptr + offset, (byte) nameBytes.length);
422-
offset += 1;
426+
Unsafe.getUnsafe().putShort(ptr + offset, (short) nameBytes.length);
427+
offset += 2;
423428
for (int j = 0; j < nameBytes.length; j++) {
424429
Unsafe.getUnsafe().putByte(ptr + offset + j, nameBytes[j]);
425430
}

‎core/src/main/java/io/questdb/client/cutlass/qwp/client/WebSocketSendQueue.java‎

Lines changed: 7 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -469,7 +469,7 @@ private void ioLoop() {
469469
client.sendPing(1000);
470470
tryReceiveAcks();
471471
} catch (Exception e) {
472-
LOG.error("Ping failed: {}", e.getMessage());
472+
failTransport(new LineSenderException("Ping failed", e));
473473
} finally {
474474
synchronized (processingLock) {
475475
pingComplete = true;
@@ -774,9 +774,14 @@ void advance(long newValue) {
774774
}
775775

776776
private static void advanceSeqTxn(ConcurrentHashMap<String, SeqTxn> map, String tableName, long seqTxn) {
777-
SeqTxn existing = map.putIfAbsent(tableName, new SeqTxn(seqTxn));
777+
SeqTxn existing = map.get(tableName);
778778
if (existing != null) {
779779
existing.advance(seqTxn);
780+
} else {
781+
existing = map.putIfAbsent(tableName, new SeqTxn(seqTxn));
782+
if (existing != null) {
783+
existing.advance(seqTxn);
784+
}
780785
}
781786
}
782787
}

‎core/src/test/java/io/questdb/client/test/cutlass/qwp/client/WebSocketResponseTest.java‎

Lines changed: 54 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -149,6 +149,60 @@ public void testSuccessRoundTripUnchanged() throws Exception {
149149
});
150150
}
151151

152+
@Test
153+
public void testSuccessWithTableEntriesRoundTrip() throws Exception {
154+
assertMemoryLeak(() -> {
155+
// Build a STATUS_OK frame with 2 table entries directly in native memory
156+
// Format: status(1) + sequence(8) + tableCount(2) + entries
157+
byte[] name1 = "trades".getBytes(java.nio.charset.StandardCharsets.UTF_8);
158+
byte[] name2 = "orders".getBytes(java.nio.charset.StandardCharsets.UTF_8);
159+
int size = 1 + 8 + 2 + (2 + name1.length + 8) + (2 + name2.length + 8);
160+
long ptr = Unsafe.malloc(size, MemoryTag.NATIVE_DEFAULT);
161+
try {
162+
int offset = 0;
163+
Unsafe.getUnsafe().putByte(ptr + offset, WebSocketResponse.STATUS_OK);
164+
offset += 1;
165+
Unsafe.getUnsafe().putLong(ptr + offset, 42L);
166+
offset += 8;
167+
Unsafe.getUnsafe().putShort(ptr + offset, (short) 2);
168+
offset += 2;
169+
// entry 1: trades, seqTxn=10
170+
Unsafe.getUnsafe().putShort(ptr + offset, (short) name1.length);
171+
offset += 2;
172+
for (int i = 0; i < name1.length; i++) {
173+
Unsafe.getUnsafe().putByte(ptr + offset + i, name1[i]);
174+
}
175+
offset += name1.length;
176+
Unsafe.getUnsafe().putLong(ptr + offset, 10L);
177+
offset += 8;
178+
// entry 2: orders, seqTxn=20
179+
Unsafe.getUnsafe().putShort(ptr + offset, (short) name2.length);
180+
offset += 2;
181+
for (int i = 0; i < name2.length; i++) {
182+
Unsafe.getUnsafe().putByte(ptr + offset + i, name2[i]);
183+
}
184+
offset += name2.length;
185+
Unsafe.getUnsafe().putLong(ptr + offset, 20L);
186+
187+
Assert.assertTrue(WebSocketResponse.isStructurallyValid(ptr, size));
188+
189+
WebSocketResponse parsed = new WebSocketResponse();
190+
Assert.assertTrue(parsed.readFrom(ptr, size));
191+
Assert.assertTrue(parsed.isSuccess());
192+
Assert.assertFalse(parsed.isDurableAck());
193+
Assert.assertEquals(42L, parsed.getSequence());
194+
Assert.assertEquals(2, parsed.getTableEntryCount());
195+
Assert.assertEquals("trades", parsed.getTableName(0));
196+
Assert.assertEquals(10L, parsed.getTableSeqTxn(0));
197+
Assert.assertEquals("orders", parsed.getTableName(1));
198+
Assert.assertEquals(20L, parsed.getTableSeqTxn(1));
199+
Assert.assertNull(parsed.getErrorMessage());
200+
} finally {
201+
Unsafe.free(ptr, size, MemoryTag.NATIVE_DEFAULT);
202+
}
203+
});
204+
}
205+
152206
@Test
153207
public void testErrorRoundTrip() throws Exception {
154208
assertMemoryLeak(() -> {

0 commit comments

Comments
 (0)