Skip to content

Commit b7c45af

Browse files
mtopolnikmtopolnik
committed
Clean up
1 parent 00d9c36 commit b7c45af

19 files changed

Lines changed: 178 additions & 208 deletions

‎core/src/main/java/io/questdb/client/BuildInformationHolder.java‎

Lines changed: 3 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -41,7 +41,8 @@ public BuildInformationHolder(Class<?> clazz) {
4141
String swVersion;
4242
try {
4343
final Attributes manifestAttributes = getManifestAttributes(clazz);
44-
swVersion = getAttr(manifestAttributes, "[DEVELOPMENT]");
44+
final String value = manifestAttributes.getValue("QuestDB-Client-Version");
45+
swVersion = value != null ? value : "[DEVELOPMENT]";
4546
} catch (IOException e) {
4647
swVersion = UNKNOWN;
4748
}
@@ -57,11 +58,6 @@ public String getSwVersion() {
5758
return swVersion;
5859
}
5960

60-
private static String getAttr(final Attributes manifestAttributes, String defaultValue) {
61-
final String value = manifestAttributes.getValue("QuestDB-Client-Version");
62-
return value != null ? value : defaultValue;
63-
}
64-
6561
private static Attributes getManifestAttributes(Class<?> clazz) throws IOException {
6662
InputStream is = clazz.getResourceAsStream("/META-INF/MANIFEST.MF");
6763
if (is != null) {
@@ -77,4 +73,4 @@ private static Attributes getManifestAttributes(Class<?> clazz) throws IOExcepti
7773
}
7874
return new Attributes();
7975
}
80-
}
76+
}

‎core/src/main/java/io/questdb/client/HttpClientConfiguration.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -81,4 +81,4 @@ default int getTimeout() {
8181
default int getWaitQueueCapacity() {
8282
return 4;
8383
}
84-
}
84+
}

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

Lines changed: 4 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -529,7 +529,7 @@ public ResponseHeaders send(CharSequence host, int port, int timeout) {
529529

530530
if (contentStart > -1) {
531531
assert state == STATE_CONTENT;
532-
sendHeaderAndContent(Integer.MAX_VALUE, timeout);
532+
sendHeaderAndContent(timeout);
533533
} else {
534534
eol();
535535
doSend(bufLo, ptr, timeout);
@@ -612,9 +612,7 @@ private void connect(CharSequence host, int port) {
612612
throw new HttpClientException("could not allocate a file descriptor").errno(nf.errno());
613613
}
614614
if (nf.setTcpNoDelay(fd, true) < 0) {
615-
// TODO: LOG
616-
// LOG.info().$("could not turn off Nagle's algorithm [fd=").$(fd)
617-
// .$(", errno=").$(nf.errno()).I$();
615+
LOG.info("could not turn off Nagle's algorithm [fd={}, errno={}", nf.errno(), fd);
618616
}
619617
socket.of(fd);
620618

@@ -803,7 +801,7 @@ private void putUrlEncoded(char c) {
803801
}
804802
}
805803

806-
private void sendHeaderAndContent(int maxContentLen, int timeout) {
804+
private void sendHeaderAndContent(int timeout) {
807805
final int contentLength = (int) (ptr - contentStart);
808806

809807
// Add content bytes into the header.
@@ -825,7 +823,7 @@ private void sendHeaderAndContent(int maxContentLen, int timeout) {
825823
doSend(bufLo, headerHi, timeout);
826824

827825
// Send content.
828-
doSend(contentStart, contentStart + Math.min(hi - contentStart, maxContentLen), timeout);
826+
doSend(contentStart, contentStart + Math.min(hi - contentStart, Integer.MAX_VALUE), timeout);
829827
}
830828
}
831829

‎core/src/main/java/io/questdb/client/cutlass/http/client/HttpClientLinux.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -63,4 +63,4 @@ protected void setupIoWait() {
6363
throw new HttpClientException("internal error: epoll_ctl failure [cmd=add, errno=").put(nf.errno()).put(']');
6464
}
6565
}
66-
}
66+
}

‎core/src/main/java/io/questdb/client/cutlass/http/client/HttpClientOsx.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -69,4 +69,4 @@ protected void ioWait(int timeout, int op) {
6969
protected void setupIoWait() {
7070
// no-op on OSX
7171
}
72-
}
72+
}

