added some tests

Author: chernser

client-v2/src/test/java/com/clickhouse/client/HttpTransportTests.java

@@ -223,7 +223,7 @@ public void testSecureConnection() {
                 .setUsername("default")
                 .setPassword("")
                 .setRootCertificate("containers/clickhouse-server/certs/localhost.crt")
-                .useNewImplementation(System.getProperty("client.tests.useNewImplementation", "true").equals("true"))
+                .compressClientRequest(true)
                 .build()) {
 
             List<GenericRecord> records = client.queryAll("SELECT timezone()");

client-v2/src/test/java/com/clickhouse/client/insert/InsertClientContentCompressionTests.java

@@ -1,7 +1,77 @@
 package com.clickhouse.client.insert;
 
+import com.clickhouse.client.ClickHouseNode;
+import com.clickhouse.client.ClickHouseProtocol;
+import com.clickhouse.client.api.Client;
+import com.clickhouse.client.api.data_formats.ClickHouseBinaryFormatReader;
+import com.clickhouse.client.api.insert.InsertResponse;
+import com.clickhouse.client.api.insert.InsertSettings;
+import com.clickhouse.client.api.query.GenericRecord;
+import com.clickhouse.client.api.query.QueryResponse;
+import org.testcontainers.shaded.org.apache.commons.lang3.RandomStringUtils;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+
 public class InsertClientContentCompressionTests extends InsertTests {
     public InsertClientContentCompressionTests() {
         super(true, false);
     }
+
+
+    @Test(groups = { "integration" })
+    public void testInsertAndReadBackWithSecureConnection() {
+        ClickHouseNode secureServer = getSecureServer(ClickHouseProtocol.HTTP);
+
+        try (Client client = new Client.Builder()
+                .addEndpoint("https://localhost:" + secureServer.getPort())
+                .setUsername("default")
+                .setPassword("")
+                .setRootCertificate("containers/clickhouse-server/certs/localhost.crt")
+                .compressClientRequest(true)
+                .build()) {
+            final String tableName = "single_pojo_table";
+            final String createSQL = SamplePOJO.generateTableCreateSQL(tableName);
+            final SamplePOJO pojo = new SamplePOJO();
+
+            initTable(tableName, createSQL);
+
+            client.register(SamplePOJO.class, client.getTableSchema(tableName, "default"));
+            InsertSettings settings = new InsertSettings()
+                    .setDeduplicationToken(RandomStringUtils.randomAlphabetic(36))
+                    .setQueryId(String.valueOf(UUID.randomUUID()));
+            System.out.println("Inserting POJO: " + pojo);
+            try (InsertResponse response = client.insert(tableName, Collections.singletonList(pojo), settings).get(10, TimeUnit.SECONDS)) {
+                Assert.assertEquals(response.getWrittenRows(), 1);
+            }
+
+            try (QueryResponse queryResponse =
+                         client.query("SELECT * FROM " + tableName + " LIMIT 1").get(10, TimeUnit.SECONDS)) {
+
+                ClickHouseBinaryFormatReader reader = client.newBinaryFormatReader(queryResponse);
+                Assert.assertNotNull(reader.next());
+
+                Assert.assertEquals(reader.getByte("byteValue"), pojo.getByteValue());
+                Assert.assertEquals(reader.getByte("int8"), pojo.getInt8());
+                Assert.assertEquals(reader.getShort("uint8"), pojo.getUint8());
+                Assert.assertEquals(reader.getShort("int16"), pojo.getInt16());
+                Assert.assertEquals(reader.getInteger("int32"), pojo.getInt32());
+                Assert.assertEquals(reader.getLong("int64"), pojo.getInt64());
+                Assert.assertEquals(reader.getFloat("float32"), pojo.getFloat32());
+                Assert.assertEquals(reader.getDouble("float64"), pojo.getFloat64());
+                Assert.assertEquals(reader.getString("string"), pojo.getString());
+                Assert.assertEquals(reader.getString("fixedString"), pojo.getFixedString());
+            }
+            List<GenericRecord> records = client.queryAll("SELECT timezone()");
+            Assert.assertTrue(records.size() > 0);
+            Assert.assertEquals(records.get(0).getString(1), "UTC");
+        } catch (Exception e) {
+            e.printStackTrace();
+            Assert.fail(e.getMessage());
+        }
+    }
 }

client-v2/src/test/java/com/clickhouse/client/insert/InsertTests.java

@@ -20,6 +20,7 @@
 import com.clickhouse.client.api.query.QuerySettings;
 import com.clickhouse.data.ClickHouseFormat;
 import com.clickhouse.data.ClickHouseVersion;
+import com.clickhouse.data.stream.ByteArrayQueueInputStream;
 import org.apache.commons.lang3.StringEscapeUtils;
 import org.testcontainers.shaded.org.apache.commons.lang3.RandomStringUtils;
 import org.testng.Assert;
@@ -37,7 +38,9 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
+import java.util.LinkedList;
 import java.util.List;
+import java.util.Queue;
 import java.util.UUID;
 import java.util.concurrent.TimeUnit;
 
@@ -66,13 +69,17 @@ public InsertTests(boolean useClientCompression, boolean useHttpCompression) {
     @BeforeMethod(groups = { "integration" })
     public void setUp() throws IOException {
         ClickHouseNode node = getServer(ClickHouseProtocol.HTTP);
+        int bufferSize = (7 * 65500);
+
         client = new Client.Builder()
                 .addEndpoint(Protocol.HTTP, node.getHost(), node.getPort(), false)
                 .setUsername("default")
                 .setPassword("")
-                .useNewImplementation(System.getProperty("client.tests.useNewImplementation", "true").equals("true"))
                 .compressClientRequest(useClientCompression)
                 .useHttpCompression(useHttpCompression)
+                .setSocketSndbuf(bufferSize)
+                .setSocketRcvbuf(bufferSize)
+                .setClientNetworkBufferSize(bufferSize)
                 .build();
         settings = new InsertSettings()
                 .setDeduplicationToken(RandomStringUtils.randomAlphabetic(36))
@@ -227,6 +234,37 @@ public void insertRawData() throws Exception {
     }
 
 
+    @Test(groups = { "integration" }, enabled = true)
+    public void insertRawDataQueued() throws Exception {
+        final String tableName = "raw_data_table";
+        final String createSQL = "CREATE TABLE " + tableName +
+                " (Id UInt32, event_ts Timestamp, name String, p1 Int64, p2 String) ENGINE = MergeTree() ORDER BY ()";
+
+        initTable(tableName, createSQL);
+        settings.setInputStreamCopyBufferSize(8198 * 2);
+        settings.compressClientRequest(true);
+        Queue<byte[]> queue = new LinkedList<>();
+        ByteArrayQueueInputStream qIn = new ByteArrayQueueInputStream(queue, null);
+        for (int i = 0; i < 10; i++) {
+            if (i > 2 && i < 5) {
+                queue.add(new byte[0]);
+            } else {
+                queue.add(String.format("{ \"Id\": %d, \"events_ts\": \"%s\", \"name\": \"%s\", \"p1\": \"%d\", \"p2\": \"%s\"}\n", i, "2021-01-01 00:00:00", "name" + i, i, "p2").getBytes());
+            }
+        }
+        InsertResponse response = client.insert(tableName, qIn,
+                ClickHouseFormat.JSONEachRow, settings).get(30, TimeUnit.SECONDS);
+
+        assertEquals((int)response.getWrittenRows(), 10 );
+
+        List<GenericRecord> records = client.queryAll("SELECT * FROM " + tableName);
+        assertEquals(records.size(), 10);
+        for (GenericRecord record : records) {
+            System.out.println(record.getString(1) + " " +record.getString(2) + " " +record.getString(3) + " " +record.getString(4) + " " +record.getString(5) + " ");
+        }
+    }
+
+
     @Test(groups = { "integration" }, enabled = true)
     public void insertRawDataSimple() throws Exception {
         insertRawDataSimple(1000);
@@ -379,7 +417,7 @@ public static Object[] logCommentDataProvider() {
         };
     }
 
-    private void initTable(String tableName, String createTableSQL) throws Exception {
+    protected void initTable(String tableName, String createTableSQL) throws Exception {
         client.execute("DROP TABLE IF EXISTS " + tableName).get(EXECUTE_CMD_TIMEOUT, TimeUnit.SECONDS);
         client.execute(createTableSQL).get(EXECUTE_CMD_TIMEOUT, TimeUnit.SECONDS);
     }