Merge pull request #1278 from zhicwu/main

Fix transaction failure since 0.4.0 and enable more tests

Author: zhicwu

.github/workflows/build.yml

@@ -176,7 +176,7 @@ jobs:
       matrix:
         # most recent LTS releases as well as latest stable builds
         # https://github.com/ClickHouse/ClickHouse/pulls?q=is%3Aopen+is%3Apr+label%3Arelease
-        clickhouse: ["21.8", "22.3", "22.8", "latest"]
+        clickhouse: ["22.3", "22.8", "latest"]
       fail-fast: false
     timeout-minutes: 15
     name: Java client + CH ${{ matrix.clickhouse }}
@@ -231,12 +231,9 @@ jobs:
     needs: compile
     strategy:
       matrix:
-        clickhouse: ["21.8", "22.3", "22.8", "latest"]
+        clickhouse: ["22.3", "22.8", "latest"]
         # here http, http_client and apache_http_client represent different value of http_connection_provider
         protocol: ["http", "http_client", "apache_http_client", "grpc"]
-        exclude:
-          - clickhouse: "21.8"
-            protocol: grpc
       fail-fast: false
     timeout-minutes: 15
     name: JDBC driver + CH ${{ matrix.clickhouse }} (${{ matrix.protocol }})
@@ -293,7 +290,7 @@ jobs:
     needs: compile
     strategy:
       matrix:
-        clickhouse: ["21.8", "22.3", "22.8", "latest"]
+        clickhouse: ["22.3", "22.8", "latest"]
         # grpc is not fully supported, and http_client and apache_http_client do not work in CI environment(due to limited threads?)
         protocol: ["http"]
         r2dbc: ["1.0.0.RELEASE", "0.9.1.RELEASE"]

CHANGELOG.md

