added dummy data processor to access response input stream

CHANGELOG.md

@@ -1,5 +1,12 @@
 ## Latest 
 
+### Bug Fixes
+- Fix getting response input stream for binary formats.
+    - https://github.com/ClickHouse/clickhouse-java/issues/1494
+    - https://github.com/ClickHouse/clickhouse-java/issues/1567
+    - https://github.com/ClickHouse/clickhouse-java/issues/1475
+    - https://github.com/ClickHouse/clickhouse-java/issues/1376
+
 ## 0.6.0-patch5
 
 ### Bug Fixes

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

@@ -14,7 +14,7 @@
 import java.util.function.Supplier;
 
 import com.clickhouse.config.ClickHouseBufferingMode;
-import com.clickhouse.data.format.ClickHouseNativeProcessor;
+import com.clickhouse.data.format.ClickHouseBinaryFormatProcessor;
 import com.clickhouse.data.format.ClickHouseRowBinaryProcessor;
 import com.clickhouse.data.format.ClickHouseTabSeparatedProcessor;
 import com.clickhouse.data.stream.BlockingPipedOutputStream;
@@ -142,8 +142,9 @@ public ClickHouseDataProcessor getProcessor(ClickHouseDataConfig config, ClickHo
                 ClickHouseFormat.RowBinaryWithDefaults == format ||
                 ClickHouseFormat.RowBinaryWithNames == format) {
             processor = new ClickHouseRowBinaryProcessor(config, input, output, columns, settings);
-        } else if (ClickHouseFormat.Native == format) {
-            processor = new ClickHouseNativeProcessor(config, input, output, columns, settings);
+        } else if (format.isBinary()) {
+            // to let outer code access input stream
+            processor = new ClickHouseBinaryFormatProcessor(config, input, output, columns, settings);
         } else if (format.isText()) {
             processor = new ClickHouseTabSeparatedProcessor(config, input, output, columns, settings);
         }

clickhouse-data/src/main/java/com/clickhouse/data/format/ClickHouseNativeProcessor.java → clickhouse-data/src/main/java/com/clickhouse/data/format/ClickHouseBinaryFormatProcessor.java

@@ -13,7 +13,7 @@
 import java.util.List;
 import java.util.Map;
 
