[client-v2] Value Reader/Writer API

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

@@ -1525,6 +1525,30 @@ public StringValue readStringValue() throws IOException {
         return new StringValue(readStringBytes(input, readVarInt(input)));
     }
 
+    /**
+     * Reads a {@code FixedString} value as raw bytes.
+     *
+     * @param length fixed length in bytes
+     * @return string value holding those bytes
+     * @throws IOException when IO error occurs
+     */
+    public StringValue readFixedStringValue(int length) throws IOException {
+        if (length < 0) {
+            throw new IllegalArgumentException("FixedString length is negative");
+        }
+        return new StringValue(readNBytes(input, length));
+    }
+
+    /**
+     * Reads a Dynamic type tag and returns the concrete column it describes.
+     *
+     * @return runtime column encoded by the tag
+     * @throws IOException when IO error occurs
+     */
+    public ClickHouseColumn readDynamicType() throws IOException {
+        return readDynamicData();
+    }
+
     /**
      * Reads the raw bytes of a string from the input stream given its length.
      *

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

@@ -372,7 +372,19 @@
                 clickHouseEnum);
     }
 
+    private static int jsonLimit(ClickHouseColumn column, String name, int fallback) {
+        List<String> parameters = column.getParameters();
+        for (int i = 0; i < parameters.size(); i++) {
+            String parameter = parameters.get(i);
+            int eq = parameter.indexOf('=');
+            if (eq > 0 && parameter.substring(0, eq).trim().equals(name)) {
+                return Integer.parseInt(parameter.substring(eq + 1).trim());
+            }
+        }
+        return fallback;
+    }
+
     public static void writeDynamicTypeTag(OutputStream stream, ClickHouseColumn typeColumn)
             throws IOException {
 
         ClickHouseDataType dt = typeColumn.getDataType();
@@ -485,7 +497,23 @@
                 stream.write(binTag);
                 break;
             case JSON:
+                // 0x30 <version> <varint max_dynamic_paths> <uint8 max_dynamic_types>
+                // <varint typed paths> (path, type)* <varint skips> <varint regexps>
+                // Version 0 has no per-path flags. SKIP paths are not retained on the column, so a
+                // tag rebuilt from a parsed column does not repeat them.
                 stream.write(binTag);
+                stream.write(0);
+                BinaryStreamUtils.writeVarInt(stream, jsonLimit(typeColumn, "max_dynamic_paths", 1024));
+                BinaryStreamUtils.writeUnsignedInt8(stream, jsonLimit(typeColumn, "max_dynamic_types", 32));
+                List<ClickHouseColumn> typedPaths = typeColumn.getNestedColumns();
+                BinaryStreamUtils.writeVarInt(stream, typedPaths.size());
+                for (int i = 0; i < typedPaths.size(); i++) {
+                    ClickHouseColumn typedPath = typedPaths.get(i);
+                    BinaryStreamUtils.writeString(stream, typedPath.getColumnName());
+                    writeDynamicTypeTag(stream, typedPath);
+                }
+                BinaryStreamUtils.writeVarInt(stream, 0);
+                BinaryStreamUtils.writeVarInt(stream, 0);
                 break;
             case SimpleAggregateFunction:
             case AggregateFunction:

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

@@ -0,0 +1,207 @@
+package com.clickhouse.client.api.data_formats.stream;
+
+import com.clickhouse.client.api.ClientException;
+import com.clickhouse.client.api.data_formats.internal.BinaryStreamReader;
+import com.clickhouse.client.api.metadata.TableSchema;
+import com.clickhouse.data.ClickHouseColumn;
+import com.clickhouse.data.ClickHouseFormat;
+
+import java.io.EOFException;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.PushbackInputStream;
+import java.time.ZoneOffset;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.TimeZone;
+
+/**
+ * Reads {@code RowBinary}, {@code RowBinaryWithNames}, or {@code RowBinaryWithNamesAndTypes}
+ * without materializing nested containers.
+ * <p>
+ * Experimental API. Not thread-safe. The caller owns the input stream. A failure desynchronizes
+ * the stream; the reader must not be reused.
+ */
+public final class RowBinaryStreamReader {
+
+    private final PushbackInputStream stream;
+
+    private final BinaryStreamReader binary;
+
+    private final ClickHouseFormat format;
+
+    private final TimeZone timeZone;
+
+    private final byte[] scratch = new byte[8];
+
+    private final byte[] skipBuffer = new byte[256];
+
+    private TableSchema schema;
+
+    private List<ClickHouseColumn> columns;
+
+    private boolean failed;
+
+    private StreamReaderScope openRow;
+
+    public RowBinaryStreamReader(InputStream in, TableSchema schema, ClickHouseFormat format) throws IOException {
+        this(in, schema, format, TimeZone.getTimeZone(ZoneOffset.UTC));
+    }
+
+    public RowBinaryStreamReader(InputStream in, TableSchema schema, ClickHouseFormat format, TimeZone timeZone)
+            throws IOException {
+        if (in == null) {
+            throw new IllegalArgumentException("Input stream is required");
+        }
+        if (format != ClickHouseFormat.RowBinary && format != ClickHouseFormat.RowBinaryWithNames
+                && format != ClickHouseFormat.RowBinaryWithNamesAndTypes) {
+            throw new IllegalArgumentException(
+                    "Only RowBinary, RowBinaryWithNames, and RowBinaryWithNamesAndTypes are supported");
+        }
+        if (format != ClickHouseFormat.RowBinaryWithNamesAndTypes && schema == null) {
+            throw new IllegalArgumentException("Schema is required");
+        }
+        this.stream = new PushbackInputStream(in, 1);
+        this.timeZone = timeZone == null ? TimeZone.getTimeZone(ZoneOffset.UTC) : timeZone;
+        this.binary = new BinaryStreamReader(stream, this.timeZone, null,
+                new BinaryStreamReader.CachingByteBufferAllocator(), true, null, true);
+        this.format = format;
+        if (schema != null) {
+            this.schema = schema;
+            this.columns = new ArrayList<ClickHouseColumn>(schema.getColumns());
+        }
+        if (format == ClickHouseFormat.RowBinaryWithNames) {
+            readNameHeader();
+        } else if (format == ClickHouseFormat.RowBinaryWithNamesAndTypes) {
+            readNamesAndTypesHeader();
+        }
+    }
+
+    public boolean hasNext() throws IOException {
+        if (failed) {
+            throw new IllegalStateException("Reader is unusable after a previous failure");
+        }
+        if (openRow != null && !openRow.isClosed()) {
+            throw new IllegalStateException("A row is still open");
+        }
+        int next = stream.read();
+        if (next < 0) {
+            return false;
+        }
+        stream.unread(next);
+        return true;
+    }
+
+    public ValueReader beginRow() throws IOException {
+        if (failed) {
+            throw new IllegalStateException("Reader is unusable after a previous failure");
+        }
+        if (openRow != null && !openRow.isClosed()) {
+            throw new IllegalStateException("A row is still open");
+        }
+        if (!hasNext()) {
+            throw new EOFException("No more rows");
+        }
+        openRow = StreamReaderScope.row(this, columns);
+        return openRow;
+    }
+
+    public TableSchema getSchema() {
+        return schema;
+    }
+
+    public ClickHouseFormat getFormat() {
+        return format;
+    }
+
+    InputStream stream() {
+        return stream;
+    }
+
+    BinaryStreamReader binary() {
+        return binary;
+    }
+
+    TimeZone timeZone() {
+        return timeZone;
+    }
+
+    byte[] scratch() {
+        return scratch;
+    }
+
+    boolean isFailed() {
+        return failed;
+    }
+
+    void fail() {
+        failed = true;
+    }
+
+    void onRowClosed() {
+        openRow = null;
+    }
+
+    void unread(int value) throws IOException {
+        stream.unread(value);
+    }
+
+    void skipFully(int length) throws IOException {
+        int left = length;
+        while (left > 0) {
+            int n = stream.read(skipBuffer, 0, Math.min(skipBuffer.length, left));
+            if (n < 0) {
+                throw new EOFException("End of stream reached before skipping " + length + " bytes");
+            }
+            left -= n;
+        }
+    }
+
+    private void readNameHeader() throws IOException {
+        List<String> names = readNames();
+        if (names.size() != columns.size()) {
+            throw new ClientException("RowBinaryWithNames column count " + names.size()
+                    + " does not match schema " + columns.size());
+        }
+        for (int i = 0; i < names.size(); i++) {
+            if (!names.get(i).equals(columns.get(i).getColumnName())) {
+                throw new ClientException("RowBinaryWithNames column '" + names.get(i)
+                        + "' does not match schema column '" + columns.get(i).getColumnName() + "'");
+            }
+        }
+    }
+
+    private void readNamesAndTypesHeader() throws IOException {
+        List<String> names = readNames();
+        List<ClickHouseColumn> header = new ArrayList<ClickHouseColumn>(names.size());
+        try {
+            for (int i = 0; i < names.size(); i++) {
+                header.add(ClickHouseColumn.of(names.get(i), binary.readString()));
+            }
+        } catch (IOException e) {
+            throw new ClientException("Failed to read header", e);
+        }
+        this.columns = header;
+        this.schema = new TableSchema(header);
+    }
+
+    private List<String> readNames() throws IOException {
+        int count;
+        try {
+            count = BinaryStreamReader.readVarInt(stream);
+        } catch (EOFException e) {
+            throw new ClientException("Failed to read header", e);
+        } catch (IOException e) {
+            throw new ClientException("Failed to read header", e);
+        }
+        List<String> names = new ArrayList<String>(count);
+        try {
+            for (int i = 0; i < count; i++) {
+                names.add(binary.readString());
+            }
+        } catch (IOException e) {
+            throw new ClientException("Failed to read header", e);
+        }
+        return names;
+    }
+}

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

