[jdbc-v2] Base implementation of RowBinaryFormatWriter support in JDBC

client-v2/src/main/java/com/clickhouse/client/api/data_formats/ClickHouseBinaryFormatWriter.java

@@ -0,0 +1,110 @@
+package com.clickhouse.client.api.data_formats;
+
+import com.clickhouse.data.ClickHouseFormat;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.io.Reader;
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.sql.Clob;
+import java.sql.NClob;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.ZonedDateTime;
+import java.util.List;
+
+/**
+ * Experimental API
+ */
+public interface ClickHouseBinaryFormatWriter {
+
+    /**
+     * Returns an output stream to which this writer is serializing values.
+     * Caution: this method is not intended for application usage.
+     * @return Output stream of the writer
+     */
+    OutputStream getOutputStream();
+
+    int getRowCount();
+
+    ClickHouseFormat getFormat();
+
+    void clearRow();
+
+    void setValue(String column, Object value);
+
+    void setValue(int colIndex, Object value);
+
+    /**
+     * Writer current row or block to the output stream.
+     * Action is idempotent: if there are no new values set - this method has no effect.
+     * @throws IOException if writing to an output stream causes an error
+     */
+    void commitRow() throws IOException;
+
+    void setByte(String column, byte value);
+
+    void setByte(int colIndex, byte value);
+
+    void setShort(String column, short value);
+
+    void setShort(int colIndex, short value);
+
+    void setInteger(String column, int value);
+
+    void setInteger(int colIndex, int value);
+
+    void setLong(String column, long value);
+
+    void setLong(int colIndex, long value);
+
+    void setBigInteger(int colIndex, BigInteger value);
+
+    void setBigInteger(String column, BigInteger value);
+
+    void setFloat(int colIndex, float value);
+
+    void setFloat(String column, float value);
+
+    void setDouble(int colIndex, double value);
+
+    void setDouble(String column, double value);
+
+    void setBigDecimal(int colIndex, BigDecimal value);
+
+    void setBigDecimal(String column, BigDecimal value);
+
+    void setBoolean(int colIndex, boolean value);
+
+    void setBoolean(String column, boolean value);
+
+    void setString(String column, String value);
+
+    void setString(int colIndex, String value);
+
+    void setDate(String column, LocalDate value);
+
+    void setDate(int colIndex, LocalDate value);
+
+    void setDateTime(String column, LocalDateTime value);
+
+    void setDateTime(int colIndex, LocalDateTime value);
+
+    void setDateTime(String column, ZonedDateTime value);
+
+    void setDateTime(int colIndex, ZonedDateTime value);
+
+    void setList(String column, List<?> value);
+
+    void setList(int colIndex, List<?> value);
+
+    void setInputStream(int colIndex, InputStream in, long len);
+
+    void setInputStream(String column, InputStream in, long len);
+
+    void setReader(int colIndex, Reader reader, long len);
+
+    void setReader(String column, Reader reader, long len);
+}

client-v2/src/main/java/com/clickhouse/client/api/data_formats/RowBinaryFormatWriter.java

@@ -6,10 +6,15 @@
 import com.clickhouse.data.ClickHouseFormat;
 
 import java.io.IOException;
+import java.io.InputStream;
 import java.io.OutputStream;
+import java.io.Reader;
+import java.math.BigDecimal;
+import java.math.BigInteger;
 import java.time.LocalDate;
 import java.time.LocalDateTime;
 import java.time.ZonedDateTime;
+import java.util.Arrays;
 import java.util.List;
 
 
@@ -21,7 +26,7 @@
  * <p>
  * Experimental API
  */
