Fix #909 and bump version

README.md

@@ -115,7 +115,7 @@ Note: in general, the new driver(v0.3.2) is a few times faster with less memory
     <groupId>com.clickhouse</groupId>
     <!-- or clickhouse-grpc-client if you prefer gRPC -->
     <artifactId>clickhouse-http-client</artifactId>
-    <version>0.3.2-patch8</version>
+    <version>0.3.2-patch9</version>
 </dependency>
 ```
 
@@ -151,7 +151,7 @@ try (ClickHouseClient client = ClickHouseClient.newInstance(preferredProtocol);
     <!-- will stop using ru.yandex.clickhouse starting from 0.4.0  -->
     <groupId>com.clickhouse</groupId>
     <artifactId>clickhouse-jdbc</artifactId>
-    <version>0.3.2-patch8</version>
+    <version>0.3.2-patch9</version>
     <!-- below is only needed when all you want is a shaded jar -->
     <classifier>http</classifier>
     <exclusions>

clickhouse-client/README.md

@@ -9,7 +9,7 @@ Async Java client for ClickHouse. `clickhouse-client` is an abstract module, so
 <dependency>
     <groupId>com.clickhouse</groupId>
     <artifactId>clickhouse-http-client</artifactId>
-    <version>0.3.2-patch8</version>
+    <version>0.3.2-patch9</version>
 </dependency>
 ```
 

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