@@ -0,0 +1,116 @@
+package com.clickhouse.client.api.data_formats.stream;
+
+import com.clickhouse.client.api.metadata.TableSchema;
+import com.clickhouse.data.ClickHouseColumn;
+import com.clickhouse.data.ClickHouseFormat;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.time.ZoneOffset;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.TimeZone;
+
+/**
+ * Writes {@code RowBinary} or {@code RowBinaryWithDefaults} directly to an output stream.
+ * Nested values are scopes: a known container size is written through, an unknown size is buffered
+ * until the scope is closed.
+ * <p>
+ * Experimental API. Not thread-safe. The caller owns {@code out}. A failure desynchronizes the
+ * stream; the writer must not be reused. Retry by writing the rows again to a new stream.
+ */
+public final class RowBinaryStreamWriter {
+
+    private final OutputStream out;
+
+    private final List<ClickHouseColumn> columns;
+
+    private final boolean defaultsSupport;
+
+    private final ClickHouseFormat format;
+
+    private final TimeZone timeZone;
+
+    private int rowCount;
+
+    private boolean failed;
+
+    private StreamWriterScope openRow;
+
+    public RowBinaryStreamWriter(OutputStream out, TableSchema schema, ClickHouseFormat format) {
+        this(out, schema, format, TimeZone.getTimeZone(ZoneOffset.UTC));
+    }
+
+    public RowBinaryStreamWriter(OutputStream out, TableSchema schema, ClickHouseFormat format, TimeZone timeZone) {
+        if (out == null) {
+            throw new IllegalArgumentException("Output stream is required");
+        }
+        if (schema == null) {
+            throw new IllegalArgumentException("Schema is required");
+        }
+        if (format != ClickHouseFormat.RowBinary && format != ClickHouseFormat.RowBinaryWithDefaults) {
+            throw new IllegalArgumentException("Only RowBinary and RowBinaryWithDefaults are supported");
+        }
+        this.out = out;
+        this.columns = writableColumns(schema);
+        this.defaultsSupport = format == ClickHouseFormat.RowBinaryWithDefaults;
+        this.format = format;
+        this.timeZone = timeZone == null ? TimeZone.getTimeZone(ZoneOffset.UTC) : timeZone;
+    }
+
+    public ValueWriter beginRow() throws IOException {
+        if (failed) {
+            throw new IllegalStateException("Writer is unusable after a previous failure");
+        }
+        if (openRow != null && !openRow.isClosed()) {
+            throw new IllegalStateException("A row is still open");
+        }
+        openRow = StreamWriterScope.row(this, columns);
+        return openRow;
+    }
+
+    public int getRowCount() {
+        return rowCount;
+    }
+
+    public ClickHouseFormat getFormat() {
+        return format;
+    }
+
+    OutputStream out() {
+        return out;
+    }
+
+    boolean defaultsSupport() {
+        return defaultsSupport;
+    }
+
+    TimeZone timeZone() {
+        return timeZone;
+    }
+
+    boolean isFailed() {
+        return failed;
+    }
+
+    void fail() {
+        failed = true;
+    }
+
+    void onRowClosed() {
+        openRow = null;
+        rowCount++;
+    }
+
+    private static List<ClickHouseColumn> writableColumns(TableSchema schema) {
+        List<ClickHouseColumn> all = schema.getColumns();
+        List<ClickHouseColumn> kept = new ArrayList<ClickHouseColumn>(all.size());
+        for (int i = 0; i < all.size(); i++) {
+            ClickHouseColumn column = all.get(i);
+            if (!StreamTypes.omitted(column)) {
+                kept.add(column);
+            }
+        }
+        return kept;
+    }
+}