-public class RowBinaryFormatWriter {
+public class RowBinaryFormatWriter implements ClickHouseBinaryFormatWriter {
 
     private final OutputStream out;
 
@@ -31,6 +36,10 @@ public class RowBinaryFormatWriter {
 
     private final boolean defaultSupport;
 
+    private int rowCount = 0;
+
+    private boolean rowStarted = false; // indicates if at least one value was written to a row
+
     public RowBinaryFormatWriter(OutputStream out, TableSchema tableSchema, ClickHouseFormat format) {
         if (format != ClickHouseFormat.RowBinary && format != ClickHouseFormat.RowBinaryWithDefaults) {
             throw new IllegalArgumentException("Only RowBinary and RowBinaryWithDefaults are supported");
@@ -42,96 +51,233 @@ public RowBinaryFormatWriter(OutputStream out, TableSchema tableSchema, ClickHou
         this.defaultSupport = format == ClickHouseFormat.RowBinaryWithDefaults;
     }
 
+    @Override
+    public OutputStream getOutputStream() {
+        return out;
+    }
+
+    @Override
+    public int getRowCount() {
+        return rowCount;
+    }
+
+    @Override
+    public ClickHouseFormat getFormat() {
+        return defaultSupport ? ClickHouseFormat.RowBinaryWithDefaults : ClickHouseFormat.RowBinary;
+    }
+
+    @Override
+    public void clearRow() {
+        Arrays.fill(row, null);
+        rowStarted = false;
+    }
+
+    @Override
     public void setValue(String column, Object value) {
         setValue(tableSchema.nameToColumnIndex(column), value);
     }
 
+    @Override
     public void setValue(int colIndex, Object value) {
         row[colIndex - 1] = value;
+        if (!rowStarted) {
+            rowStarted = true;
+        }
     }
 
+    @Override
     public void commitRow() throws IOException {
-        List<ClickHouseColumn> columnList = tableSchema.getColumns();
-        for (int i = 0; i < row.length; i++) {
-            ClickHouseColumn column = columnList.get(i);
-            // here we skip if we have a default value that is MATERIALIZED or ALIAS or ...
-            if (column.hasDefault() && column.getDefaultValue() != ClickHouseColumn.DefaultValue.DEFAULT)
-                continue;
-            if (RowBinaryFormatSerializer.writeValuePreamble(out, defaultSupport, column, row[i])) {
-                SerializerUtils.serializeData(out, row[i], column);
+        if (rowStarted) {
+            List<ClickHouseColumn> columnList = tableSchema.getColumns();
+            for (int i = 0; i < row.length; i++) {
+                ClickHouseColumn column = columnList.get(i);
+                // here we skip if we have a default value that is MATERIALIZED or ALIAS or ...
+                if (column.hasDefault() && column.getDefaultValue() != ClickHouseColumn.DefaultValue.DEFAULT)
+                    continue;
+                if (RowBinaryFormatSerializer.writeValuePreamble(out, defaultSupport, column, row[i])) {
+                    SerializerUtils.serializeData(out, row[i], column);
+                }
             }
+            clearRow();
+            rowCount++;
         }
     }
 
+    @Override
     public void setByte(String column, byte value) {
         setValue(column, value);
     }
 
+    @Override
     public void setByte(int colIndex, byte value) {
         setValue(colIndex, value);
     }
 
+    @Override
     public void setShort(String column, short value) {
         setValue(column, value);
     }
 
+    @Override
     public void setShort(int colIndex, short value) {
         setValue(colIndex, value);
     }
 
+    @Override
     public void setInteger(String column, int value) {
         setValue(column, value);
     }
 
+    @Override
     public void setInteger(int colIndex, int value) {
         setValue(colIndex, value);
     }
 
+    @Override
     public void setLong(String column, long value) {
         setValue(column, value);
     }
 
+    @Override
     public void setLong(int colIndex, long value) {
         setValue(colIndex, value);
     }
 
+    @Override
+    public void setBigInteger(int colIndex, BigInteger value) {
+        setValue(colIndex, value);
+    }
+
+    @Override
+    public void setBigInteger(String column, BigInteger value) {
+        setValue(column, value);
+    }
+
+    @Override
+    public void setFloat(int colIndex, float value) {
+        setValue(colIndex, value);
+    }
+
+    @Override
+    public void setFloat(String column, float value) {
+        setValue(column, value);
+    }
+
+    @Override
+    public void setDouble(int colIndex, double value) {
+        setValue(colIndex, value);
+    }
+
+    @Override
+    public void setDouble(String column, double value) {
+        setValue(column, value);
+    }
+
+    @Override
+    public void setBigDecimal(int colIndex, BigDecimal value) {
+        setValue(colIndex, value);
+    }
+
+    @Override
+    public void setBigDecimal(String column, BigDecimal value) {
+        setValue(column, value);
+    }
+
+    @Override
+    public void setBoolean(int colIndex, boolean value) {
+        setValue(colIndex, value);
+    }
+
+    @Override
+    public void setBoolean(String column, boolean value) {
+        setValue(column, value);
+    }
+
+    @Override
     public void setString(String column, String value) {
         setValue(column, value);
     }
 
+    @Override
     public void setString(int colIndex, String value) {
         setValue(colIndex, value);
     }
 
+    @Override
     public void setDate(String column, LocalDate value) {
         setValue(column, value);
     }
 
+    @Override
     public void setDate(int colIndex, LocalDate value) {
         setValue(colIndex, value);
     }
 
+    @Override
     public void setDateTime(String column, LocalDateTime value) {
         setValue(column, value);
     }
 
+    @Override
     public void setDateTime(int colIndex, LocalDateTime value) {
         setValue(colIndex, value);
     }
 
+    @Override
     public void setDateTime(String column, ZonedDateTime value) {
         setValue(column, value);
     }
 
+    @Override
     public void setDateTime(int colIndex, ZonedDateTime value) {
         setValue(colIndex, value);
     }
 
+    @Override
     public void setList(String column, List<?> value) {
         setValue(column, value);
     }
 
+    @Override
     public void setList(int colIndex, List<?> value) {
         setValue(colIndex, value);
     }
+
+    @Override
+    public void setInputStream(int colIndex, InputStream in, long len) {
+        setValue(colIndex, new InputStreamHolder(in, len));
+    }
+
+    @Override
+    public void setInputStream(String column, InputStream in, long len) {
+        setValue(column, new InputStreamHolder(in, len));
+    }
+
+    @Override
+    public void setReader(int colIndex, Reader reader, long len) {
+        setValue(colIndex, new ReaderHolder(reader, len));
+    }
+
+    @Override
+    public void setReader(String column, Reader reader, long len) {
+        setValue(column, new ReaderHolder(reader, len));
+    }
+
+    private static class InputStreamHolder {
+        final InputStream stream;
+        final long length;
+        InputStreamHolder(InputStream stream, long length) {
+            this.stream = stream;
+            this.length = length;
+        }
+    }
+
+    private static class ReaderHolder {
+        final Reader read;
+        final long length;
+        ReaderHolder(Reader reader, long length) {
+            this.read = reader;
+            this.length = length;
+        }
+    }
 }