Merge pull request #1000 from zhicwu/develop

Use unbounded queue for batch insert

Author: zhicwu

clickhouse-jdbc/src/main/java/com/clickhouse/jdbc/internal/InputBasedPreparedStatement.java

@@ -77,7 +77,8 @@ protected InputBasedPreparedStatement(ClickHouseConnectionImpl connection, Click
 
         counter = 0;
         // it's important to make sure the queue has unlimited length
-        stream = ClickHouseDataStreamFactory.getInstance().createPipedOutputStream(config, null);
+        stream = ClickHouseDataStreamFactory.getInstance().createPipedOutputStream(config.getWriteBufferSize(), 0,
+                config.getSocketTimeout(), null);
     }
 
     protected void ensureParams() throws SQLException {
@@ -350,7 +351,10 @@ public void clearBatch() throws SQLException {
             // ignore
         }
         counter = 0;
-        stream = ClickHouseDataStreamFactory.getInstance().createPipedOutputStream(getConfig(), null);
+
+        ClickHouseConfig config = getConfig();
+        stream = ClickHouseDataStreamFactory.getInstance().createPipedOutputStream(config.getWriteBufferSize(), 0,
+                config.getSocketTimeout(), null);
     }
 
     @Override

clickhouse-jdbc/src/test/java/com/clickhouse/jdbc/ClickHousePreparedStatementTest.java

@@ -710,6 +710,49 @@ public void testBatchInsert() throws SQLException {
         }
     }
 
+    @Test(groups = "integration")
+    public void testBatchInsertWithoutUnboundedQueue() throws Exception {
+        Properties props = new Properties();
+        props.setProperty(ClickHouseClientOption.WRITE_BUFFER_SIZE.getKey(), "1");
+        props.setProperty(ClickHouseClientOption.MAX_QUEUED_BUFFERS.getKey(), "1");
+        try (ClickHouseConnection conn = newConnection(new Properties());
+                Statement s = conn.createStatement()) {
+            s.execute("drop table if exists test_insert_buffer_size; "
+                    + "CREATE TABLE test_insert_buffer_size(value String) ENGINE=Memory");
+            try (PreparedStatement ps = conn.prepareStatement(
+                    "INSERT INTO test_insert_buffer_size")) {
+                ps.setString(1, "1");
+                ps.addBatch();
+                ps.setString(1, "2");
+                ps.addBatch();
+                ps.setString(1, "3");
+                ps.addBatch();
+                ps.executeBatch();
+
+                ps.setString(1, "4");
+                ps.addBatch();
+                ps.executeBatch();
+
+                ps.setString(1, "4");
+                ps.addBatch();
+                ps.clearBatch();
+                ps.setString(1, "5");
+                ps.addBatch();
+                ps.setString(1, "6");
+                ps.addBatch();
+                ps.executeBatch();
+            }
+
+            try (ResultSet rs = s.executeQuery("select * from test_insert_buffer_size order by value")) {
+                int count = 1;
+                while (rs.next()) {
+                    Assert.assertEquals(rs.getInt(1), count++);
+                }
+                Assert.assertEquals(count, 7);
+            }
+        }
+    }
+
     @Test(groups = "integration")
     public void testQueryWithDateTime() throws SQLException {
         try (ClickHouseConnection conn = newConnection(new Properties());