Skip to content

Commit 3f2cfed

Browse files
kafka1991kafka1991
committed
high quality example
1 parent c7012df commit 3f2cfed

9 files changed

Lines changed: 99 additions & 21 deletions

‎examples.manifest.yaml‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -58,7 +58,7 @@
5858
```
5959
- name: ilp
6060
lang: java
61-
path: examples/src/main/java/com/example/sender/BasicExample.java
61+
path: examples/src/main/java/com/example/sender/TcpExample.java
6262
header: |-
6363
Java client library [docs](https://questdb.io/docs/reference/clients/java_ilp)
6464
and [Maven artifact](https://search.maven.org/artifact/org.questdb/questdb).
@@ -130,7 +130,7 @@
130130
port: 9009
131131
- name: ilp-from-conf
132132
lang: java
133-
path: examples/src/main/java/com/example/sender/BasicExample.java
133+
path: examples/src/main/java/com/example/sender/TcpExample.java
134134
header: |-
135135
Java client library [docs](https://questdb.io/docs/reference/clients/java_ilp)
136136
and [Maven artifact](https://search.maven.org/artifact/org.questdb/questdb).

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

Lines changed: 14 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -18,10 +18,13 @@
1818
* compact; the underlying batch is still column-major and the {@code (col, row)}
1919
* primitives remain available on {@code batch} for callers that prefer them.
2020
* <p>
21-
* Assumes a table exists:
21+
* Queries the {@code trades} table the ingest examples
22+
* ({@code com.example.sender.WsExample}, {@code com.example.QuestDBExample})
23+
* write -- auto-created on first ingest as:
2224
* <pre>
23-
* CREATE TABLE trades (ts TIMESTAMP, sym SYMBOL, price DOUBLE, qty LONG)
24-
* TIMESTAMP(ts) PARTITION BY DAY WAL;
25+
* CREATE TABLE trades (
26+
* symbol SYMBOL, side SYMBOL, price DOUBLE, amount DOUBLE, timestamp TIMESTAMP
27+
* ) TIMESTAMP(timestamp) PARTITION BY DAY WAL;
2528
* </pre>
2629
*/
2730
public class BasicQueryExample {
@@ -30,22 +33,23 @@ public static void main(String[] args) throws InterruptedException {
3033
try (QuestDB db = QuestDB.connect("ws::addr=localhost:9000;")) {
3134
try {
3235
db.executeSql(
33-
"SELECT ts, sym, price, qty FROM trades WHERE sym = 'AAPL' LIMIT 1000",
36+
"SELECT timestamp, symbol, side, price, amount FROM trades WHERE symbol = 'ETH-USD' LIMIT 1000",
3437
new QwpColumnBatchHandler() {
3538
@Override
3639
public void onBatch(QwpColumnBatch batch) {
3740
// The RowView handed to the lambda is reusable and pinned to the
3841
// current row; copy values out before the callback returns if you
3942
// need to retain them past the surrounding onBatch call.
4043
batch.forEachRow(row -> {
41-
long timestamp = row.getLongValue(0); // TIMESTAMP -> microseconds since epoch
42-
String symbol = row.getSymbol(1); // SYMBOL -> String
43-
double price = row.getDoubleValue(2); // DOUBLE
44-
long qty = row.getLongValue(3); // LONG
44+
long timestamp = row.getLongValue(0); // TIMESTAMP -> microseconds since epoch
45+
String symbol = row.getSymbol(1); // SYMBOL -> String
46+
String side = row.getSymbol(2); // SYMBOL -> String
47+
double price = row.getDoubleValue(3); // DOUBLE
48+
double amount = row.getDoubleValue(4); // DOUBLE
4549

4650
System.out.printf(
47-
"ts=%d sym=%s price=%.4f qty=%d%n",
48-
timestamp, symbol, price, qty
51+
"ts=%d symbol=%s side=%s price=%.4f amount=%.5f%n",
52+
timestamp, symbol, side, price, amount
4953
);
5054
});
5155
}

examples/src/main/java/com/example/BinaryColumnExample.java renamed to examples/src/main/java/com/example/query/BinaryResultExample.java

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
package com.example;
1+
package com.example.query;
22

33
import io.questdb.client.QueryException;
44
import io.questdb.client.QuestDB;
@@ -17,7 +17,7 @@
1717
* {@code byte[]}; {@code getBinaryA}/{@code getBinaryB} give reusable native
1818
* views for allocation-free reads.
1919
*/
20-
public class BinaryColumnExample {
20+
public class BinaryResultExample {
2121

2222
public static void main(String[] args) throws InterruptedException {
2323
byte[] payload = {0x00, 0x01, 0x02, (byte) 0xFF};

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

Lines changed: 74 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,74 @@
1+
package com.example.query;
2+
3+
import io.questdb.client.QueryException;
4+
import io.questdb.client.QuestDB;
5+
import io.questdb.client.cutlass.qwp.client.QwpColumnBatch;
6+
import io.questdb.client.cutlass.qwp.client.QwpColumnBatchHandler;
7+
8+
/**
9+
* Sharing one {@link QuestDB} handle across many query threads.
10+
* <p>
11+
* The query side of the pool is thread-safe the same way ingest is (see
12+
* {@link com.example.sender.WsConcurrentIngestExample}): create one
13+
* {@code QuestDB}, hand the same instance to every thread, and let each call
14+
* {@link QuestDB#query()}. Every call returns that thread's own cached
15+
* {@link io.questdb.client.Query} instance -- no external synchronization, no
16+
* per-query allocation, one in-flight query per thread. Each {@code submit()}
17+
* acquires a worker from the query pool, so {@code queryPoolSize} caps how
18+
* many queries run in parallel; extra submits block on the acquire timeout
19+
* until a worker frees up.
20+
* <p>
21+
* For several in-flight queries from a <em>single</em> thread, see
22+
* {@link MultiInFlightQueryExample} ({@link QuestDB#newQuery()}).
23+
*/
24+
public class ConcurrentQueryExample {
25+
26+
private static final String[] SYMBOLS = {"ETH-USD", "BTC-USD", "SOL-USD", "ADA-USD"};
27+
28+
public static void main(String[] args) throws InterruptedException {
29+
try (QuestDB db = QuestDB.builder()
30+
.fromConfig("ws::addr=localhost:9000;")
31+
.queryPoolSize(SYMBOLS.length)
32+
.build()) {
33+
34+
Thread[] readers = new Thread[SYMBOLS.length];
35+
for (int i = 0; i < SYMBOLS.length; i++) {
36+
final String symbol = SYMBOLS[i];
37+
readers[i] = new Thread(() -> countTrades(db, symbol), "reader-" + symbol);
38+
readers[i].start();
39+
}
40+
for (Thread reader : readers) {
41+
reader.join();
42+
}
43+
}
44+
}
45+
46+
private static void countTrades(QuestDB db, final String symbol) {
47+
try {
48+
db.query()
49+
.sql("SELECT count() FROM trades WHERE symbol = $1")
50+
.binds(binds -> binds.setVarchar(0, symbol))
51+
.handler(new QwpColumnBatchHandler() {
52+
@Override
53+
public void onBatch(QwpColumnBatch batch) {
54+
batch.forEachRow(row -> System.out.println(symbol + " count = " + row.getLongValue(0)));
55+
}
56+
57+
@Override
58+
public void onEnd(long totalRows) {
59+
}
60+
61+
@Override
62+
public void onError(byte status, String message) {
63+
System.err.printf("%s error: 0x%02X %s%n", symbol, status & 0xFF, message);
64+
}
65+
})
66+
.submit()
67+
.await();
68+
} catch (QueryException e) {
69+
System.err.printf("query failed: status=0x%02X %s%n", e.getStatus() & 0xFF, e.getMessage());
70+
} catch (InterruptedException e) {
71+
Thread.currentThread().interrupt();
72+
}
73+
}
74+
}

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,9 +1,9 @@
11
package com.example.query;
22

33
import io.questdb.client.Completion;
4+
import io.questdb.client.Query;
45
import io.questdb.client.QueryException;
56
import io.questdb.client.QuestDB;
6-
import io.questdb.client.Query;
77
import io.questdb.client.cutlass.qwp.client.QwpColumnBatch;
88
import io.questdb.client.cutlass.qwp.client.QwpColumnBatchHandler;
99

‎examples/src/main/java/com/example/sender/HttpExample.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -12,7 +12,7 @@ public static void main(String[] args) {
1212
.doubleColumn("amount", 0.00044)
1313
.atNow();
1414
sender.table("trades")
15-
.symbol("symbol", "TC-USD")
15+
.symbol("symbol", "BTC-USD")
1616
.symbol("side", "sell")
1717
.doubleColumn("price", 39269.98)
1818
.doubleColumn("amount", 0.001)

examples/src/main/java/com/example/sender/BasicExample.java renamed to examples/src/main/java/com/example/sender/TcpExample.java

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,7 @@
22

33
import io.questdb.client.Sender;
44

5-
public class BasicExample {
5+
public class TcpExample {
66
public static void main(String[] args) {
77
try (Sender sender = Sender.fromConfig("tcp::addr=localhost:9009;")) {
88
sender.table("trades")
@@ -12,11 +12,11 @@ public static void main(String[] args) {
1212
.doubleColumn("amount", 0.00044)
1313
.atNow();
1414
sender.table("trades")
15-
.symbol("symbol", "TC-USD")
15+
.symbol("symbol", "BTC-USD")
1616
.symbol("side", "sell")
1717
.doubleColumn("price", 39269.98)
1818
.doubleColumn("amount", 0.001)
1919
.atNow();
2020
}
2121
}
22-
}
22+
}

‎examples/src/main/java/com/example/sender/WsColumnTypesExample.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -53,7 +53,7 @@ public static void main(String[] args) {
5353
// microsecond TIMESTAMP.
5454
sender.table("ticks")
5555
.doubleColumn("d", 1.0)
56-
.at(System.nanoTime(), ChronoUnit.NANOS);
56+
.at(System.currentTimeMillis() * 1_000_000L, ChronoUnit.NANOS);
5757

5858
// Server-assigned wall-clock time.
5959
sender.table("all_types")

‎examples/src/main/java/com/example/sender/WsExample.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -20,7 +20,7 @@
2020
* </ul>
2121
* A {@code Sender} is not thread-safe, so no two threads may share one -- the
2222
* pool hands each thread its own. Contrast with the low-level ILP examples in
23-
* this package ({@link BasicExample}, {@link HttpExample}, ...), which open a
23+
* this package ({@link TcpExample}, {@link HttpExample}, ...), which open a
2424
* dedicated {@code Sender.fromConfig(...)} connection per use over HTTP/TCP.
2525
*/
2626
public class WsExample {

0 commit comments

Comments
 (0)