@@ -3,6 +3,7 @@
 ### Bug Fixes
 * error while converting Nested values to Java maps.
 * incorrect algorithm extracted from PEM. [#1274](https://github.com/ClickHouse/clickhouse-java/issues/1274)
+* transaction failure introduced 0.4.0.
 
 ## 0.4.1, 2023-02-19
 ### Breaking Changes

clickhouse-cli-client/src/test/java/com/clickhouse/client/cli/ClickHouseCommandLineClientTest.java

@@ -135,4 +135,52 @@ public void testSessionLock() {
     public void testTempTable() {
         throw new SkipException("Skip due to session is not supported");
     }
+
+    @Test(groups = { "integration" })
+    @Override
+    public void testAbortTransaction() throws ClickHouseException {
+        throw new SkipException("Skip due to session is not supported");
+    }
+
+    @Test(groups = { "integration" })
+    @Override
+    public void testCommitTransaction() throws ClickHouseException {
+        throw new SkipException("Skip due to session is not supported");
+    }
+
+    @Test(groups = { "integration" })
+    @Override
+    public void testImplicitTransaction() throws ClickHouseException {
+        throw new SkipException("Skip due to session is not supported");
+    }
+
+    @Test(groups = { "integration" })
+    @Override
+    public void testJoinTransaction() throws ClickHouseException {
+        throw new SkipException("Skip due to session is not supported");
+    }
+
+    @Test(groups = { "integration" })
+    @Override
+    public void testNewTransaction() throws ClickHouseException {
+        throw new SkipException("Skip due to session is not supported");
+    }
+
+    @Test(groups = { "integration" })
+    @Override
+    public void testRollbackTransaction() throws ClickHouseException {
+        throw new SkipException("Skip due to session is not supported");
+    }
+
+    @Test(groups = { "integration" })
+    @Override
+    public void testTransactionSnapshot() throws ClickHouseException {
+        throw new SkipException("Skip due to session is not supported");
+    }
+
+    @Test(groups = { "integration" })
+    @Override
+    public void testTransactionTimeout() throws ClickHouseException {
+        throw new SkipException("Skip due to session is not supported");
+    }
 }

clickhouse-client/src/main/java/com/clickhouse/client/AbstractClient.java

@@ -1,7 +1,9 @@
 package com.clickhouse.client;
 
 import java.io.IOException;
+import java.io.Serializable;
 import java.util.Collection;
+import java.util.Map;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionException;
 import java.util.concurrent.ExecutorService;
@@ -11,6 +13,7 @@
 
 import com.clickhouse.client.config.ClickHouseClientOption;
 import com.clickhouse.client.config.ClickHouseHealthCheckMethod;
+import com.clickhouse.config.ClickHouseOption;
 import com.clickhouse.data.ClickHouseChecker;
 import com.clickhouse.data.ClickHouseUtils;
 import com.clickhouse.logging.Logger;
@@ -212,33 +215,12 @@ public boolean accept(ClickHouseProtocol protocol) {
     }
 
     @Override
-    public ClickHouseRequest<?> connect(ClickHouseNode node) {
+    public ClickHouseRequest<?> read(Function<ClickHouseNodeSelector, ClickHouseNode> nodeFunc,
+            Map<ClickHouseOption, Serializable> options) {
         lock.readLock().lock();
         try {
             ensureInitialized();
-            return ClickHouseClient.super.connect(node);
-        } finally {
-            lock.readLock().unlock();
-        }
-    }
-
-    @Override
-    public ClickHouseRequest<?> connect(ClickHouseNodes nodes) {
-        lock.readLock().lock();
-        try {
-            ensureInitialized();
-            return ClickHouseClient.super.connect(nodes);
-        } finally {
-            lock.readLock().unlock();
-        }
-    }
-
-    @Override
-    public ClickHouseRequest<?> connect(Function<ClickHouseNodeSelector, ClickHouseNode> nodeFunc) {
-        lock.readLock().lock();
-        try {
-            ensureInitialized();
-            return ClickHouseClient.super.connect(nodeFunc);
+            return ClickHouseClient.super.read(nodeFunc, options);
         } finally {
             lock.readLock().unlock();
         }

clickhouse-client/src/main/java/com/clickhouse/client/ClickHouseClient.java

@@ -3,6 +3,7 @@
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
+import java.io.Serializable;
 import java.io.UncheckedIOException;
 import java.util.ArrayList;
 import java.util.List;
@@ -658,11 +659,13 @@ static ClickHouseClient newInstance(ClickHouseProtocol... preferredProtocols) {
     }
 
     /**
-     * Creates a new instance with default credentials compatible with any of the given protocols.
+     * Creates a new instance with default credentials compatible with any of the
+     * given protocols.
      *
      * @param defaultCredentials default credentials
      * @param preferredProtocols preferred protocols
-     * @return new instance compatible with any of the given protocols using default credentials
+     * @return new instance compatible with any of the given protocols using default
+     *         credentials
      */
     static ClickHouseClient newInstance(ClickHouseCredentials defaultCredentials,
             ClickHouseProtocol... preferredProtocols) {
@@ -946,7 +949,29 @@ default ClickHouseRequest<?> connect(ClickHouseNode node) {
     }
 
     /**
-     * Reads from one or more ClickHouse servers. Same as
+     * Connects to a ClickHouse server defined by the given
+     * {@link java.util.function.Function}. You can pass either
+     * {@link ClickHouseCluster}, {@link ClickHouseNodes} or {@link ClickHouseNode}
+     * here, as all of them implemented the same interface.
+     *
+     * <p>
+     * Please be aware that this is nothing but an intention, so no network
+     * communication happens until {@link #execute(ClickHouseRequest)} is
+     * invoked(usually triggered by {@code request.execute()}).
+     *
+     * @param nodeFunc function to get a {@link ClickHouseNode} to connect to
+     * @return non-null request object holding references to this client and node
+     *         provider
+     * @deprecated will be dropped in 0.5, please use {@link #read(Function, Map)}
+     *             instead
+     */
+    @Deprecated
+    default ClickHouseRequest<?> connect(Function<ClickHouseNodeSelector, ClickHouseNode> nodeFunc) {
+        return read(nodeFunc, null);
+    }
+
+    /**
+     * Connects to one or more ClickHouse servers to read. Same as
      * {@code read(ClickHouseNodes.of(uri))}.
      *
      * @param endpoints non-empty connection string separated by comma
@@ -958,55 +983,46 @@ default ClickHouseRequest<?> read(String endpoints) {
     }
 
     /**
-     * Reads from a list of managed ClickHouse servers.
+     * Connects to a list of managed ClickHouse servers to read.
      *
      * @param nodes non-null list of servers for read
      * @return non-null request object holding references to this client and node
      *         provider
      */
     default ClickHouseRequest<?> read(ClickHouseNodes nodes) {
-        return new ClickHouseRequest<>(this, ClickHouseChecker.nonNull(nodes, "Nodes"),
-                null, nodes.template.config.getAllOptions(), false);
+        return read(nodes, nodes.template.config.getAllOptions());
     }
 
     /**
-     * Reads from a ClickHouse server.
+     * Connects to a ClickHouse server to read.
      *
      * @param node non-null server for read
      * @return non-null request object holding references to this client and server
      */
     default ClickHouseRequest<?> read(ClickHouseNode node) {
-        return new ClickHouseRequest<>(this, ClickHouseChecker.nonNull(node, "Node"), null, node.config.getAllOptions(),
-                false);
+        return read(node, node.config.getAllOptions());
     }
 
     /**
-     * Writes into a ClickHouse server.
+     * Connects to a ClickHouse server to read.
      *
-     * @param server non-null server for write
+     * @param nodeFunc function to get a {@link ClickHouseNode} to connect to
+     * @param options  optional client options for connecting to the server
      * @return non-null request object holding references to this client and server
      */
-    default Mutation write(ClickHouseNode server) {
-        return read(server).write();
+    default ClickHouseRequest<?> read(Function<ClickHouseNodeSelector, ClickHouseNode> nodeFunc,
+            Map<ClickHouseOption, Serializable> options) {
+        return new ClickHouseRequest<>(this, ClickHouseChecker.nonNull(nodeFunc, "Node"), null, options, false);
     }
 
     /**
-     * Connects to a ClickHouse server defined by the given
-     * {@link java.util.function.Function}. You can pass either
-     * {@link ClickHouseCluster}, {@link ClickHouseNodes} or {@link ClickHouseNode}
-     * here, as all of them implemented the same interface.
-     *
-     * <p>
-     * Please be aware that this is nothing but an intention, so no network
-     * communication happens until {@link #execute(ClickHouseRequest)} is
-     * invoked(usually triggered by {@code request.execute()}).
+     * Writes into a ClickHouse server.
      *
-     * @param nodeFunc function to get a {@link ClickHouseNode} to connect to
-     * @return non-null request object holding references to this client and node
-     *         provider
+     * @param node non-null server for write
+     * @return non-null request object holding references to this client and server
      */
-    default ClickHouseRequest<?> connect(Function<ClickHouseNodeSelector, ClickHouseNode> nodeFunc) {
-        return new ClickHouseRequest<>(this, ClickHouseChecker.nonNull(nodeFunc, "nodeFunc"), null, null, false);
+    default Mutation write(ClickHouseNode node) {
+        return read(node).write();
     }
 
     /**

clickhouse-client/src/main/java/com/clickhouse/client/ClickHouseTransaction.java

@@ -12,6 +12,7 @@
 import com.clickhouse.data.ClickHouseFormat;
 import com.clickhouse.data.ClickHouseRecord;
 import com.clickhouse.data.ClickHouseUtils;
+import com.clickhouse.data.value.UnsignedLong;
 import com.clickhouse.logging.Logger;
 import com.clickhouse.logging.LoggerFactory;
 
@@ -42,8 +43,8 @@ public static XID of(List<?> list) {
                 throw new IllegalArgumentException(
                         "Non-null tuple with 3 elements(long, long, String) is required");
             }
-            long snapshotVersion = (Long) list.get(0);
-            long localTxCounter = (Long) list.get(1);
+            long snapshotVersion = ((UnsignedLong) list.get(0)).longValue();
+            long localTxCounter = ((UnsignedLong) list.get(1)).longValue();
             String hostId = String.valueOf(list.get(2));
             if (EMPTY.snapshotVersion == snapshotVersion && EMPTY.localTxCounter == localTxCounter
                     && EMPTY.hostId.equals(hostId)) {
@@ -278,7 +279,7 @@ protected final ClickHouseRecord issue(String command) throws ClickHouseExceptio
     protected ClickHouseRecord issue(String command, boolean sessionCheck, Map<String, Serializable> settings)
             throws ClickHouseException {
         ClickHouseRecord result = ClickHouseRecord.EMPTY;
-        try (ClickHouseResponse response = ClickHouseClient.newInstance(server.getProtocol()).connect(server)
+        try (ClickHouseResponse response = ClickHouseClient.newInstance(server.getProtocol()).read(server)
                 .format(ClickHouseFormat.RowBinaryWithNamesAndTypes)
                 .settings(settings).session(sessionId, sessionCheck, timeout > 0 ? timeout : null)
                 .query(command).executeAndWait()) {