Merge branch 'main' into adjusting-kafka-feedback

Author: Paultagoras

client-v2/pom.xml

@@ -142,6 +142,7 @@
                             <version>1.18.32</version>
                         </path>
                     </annotationProcessorPaths>
+                    <release>8</release>
                 </configuration>
             </plugin>
             <plugin>

client-v2/src/main/java/com/clickhouse/client/api/Client.java

@@ -23,6 +23,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;
@@ -106,7 +108,7 @@ public class Client {
     private static final Logger LOG = LoggerFactory.getLogger(Client.class);
     private ExecutorService queryExecutor;
 
-    private Map<String, OperationStatistics.ClientStatistics> globalClientStats = new ConcurrentHashMap<>();
+    private Map<String, ClientStatisticsHolder> globalClientStats = new ConcurrentHashMap<>();
 
     private Client(Set<String> endpoints, Map<String,String> configuration) {
         this.endpoints = endpoints;
@@ -540,7 +542,7 @@ public CompletableFuture<InsertResponse> 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");
@@ -574,7 +576,7 @@ public CompletableFuture<InsertResponse> 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);
     }
@@ -609,8 +611,8 @@ public CompletableFuture<InsertResponse> 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<InsertResponse> responseFuture = new CompletableFuture<>();
 
@@ -638,7 +640,6 @@ public CompletableFuture<InsertResponse> 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"));
         }
 
@@ -706,8 +707,8 @@ public CompletableFuture<QueryResponse> query(String sqlQuery, Map<String, Objec
         if (settings.getFormat() == null) {
             settings.setFormat(ClickHouseFormat.RowBinaryWithNamesAndTypes);
         }
-        OperationStatistics.ClientStatistics clientStats = new OperationStatistics.ClientStatistics();
-        clientStats.start("query");
+        ClientStatisticsHolder clientStats = new ClientStatisticsHolder();
+        clientStats.start(ClientMetrics.OP_DURATION);
         ClickHouseClient client = createClient();
         ClickHouseRequest<?> request = client.read(getServerNode());
 
@@ -762,7 +763,7 @@ public CompletableFuture<Records> 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());
@@ -876,7 +877,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;
     }
 

client-v2/src/main/java/com/clickhouse/client/api/OperationStatistics.java

@@ -2,25 +2,22 @@
 
 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;
+import com.clickhouse.client.api.metrics.ServerMetrics;
 
 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 +29,60 @@ public void close() {
         }
     }
 
-    public OperationStatistics getOperationStatistics() {
-        return operationStatistics;
+    /**
+     * 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();
     }
 }

client-v2/src/main/java/com/clickhouse/client/api/insert/InsertResponse.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<String, StopWatch> 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<String, StopWatch> getStopWatches() {
+        return stopWatches;
+    }
+
+    @Override
+    public String toString() {
+        return "ClientStatistics{" +
+                "\"spans\"=" + stopWatches +
+                '}';
+    }
+}