-public class ClickHouseNativeProcessor extends ClickHouseDataProcessor {
+public class ClickHouseBinaryFormatProcessor extends ClickHouseDataProcessor {
 
 
     /**
@@ -28,9 +28,9 @@ public class ClickHouseNativeProcessor extends ClickHouseDataProcessor {
      * @param settings nullable settings
      * @throws IOException when failed to read columns from input stream
      */
-    public ClickHouseNativeProcessor(ClickHouseDataConfig config, ClickHouseInputStream input,
-                                        ClickHouseOutputStream output, List<ClickHouseColumn> columns,
-                                        Map<String, Serializable> settings) throws IOException {
+    public ClickHouseBinaryFormatProcessor(ClickHouseDataConfig config, ClickHouseInputStream input,
+                                           ClickHouseOutputStream output, List<ClickHouseColumn> columns,
+                                           Map<String, Serializable> settings) throws IOException {
         super(config, input, output, columns, settings);
     }
 

clickhouse-data/src/main/java/com/clickhouse/data/stream/ByteArrayQueueInputStream.java

@@ -0,0 +1,37 @@
+package com.clickhouse.data.stream;
+
+import com.clickhouse.data.ClickHouseByteBuffer;
+import com.clickhouse.data.ClickHouseChecker;
+
+import java.io.IOException;
+import java.util.Queue;
+
+public class ByteArrayQueueInputStream extends AbstractByteArrayInputStream {
+
+    private final Queue<byte[]> queue;
+
+    public ByteArrayQueueInputStream(Queue<byte[]> queue, Runnable postCloseAction) {
+        super(null, null, postCloseAction);
+        this.queue =  ClickHouseChecker.nonNull(queue, "queue");;
+    }
+
+    @Override
+    protected int updateBuffer() throws IOException {
+        position = 0;
+
+        while (!queue.isEmpty()) {
+            byte[] bytes = queue.poll();
+            int len = bytes != null ? bytes.length : 0;
+            if (len > 0) {
+                buffer = bytes;
+                if (copyTo != null) {
+                    copyTo.write(bytes);
+                }
+                return limit = len;
+            }
+        }
+
+        buffer = ClickHouseByteBuffer.EMPTY_BYTES;
+        return limit = 0;
+    }
+}

clickhouse-data/src/test/java/com/clickhouse/data/stream/IterableByteArrayInputStreamTests.java

@@ -0,0 +1,39 @@
+package com.clickhouse.data.stream;
+
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+import java.util.LinkedList;
+import java.util.Queue;
+
+public class IterableByteArrayInputStreamTests {
+
+    @Test(groups = { "unit" })
+    public void testIterableByteArrayInputStream() throws Exception {
+        LinkedList<byte[]> buffers = new LinkedList<>();
+        buffers.add("Hello".getBytes());
+        buffers.add("World".getBytes());
+        IterableByteArrayInputStream is = new IterableByteArrayInputStream(buffers, null);
+        byte[] buffer = new byte[5];
+        is.read(buffer, 0, 5);
+        Assert.assertEquals(new String(buffer), "Hello");
+        is.read(buffer, 0, 5);
+        Assert.assertEquals(new String(buffer) ,"World");
+        is.close();
+    }
+
+    @Test(groups = { "unit" })
+    public void testByteArrayQueueInputStream() throws Exception {
+        Queue<byte[]> buffers = new LinkedList<>();
+        buffers.add("Hello".getBytes());
+        buffers.add("World".getBytes());
+        ByteArrayQueueInputStream is = new ByteArrayQueueInputStream(buffers, null);
+        byte[] buffer = new byte[5];
+        is.read(buffer, 0, 5);
+        Assert.assertEquals(new String(buffer), "Hello");
+        is.read(buffer, 0, 5);
+        Assert.assertEquals(new String(buffer) ,"World");
+        Assert.assertTrue(buffers.isEmpty());
+        is.close();
+    }
+}

clickhouse-http-client/src/test/java/com/clickhouse/client/http/ClickHouseHttpClientTest.java

@@ -1,7 +1,12 @@
 package com.clickhouse.client.http;
 
+import java.io.BufferedReader;
 import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
 import java.io.Serializable;
+import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
 import java.util.*;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
@@ -24,7 +29,6 @@
 import com.clickhouse.client.ClickHouseServerForTest;
 import com.clickhouse.client.ClientIntegrationTest;
 import com.clickhouse.client.config.ClickHouseClientOption;
-import com.clickhouse.client.config.ClickHouseHealthCheckMethod;
 import com.clickhouse.client.config.ClickHouseSslMode;
 import com.clickhouse.client.http.config.ClickHouseHttpOption;
 import com.clickhouse.client.http.config.HttpConnectionProvider;
@@ -555,4 +559,26 @@ public void testLongHttpHeaderReferer() throws ClickHouseException {
             Assert.assertEquals(summary.getReadRows(), 1L);
         }
     }
+
+    @Test(groups = {"integration"})
+    public void testReadingBinaryFromRespose() throws Exception {
+        final ClickHouseNode server = getServer();
+        String tableName = "test_protobuf_format";
+        String tableColumns = String.format("id Int64, raw String");
+        sendAndWait(server, "drop table if exists " + tableName,
+                "create table " + tableName + " (" + tableColumns + ") engine=MergeTree order by tuple()");
+
+        try (ClickHouseClient client = getClient()) {
+            ClickHouseResponse response = client.read(server).query("select hostname()")
+                    .format(ClickHouseFormat.RawBLOB)
+                    .executeAndWait();
+
+            try (InputStream responseBody = response.getInputStream()) {
+                byte[] buffer = new byte[responseBody.available()];
+                Assert.assertTrue(responseBody.read(buffer) > 0);
+            } finally {
+                response.close();
+            }
+        }
+    }
 }