From ede4718d71d84cb8f6a13b86085a2efce52ec1f1 Mon Sep 17 00:00:00 2001 From: Sergey Chernov Date: Wed, 5 Jun 2024 22:31:51 -0700 Subject: [PATCH 1/4] refactored metrics to make them more generic --- .../com/clickhouse/client/api/Client.java | 21 +-- .../client/api/OperationStatistics.java | 156 ------------------ .../client/api/insert/InsertResponse.java | 20 +-- .../api/internal/ClientStatisticsHolder.java | 45 +++++ .../clickhouse/client/api/internal/Gauge.java | 21 +++ .../client/api/internal/StopWatch.java | 9 +- .../client/api/metrics/ClientMetrics.java | 24 +++ .../clickhouse/client/api/metrics/Metric.java | 10 ++ .../client/api/metrics/OperationMetrics.java | 54 ++++++ .../client/api/metrics/ServerMetrics.java | 55 ++++++ .../client/api/query/QueryResponse.java | 17 +- .../clickhouse/client/insert/InsertTests.java | 10 +- .../clickhouse/client/query/QueryTests.java | 37 ++--- 13 files changed, 266 insertions(+), 213 deletions(-) delete mode 100644 client-v2/src/main/java/com/clickhouse/client/api/OperationStatistics.java create mode 100644 client-v2/src/main/java/com/clickhouse/client/api/internal/ClientStatisticsHolder.java create mode 100644 client-v2/src/main/java/com/clickhouse/client/api/internal/Gauge.java create mode 100644 client-v2/src/main/java/com/clickhouse/client/api/metrics/ClientMetrics.java create mode 100644 client-v2/src/main/java/com/clickhouse/client/api/metrics/Metric.java create mode 100644 client-v2/src/main/java/com/clickhouse/client/api/metrics/OperationMetrics.java create mode 100644 client-v2/src/main/java/com/clickhouse/client/api/metrics/ServerMetrics.java diff --git a/client-v2/src/main/java/com/clickhouse/client/api/Client.java b/client-v2/src/main/java/com/clickhouse/client/api/Client.java index 5d3a6c465..184464067 100644 --- a/client-v2/src/main/java/com/clickhouse/client/api/Client.java +++ b/client-v2/src/main/java/com/clickhouse/client/api/Client.java @@ -21,6 +21,8 @@ import com.clickhouse.client.api.internal.TableSchemaParser; import com.clickhouse.client.api.internal.ValidationUtils; import com.clickhouse.client.api.metadata.TableSchema; +import com.clickhouse.client.api.internal.ClientStatisticsHolder; +import com.clickhouse.client.api.metrics.ClientMetrics; import com.clickhouse.client.api.query.GenericRecord; import com.clickhouse.client.api.query.QueryResponse; import com.clickhouse.client.api.query.QuerySettings; @@ -103,7 +105,7 @@ public class Client { private static final Logger LOG = LoggerFactory.getLogger(Client.class); private ExecutorService queryExecutor; - private Map globalClientStats = new ConcurrentHashMap<>(); + private Map globalClientStats = new ConcurrentHashMap<>(); private Client(Set endpoints, Map configuration) { this.endpoints = endpoints; @@ -526,7 +528,7 @@ public CompletableFuture insert(String tableName, List data, String operationId = startOperation(); settings.setOperationId(operationId); - globalClientStats.get(operationId).start("serialization"); + globalClientStats.get(operationId).start(ClientMetrics.OP_SERIALIZATION); if (data == null || data.isEmpty()) { throw new IllegalArgumentException("Data cannot be empty"); @@ -560,7 +562,7 @@ public CompletableFuture insert(String tableName, List data, } } - globalClientStats.get(operationId).stop("serialization"); + globalClientStats.get(operationId).stop(ClientMetrics.OP_SERIALIZATION); LOG.debug("Total serialization time: {}", globalClientStats.get(operationId).getElapsedTime("serialization")); return insert(tableName, new ByteArrayInputStream(stream.toByteArray()), format, settings); } @@ -598,8 +600,8 @@ public CompletableFuture insert(String tableName, operationId = startOperation(); settings.setOperationId(operationId); } - OperationStatistics.ClientStatistics clientStats = globalClientStats.remove(operationId); - clientStats.start("insert"); + ClientStatisticsHolder clientStats = globalClientStats.remove(operationId); + clientStats.start(ClientMetrics.OP_DURATION); CompletableFuture responseFuture = new CompletableFuture<>(); @@ -627,7 +629,6 @@ public CompletableFuture insert(String tableName, responseFuture.completeExceptionally(new ClientException("Operation has likely timed out.", e)); } } - clientStats.stop("insert"); LOG.debug("Total insert (InputStream) time: {}", clientStats.getElapsedTime("insert")); } @@ -695,8 +696,8 @@ public CompletableFuture query(String sqlQuery, Map request = client.read(getServerNode()); @@ -751,7 +752,7 @@ public CompletableFuture queryRecords(String sqlQuery, QuerySettings se settings = new QuerySettings(); } settings.setFormat(ClickHouseFormat.RowBinaryWithNamesAndTypes); - OperationStatistics.ClientStatistics clientStats = new OperationStatistics.ClientStatistics(); + ClientStatisticsHolder clientStats = new ClientStatisticsHolder(); clientStats.start("query"); ClickHouseClient client = createClient(); ClickHouseRequest request = client.read(getServerNode()); @@ -865,7 +866,7 @@ private ClickHouseRequest.Mutation createMutationRequest(ClickHouseRequest.Mutat private String startOperation() { String operationId = UUID.randomUUID().toString(); - globalClientStats.put(operationId, new OperationStatistics.ClientStatistics()); + globalClientStats.put(operationId, new ClientStatisticsHolder()); return operationId; } } diff --git a/client-v2/src/main/java/com/clickhouse/client/api/OperationStatistics.java b/client-v2/src/main/java/com/clickhouse/client/api/OperationStatistics.java deleted file mode 100644 index 57d642a7c..000000000 --- a/client-v2/src/main/java/com/clickhouse/client/api/OperationStatistics.java +++ /dev/null @@ -1,156 +0,0 @@ -package com.clickhouse.client.api; - -import com.clickhouse.client.ClickHouseResponseSummary; -import com.clickhouse.client.api.internal.StopWatch; - -import java.util.HashMap; -import java.util.Map; - -/** - * OperationStatistics objects hold various stats for complete operations. - *

- * It can be used for logging or monitoring purposes. - */ -public class OperationStatistics { - - public static final ServerStatistics EMPTY_SERVER_STATS = new ServerStatistics(-1, -1, -1, -1, -1, -1, -1); - - public ServerStatistics serverStatistics; - - public ClientStatistics clientStatistics; - - public OperationStatistics(ClientStatistics clientStatistics) { - this.serverStatistics = EMPTY_SERVER_STATS; - this.clientStatistics = clientStatistics; - } - - public void setClientStatistics(ClientStatistics clientStatistics) { - this.clientStatistics = clientStatistics; - } - - public ClientStatistics getClientStatistics() { - return clientStatistics; - } - - public ServerStatistics getServerStatistics() { - return serverStatistics; - } - - public void updateServerStats(ClickHouseResponseSummary summary) { - if (summary == null || summary.equals(ClickHouseResponseSummary.EMPTY)) { - this.serverStatistics = EMPTY_SERVER_STATS; - return; - } - - this.serverStatistics = new ServerStatistics( - summary.getReadRows(), - summary.getReadBytes(), - summary.getTotalRowsToRead(), - summary.getWrittenRows(), - summary.getWrittenBytes(), - summary.getResultRows(), - summary.getElapsedTime() - ); - } - - @Override - public String toString() { - return "OperationStatistics{" + - "\"serverStatistics\"=" + serverStatistics + - ", \"clientStatistics\"=" + clientStatistics + - '}'; - } - - /** - * Stats returned by the server. - *

- * `-1` means the value is not available. - */ - public static class ServerStatistics { - - /** - * Number of rows read by server from the storage. - */ - public final long numRowsRead; - - /** - * Number of rows written by server to the storage. - */ - public final long numRowsWritten; - - /** - * Estimated number of rows to read from the storage. - *

- */ - public final long totalRowsToRead; - - /** - * Number of bytes read by server from the storage. - */ - public final long numBytesRead; - - /** - * Number of bytes written by server to the storage. - */ - public final long numBytesWritten; - - /** - * Number of returned rows. - */ - public final long resultRows; - - /** - * Elapsed time in nanoseconds. - */ - public final long elapsedTime; - - - public ServerStatistics(long numRowsRead, long numBytesRead, long totalRowsToRead, long numRowsWritten, long numBytesWritten, long resultRows, long elapsedTime) { - this.numRowsRead = numRowsRead; - this.numBytesRead = numBytesRead; - this.totalRowsToRead = totalRowsToRead; - this.numRowsWritten = numRowsWritten; - this.numBytesWritten = numBytesWritten; - this.resultRows = resultRows; - this.elapsedTime = elapsedTime; - } - - @Override - public String toString() { - return "ServerStatistics{" + - "\"numRowsRead\"=" + numRowsRead + - ", \"numRowsWritten\"=" + numRowsWritten + - ", \"totalRowsToRead\"=" + totalRowsToRead + - ", \"numBytesRead\"=" + numBytesRead + - ", \"numBytesWritten\"=" + numBytesWritten + - ", \"resultRows\"=" + resultRows + - ", \"elapsedTime\"=\"" + elapsedTime + "ns\"" + - '}'; - } - } - - public static class ClientStatistics { - private final Map spans = new HashMap<>(); - - public void start(String spanName) { - spans.computeIfAbsent(spanName, k -> new StopWatch()).start(); - } - - public void stop(String spanName) { - spans.computeIfAbsent(spanName, k -> new StopWatch()).stop(); - } - - public long getElapsedTime(String spanName) { - StopWatch sw = spans.get(spanName); - return sw == null ? -1 : sw.getElapsedTime(); - } - - @Override - public String toString() { - return "ClientStatistics{" + - "\"spans\"=" + spans + - '}'; - } - } - -} diff --git a/client-v2/src/main/java/com/clickhouse/client/api/insert/InsertResponse.java b/client-v2/src/main/java/com/clickhouse/client/api/insert/InsertResponse.java index 41fa0fdb4..85ff85b1d 100644 --- a/client-v2/src/main/java/com/clickhouse/client/api/insert/InsertResponse.java +++ b/client-v2/src/main/java/com/clickhouse/client/api/insert/InsertResponse.java @@ -2,25 +2,21 @@ import com.clickhouse.client.ClickHouseClient; import com.clickhouse.client.ClickHouseResponse; -import com.clickhouse.client.ClickHouseResponseSummary; -import com.clickhouse.client.api.OperationStatistics; - -import java.util.concurrent.ExecutionException; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.TimeoutException; +import com.clickhouse.client.api.internal.ClientStatisticsHolder; +import com.clickhouse.client.api.metrics.OperationMetrics; public class InsertResponse implements AutoCloseable { private final ClickHouseResponse responseRef; private final ClickHouseClient client; - private OperationStatistics operationStatistics; + private OperationMetrics operationMetrics; public InsertResponse(ClickHouseClient client, ClickHouseResponse responseRef, - OperationStatistics.ClientStatistics clientStatistics) { + ClientStatisticsHolder clientStatisticsHolder) { this.responseRef = responseRef; this.client = client; - this.operationStatistics = new OperationStatistics(clientStatistics); - this.operationStatistics.updateServerStats(responseRef.getSummary()); + this.operationMetrics = new OperationMetrics(clientStatisticsHolder); + this.operationMetrics.operationComplete(responseRef.getSummary()); } @Override @@ -32,7 +28,7 @@ public void close() { } } - public OperationStatistics getOperationStatistics() { - return operationStatistics; + public OperationMetrics getMetrics() { + return operationMetrics; } } diff --git a/client-v2/src/main/java/com/clickhouse/client/api/internal/ClientStatisticsHolder.java b/client-v2/src/main/java/com/clickhouse/client/api/internal/ClientStatisticsHolder.java new file mode 100644 index 000000000..3d96c880f --- /dev/null +++ b/client-v2/src/main/java/com/clickhouse/client/api/internal/ClientStatisticsHolder.java @@ -0,0 +1,45 @@ +package com.clickhouse.client.api.internal; + +import com.clickhouse.client.api.metrics.ClientMetrics; + +import java.util.HashMap; +import java.util.Map; + +public class ClientStatisticsHolder { + + private final Map stopWatches = new HashMap<>(); + + public void start(ClientMetrics metric) { + start(metric.getKey()); + } + + public void start(String spanName) { + stopWatches.computeIfAbsent(spanName, k -> new StopWatch()).start(); + } + + public StopWatch stop(ClientMetrics metric) { + return stop(metric.getKey()); + } + + public StopWatch stop(String spanName) { + StopWatch timer = stopWatches.computeIfAbsent(spanName, k -> new StopWatch()); + timer.stop(); + return timer; + } + + public long getElapsedTime(String spanName) { + StopWatch sw = stopWatches.get(spanName); + return sw == null ? -1 : sw.getElapsedTime(); + } + + public Map getStopWatches() { + return stopWatches; + } + + @Override + public String toString() { + return "ClientStatistics{" + + "\"spans\"=" + stopWatches + + '}'; + } +} diff --git a/client-v2/src/main/java/com/clickhouse/client/api/internal/Gauge.java b/client-v2/src/main/java/com/clickhouse/client/api/internal/Gauge.java new file mode 100644 index 000000000..f83f653fc --- /dev/null +++ b/client-v2/src/main/java/com/clickhouse/client/api/internal/Gauge.java @@ -0,0 +1,21 @@ +package com.clickhouse.client.api.internal; + +import com.clickhouse.client.api.metrics.Metric; + +public class Gauge implements Metric { + + private volatile long value; + + public Gauge(long readRows) { + this.value = readRows; + } + + public void set(long value) { + this.value = value; + } + + @Override + public long getLong() { + return value; + } +} diff --git a/client-v2/src/main/java/com/clickhouse/client/api/internal/StopWatch.java b/client-v2/src/main/java/com/clickhouse/client/api/internal/StopWatch.java index 70c2b76f9..ac686f6ef 100644 --- a/client-v2/src/main/java/com/clickhouse/client/api/internal/StopWatch.java +++ b/client-v2/src/main/java/com/clickhouse/client/api/internal/StopWatch.java @@ -1,8 +1,10 @@ package com.clickhouse.client.api.internal; +import com.clickhouse.client.api.metrics.Metric; + import java.util.concurrent.TimeUnit; -public class StopWatch { +public class StopWatch implements Metric { long elapsedNanoTime = 0; long startNanoTime; @@ -38,4 +40,9 @@ public String toString() { ", \"elapsedTime\"=" + getElapsedTime() + '}'; } + + @Override + public long getLong() { + return getElapsedTime(); + } } diff --git a/client-v2/src/main/java/com/clickhouse/client/api/metrics/ClientMetrics.java b/client-v2/src/main/java/com/clickhouse/client/api/metrics/ClientMetrics.java new file mode 100644 index 000000000..c21becbcf --- /dev/null +++ b/client-v2/src/main/java/com/clickhouse/client/api/metrics/ClientMetrics.java @@ -0,0 +1,24 @@ +package com.clickhouse.client.api.metrics; + +public enum ClientMetrics { + + /** + * Operation duration in nanoseconds. + */ + OP_DURATION("client.opDuration"), + + /** + * Duration of the operation serialization step in nanoseconds. + */ + OP_SERIALIZATION("client.opSerialization"); + + private final String key; + + ClientMetrics(String key) { + this.key = key; + } + + public String getKey() { + return key; + } +} diff --git a/client-v2/src/main/java/com/clickhouse/client/api/metrics/Metric.java b/client-v2/src/main/java/com/clickhouse/client/api/metrics/Metric.java new file mode 100644 index 000000000..e9628e13c --- /dev/null +++ b/client-v2/src/main/java/com/clickhouse/client/api/metrics/Metric.java @@ -0,0 +1,10 @@ +package com.clickhouse.client.api.metrics; + +public interface Metric { + + /** + * Returns value of the metric as a long. + * @return value of the metric as a long + */ + long getLong(); +} diff --git a/client-v2/src/main/java/com/clickhouse/client/api/metrics/OperationMetrics.java b/client-v2/src/main/java/com/clickhouse/client/api/metrics/OperationMetrics.java new file mode 100644 index 000000000..723a8a186 --- /dev/null +++ b/client-v2/src/main/java/com/clickhouse/client/api/metrics/OperationMetrics.java @@ -0,0 +1,54 @@ +package com.clickhouse.client.api.metrics; + +import com.clickhouse.client.ClickHouseResponseSummary; +import com.clickhouse.client.api.internal.ClientStatisticsHolder; +import com.clickhouse.client.api.internal.Gauge; +import com.clickhouse.client.api.internal.StopWatch; + +import java.util.HashMap; +import java.util.Map; + +/** + * OperationStatistics objects hold various stats for complete operations. + *

+ * It can be used for logging or monitoring purposes. + */ +public class OperationMetrics { + + public Map metrics = new HashMap<>(); + + private final ClientStatisticsHolder clientStatistics; + + public OperationMetrics(ClientStatisticsHolder clientStatisticsHolder) { + this.clientStatistics = clientStatisticsHolder; + } + + public Metric getMetric(ServerMetrics metric) { + return metrics.get(metric.getKey()); + } + + public Metric getMetric(ClientMetrics metric) { + return metrics.get(metric.getKey()); + } + + public void operationComplete(ClickHouseResponseSummary serverStats) { + for (Map.Entry sw : clientStatistics.getStopWatches().entrySet()) { + sw.getValue().stop(); + metrics.put(sw.getKey(), sw.getValue()); + } + metrics.put(ServerMetrics.NUM_ROWS_READ.getKey(), new Gauge(serverStats.getReadRows())); + metrics.put(ServerMetrics.NUM_ROWS_WRITTEN.getKey(), new Gauge(serverStats.getWrittenRows())); + metrics.put(ServerMetrics.TOTAL_ROWS_TO_READ.getKey(), new Gauge(serverStats.getTotalRowsToRead())); + metrics.put(ServerMetrics.NUM_BYTES_READ.getKey(), new Gauge(serverStats.getReadBytes())); + metrics.put(ServerMetrics.NUM_BYTES_WRITTEN.getKey(), new Gauge(serverStats.getWrittenBytes())); + metrics.put(ServerMetrics.RESULT_ROWS.getKey(), new Gauge(serverStats.getResultRows())); + metrics.put(ServerMetrics.ELAPSED_TIME.getKey(), new Gauge(serverStats.getElapsedTime())); + } + + @Override + public String toString() { + return "OperationStatistics{" + + "\"metrics\"=" + metrics + + '}'; + } +} diff --git a/client-v2/src/main/java/com/clickhouse/client/api/metrics/ServerMetrics.java b/client-v2/src/main/java/com/clickhouse/client/api/metrics/ServerMetrics.java new file mode 100644 index 000000000..f601f0d2d --- /dev/null +++ b/client-v2/src/main/java/com/clickhouse/client/api/metrics/ServerMetrics.java @@ -0,0 +1,55 @@ +package com.clickhouse.client.api.metrics; + +/** + * Stats returned by the server. + *

+ * `-1` means the value is not available. + */ +public enum ServerMetrics { + + /** + * Number of rows read by server from the storage. + */ + NUM_ROWS_READ("server.numRowsRead"), + + /** + * Number of rows written by server to the storage. + */ + NUM_ROWS_WRITTEN("server.numRowsWritten"), + + /** + * Estimated number of rows to read from the storage. + *

+ */ + TOTAL_ROWS_TO_READ("server.totalRowsToRead"), + + /** + * Number of bytes read by server from the storage. + */ + NUM_BYTES_READ("server.numBytesRead"), + + /** + * Number of bytes written by server to the storage. + */ + NUM_BYTES_WRITTEN("server.numBytesWritten"), + + /** + * Number of returned rows. + */ + RESULT_ROWS("server.resultRows"), + + /** + * Elapsed time in nanoseconds. + */ + ELAPSED_TIME("server.elapsedTime"); + + private final String key; + + ServerMetrics(String key) { + this.key = key; + } + + public String getKey() { + return key; + } +} diff --git a/client-v2/src/main/java/com/clickhouse/client/api/query/QueryResponse.java b/client-v2/src/main/java/com/clickhouse/client/api/query/QueryResponse.java index e3e339275..c4056df4e 100644 --- a/client-v2/src/main/java/com/clickhouse/client/api/query/QueryResponse.java +++ b/client-v2/src/main/java/com/clickhouse/client/api/query/QueryResponse.java @@ -1,10 +1,10 @@ package com.clickhouse.client.api.query; import com.clickhouse.client.ClickHouseClient; -import com.clickhouse.client.ClickHouseException; import com.clickhouse.client.ClickHouseResponse; import com.clickhouse.client.api.ClientException; -import com.clickhouse.client.api.OperationStatistics; +import com.clickhouse.client.api.internal.ClientStatisticsHolder; +import com.clickhouse.client.api.metrics.OperationMetrics; import com.clickhouse.data.ClickHouseFormat; import com.clickhouse.data.ClickHouseInputStream; @@ -37,18 +37,18 @@ public class QueryResponse implements AutoCloseable { private QuerySettings settings; - private OperationStatistics operationStatistics; + private OperationMetrics operationMetrics; private volatile boolean completed = false; public QueryResponse(ClickHouseClient client, Future responseRef, QuerySettings settings, ClickHouseFormat format, - OperationStatistics.ClientStatistics clientStatistics) { + ClientStatisticsHolder clientStatisticsHolder) { this.client = client; this.responseRef = responseRef; this.format = format; this.settings = settings; - this.operationStatistics = new OperationStatistics(clientStatistics); + this.operationMetrics = new OperationMetrics(clientStatisticsHolder); } /** @@ -66,8 +66,7 @@ private void makeComplete() { try { ClickHouseResponse response = responseRef.get(completeTimeout, TimeUnit.MILLISECONDS); completed = true; - operationStatistics.clientStatistics.stop("query"); - this.operationStatistics.updateServerStats(response.getSummary()); + operationMetrics.operationComplete(response.getSummary()); } catch (TimeoutException | InterruptedException | ExecutionException e) { throw new ClientException("Query request failed", e); } @@ -95,8 +94,8 @@ public ClickHouseFormat getFormat() { return format; } - public OperationStatistics getOperationStatistics() { + public OperationMetrics getMetrics() { ensureDone(); - return operationStatistics; + return operationMetrics; } } diff --git a/client-v2/src/test/java/com/clickhouse/client/insert/InsertTests.java b/client-v2/src/test/java/com/clickhouse/client/insert/InsertTests.java index 68de13c93..691d6bbf8 100644 --- a/client-v2/src/test/java/com/clickhouse/client/insert/InsertTests.java +++ b/client-v2/src/test/java/com/clickhouse/client/insert/InsertTests.java @@ -11,6 +11,9 @@ import com.clickhouse.client.api.Protocol; import com.clickhouse.client.api.insert.InsertResponse; import com.clickhouse.client.api.insert.InsertSettings; +import com.clickhouse.client.api.metrics.ClientMetrics; +import com.clickhouse.client.api.metrics.OperationMetrics; +import com.clickhouse.client.api.metrics.ServerMetrics; import org.testcontainers.shaded.org.apache.commons.lang3.RandomStringUtils; import org.testng.annotations.BeforeMethod; import org.testng.annotations.Test; @@ -63,8 +66,9 @@ public void insertSimplePOJOs() throws Exception { } InsertResponse response = client.insert(tableName, simplePOJOs, settings).get(30, TimeUnit.SECONDS); - assertEquals(simplePOJOs.size(), response.getOperationStatistics().getServerStatistics().numRowsWritten); - assertTrue(response.getOperationStatistics().getClientStatistics().getElapsedTime("insert") > 0); - assertTrue(response.getOperationStatistics().getClientStatistics().getElapsedTime("serialization") > 0); + OperationMetrics metrics = response.getMetrics(); + assertEquals(simplePOJOs.size(), metrics.getMetric(ServerMetrics.NUM_ROWS_WRITTEN).getLong()); + assertTrue(metrics.getMetric(ClientMetrics.OP_DURATION).getLong() > 0); + assertTrue(metrics.getMetric(ClientMetrics.OP_SERIALIZATION).getLong() > 0); } } diff --git a/client-v2/src/test/java/com/clickhouse/client/query/QueryTests.java b/client-v2/src/test/java/com/clickhouse/client/query/QueryTests.java index ad626ce13..2dd27a73d 100644 --- a/client-v2/src/test/java/com/clickhouse/client/query/QueryTests.java +++ b/client-v2/src/test/java/com/clickhouse/client/query/QueryTests.java @@ -11,7 +11,6 @@ import com.clickhouse.client.ClickHouseResponse; import com.clickhouse.client.api.Client; import com.clickhouse.client.api.DataTypeUtils; -import com.clickhouse.client.api.OperationStatistics; import com.clickhouse.client.api.Protocol; import com.clickhouse.client.api.data_formats.ClickHouseBinaryFormatReader; import com.clickhouse.client.api.data_formats.NativeFormatReader; @@ -20,6 +19,9 @@ import com.clickhouse.client.api.data_formats.RowBinaryWithNamesFormatReader; import com.clickhouse.client.api.insert.InsertSettings; import com.clickhouse.client.api.metadata.TableSchema; +import com.clickhouse.client.api.metrics.ClientMetrics; +import com.clickhouse.client.api.metrics.OperationMetrics; +import com.clickhouse.client.api.metrics.ServerMetrics; import com.clickhouse.client.api.query.GenericRecord; import com.clickhouse.client.api.query.NullValueException; import com.clickhouse.client.api.query.QueryResponse; @@ -898,16 +900,12 @@ public void testQueryMetrics() throws Exception { QueryResponse response = client.query("SELECT * FROM " + DATASET_TABLE + " LIMIT 3", settings).get(); // Stats should be available after the query is done - OperationStatistics stats = response.getOperationStatistics(); - OperationStatistics.ServerStatistics serverStats = stats.serverStatistics; - System.out.println("Server stats: " + serverStats); - System.out.println("Client stats: " + stats.clientStatistics); + OperationMetrics metrics = response.getMetrics(); + System.out.println("Server read rows: " + metrics.getMetric(ServerMetrics.NUM_ROWS_READ).getLong()); + System.out.println("Client stats: " + metrics.getMetric(ClientMetrics.OP_DURATION).getLong()); - Assert.assertTrue(serverStats.numBytesRead > 0); - Assert.assertEquals(serverStats.numBytesWritten, 0); - Assert.assertEquals(serverStats.numRowsRead, 10); // 10 rows in the table - Assert.assertEquals(serverStats.numRowsWritten, 0); - Assert.assertEquals(serverStats.resultRows, 3); + Assert.assertEquals(metrics.getMetric(ServerMetrics.NUM_ROWS_READ).getLong(), 10); // 10 rows in the table + Assert.assertEquals(metrics.getMetric(ServerMetrics.RESULT_ROWS).getLong(), 3); StringBuilder insertStmtBuilder = new StringBuilder(); insertStmtBuilder.append("INSERT INTO default.").append(DATASET_TABLE).append(" VALUES "); @@ -920,18 +918,13 @@ public void testQueryMetrics() throws Exception { } response = client.query(insertStmtBuilder.toString(), settings).get(); - serverStats = response.getOperationStatistics().serverStatistics; - System.out.println("Server stats: " + serverStats); - System.out.println("Client stats: " + stats.clientStatistics); - - // Server stats: ServerStatistics{"numRowsRead"=10, "numRowsWritten"=10, "totalRowsToRead"=0, "numBytesRead"=651, "numBytesWritten"=651} - Assert.assertTrue(serverStats.numBytesRead > 0); - Assert.assertTrue(serverStats.numBytesWritten > 0); - Assert.assertEquals(serverStats.numRowsRead, rowsToInsert); // 10 rows in the table - Assert.assertEquals(serverStats.numRowsWritten, rowsToInsert); // 10 rows inserted - Assert.assertEquals(serverStats.totalRowsToRead, 0); - Assert.assertEquals(serverStats.resultRows, rowsToInsert); - Assert.assertTrue(stats.clientStatistics.getElapsedTime("query") > 0); + metrics = response.getMetrics(); + System.out.println("Server read rows: " + metrics.getMetric(ServerMetrics.NUM_ROWS_READ).getLong()); + System.out.println("Client stats: " + metrics.getMetric(ClientMetrics.OP_DURATION).getLong()); + + Assert.assertEquals(metrics.getMetric(ServerMetrics.NUM_ROWS_READ).getLong(), rowsToInsert); // 10 rows in the table + Assert.assertEquals(metrics.getMetric(ServerMetrics.RESULT_ROWS).getLong(), rowsToInsert); + Assert.assertTrue(metrics.getMetric(ClientMetrics.OP_DURATION).getLong() > 0); } private final static List DATASET_COLUMNS = Arrays.asList( From 93ad43bc3ec5b63741e4cb64706449d3a14bcaf9 Mon Sep 17 00:00:00 2001 From: Sergey Chernov Date: Wed, 5 Jun 2024 23:12:59 -0700 Subject: [PATCH 2/4] fix client example --- .../java/com/clickhouse/examples/client_v2/SimpleReader.java | 3 ++- .../com/clickhouse/examples/client_v2/Stream2DbWriter.java | 4 +++- 2 files changed, 5 insertions(+), 2 deletions(-) diff --git a/examples/client-v2/src/main/java/com/clickhouse/examples/client_v2/SimpleReader.java b/examples/client-v2/src/main/java/com/clickhouse/examples/client_v2/SimpleReader.java index 866ec3c77..561f17c95 100644 --- a/examples/client-v2/src/main/java/com/clickhouse/examples/client_v2/SimpleReader.java +++ b/examples/client-v2/src/main/java/com/clickhouse/examples/client_v2/SimpleReader.java @@ -3,6 +3,7 @@ import com.clickhouse.client.api.Client; import com.clickhouse.client.api.data_formats.ClickHouseBinaryFormatReader; import com.clickhouse.client.api.data_formats.RowBinaryWithNamesAndTypesFormatReader; +import com.clickhouse.client.api.metrics.ClientMetrics; import com.clickhouse.client.api.query.QueryResponse; import com.clickhouse.client.api.query.QuerySettings; import com.clickhouse.data.ClickHouseFormat; @@ -72,7 +73,7 @@ public void readData() { log.info("id: {}, title: {}, url: {}", id, title, url); } - log.info("Data read successfully: {}", response.getOperationStatistics()); + log.info("Data read successfully: {} ms", response.getMetrics().getMetric(ClientMetrics.OP_DURATION).getLong()); } catch (Exception e) { log.error("Failed to read data", e); } diff --git a/examples/client-v2/src/main/java/com/clickhouse/examples/client_v2/Stream2DbWriter.java b/examples/client-v2/src/main/java/com/clickhouse/examples/client_v2/Stream2DbWriter.java index 7868106d0..175a40e0f 100644 --- a/examples/client-v2/src/main/java/com/clickhouse/examples/client_v2/Stream2DbWriter.java +++ b/examples/client-v2/src/main/java/com/clickhouse/examples/client_v2/Stream2DbWriter.java @@ -3,6 +3,7 @@ import com.clickhouse.client.api.Client; import com.clickhouse.client.api.insert.InsertResponse; import com.clickhouse.client.api.insert.InsertSettings; +import com.clickhouse.client.api.metrics.ServerMetrics; import com.clickhouse.data.ClickHouseFormat; import lombok.extern.slf4j.Slf4j; @@ -67,6 +68,7 @@ public void resetTable() { * *

Any other format can be used instead of JSONEachRow, such as TabSeparated, RowBinary, etc. As data stream * is passed almost directly to a transport layer.

+ * * @param inputStream - input stream of JSONEachRow formatted data */ public void insertData_JSONEachRowFormat(InputStream inputStream) { @@ -74,7 +76,7 @@ public void insertData_JSONEachRowFormat(InputStream inputStream) { try { InsertResponse response = client.insert(TABLE_NAME, inputStream, ClickHouseFormat.JSONEachRow, insertSettings).get(3, TimeUnit.SECONDS); - log.info("Insert finished: {}", response.getOperationStatistics()); + log.info("Insert finished: {} rows written", response.getMetrics().getMetric(ServerMetrics.NUM_ROWS_WRITTEN).getLong()); } catch (Exception e) { log.error("Failed to write JSONEachRow data", e); throw new RuntimeException(e); From ad346867ef7881077bb4722f0d1d1d4b228efe59 Mon Sep 17 00:00:00 2001 From: Sergey Chernov Date: Wed, 5 Jun 2024 23:33:01 -0700 Subject: [PATCH 3/4] fix build for jdk8 --- pom.xml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pom.xml b/pom.xml index 1c44f7928..1f1369aea 100644 --- a/pom.xml +++ b/pom.xml @@ -127,7 +127,7 @@ 3.1.0 3.6.0 3.3.1 - 3.11.0 + 3.13.0 3.1.1 3.3.0 3.1.0 From 9685e5fa4c5237236cdfde42843c19eaafdf876d Mon Sep 17 00:00:00 2001 From: Sergey Chernov Date: Thu, 6 Jun 2024 12:29:49 -0700 Subject: [PATCH 4/4] added alias methods for server stats --- .../client/api/insert/InsertResponse.java | 54 +++++++++++++++++++ .../client/api/query/QueryResponse.java | 54 +++++++++++++++++++ .../clickhouse/client/insert/InsertTests.java | 1 + .../clickhouse/client/query/QueryTests.java | 2 + 4 files changed, 111 insertions(+) diff --git a/client-v2/src/main/java/com/clickhouse/client/api/insert/InsertResponse.java b/client-v2/src/main/java/com/clickhouse/client/api/insert/InsertResponse.java index 85ff85b1d..72582bb1e 100644 --- a/client-v2/src/main/java/com/clickhouse/client/api/insert/InsertResponse.java +++ b/client-v2/src/main/java/com/clickhouse/client/api/insert/InsertResponse.java @@ -4,6 +4,7 @@ import com.clickhouse.client.ClickHouseResponse; import com.clickhouse.client.api.internal.ClientStatisticsHolder; import com.clickhouse.client.api.metrics.OperationMetrics; +import com.clickhouse.client.api.metrics.ServerMetrics; public class InsertResponse implements AutoCloseable { private final ClickHouseResponse responseRef; @@ -28,7 +29,60 @@ public void close() { } } + /** + * Returns the metrics of this operation. + * + * @return metrics of this operation + */ public OperationMetrics getMetrics() { return operationMetrics; } + + /** + * Alias for {@link ServerMetrics#NUM_ROWS_READ} + * @return number of rows read by server from the storage + */ + public long getReadRows() { + return operationMetrics.getMetric(ServerMetrics.NUM_ROWS_READ).getLong(); + } + + /** + * Alias for {@link ServerMetrics#NUM_BYTES_READ} + * @return number of bytes read by server from the storage + */ + public long getReadBytes() { + return operationMetrics.getMetric(ServerMetrics.NUM_BYTES_READ).getLong(); + } + + /** + * Alias for {@link ServerMetrics#NUM_ROWS_WRITTEN} + * @return number of rows written by server to the storage + */ + public long getWrittenRows() { + return operationMetrics.getMetric(ServerMetrics.NUM_ROWS_WRITTEN).getLong(); + } + + /** + * Alias for {@link ServerMetrics#NUM_BYTES_WRITTEN} + * @return number of bytes written by server to the storage + */ + public long getWrittenBytes() { + return operationMetrics.getMetric(ServerMetrics.NUM_BYTES_WRITTEN).getLong(); + } + + /** + * Alias for {@link ServerMetrics#ELAPSED_TIME} + * @return elapsed time in nanoseconds + */ + public long getServerTime() { + return operationMetrics.getMetric(ServerMetrics.ELAPSED_TIME).getLong(); + } + + /** + * Alias for {@link ServerMetrics#RESULT_ROWS} + * @return number of returned rows + */ + public long getResultRows() { + return operationMetrics.getMetric(ServerMetrics.RESULT_ROWS).getLong(); + } } diff --git a/client-v2/src/main/java/com/clickhouse/client/api/query/QueryResponse.java b/client-v2/src/main/java/com/clickhouse/client/api/query/QueryResponse.java index c4056df4e..78dd026e6 100644 --- a/client-v2/src/main/java/com/clickhouse/client/api/query/QueryResponse.java +++ b/client-v2/src/main/java/com/clickhouse/client/api/query/QueryResponse.java @@ -5,6 +5,7 @@ import com.clickhouse.client.api.ClientException; import com.clickhouse.client.api.internal.ClientStatisticsHolder; import com.clickhouse.client.api.metrics.OperationMetrics; +import com.clickhouse.client.api.metrics.ServerMetrics; import com.clickhouse.data.ClickHouseFormat; import com.clickhouse.data.ClickHouseInputStream; @@ -94,8 +95,61 @@ public ClickHouseFormat getFormat() { return format; } + /** + * Returns the metrics of this operation. + * + * @return metrics of this operation + */ public OperationMetrics getMetrics() { ensureDone(); return operationMetrics; } + + /** + * Alias for {@link ServerMetrics#NUM_ROWS_READ} + * @return number of rows read by server from the storage + */ + public long getReadRows() { + return operationMetrics.getMetric(ServerMetrics.NUM_ROWS_READ).getLong(); + } + + /** + * Alias for {@link ServerMetrics#NUM_BYTES_READ} + * @return number of bytes read by server from the storage + */ + public long getReadBytes() { + return operationMetrics.getMetric(ServerMetrics.NUM_BYTES_READ).getLong(); + } + + /** + * Alias for {@link ServerMetrics#NUM_ROWS_WRITTEN} + * @return number of rows written by server to the storage + */ + public long getWrittenRows() { + return operationMetrics.getMetric(ServerMetrics.NUM_ROWS_WRITTEN).getLong(); + } + + /** + * Alias for {@link ServerMetrics#NUM_BYTES_WRITTEN} + * @return number of bytes written by server to the storage + */ + public long getWrittenBytes() { + return operationMetrics.getMetric(ServerMetrics.NUM_BYTES_WRITTEN).getLong(); + } + + /** + * Alias for {@link ServerMetrics#ELAPSED_TIME} + * @return elapsed time in nanoseconds + */ + public long getServerTime() { + return operationMetrics.getMetric(ServerMetrics.ELAPSED_TIME).getLong(); + } + + /** + * Alias for {@link ServerMetrics#RESULT_ROWS} + * @return number of returned rows + */ + public long getResultRows() { + return operationMetrics.getMetric(ServerMetrics.RESULT_ROWS).getLong(); + } } diff --git a/client-v2/src/test/java/com/clickhouse/client/insert/InsertTests.java b/client-v2/src/test/java/com/clickhouse/client/insert/InsertTests.java index 691d6bbf8..3d4e63020 100644 --- a/client-v2/src/test/java/com/clickhouse/client/insert/InsertTests.java +++ b/client-v2/src/test/java/com/clickhouse/client/insert/InsertTests.java @@ -68,6 +68,7 @@ public void insertSimplePOJOs() throws Exception { OperationMetrics metrics = response.getMetrics(); assertEquals(simplePOJOs.size(), metrics.getMetric(ServerMetrics.NUM_ROWS_WRITTEN).getLong()); + assertEquals(simplePOJOs.size(), response.getWrittenRows()); assertTrue(metrics.getMetric(ClientMetrics.OP_DURATION).getLong() > 0); assertTrue(metrics.getMetric(ClientMetrics.OP_SERIALIZATION).getLong() > 0); } diff --git a/client-v2/src/test/java/com/clickhouse/client/query/QueryTests.java b/client-v2/src/test/java/com/clickhouse/client/query/QueryTests.java index 2dd27a73d..d70bcf2f4 100644 --- a/client-v2/src/test/java/com/clickhouse/client/query/QueryTests.java +++ b/client-v2/src/test/java/com/clickhouse/client/query/QueryTests.java @@ -924,7 +924,9 @@ public void testQueryMetrics() throws Exception { Assert.assertEquals(metrics.getMetric(ServerMetrics.NUM_ROWS_READ).getLong(), rowsToInsert); // 10 rows in the table Assert.assertEquals(metrics.getMetric(ServerMetrics.RESULT_ROWS).getLong(), rowsToInsert); + Assert.assertEquals(response.getReadRows(), rowsToInsert); Assert.assertTrue(metrics.getMetric(ClientMetrics.OP_DURATION).getLong() > 0); + } private final static List DATASET_COLUMNS = Arrays.asList(