@@ -51,7 +51,7 @@ public ClickHouseDataProcessor getProcessor(ClickHouseConfig config, ClickHouseI
             ClickHouseOutputStream output, Map<String, Object> settings, List<ClickHouseColumn> columns)
             throws IOException {
         ClickHouseFormat format = ClickHouseChecker.nonNull(config, "config").getFormat();
-        ClickHouseDataProcessor processor;
+        ClickHouseDataProcessor processor = null;
         if (ClickHouseFormat.RowBinary == format || ClickHouseFormat.RowBinaryWithNamesAndTypes == format) {
             processor = new ClickHouseRowBinaryProcessor(config, input, output, columns, settings);
         } else if (ClickHouseFormat.TSVWithNames == format || ClickHouseFormat.TSVWithNamesAndTypes == format
@@ -61,10 +61,7 @@ public ClickHouseDataProcessor getProcessor(ClickHouseConfig config, ClickHouseI
         } else if (format != null && format.isText()) {
             processor = new ClickHouseTabSeparatedProcessor(config, input, output,
                     ClickHouseDataProcessor.DEFAULT_COLUMNS, settings);
-        } else {
-            throw new IllegalArgumentException(ERROR_UNSUPPORTED_FORMAT + format);
         }
-
         return processor;
     }
 

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

@@ -48,7 +48,50 @@ protected Mutation(ClickHouseRequest<?> request, boolean sealed) {
         @Override
         protected String getQuery() {
             if (input != null && sql != null) {
-                return new StringBuilder().append(sql).append(" FORMAT ").append(getInputFormat().name()).toString();
+                int tmp = 0;
+                int len = sql.length();
+                int index = len;
+                while ((tmp = ClickHouseUtils.skipContentsUntil(sql, tmp, sql.length(), "format", false)) < len) {
+                    index = tmp;
+                }
+
+                StringBuilder builder = new StringBuilder();
+                boolean spaces = false;
+                for (; index < len; index++) {
+                    char ch = sql.charAt(index);
+                    if (ClickHouseUtils.isQuote(ch) || ClickHouseUtils.isOpenBracket(ch)) {
+                        break;
+                    } else if (Character.isWhitespace(ch)) {
+                        if (builder.length() > 0) {
+                            spaces = true;
+                        }
+                    } else if (index + 1 < len) {
+                        char nextCh = sql.charAt(index + 1);
+                        if (ch == '-' && nextCh == '-') {
+                            index = ClickHouseUtils.skipSingleLineComment(sql, index + 2, len) - 1;
+                        } else if (ch == '/' && nextCh == '*') {
+                            index = ClickHouseUtils.skipMultiLineComment(sql, index + 2, len) - 1;
+                        } else if (ch == '\\') {
+                            index++;
+                        } else {
+                            if (spaces) {
+                                break;
+                            } else {
+                                builder.append(ch);
+                            }
+                        }
+                    } else {
+                        if (spaces) {
+                            break;
+                        } else {
+                            builder.append(ch);
+                        }
+                    }
+                }
+
+                return builder.length() > 0 && index == len ? sql
+                        : new StringBuilder().append(sql).append("\n FORMAT ").append(getInputFormat().name())
+                                .toString();
             }
 
             return super.getQuery();

clickhouse-client/src/main/java/com/clickhouse/client/stream/BlockingPipedOutputStream.java

@@ -102,7 +102,10 @@ public void close() throws IOException {
             Thread.currentThread().interrupt();
             throw new IOException("Thread was interrupted when putting EMPTY buffer into queue", e);
         } finally {
-            super.close();
+            closed = true;
+            if (postCloseAction != null) {
+                postCloseAction.run();
+            }
         }
     }
 

clickhouse-client/src/main/java/com/clickhouse/client/stream/NonBlockingPipedOutputStream.java

@@ -116,7 +116,11 @@ public void close() throws IOException {
             for (int i = 0, len = buckets.length; i < len; i++) {
                 buckets[i] = null;
             }
-            super.close();
+
+            closed = true;
+            if (postCloseAction != null) {
+                postCloseAction.run();
+            }
         }
     }
 

clickhouse-client/src/test/java/com/clickhouse/client/ClickHouseRequestTest.java

@@ -360,11 +360,39 @@ public void testMutation() {
         Mutation request = ClickHouseClient.newInstance().connect(ClickHouseNode.builder().build()).write();
         request.table("test_table").format(ClickHouseFormat.Arrow).data(new ByteArrayInputStream(new byte[0]));
 
-        String expectedSql = "INSERT INTO test_table FORMAT Arrow";
+        String expectedSql = "INSERT INTO test_table\n FORMAT Arrow";
         Assert.assertEquals(request.getQuery(), expectedSql);
         Assert.assertEquals(request.getStatements().get(0), expectedSql);
 
-        request = request.seal();
+        ClickHouseRequest<?> sealedRequest = request.seal();
+        Assert.assertEquals(sealedRequest.getQuery(), expectedSql);
+        Assert.assertEquals(sealedRequest.getStatements().get(0), expectedSql);
+
+        request.query(expectedSql = "select 1 format CSV");
+        Assert.assertEquals(request.getQuery(), expectedSql);
+        Assert.assertEquals(request.getStatements().get(0), expectedSql);
+
+        request.query(expectedSql = "select format tsv from table format CSV ");
+        Assert.assertEquals(request.getQuery(), expectedSql);
+        Assert.assertEquals(request.getStatements().get(0), expectedSql);
+
+        request.query(expectedSql = "select 1 -- format CSV ");
+        expectedSql += "\n FORMAT Arrow";
+        Assert.assertEquals(request.getQuery(), expectedSql);
+        Assert.assertEquals(request.getStatements().get(0), expectedSql);
+
+        request.query(expectedSql = "select format CSV from table /* ccc */");
+        expectedSql += "\n FORMAT Arrow";
+        Assert.assertEquals(request.getQuery(), expectedSql);
+        Assert.assertEquals(request.getStatements().get(0), expectedSql);
+
+        request.query(expectedSql = "select /* format CSV */");
+        expectedSql += "\n FORMAT Arrow";
+        Assert.assertEquals(request.getQuery(), expectedSql);
+        Assert.assertEquals(request.getStatements().get(0), expectedSql);
+
+        request.query(expectedSql = "select 1 format CSV a");
+        expectedSql += "\n FORMAT Arrow";
         Assert.assertEquals(request.getQuery(), expectedSql);
         Assert.assertEquals(request.getStatements().get(0), expectedSql);
     }

clickhouse-jdbc/README.md

@@ -11,7 +11,7 @@ Keep in mind that `clickhouse-jdbc` is synchronous, and in general it has more o
     <!-- will stop using ru.yandex.clickhouse starting from 0.4.0 -->
     <groupId>com.clickhouse</groupId>
     <artifactId>clickhouse-jdbc</artifactId>
-    <version>0.3.2-patch8</version>
+    <version>0.3.2-patch9</version>
 </dependency>
 ```
 

examples/grpc/pom.xml

@@ -56,7 +56,7 @@
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
         <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
 
-        <clickhouse-grpc.version>0.3.2-patch8</clickhouse-grpc.version>
+        <clickhouse-grpc.version>0.3.2-patch9</clickhouse-grpc.version>
 
         <compiler-plugin.version>3.8.1</compiler-plugin.version>
 

examples/jdbc/pom.xml

@@ -56,7 +56,7 @@
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
         <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
 
-        <clickhouse-jdbc.version>0.3.2-patch8</clickhouse-jdbc.version>
+        <clickhouse-jdbc.version>0.3.2-patch9</clickhouse-jdbc.version>
 
         <compiler-plugin.version>3.8.1</compiler-plugin.version>