‎core/src/main/java/io/questdb/client/cutlass/http/client/HttpClientWindows.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -67,4 +67,4 @@ protected void ioWait(int timeout, int op) {
6767
@Override
6868
protected void setupIoWait() {
6969
}
70-
}
70+
}

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

Lines changed: 45 additions & 44 deletions
Original file line numberDiff line numberDiff line change
@@ -71,6 +71,7 @@ public abstract class WebSocketClient implements QuietCloseable {
7171
private static final int DEFAULT_RECV_BUFFER_SIZE = 65536;
7272
private static final int DEFAULT_SEND_BUFFER_SIZE = 65536;
7373
private static final Logger LOG = LoggerFactory.getLogger(WebSocketClient.class);
74+
private static final String QWP_VERSION_HEADER_NAME = "X-QWP-Version:";
7475
private static final ThreadLocal<MessageDigest> SHA1_DIGEST = ThreadLocal.withInitial(() -> {
7576
try {
7677
return MessageDigest.getInstance("SHA-1");
@@ -101,12 +102,12 @@ public abstract class WebSocketClient implements QuietCloseable {
101102
// QWP version negotiation
102103
private String qwpClientId;
103104
private int qwpMaxVersion = 1;
104-
private int serverQwpVersion = 1;
105105
// Receive buffer (native memory)
106106
private long recvBufPtr;
107107
private int recvBufSize;
108108
private int recvPos; // Write position
109109
private int recvReadPos; // Read position
110+
private int serverQwpVersion = 1;
110111
private boolean upgraded;
111112

112113
public WebSocketClient(HttpClientConfiguration configuration, SocketFactory socketFactory) {
@@ -235,13 +236,6 @@ public int getPort() {
235236
return port;
236237
}
237238

238-
/**
239-
* Returns the QWP version selected by the server during the upgrade handshake.
240-
*/
241-
public int getServerQwpVersion() {
242-
return serverQwpVersion;
243-
}
244-
245239
/**
246240
* Gets the send buffer for building WebSocket frames.
247241
* <p>
@@ -259,6 +253,13 @@ public WebSocketSendBuffer getSendBuffer() {
259253
return sendBuffer;
260254
}
261255

256+
/**
257+
* Returns the QWP version selected by the server during the upgrade handshake.
258+
*/
259+
public int getServerQwpVersion() {
260+
return serverQwpVersion;
261+
}
262+
262263
/**
263264
* Returns whether the WebSocket is connected and upgraded.
264265
*/
@@ -356,6 +357,20 @@ public void sendPing(int timeout) {
356357
controlFrameBuffer.reset();
357358
}
358359

360+
/**
361+
* Sets the QWP client identifier sent in the X-QWP-Client-Id upgrade header.
362+
*/
363+
public void setQwpClientId(String clientId) {
364+
this.qwpClientId = clientId;
365+
}
366+
367+
/**
368+
* Sets the maximum QWP version this client supports, sent in the X-QWP-Max-Version upgrade header.
369+
*/
370+
public void setQwpMaxVersion(int maxVersion) {
371+
this.qwpMaxVersion = maxVersion;
372+
}
373+
359374
/**
360375
* Non-blocking attempt to receive a WebSocket frame.
361376
* Returns immediately if no complete frame is available.
@@ -391,20 +406,6 @@ public boolean tryReceiveFrame(WebSocketFrameHandler handler) {
391406
return result != null && result;
392407
}
393408

394-
/**
395-
* Sets the QWP client identifier sent in the X-QWP-Client-Id upgrade header.
396-
*/
397-
public void setQwpClientId(String clientId) {
398-
this.qwpClientId = clientId;
399-
}
400-
401-
/**
402-
* Sets the maximum QWP version this client supports, sent in the X-QWP-Max-Version upgrade header.
403-
*/
404-
public void setQwpMaxVersion(int maxVersion) {
405-
this.qwpMaxVersion = maxVersion;
406-
}
407-
408409
/**
409410
* Performs WebSocket upgrade handshake.
410411
*
@@ -526,6 +527,27 @@ private static boolean containsHeaderValue(String response, String headerName, S
526527
return false;
527528
}
528529

530+
private static int extractQwpVersion(String response) {
531+
int headerLen = QWP_VERSION_HEADER_NAME.length();
532+
int responseLen = response.length();
533+
for (int i = 0; i <= responseLen - headerLen; i++) {
534+
if (response.regionMatches(true, i, QWP_VERSION_HEADER_NAME, 0, headerLen)) {
535+
int valueStart = i + headerLen;
536+
int lineEnd = response.indexOf('\r', valueStart);
537+
if (lineEnd < 0) {
538+
lineEnd = responseLen;
539+
}
540+
String value = response.substring(valueStart, lineEnd).trim();
541+
try {
542+
return Integer.parseInt(value);
543+
} catch (NumberFormatException e) {
544+
return 1;
545+
}
546+
}
547+
}
548+
return 1;
549+
}
550+
529551
private static int remainingTime(int timeoutMillis, long startTimeNanos) {
530552
return timeoutMillis - (int) NANOSECONDS.toMillis(System.nanoTime() - startTimeNanos);
531553
}
@@ -924,28 +946,7 @@ private void validateUpgradeResponse(int headerEnd) {
924946
}
925947

926948
// Extract X-QWP-Version (optional — defaults to 1 if absent)
927-
serverQwpVersion = extractIntHeader(response, "X-QWP-Version:", 1);
928-
}
929-
930-
private static int extractIntHeader(String response, String headerName, int defaultValue) {
931-
int headerLen = headerName.length();
932-
int responseLen = response.length();
933-
for (int i = 0; i <= responseLen - headerLen; i++) {
934-
if (response.regionMatches(true, i, headerName, 0, headerLen)) {
935-
int valueStart = i + headerLen;
936-
int lineEnd = response.indexOf('\r', valueStart);
937-
if (lineEnd < 0) {
938-
lineEnd = responseLen;
939-
}
940-
String value = response.substring(valueStart, lineEnd).trim();
941-
try {
942-
return Integer.parseInt(value);
943-
} catch (NumberFormatException e) {
944-
return defaultValue;
945-
}
946-
}
947-
}
948-
return defaultValue;
949+
serverQwpVersion = extractQwpVersion(response);
949950
}
950951

951952
protected void dieWaiting(int n) {

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

Lines changed: 1 addition & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -36,11 +36,7 @@ final class NativeSegmentList implements QuietCloseable {
3636
private long totalLength;
3737

3838
NativeSegmentList() {
39-
this(16);
40-
}
41-
42-
NativeSegmentList(int initialCapacity) {
43-
this.capacity = Math.max(initialCapacity, 4);
39+
this.capacity = Math.max(16, 4);
4440
this.ptr = Unsafe.malloc((long) capacity * ENTRY_SIZE, MemoryTag.NATIVE_DEFAULT);
4541
}
4642

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

Lines changed: 54 additions & 54 deletions
Original file line numberDiff line numberDiff line change
@@ -126,67 +126,14 @@ private void encodeColumn(
126126
}
127127
}
128128

129-
void encodeTable(QwpTableBuffer tableBuffer, boolean useSchemaRef, boolean useGlobalSymbols, boolean useGorilla) {
130-
int schemaId = tableBuffer.getSchemaId();
131-
if (schemaId < 0) {
132-
schemaId = 0;
133-
}
134-
encodeTable(tableBuffer, tableBuffer.getRowCount(), null, null, null, useSchemaRef, useGlobalSymbols, useGorilla, schemaId);
135-
}
136-
137-
void encodeTable(
138-
QwpTableBuffer tableBuffer,
139-
int rowCount,
140-
int[] limitedValueCounts,
141-
long[] limitedStringDataSizes,
142-
int[] limitedSymbolDictionarySizes,
143-
boolean useSchemaRef,
144-
boolean useGlobalSymbols,
145-
boolean useGorilla,
146-
int schemaId
147-
) {
148-
QwpColumnDef[] columnDefs = tableBuffer.getColumnDefs();
149-
150-
if (useSchemaRef) {
151-
writeTableHeaderWithSchemaRef(
152-
tableBuffer.getTableName(),
153-
rowCount,
154-
schemaId,
155-
columnDefs.length
156-
);
157-
} else {
158-
writeTableHeaderWithSchema(tableBuffer.getTableName(), rowCount, schemaId, columnDefs);
159-
}
160-
161-
for (int i = 0; i < tableBuffer.getColumnCount(); i++) {
162-
QwpTableBuffer.ColumnBuffer col = tableBuffer.getColumn(i);
163-
QwpColumnDef colDef = columnDefs[i];
164-
int valueCount = col.getValueCount();
165-
long stringDataSize = col.getStringDataSize();
166-
int symbolDictionarySize = col.getSymbolDictionarySize();
167-
168-
if (limitedValueCounts != null && limitedValueCounts[i] > -1) {
169-
valueCount = limitedValueCounts[i];
170-
stringDataSize = limitedStringDataSizes[i];
171-
symbolDictionarySize = limitedSymbolDictionarySizes[i];
172-
}
173-
174-
encodeColumn(col, colDef, rowCount, valueCount, stringDataSize, symbolDictionarySize, useGorilla, useGlobalSymbols);
175-
}
176-
}
177-
178-
void setBuffer(QwpBufferWriter buffer) {
179-
this.buffer = buffer;
180-
}
181-
182129
private void writeBooleanColumn(long addr, int count) {
183130
int packedSize = (count + 7) / 8;
184131
for (int i = 0; i < packedSize; i++) {
185132
byte b = 0;
186133
for (int bit = 0; bit < 8; bit++) {
187134
int idx = i * 8 + bit;
188135
if (idx < count && Unsafe.getUnsafe().getByte(addr + idx) != 0) {
189-
b |= (1 << bit);
136+
b |= (byte) (1 << bit);
190137
}
191138
}
192139
buffer.putByte(b);
@@ -374,4 +321,57 @@ private void writeTimestampColumn(long addr, int count, boolean useGorilla) {
374321
buffer.putBlockOfBytes(addr, (long) count * 8);
375322
}
376323
}
324+
325+
void encodeTable(QwpTableBuffer tableBuffer, boolean useSchemaRef, boolean useGlobalSymbols, boolean useGorilla) {
326+
int schemaId = tableBuffer.getSchemaId();
327+
if (schemaId < 0) {
328+
schemaId = 0;
329+
}
330+
encodeTable(tableBuffer, tableBuffer.getRowCount(), null, null, null, useSchemaRef, useGlobalSymbols, useGorilla, schemaId);
331+
}
332+
333+
void encodeTable(
334+
QwpTableBuffer tableBuffer,
335+
int rowCount,
336+
int[] limitedValueCounts,
337+
long[] limitedStringDataSizes,
338+
int[] limitedSymbolDictionarySizes,
339+
boolean useSchemaRef,
340+
boolean useGlobalSymbols,
341+
boolean useGorilla,
342+
int schemaId
343+
) {
344+
QwpColumnDef[] columnDefs = tableBuffer.getColumnDefs();
345+
346+
if (useSchemaRef) {
347+
writeTableHeaderWithSchemaRef(
348+
tableBuffer.getTableName(),
349+
rowCount,
350+
schemaId,
351+
columnDefs.length
352+
);
353+
} else {
354+
writeTableHeaderWithSchema(tableBuffer.getTableName(), rowCount, schemaId, columnDefs);
355+
}
356+
357+
for (int i = 0; i < tableBuffer.getColumnCount(); i++) {
358+
QwpTableBuffer.ColumnBuffer col = tableBuffer.getColumn(i);
359+
QwpColumnDef colDef = columnDefs[i];
360+
int valueCount = col.getValueCount();
361+
long stringDataSize = col.getStringDataSize();
362+
int symbolDictionarySize = col.getSymbolDictionarySize();
363+
364+
if (limitedValueCounts != null && limitedValueCounts[i] > -1) {
365+
valueCount = limitedValueCounts[i];
366+
stringDataSize = limitedStringDataSizes[i];
367+
symbolDictionarySize = limitedSymbolDictionarySizes[i];
368+
}
369+
370+
encodeColumn(col, colDef, rowCount, valueCount, stringDataSize, symbolDictionarySize, useGorilla, useGlobalSymbols);
371+
}
372+
}
373+
374+
void setBuffer(QwpBufferWriter buffer) {
375+
this.buffer = buffer;
376+
}
377377
}

0 commit comments

Comments
 (0)