Skip to content

Commit 6c336d1

Browse files
kafka1991kafka1991
authored
docs(qwp): make the pooled facade first-class and add more ws ingest/query examples (#61)
1 parent f0b8b21 commit 6c336d1

42 files changed

Lines changed: 2190 additions & 890 deletions

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎examples.manifest.yaml‎

Lines changed: 20 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,21 @@
1+
- name: qwp-pool
2+
lang: java
3+
path: examples/src/main/java/com/example/QuestDBExample.java
4+
header: |-
5+
QuestDB Java client [docs](https://questdb.io/docs/connect/clients/java/)
6+
and [Maven artifact](https://central.sonatype.com/artifact/org.questdb/questdb-client).
7+
- name: qwp-ingest
8+
lang: java
9+
path: examples/src/main/java/com/example/sender/WsExample.java
10+
header: |-
11+
QuestDB Java client [docs](https://questdb.io/docs/connect/clients/java/)
12+
and [Maven artifact](https://central.sonatype.com/artifact/org.questdb/questdb-client).
13+
- name: qwp-query
14+
lang: java
15+
path: examples/src/main/java/com/example/query/BasicQueryExample.java
16+
header: |-
17+
QuestDB Java client [docs](https://questdb.io/docs/connect/clients/java/)
18+
and [Maven artifact](https://central.sonatype.com/artifact/org.questdb/questdb-client).
119
- name: ilp-http
220
lang: java
321
path: examples/src/main/java/com/example/sender/HttpExample.java
@@ -40,7 +58,7 @@
4058
```
4159
- name: ilp
4260
lang: java
43-
path: examples/src/main/java/com/example/sender/BasicExample.java
61+
path: examples/src/main/java/com/example/sender/TcpExample.java
4462
header: |-
4563
Java client library [docs](https://questdb.io/docs/reference/clients/java_ilp)
4664
and [Maven artifact](https://search.maven.org/artifact/org.questdb/questdb).
@@ -112,7 +130,7 @@
112130
port: 9009
113131
- name: ilp-from-conf
114132
lang: java
115-
path: examples/src/main/java/com/example/sender/BasicExample.java
133+
path: examples/src/main/java/com/example/sender/TcpExample.java
116134
header: |-
117135
Java client library [docs](https://questdb.io/docs/reference/clients/java_ilp)
118136
and [Maven artifact](https://search.maven.org/artifact/org.questdb/questdb).

‎examples/src/main/java/com/example/ConnectionErrorExample.java‎

Lines changed: 74 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,74 @@
1+
package com.example;
2+
3+
import io.questdb.client.Query;
4+
import io.questdb.client.QueryException;
5+
import io.questdb.client.QuestDB;
6+
import io.questdb.client.Sender;
7+
import io.questdb.client.cutlass.http.client.HttpClientException;
8+
import io.questdb.client.cutlass.line.LineSenderException;
9+
import io.questdb.client.cutlass.qwp.client.QwpColumnBatch;
10+
import io.questdb.client.cutlass.qwp.client.QwpColumnBatchHandler;
11+
import io.questdb.client.cutlass.qwp.client.QwpRoleMismatchException;
12+
13+
/**
14+
* Connection-level failures on the pooled facade.
15+
* <p>
16+
* Some failures happen when the pool opens a WebSocket, before any row or query
17+
* is exchanged. The pool warms up its minimum connections at {@code connect()}
18+
* and opens more on demand, so a connect-level error can surface either from
19+
* {@link QuestDB#connect} or from the first {@link QuestDB#borrowSender()} /
20+
* query -- wrap both.
21+
* <ul>
22+
* <li><b>Authentication failure</b> (401/403 on the upgrade) is terminal
23+
* across all endpoints and surfaces as {@link LineSenderException} on the
24+
* ingest side.</li>
25+
* <li><b>Role mismatch</b>: when every listed endpoint reports a role that
26+
* doesn't match the query-side {@code target=} filter, the client raises
27+
* {@link QwpRoleMismatchException}.</li>
28+
* <li><b>Unreachable / unresolvable host</b>: a generic connect failure
29+
* surfaces as {@link LineSenderException} on the ingest side and as
30+
* {@link HttpClientException} (the {@code QwpRoleMismatchException}
31+
* supertype) on the query side.</li>
32+
* </ul>
33+
*/
34+
public class ConnectionErrorExample {
35+
36+
public static void main(String[] args) throws InterruptedException {
37+
38+
// 1. Bad credentials -> auth failure on the ingest connection.
39+
try (QuestDB db = QuestDB.connect(
40+
"wss::addr=db.example.com:9000;token=INVALID_TOKEN;")) {
41+
try (Sender sender = db.borrowSender()) {
42+
sender.table("trades").doubleColumn("price", 1.0).atNow();
43+
}
44+
} catch (LineSenderException | HttpClientException e) {
45+
System.err.println("ingest connection failed (auth/unreachable?): " + e.getMessage());
46+
}
47+
48+
// 2. target=replica but no listed endpoint is a replica -> role mismatch.
49+
try (QuestDB db = QuestDB.connect(
50+
"wss::addr=db-primary:9000;token=YOUR_TOKEN;target=replica;")) {
51+
try (Query q = db.borrowQuery()) {
52+
q.sql("SELECT 1").handler(new QwpColumnBatchHandler() {
53+
@Override
54+
public void onBatch(QwpColumnBatch batch) {
55+
}
56+
57+
@Override
58+
public void onEnd(long totalRows) {
59+
}
60+
61+
@Override
62+
public void onError(byte status, String message) {
63+
}
64+
}).submit().await();
65+
} catch (QueryException e) {
66+
System.err.printf("query failed: status=0x%02X %s%n", e.getStatus() & 0xFF, e.getMessage());
67+
}
68+
} catch (QwpRoleMismatchException e) {
69+
System.err.println("no endpoint matched target=replica: " + e.getMessage());
70+
} catch (LineSenderException | HttpClientException e) {
71+
System.err.println("connection failed: " + e.getMessage());
72+
}
73+
}
74+
}

‎examples/src/main/java/com/example/QuestDBExample.java‎

Lines changed: 97 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,97 @@
1+
package com.example;
2+
3+
import io.questdb.client.Completion;
4+
import io.questdb.client.Query;
5+
import io.questdb.client.QueryException;
6+
import io.questdb.client.QuestDB;
7+
import io.questdb.client.Sender;
8+
import io.questdb.client.cutlass.qwp.client.QwpColumnBatch;
9+
import io.questdb.client.cutlass.qwp.client.QwpColumnBatchHandler;
10+
11+
/**
12+
* The recommended entry point: the pooled {@link QuestDB} handle.
13+
* <p>
14+
* {@code QuestDB.connect(...)} owns connection pools for <b>both</b> directions
15+
* over QWP (the QuestDB WebSocket protocol): a {@link Sender} pool for ingest
16+
* and a query-client pool for egress. One {@code ws}/{@code wss} connect string
17+
* configures both -- each side reads the keys it owns and ignores the rest.
18+
* <p>
19+
* Construct one {@code QuestDB} per deployment, share it across threads, and
20+
* close it at shutdown. This example does the whole round trip on one thread:
21+
* borrow a {@link Sender} to ingest a couple of rows, then run a {@code SELECT}
22+
* to read them back.
23+
* <p>
24+
* For ingest-only and query-only variants, see
25+
* {@code com.example.sender.WsExample} and
26+
* {@code com.example.query.BasicQueryExample}.
27+
*/
28+
public class QuestDBExample {
29+
30+
public static void main(String[] args) throws InterruptedException {
31+
// One handle for the whole deployment. close() (try-with-resources)
32+
// shuts the pools down and disconnects every underlying client.
33+
try (QuestDB db = QuestDB.connect("ws::addr=localhost:9000;");
34+
Query q = db.borrowQuery()) {
35+
36+
// Ingest: borrow a Sender, write rows, close() flushes it and
37+
// returns it to the pool. The WebSocket is NOT torn down here --
38+
// a real disconnect only happens at db.close().
39+
try (Sender sender = db.borrowSender()) {
40+
sender.table("trades")
41+
.symbol("symbol", "ETH-USD")
42+
.symbol("side", "sell")
43+
.doubleColumn("price", 2615.54)
44+
.doubleColumn("amount", 0.00044)
45+
.atNow();
46+
sender.table("trades")
47+
.symbol("symbol", "BTC-USD")
48+
.symbol("side", "sell")
49+
.doubleColumn("price", 39269.98)
50+
.doubleColumn("amount", 0.001)
51+
.atNow();
52+
}
53+
54+
// Query: one-shot SELECT. Borrow a Query handle from the egress
55+
// pool, submit the SQL, and await the returned Completion.
56+
// (Freshly ingested rows land in the WAL asynchronously, so an
57+
// immediate read-back may not see them yet -- that is expected.)
58+
Completion c = q.sql(
59+
"SELECT symbol, side, price, amount FROM trades "
60+
+ "WHERE symbol = 'ETH-USD' LIMIT 10")
61+
.handler(new PrintingHandler())
62+
.submit();
63+
try {
64+
c.await(); // blocks until onEnd / onError has run
65+
} catch (QueryException e) {
66+
// Server reported an error (parse failure, schema mismatch, ...).
67+
System.err.printf("query failed: status=0x%02X %s%n",
68+
e.getStatus() & 0xFF, e.getMessage());
69+
}
70+
}
71+
}
72+
73+
/**
74+
* Prints each row. The {@code QwpColumnBatch} is valid only during the
75+
* {@code onBatch} callback -- copy values out if you need them later.
76+
*/
77+
private static final class PrintingHandler implements QwpColumnBatchHandler {
78+
@Override
79+
public void onBatch(QwpColumnBatch batch) {
80+
batch.forEachRow(row -> System.out.printf(
81+
"symbol=%s side=%s price=%.4f amount=%.5f%n",
82+
row.getSymbol(0),
83+
row.getSymbol(1),
84+
row.getDoubleValue(2),
85+
row.getDoubleValue(3)));
86+
}
87+
88+
@Override
89+
public void onEnd(long totalRows) {
90+
System.out.println("done: " + totalRows + " rows");
91+
}
92+
93+
@Override
94+
public void onError(byte status, String message) {
95+
}
96+
}
97+
}

‎examples/src/main/java/com/example/query/ArrayResultExample.java‎

Lines changed: 66 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,66 @@
1+
package com.example.query;
2+
3+
import io.questdb.client.Query;
4+
import io.questdb.client.QueryException;
5+
import io.questdb.client.QuestDB;
6+
import io.questdb.client.cutlass.qwp.client.QwpColumnBatch;
7+
import io.questdb.client.cutlass.qwp.client.QwpColumnBatchHandler;
8+
9+
import java.util.Arrays;
10+
11+
/**
12+
* Reading array columns from a result set.
13+
* <p>
14+
* {@link QwpColumnBatch#getDoubleArrayElements(int, int)} returns a flattened
15+
* {@code double[]} in row-major order; {@link QwpColumnBatch#getArrayNDims(int, int)}
16+
* reports the dimensionality so you can interpret that flat layout. This is the
17+
* egress counterpart to {@link com.example.sender.WsArrayExample} (ingest).
18+
* <p>
19+
* Alternatively, extract individual elements in SQL (e.g. {@code SELECT
20+
* bids[1][1] FROM market_data}) and read them as scalar doubles.
21+
* <p>
22+
* Assumes a table with a {@code DOUBLE[]} column, e.g. the one
23+
* {@code WsArrayExample} writes:
24+
* <pre>
25+
* CREATE TABLE book (levels DOUBLE[], timestamp TIMESTAMP)
26+
* TIMESTAMP(timestamp) PARTITION BY DAY WAL;
27+
* </pre>
28+
*/
29+
public class ArrayResultExample {
30+
31+
public static void main(String[] args) throws InterruptedException {
32+
try (QuestDB db = QuestDB.connect("ws::addr=localhost:9000;")) {
33+
try (Query q = db.borrowQuery()) {
34+
q.sql(
35+
"SELECT levels FROM book LIMIT 10")
36+
.handler(new QwpColumnBatchHandler() {
37+
@Override
38+
public void onBatch(QwpColumnBatch batch) {
39+
for (int row = 0; row < batch.getRowCount(); row++) {
40+
if (batch.isNull(0, row)) {
41+
System.out.println("levels: null");
42+
continue;
43+
}
44+
int nDims = batch.getArrayNDims(0, row);
45+
double[] flat = batch.getDoubleArrayElements(0, row);
46+
System.out.printf("levels: nDims=%d values=%s%n",
47+
nDims, Arrays.toString(flat));
48+
}
49+
}
50+
51+
@Override
52+
public void onEnd(long totalRows) {
53+
System.out.println("done: " + totalRows + " rows");
54+
}
55+
56+
@Override
57+
public void onError(byte status, String message) {
58+
}
59+
}
60+
).submit().await();
61+
} catch (QueryException e) {
62+
System.err.printf("query failed: status=0x%02X %s%n", e.getStatus() & 0xFF, e.getMessage());
63+
}
64+
}
65+
}
66+
}

‎examples/src/main/java/com/example/query/BasicQueryExample.java‎

Lines changed: 47 additions & 38 deletions
Original file line numberDiff line numberDiff line change
@@ -1,64 +1,73 @@
11
package com.example.query;
22

3+
import io.questdb.client.Query;
4+
import io.questdb.client.QueryException;
5+
import io.questdb.client.QuestDB;
36
import io.questdb.client.cutlass.qwp.client.QwpColumnBatch;
47
import io.questdb.client.cutlass.qwp.client.QwpColumnBatchHandler;
5-
import io.questdb.client.cutlass.qwp.client.QwpQueryClient;
68

79
/**
810
* Minimal QWP egress query example.
911
* <p>
10-
* Connects to a QuestDB server over the /read/v1 WebSocket endpoint,
11-
* runs a SELECT query, and prints each row as the batches arrive.
12+
* Opens a pooled {@link QuestDB} handle over QWP (WebSocket), borrows a
13+
* {@link Query} to run a SELECT, and prints each row as the batches
14+
* arrive. {@code Completion.await()} blocks until the query finishes and
15+
* rethrows any server error as a {@link QueryException}.
1216
* <p>
1317
* Iterates rows via {@link QwpColumnBatch#forEachRow}, which hands a reusable
1418
* row-pinned view to the lambda. Single-arg accessors keep the read path
1519
* compact; the underlying batch is still column-major and the {@code (col, row)}
1620
* primitives remain available on {@code batch} for callers that prefer them.
1721
* <p>
18-
* Assumes a table exists:
22+
* Queries the {@code trades} table the ingest examples
23+
* ({@code com.example.sender.WsExample}, {@code com.example.QuestDBExample})
24+
* write -- auto-created on first ingest as:
1925
* <pre>
20-
* CREATE TABLE trades (ts TIMESTAMP, sym SYMBOL, price DOUBLE, qty LONG)
21-
* TIMESTAMP(ts) PARTITION BY DAY WAL;
26+
* CREATE TABLE trades (
27+
* symbol SYMBOL, side SYMBOL, price DOUBLE, amount DOUBLE, timestamp TIMESTAMP
28+
* ) TIMESTAMP(timestamp) PARTITION BY DAY WAL;
2229
* </pre>
2330
*/
2431
public class BasicQueryExample {
2532

26-
public static void main(String[] args) {
27-
try (QwpQueryClient client = QwpQueryClient.newPlainText("localhost", 9000)) {
28-
client.connect();
33+
public static void main(String[] args) throws InterruptedException {
34+
try (QuestDB db = QuestDB.connect("ws::addr=localhost:9000;")) {
35+
try (Query q = db.borrowQuery()) {
36+
q.sql(
37+
"SELECT timestamp, symbol, side, price, amount FROM trades WHERE symbol = 'ETH-USD' LIMIT 1000")
38+
.handler(new QwpColumnBatchHandler() {
39+
@Override
40+
public void onBatch(QwpColumnBatch batch) {
41+
// The RowView handed to the lambda is reusable and pinned to the
42+
// current row; copy values out before the callback returns if you
43+
// need to retain them past the surrounding onBatch call.
44+
batch.forEachRow(row -> {
45+
long timestamp = row.getLongValue(0); // TIMESTAMP -> microseconds since epoch
46+
String symbol = row.getSymbol(1); // SYMBOL -> String
47+
String side = row.getSymbol(2); // SYMBOL -> String
48+
double price = row.getDoubleValue(3); // DOUBLE
49+
double amount = row.getDoubleValue(4); // DOUBLE
2950

30-
client.execute(
31-
"SELECT ts, sym, price, qty FROM trades WHERE sym = 'AAPL' LIMIT 1000",
32-
new QwpColumnBatchHandler() {
33-
@Override
34-
public void onBatch(QwpColumnBatch batch) {
35-
// The RowView handed to the lambda is reusable and pinned to the
36-
// current row; copy values out before the callback returns if you
37-
// need to retain them past the surrounding onBatch call.
38-
batch.forEachRow(row -> {
39-
long timestamp = row.getLongValue(0); // TIMESTAMP -> microseconds since epoch
40-
String symbol = row.getSymbol(1); // SYMBOL -> String
41-
double price = row.getDoubleValue(2); // DOUBLE
42-
long qty = row.getLongValue(3); // LONG
51+
System.out.printf(
52+
"timestamp=%d symbol=%s side=%s price=%.4f amount=%.5f%n",
53+
timestamp, symbol, side, price, amount
54+
);
55+
});
56+
}
4357

44-
System.out.printf(
45-
"ts=%d sym=%s price=%.4f qty=%d%n",
46-
timestamp, symbol, price, qty
47-
);
48-
});
49-
}
50-
51-
@Override
52-
public void onEnd(long totalRows) {
53-
System.out.println("query finished");
54-
}
58+
@Override
59+
public void onEnd(long totalRows) {
60+
System.out.println("query finished");
61+
}
5562

56-
@Override
57-
public void onError(byte status, String message) {
58-
System.err.println("query failed: status=" + status + " msg=" + message);
63+
@Override
64+
public void onError(byte status, String message) {
65+
}
5966
}
60-
}
61-
);
67+
).submit().await();
68+
} catch (QueryException e) {
69+
System.err.printf("query failed: status=0x%02X %s%n", e.getStatus() & 0xFF, e.getMessage());
70+
}
6271
}
6372
}
6473
}

0 commit comments

Comments
 (0)