override readCustom to stop reusing same byte array for mutliple reads

Author: zhicwu

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

@@ -4,11 +4,14 @@
 import java.io.InputStream;
 import java.io.InvalidObjectException;
 import java.io.StreamCorruptedException;
+import java.util.Arrays;
+import java.util.LinkedList;
 
 import com.clickhouse.data.ClickHouseByteUtils;
 import com.clickhouse.data.ClickHouseByteBuffer;
 import com.clickhouse.data.ClickHouseChecker;
 import com.clickhouse.data.ClickHouseCityHash;
+import com.clickhouse.data.ClickHouseDataUpdater;
 import com.clickhouse.data.ClickHouseInputStream;
 import com.clickhouse.data.ClickHousePassThruStream;
 import com.clickhouse.data.ClickHouseUtils;
@@ -105,6 +108,46 @@ public Lz4InputStream(ClickHousePassThruStream stream, InputStream input, Runnab
         this.compressedBlock = ClickHouseByteBuffer.EMPTY_BYTES;
     }
 
+    @Override
+    public ClickHouseByteBuffer readCustom(ClickHouseDataUpdater reader) throws IOException {
+        if (reader == null) {
+            return byteBuffer.reset();
+        }
+        ensureOpen();
+
+        LinkedList<byte[]> list = new LinkedList<>();
+        int length = 0;
+        boolean more = true;
+        while (more) {
+            int remain = limit - position;
+            if (remain < 1) {
+                closeQuietly();
+                more = false;
+            } else {
+                int read = reader.update(buffer, position, limit);
+                if (read == -1) {
+                    list.add(Arrays.copyOfRange(buffer, position, limit));
+                    length += remain;
+                    position = limit;
+                    if (updateBuffer() < 1) {
+                        closeQuietly();
+                        more = false;
+                    }
+                } else {
+                    if (read > 0) {
+                        byte[] bytes = new byte[read];
+                        System.arraycopy(buffer, position, bytes, 0, read);
+                        list.add(bytes);
+                        length += read;
+                        position += read;
+                    }
+                    more = false;
+                }
+            }
+        }
+        return byteBuffer.update(list, 0, length);
+    }
+
     @Override
     public void close() throws IOException {
         if (!closed) {