From 5646ea25a840ae8e228e77cc820bbbc31dcd21a6 Mon Sep 17 00:00:00 2001 From: Sergey Chernov Date: Wed, 28 Oct 2026 13:03:30 -0700 Subject: [PATCH 1/2] Draft writer/reader API --- CHANGELOG.md | 16 + .../internal/BinaryStreamReader.java | 24 + .../internal/SerializerUtils.java | 28 + .../stream/RowBinaryStreamReader.java | 207 +++ .../stream/RowBinaryStreamWriter.java | 116 ++ .../api/data_formats/stream/SlotCursor.java | 182 +++ .../stream/StreamReaderScope.java | 829 +++++++++++ .../api/data_formats/stream/StreamTypes.java | 137 ++ .../stream/StreamWriterScope.java | 1228 +++++++++++++++++ .../api/data_formats/stream/ValueReader.java | 126 ++ .../api/data_formats/stream/ValueWriter.java | 202 +++ .../stream/RowBinaryStreamRoundTripTest.java | 672 +++++++++ .../stream/RowBinaryStreamServerTest.java | 297 ++++ docs/features.md | 1 + docs/integration-client.md | 86 ++ docs/reader_writer_api.md | 247 ++++ 16 files changed, 4398 insertions(+) create mode 100644 client-v2/src/main/java/com/clickhouse/client/api/data_formats/stream/RowBinaryStreamReader.java create mode 100644 client-v2/src/main/java/com/clickhouse/client/api/data_formats/stream/RowBinaryStreamWriter.java create mode 100644 client-v2/src/main/java/com/clickhouse/client/api/data_formats/stream/SlotCursor.java create mode 100644 client-v2/src/main/java/com/clickhouse/client/api/data_formats/stream/StreamReaderScope.java create mode 100644 client-v2/src/main/java/com/clickhouse/client/api/data_formats/stream/StreamTypes.java create mode 100644 client-v2/src/main/java/com/clickhouse/client/api/data_formats/stream/StreamWriterScope.java create mode 100644 client-v2/src/main/java/com/clickhouse/client/api/data_formats/stream/ValueReader.java create mode 100644 client-v2/src/main/java/com/clickhouse/client/api/data_formats/stream/ValueWriter.java create mode 100644 client-v2/src/test/java/com/clickhouse/client/api/data_formats/stream/RowBinaryStreamRoundTripTest.java create mode 100644 client-v2/src/test/java/com/clickhouse/client/api/data_formats/stream/RowBinaryStreamServerTest.java create mode 100644 docs/reader_writer_api.md diff --git a/CHANGELOG.md b/CHANGELOG.md index 6635e0627..884d36578 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -26,6 +26,22 @@ ### New Features +- **[client-v2]** Added an experimental streaming RowBinary reader and writer + (`com.clickhouse.client.api.data_formats.stream`). Nested values used to be written only after the whole + container existed in memory, because `Array`, `Map`, unflattened `Nested`, and `QBit` put a length in front of + their elements, and they were read only after the whole container had been allocated. `RowBinaryStreamWriter` + and `RowBinaryStreamReader` treat a row and each nested value as a scope that is opened and closed explicitly. + A container opened with a known size writes that length immediately and sends the element bytes straight to the + stream; a container opened without a size buffers the encoded bytes and writes the length when the scope closes. + `writeString(byte[])` keeps bytes that are not valid UTF-8. The writer accepts `RowBinary` and + `RowBinaryWithDefaults`. The reader accepts `RowBinary` (schema supplied by the caller), `RowBinaryWithNames`, + and `RowBinaryWithNamesAndTypes` (schema taken from the header). A column that was not written is an error. + `RowBinaryFormatWriter`, POJO insert, and JDBC are unchanged, and `Native` is not supported on this path. + A `JSON` column can be written as a binary object with `beginJson`: each path is a name plus a value, a nested + object is dotted paths, a JSON null is `writeNull(path)`, and an array of objects is an `Array(JSON)` path whose + elements are themselves JSON documents. `writeString` without a path is still the string form. + See `docs/reader_writer_api.md`. + - **[migration-helpers]** Added `migration-helpers` module containing `ConfigurationMigrationHelper` and `ConfigPropertyCache` to convert configuration properties and connection URLs from v1 (0.7.1) format to v2 (0.9.8+) format (automatically prefixing ClickHouse server settings with `clickhouse_setting_`, custom headers with diff --git a/client-v2/src/main/java/com/clickhouse/client/api/data_formats/internal/BinaryStreamReader.java b/client-v2/src/main/java/com/clickhouse/client/api/data_formats/internal/BinaryStreamReader.java index dbdbd559b..785f7f61d 100644 --- a/client-v2/src/main/java/com/clickhouse/client/api/data_formats/internal/BinaryStreamReader.java +++ b/client-v2/src/main/java/com/clickhouse/client/api/data_formats/internal/BinaryStreamReader.java @@ -1399,6 +1399,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. * diff --git a/client-v2/src/main/java/com/clickhouse/client/api/data_formats/internal/SerializerUtils.java b/client-v2/src/main/java/com/clickhouse/client/api/data_formats/internal/SerializerUtils.java index 47a460a76..512845dac 100644 --- a/client-v2/src/main/java/com/clickhouse/client/api/data_formats/internal/SerializerUtils.java +++ b/client-v2/src/main/java/com/clickhouse/client/api/data_formats/internal/SerializerUtils.java @@ -372,6 +372,18 @@ private static ClickHouseColumn enumValue2Column(Enum enumValue) { clickHouseEnum); } + private static int jsonLimit(ClickHouseColumn column, String name, int fallback) { + List 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 { @@ -485,7 +497,23 @@ public static void writeDynamicTypeTag(OutputStream stream, ClickHouseColumn typ stream.write(binTag); break; case JSON: + // 0x30 + // (path, type)* + // 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 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: diff --git a/client-v2/src/main/java/com/clickhouse/client/api/data_formats/stream/RowBinaryStreamReader.java b/client-v2/src/main/java/com/clickhouse/client/api/data_formats/stream/RowBinaryStreamReader.java new file mode 100644 index 000000000..18830f142 --- /dev/null +++ b/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. + *

+ * 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 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(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 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 names = readNames(); + List header = new ArrayList(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 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 names = new ArrayList(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; + } +} diff --git a/client-v2/src/main/java/com/clickhouse/client/api/data_formats/stream/RowBinaryStreamWriter.java b/client-v2/src/main/java/com/clickhouse/client/api/data_formats/stream/RowBinaryStreamWriter.java new file mode 100644 index 000000000..0867b5b06 --- /dev/null +++ b/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. + *

+ * 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 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 writableColumns(TableSchema schema) { + List all = schema.getColumns(); + List kept = new ArrayList(all.size()); + for (int i = 0; i < all.size(); i++) { + ClickHouseColumn column = all.get(i); + if (!StreamTypes.omitted(column)) { + kept.add(column); + } + } + return kept; + } +} diff --git a/client-v2/src/main/java/com/clickhouse/client/api/data_formats/stream/SlotCursor.java b/client-v2/src/main/java/com/clickhouse/client/api/data_formats/stream/SlotCursor.java new file mode 100644 index 000000000..f506fdc22 --- /dev/null +++ b/client-v2/src/main/java/com/clickhouse/client/api/data_formats/stream/SlotCursor.java @@ -0,0 +1,182 @@ +package com.clickhouse.client.api.data_formats.stream; + +import com.clickhouse.data.ClickHouseColumn; + +import java.util.List; + +/** + * Position inside one scope: a row, array, tuple, map, or a single variant/dynamic value. + */ +final class SlotCursor { + + static final int UNKNOWN = -1; + + enum Kind { + ROW, + ARRAY, + TUPLE, + MAP, + SINGLE, + JSON + } + + final Kind kind; + + private final List slots; + + private final ClickHouseColumn repeated; + + private final ClickHouseColumn mapKey; + + private final ClickHouseColumn mapValue; + + /** Expected number of completed values, or {@link #UNKNOWN}. */ + final int declared; + + private int index; + + private boolean expectValue; + + private SlotCursor(Kind kind, List slots, ClickHouseColumn repeated, + ClickHouseColumn mapKey, ClickHouseColumn mapValue, int declared) { + this.kind = kind; + this.slots = slots; + this.repeated = repeated; + this.mapKey = mapKey; + this.mapValue = mapValue; + this.declared = declared; + } + + static SlotCursor row(List columns) { + return new SlotCursor(Kind.ROW, columns, null, null, null, columns.size()); + } + + static SlotCursor array(ClickHouseColumn element, int size) { + return new SlotCursor(Kind.ARRAY, null, element, null, null, size); + } + + static SlotCursor tuple(List fields) { + return new SlotCursor(Kind.TUPLE, fields, null, null, null, fields.size()); + } + + static SlotCursor map(ClickHouseColumn key, ClickHouseColumn value, int pairs) { + return new SlotCursor(Kind.MAP, null, null, key, value, pairs); + } + + static SlotCursor single(ClickHouseColumn value) { + return new SlotCursor(Kind.SINGLE, null, value, null, null, 1); + } + + /** A JSON object. {@code paths} is the number of leaf paths, or {@link #UNKNOWN}. */ + static SlotCursor json(int paths) { + return new SlotCursor(Kind.JSON, null, null, null, null, paths); + } + + ClickHouseColumn current() { + switch (kind) { + case ROW: + case TUPLE: + if (index >= slots.size()) { + throw new IllegalStateException("No more values in this " + kind.name().toLowerCase()); + } + return slots.get(index); + case ARRAY: + if (declared != UNKNOWN && index >= declared) { + throw new IllegalStateException("No more array elements"); + } + return repeated; + case MAP: + if (expectValue) { + return mapValue; + } + if (declared != UNKNOWN && index >= declared) { + throw new IllegalStateException("No more map entries"); + } + return mapKey; + case SINGLE: + if (index >= 1) { + throw new IllegalStateException("The value has already been written"); + } + return repeated; + case JSON: + throw new IllegalStateException("A JSON value is written by path"); + default: + throw new IllegalStateException("Unknown scope"); + } + } + + ClickHouseColumn repeated() { + return repeated; + } + + boolean expectingValue() { + return expectValue; + } + + int completed() { + return index; + } + + int size() { + if (kind == Kind.ARRAY || kind == Kind.MAP) { + return declared; + } + if (kind == Kind.SINGLE) { + return 1; + } + if (kind == Kind.JSON) { + return declared; + } + return slots.size(); + } + + boolean hasNext() { + if (kind == Kind.MAP) { + return !expectValue && (declared == UNKNOWN || index < declared); + } + if (kind == Kind.JSON) { + return declared == UNKNOWN || index < declared; + } + int limit = kind == Kind.ARRAY ? declared : (kind == Kind.SINGLE ? 1 : slots.size()); + return index < limit; + } + + int remaining() { + if (kind == Kind.ARRAY) { + return declared - index; + } + throw new IllegalStateException("remaining() is only valid inside an array"); + } + + void advance() { + if (kind == Kind.MAP) { + if (!expectValue) { + expectValue = true; + } else { + expectValue = false; + index++; + } + return; + } + index++; + } + + /** Adds leaf paths contributed by a nested object that was flattened into this one. */ + void credit(int paths) { + index += paths; + } + + void forceComplete() { + index = declared == UNKNOWN ? 1 : declared; + expectValue = false; + } + + void checkComplete() { + if (kind == Kind.MAP && expectValue) { + throw new IllegalStateException("Map entry is missing a value"); + } + if (declared != UNKNOWN && index != declared) { + throw new IllegalStateException("Expected " + declared + " values but completed " + index); + } + } +} diff --git a/client-v2/src/main/java/com/clickhouse/client/api/data_formats/stream/StreamReaderScope.java b/client-v2/src/main/java/com/clickhouse/client/api/data_formats/stream/StreamReaderScope.java new file mode 100644 index 000000000..4afceaf38 --- /dev/null +++ b/client-v2/src/main/java/com/clickhouse/client/api/data_formats/stream/StreamReaderScope.java @@ -0,0 +1,829 @@ +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.data_formats.internal.StringValue; +import com.clickhouse.client.api.query.NullValueException; +import com.clickhouse.data.ClickHouseAggregateFunction; +import com.clickhouse.data.ClickHouseColumn; +import com.clickhouse.data.ClickHouseDataType; +import com.clickhouse.data.value.ClickHouseBitmap; + +import java.io.EOFException; +import java.io.IOException; +import java.math.BigDecimal; +import java.math.BigInteger; +import java.net.Inet4Address; +import java.net.Inet6Address; +import java.net.InetAddress; +import java.time.LocalDate; +import java.time.ZonedDateTime; +import java.util.List; +import java.util.TimeZone; +import java.util.UUID; + +/** + * Scope implementation shared by a row and every nested container on the read path. + */ +final class StreamReaderScope implements ValueReader { + + private final RowBinaryStreamReader owner; + + private final StreamReaderScope parent; + + private final SlotCursor cursor; + + private final int variantOrdinal; + + private StreamReaderScope child; + + private boolean markerConsumed; + + private boolean closed; + + private StreamReaderScope(RowBinaryStreamReader owner, StreamReaderScope parent, SlotCursor cursor, int variantOrdinal) { + this.owner = owner; + this.parent = parent; + this.cursor = cursor; + this.variantOrdinal = variantOrdinal; + } + + static StreamReaderScope row(RowBinaryStreamReader owner, List columns) { + return new StreamReaderScope(owner, null, SlotCursor.row(columns), -1); + } + + boolean isClosed() { + return closed; + } + + @Override + public byte readInt8() throws IOException { + prepare(ClickHouseDataType.Int8); + return finishByte(owner.binary().readByte()); + } + + @Override + public short readInt16() throws IOException { + prepare(ClickHouseDataType.Int16); + return finishShort(owner.binary().readShortLE()); + } + + @Override + public int readInt32() throws IOException { + prepare(ClickHouseDataType.Int32); + return finishInt(owner.binary().readIntLE()); + } + + @Override + public long readInt64() throws IOException { + prepare(ClickHouseDataType.Int64); + return finishLong(owner.binary().readLongLE()); + } + + @Override + public BigInteger readInt128() throws IOException { + prepare(ClickHouseDataType.Int128); + return finish(owner.binary().readBigIntegerLE(BinaryStreamReader.INT128_SIZE, false)); + } + + @Override + public BigInteger readInt256() throws IOException { + prepare(ClickHouseDataType.Int256); + return finish(owner.binary().readBigIntegerLE(BinaryStreamReader.INT256_SIZE, false)); + } + + @Override + public int readUInt8() throws IOException { + prepare(ClickHouseDataType.UInt8); + return finishInt(owner.binary().readUnsignedByte()); + } + + @Override + public int readUInt16() throws IOException { + prepare(ClickHouseDataType.UInt16); + return finishInt(owner.binary().readUnsignedShortLE()); + } + + @Override + public long readUInt32() throws IOException { + prepare(ClickHouseDataType.UInt32); + return finishLong(owner.binary().readUnsignedIntLE()); + } + + @Override + public BigInteger readUInt64() throws IOException { + prepare(ClickHouseDataType.UInt64); + return finish(owner.binary().readBigIntegerLE(BinaryStreamReader.INT64_SIZE, true)); + } + + @Override + public BigInteger readUInt128() throws IOException { + prepare(ClickHouseDataType.UInt128); + return finish(owner.binary().readBigIntegerLE(BinaryStreamReader.INT128_SIZE, true)); + } + + @Override + public BigInteger readUInt256() throws IOException { + prepare(ClickHouseDataType.UInt256); + return finish(owner.binary().readBigIntegerLE(BinaryStreamReader.INT256_SIZE, true)); + } + + @Override + public float readFloat32() throws IOException { + prepare(ClickHouseDataType.Float32); + return finishFloat(owner.binary().readFloatLE()); + } + + @Override + public double readFloat64() throws IOException { + prepare(ClickHouseDataType.Float64); + return finishDouble(owner.binary().readDoubleLE()); + } + + @Override + public float readBFloat16() throws IOException { + prepare(ClickHouseDataType.BFloat16); + return finishFloat(owner.binary().readBFloat16LE()); + } + + @Override + public boolean readBool() throws IOException { + prepare(ClickHouseDataType.Bool); + int value = owner.binary().readByte() & 0xFF; + cursor.advance(); + return value == 1; + } + + @Override + public String readString() throws IOException { + return readStringValue().asString(); + } + + @Override + public StringValue readStringValue() throws IOException { + ensureReadable(); + ClickHouseColumn effective = StreamTypes.unwrap(slot()); + ClickHouseDataType type = effective.getDataType(); + if (type != ClickHouseDataType.String && type != ClickHouseDataType.FixedString + && type != ClickHouseDataType.JSON) { + owner.fail(); + throw new IllegalStateException("Cannot read a string from " + StreamTypes.describe(effective)); + } + consumePresent(effective); + try { + StringValue value = type == ClickHouseDataType.FixedString + ? owner.binary().readFixedStringValue(effective.getPrecision()) + : owner.binary().readStringValue(); + cursor.advance(); + return value; + } catch (IOException e) { + owner.fail(); + throw e; + } + } + + @Override + public String readFixedString() throws IOException { + ensureReadable(); + ClickHouseColumn effective = StreamTypes.unwrap(slot()); + if (effective.getDataType() != ClickHouseDataType.FixedString) { + owner.fail(); + throw new IllegalStateException("Cannot read FixedString from " + StreamTypes.describe(effective)); + } + return readStringValue().asString(); + } + + @Override + public LocalDate readDate() throws IOException { + prepare(ClickHouseDataType.Date); + return finish(owner.binary().readDateAsLocalDate()); + } + + @Override + public LocalDate readDate32() throws IOException { + prepare(ClickHouseDataType.Date32); + return finish(owner.binary().readDate32AsLocalDate()); + } + + @Override + public ZonedDateTime readDateTime() throws IOException { + ClickHouseColumn column = prepare(ClickHouseDataType.DateTime, ClickHouseDataType.DateTime32); + TimeZone zone = column.getTimeZone() != null ? column.getTimeZone() : owner.timeZone(); + return finish(BinaryStreamReader.readDateTime32(owner.stream(), owner.scratch(), zone)); + } + + @Override + public ZonedDateTime readDateTime64() throws IOException { + ClickHouseColumn column = prepare(ClickHouseDataType.DateTime64); + TimeZone zone = column.getTimeZone() != null ? column.getTimeZone() : owner.timeZone(); + return finish(BinaryStreamReader.readDateTime64(owner.stream(), owner.scratch(), column.getScale(), zone)); + } + + @Override + public UUID readUUID() throws IOException { + prepare(ClickHouseDataType.UUID); + return finish(new UUID(owner.binary().readLongLE(), owner.binary().readLongLE())); + } + + @Override + public Inet4Address readIPv4() throws IOException { + prepare(ClickHouseDataType.IPv4); + byte[] raw = new byte[4]; + BinaryStreamReader.readNBytesLE(owner.stream(), raw, 0, 4); + return finish((Inet4Address) InetAddress.getByAddress(raw)); + } + + @Override + public Inet6Address readIPv6() throws IOException { + prepare(ClickHouseDataType.IPv6); + byte[] raw = BinaryStreamReader.readNBytes(owner.stream(), 16); + return finish((Inet6Address) InetAddress.getByAddress(raw)); + } + + @Override + public BigDecimal readDecimal() throws IOException { + ClickHouseColumn column = prepare(ClickHouseDataType.Decimal, ClickHouseDataType.Decimal32, + ClickHouseDataType.Decimal64, ClickHouseDataType.Decimal128, ClickHouseDataType.Decimal256); + return finish(owner.binary().readDecimal(column.getPrecision(), column.getScale())); + } + + @Override + public int readEnum8() throws IOException { + prepare(ClickHouseDataType.Enum8); + return finishInt(owner.binary().readByte()); + } + + @Override + public int readEnum16() throws IOException { + prepare(ClickHouseDataType.Enum16); + return finishInt(owner.binary().readShortLE()); + } + + @Override + public ClickHouseBitmap readBitmap() throws IOException { + ensureReadable(); + ClickHouseColumn effective = StreamTypes.unwrap(slot()); + if (effective.getDataType() != ClickHouseDataType.AggregateFunction + || effective.getAggregateFunction() != ClickHouseAggregateFunction.groupBitmap) { + owner.fail(); + throw new IllegalStateException("readBitmap is only valid for AggregateFunction(groupBitmap, ...) but was " + + StreamTypes.describe(effective)); + } + consumePresent(effective); + try { + ClickHouseBitmap value = ClickHouseBitmap.deserialize(owner.stream(), + effective.getNestedColumns().get(0).getDataType()); + cursor.advance(); + return value; + } catch (IOException e) { + owner.fail(); + throw e; + } + } + + @Override + public boolean isNull() throws IOException { + ensureReadable(); + if (markerConsumed) { + return false; + } + if (cursor.kind == SlotCursor.Kind.MAP && !cursor.expectingValue()) { + return false; + } + ClickHouseColumn effective = StreamTypes.unwrap(slot()); + if (effective.isNullable()) { + int marker = readMarker(); + if (marker == 1) { + cursor.advance(); + return true; + } + markerConsumed = true; + return false; + } + if (effective.getDataType() == ClickHouseDataType.Variant + || effective.getDataType() == ClickHouseDataType.Geometry) { + int ordinal = owner.binary().readByte() & 0xFF; + if (ordinal == 0xFF) { + cursor.advance(); + return true; + } + owner.unread(ordinal); + return false; + } + if (effective.getDataType() == ClickHouseDataType.Dynamic) { + return false; + } + return false; + } + + @Override + public boolean hasNext() { + return cursor.hasNext(); + } + + @Override + public int size() { + return cursor.size(); + } + + @Override + public int variantOrdinal() { + if (variantOrdinal < 0) { + throw new IllegalStateException("variantOrdinal is only valid on a variant scope"); + } + return variantOrdinal; + } + + @Override + public ClickHouseColumn runtimeType() { + if (cursor.kind != SlotCursor.Kind.SINGLE) { + throw new IllegalStateException("runtimeType is only valid on a variant or dynamic scope"); + } + return cursor.repeated(); + } + + @Override + public ValueReader beginArray() throws IOException { + return openCounted(false, false); + } + + @Override + public ValueReader beginTuple() throws IOException { + ensureReadable(); + ClickHouseColumn slot = slot(); + ClickHouseColumn effective = StreamTypes.unwrap(slot); + List fields = StreamTypes.tupleFields(effective); + if (fields == null) { + owner.fail(); + throw new IllegalStateException("beginTuple is not valid for " + StreamTypes.describe(effective)); + } + consumePresent(effective); + return nest(new StreamReaderScope(owner, this, SlotCursor.tuple(fields), -1)); + } + + @Override + public ValueReader beginMap() throws IOException { + ensureReadable(); + ClickHouseColumn slot = slot(); + ClickHouseColumn effective = StreamTypes.unwrap(slot); + if (effective.getDataType() != ClickHouseDataType.Map) { + owner.fail(); + throw new IllegalStateException("beginMap is not valid for " + StreamTypes.describe(effective)); + } + consumePresent(effective); + int pairs = readCount(); + return nest(new StreamReaderScope(owner, this, + SlotCursor.map(effective.getKeyInfo(), effective.getValueInfo(), pairs), -1)); + } + + @Override + public ValueReader beginNested() throws IOException { + return openCounted(true, false); + } + + @Override + public ValueReader beginVariant() throws IOException { + ensureReadable(); + ClickHouseColumn slot = slot(); + ClickHouseColumn effective = StreamTypes.unwrap(slot); + if (effective.getDataType() != ClickHouseDataType.Variant + && effective.getDataType() != ClickHouseDataType.Geometry) { + owner.fail(); + throw new IllegalStateException("beginVariant is not valid for " + StreamTypes.describe(effective)); + } + consumePresent(effective); + int ordinal = owner.binary().readByte() & 0xFF; + if (ordinal == 0xFF) { + cursor.advance(); + throw new NullValueException("Variant value is null"); + } + List alternatives = effective.getNestedColumns(); + if (ordinal >= alternatives.size()) { + owner.fail(); + throw new ClientException("Variant ordinal " + ordinal + " is out of range"); + } + return nest(new StreamReaderScope(owner, this, SlotCursor.single(alternatives.get(ordinal)), ordinal)); + } + + @Override + public ValueReader beginDynamic() throws IOException { + ensureReadable(); + ClickHouseColumn slot = slot(); + ClickHouseColumn effective = StreamTypes.unwrap(slot); + if (effective.getDataType() != ClickHouseDataType.Dynamic) { + owner.fail(); + throw new IllegalStateException("beginDynamic is not valid for " + StreamTypes.describe(effective)); + } + consumePresent(effective); + ClickHouseColumn runtime = owner.binary().readDynamicType(); + StreamReaderScope childScope = new StreamReaderScope(owner, this, SlotCursor.single(runtime), -1); + if (runtime.getDataType() == ClickHouseDataType.Nothing) { + childScope.cursor.forceComplete(); + } + return nest(childScope); + } + + @Override + public ValueReader beginQBit() throws IOException { + return openCounted(false, true); + } + + @Override + public int readInto(int[] dst, int off, int len) throws IOException { + ensureReadable(); + if (dst == null) { + throw new IllegalArgumentException("Destination array is required"); + } + if (off < 0 || len < 0 || off > dst.length - len) { + throw new IllegalArgumentException("readInto range is outside the destination array"); + } + if (cursor.kind != SlotCursor.Kind.ARRAY) { + owner.fail(); + throw new IllegalStateException("readInto is only valid inside an array"); + } + ClickHouseColumn element = StreamTypes.unwrap(cursor.repeated()); + if (element.getDataType() != ClickHouseDataType.Int32) { + owner.fail(); + throw new IllegalStateException("readInto expects Array(Int32) but was " + StreamTypes.describe(element)); + } + int count = Math.min(len, cursor.remaining()); + try { + for (int i = 0; i < count; i++) { + if (element.isNullable()) { + int marker = readMarker(); + if (marker == 1) { + owner.fail(); + throw new NullValueException("Array element is null"); + } + } + dst[off + i] = owner.binary().readIntLE(); + cursor.advance(); + } + } catch (IOException e) { + owner.fail(); + throw e; + } + return count; + } + + @Override + public void skip() throws IOException { + ensureReadable(); + ClickHouseColumn effective = StreamTypes.unwrap(slot()); + if (markerConsumed) { + markerConsumed = false; + } else if (effective.isNullable()) { + int marker = readMarker(); + if (marker == 1) { + cursor.advance(); + return; + } + } + try { + skipBody(effective); + } catch (IOException e) { + owner.fail(); + throw e; + } catch (RuntimeException e) { + owner.fail(); + throw e; + } + cursor.advance(); + } + + @Override + public void close() throws IOException { + if (closed) { + return; + } + if (owner.isFailed()) { + closed = true; + return; + } + if (child != null) { + owner.fail(); + closed = true; + throw new IllegalStateException("Unclosed nested value"); + } + try { + cursor.checkComplete(); + } catch (RuntimeException e) { + owner.fail(); + closed = true; + throw e; + } + closed = true; + if (parent != null) { + parent.child = null; + parent.cursor.advance(); + } else { + owner.onRowClosed(); + } + } + + private ValueReader openCounted(boolean nested, boolean qbit) throws IOException { + ensureReadable(); + ClickHouseColumn slot = slot(); + ClickHouseColumn effective = StreamTypes.unwrap(slot); + ClickHouseColumn element; + if (nested) { + if (effective.getDataType() != ClickHouseDataType.Nested) { + owner.fail(); + throw new IllegalStateException("beginNested is not valid for " + StreamTypes.describe(effective)); + } + element = ClickHouseColumn.tuple(effective.getColumnName(), false, + effective.getNestedColumns().toArray(new ClickHouseColumn[0])); + } else if (qbit) { + if (effective.getDataType() != ClickHouseDataType.QBit) { + owner.fail(); + throw new IllegalStateException("beginQBit is not valid for " + StreamTypes.describe(effective)); + } + element = StreamTypes.arrayElement(effective); + } else { + element = StreamTypes.arrayElement(effective); + if (element == null || effective.getDataType() == ClickHouseDataType.QBit) { + owner.fail(); + throw new IllegalStateException("beginArray is not valid for " + StreamTypes.describe(effective)); + } + } + consumePresent(effective); + int count = readCount(); + if (qbit && count != effective.getPrecision()) { + owner.fail(); + throw new ClientException("QBit column '" + effective.getColumnName() + "' expected exactly " + + effective.getPrecision() + " elements but received " + count); + } + return nest(new StreamReaderScope(owner, this, SlotCursor.array(element, count), -1)); + } + + private ClickHouseColumn prepare(ClickHouseDataType... allowed) throws IOException { + ensureReadable(); + ClickHouseColumn effective = StreamTypes.unwrap(slot()); + if (!StreamTypes.isOneOf(effective.getDataType(), allowed)) { + owner.fail(); + throw new IllegalStateException("Cannot read " + allowed[0] + " from " + StreamTypes.describe(effective)); + } + consumePresent(effective); + return effective; + } + + private void consumePresent(ClickHouseColumn effective) throws IOException { + if (markerConsumed) { + markerConsumed = false; + return; + } + if (effective.isNullable()) { + int marker = readMarker(); + if (marker == 1) { + cursor.advance(); + throw new NullValueException("Column '" + effective.getColumnName() + "' is null"); + } + } + } + + private int readMarker() throws IOException { + int marker = owner.binary().readByte() & 0xFF; + if (marker != 0 && marker != 1) { + owner.fail(); + throw new ClientException("Invalid null marker: " + marker); + } + return marker; + } + + private int readCount() throws IOException { + try { + return BinaryStreamReader.readVarInt(owner.stream()); + } catch (IOException e) { + owner.fail(); + throw e; + } + } + + private ClickHouseColumn slot() { + try { + return cursor.current(); + } catch (RuntimeException e) { + owner.fail(); + throw e; + } + } + + private void ensureReadable() { + if (owner.isFailed()) { + throw new IllegalStateException("Reader is unusable after a previous failure"); + } + if (closed) { + throw new IllegalStateException("Scope is closed"); + } + if (child != null) { + owner.fail(); + throw new IllegalStateException("A nested value is still open"); + } + } + + private byte finishByte(byte value) { + cursor.advance(); + return value; + } + + private short finishShort(short value) { + cursor.advance(); + return value; + } + + private int finishInt(int value) { + cursor.advance(); + return value; + } + + private long finishLong(long value) { + cursor.advance(); + return value; + } + + private float finishFloat(float value) { + cursor.advance(); + return value; + } + + private double finishDouble(double value) { + cursor.advance(); + return value; + } + + private T finish(T value) { + cursor.advance(); + return value; + } + + private StreamReaderScope nest(StreamReaderScope childScope) { + this.child = childScope; + return childScope; + } + + private void skipBody(ClickHouseColumn column) throws IOException { + ClickHouseDataType type = column.getDataType(); + switch (type) { + case Int8: + case UInt8: + case Bool: + case Enum8: + owner.binary().readByte(); + return; + case Int16: + case UInt16: + case Date: + case Enum16: + case BFloat16: + owner.skipFully(2); + return; + case Int32: + case UInt32: + case Float32: + case Date32: + case DateTime: + case DateTime32: + case Decimal32: + case IPv4: + case Time: + owner.skipFully(4); + return; + case Int64: + case UInt64: + case Float64: + case DateTime64: + case Decimal64: + case Time64: + case IntervalYear: + case IntervalQuarter: + case IntervalMonth: + case IntervalWeek: + case IntervalDay: + case IntervalHour: + case IntervalMinute: + case IntervalSecond: + case IntervalMicrosecond: + case IntervalMillisecond: + case IntervalNanosecond: + owner.skipFully(8); + return; + case Int128: + case UInt128: + case Decimal128: + case UUID: + case IPv6: + owner.skipFully(16); + return; + case Int256: + case UInt256: + case Decimal256: + owner.skipFully(32); + return; + case Decimal: + owner.skipFully(StreamTypes.decimalBytes(column.getPrecision())); + return; + case String: + case JSON: + owner.skipFully(BinaryStreamReader.readVarInt(owner.stream())); + return; + case FixedString: + owner.skipFully(column.getPrecision()); + return; + case Nothing: + return; + case Array: + case QBit: + skipRepeated(column.getNestedColumns().get(0), BinaryStreamReader.readVarInt(owner.stream())); + return; + case Nested: + int rows = BinaryStreamReader.readVarInt(owner.stream()); + List fields = column.getNestedColumns(); + for (int r = 0; r < rows; r++) { + for (int f = 0; f < fields.size(); f++) { + skipValue(fields.get(f)); + } + } + return; + case Map: + int pairs = BinaryStreamReader.readVarInt(owner.stream()); + for (int i = 0; i < pairs; i++) { + skipValue(column.getKeyInfo()); + skipValue(column.getValueInfo()); + } + return; + case Tuple: + List tupleFields = column.getNestedColumns(); + for (int i = 0; i < tupleFields.size(); i++) { + skipValue(tupleFields.get(i)); + } + return; + case Variant: + case Geometry: + int ordinal = owner.binary().readByte() & 0xFF; + if (ordinal == 0xFF) { + return; + } + skipValue(column.getNestedColumns().get(ordinal)); + return; + case Point: + owner.skipFully(16); + return; + case Ring: + case LineString: + case MultiPoint: + owner.skipFully(BinaryStreamReader.readVarInt(owner.stream()) * 16); + return; + case Polygon: + case MultiLineString: + skipRings(BinaryStreamReader.readVarInt(owner.stream())); + return; + case MultiPolygon: + int polygons = BinaryStreamReader.readVarInt(owner.stream()); + for (int i = 0; i < polygons; i++) { + skipRings(BinaryStreamReader.readVarInt(owner.stream())); + } + return; + case SimpleAggregateFunction: + case LowCardinality: + skipValue(column.getNestedColumns().get(0)); + return; + case Dynamic: + skipValue(owner.binary().readDynamicType()); + return; + case AggregateFunction: + if (column.getAggregateFunction() != ClickHouseAggregateFunction.groupBitmap) { + throw new ClientException("Cannot skip aggregate function " + column.getAggregateFunction()); + } + ClickHouseBitmap.deserialize(owner.stream(), column.getNestedColumns().get(0).getDataType()); + return; + default: + throw new ClientException("Cannot skip " + StreamTypes.describe(column)); + } + } + + private void skipRepeated(ClickHouseColumn element, int count) throws IOException { + for (int i = 0; i < count; i++) { + skipValue(element); + } + } + + private void skipRings(int count) throws IOException { + for (int i = 0; i < count; i++) { + owner.skipFully(BinaryStreamReader.readVarInt(owner.stream()) * 16); + } + } + + private void skipValue(ClickHouseColumn column) throws IOException { + ClickHouseColumn effective = StreamTypes.unwrap(column); + if (effective.isNullable()) { + int marker = owner.binary().readByte() & 0xFF; + if (marker == 1) { + return; + } + if (marker != 0) { + throw new ClientException("Invalid null marker: " + marker); + } + } + skipBody(effective); + } +} diff --git a/client-v2/src/main/java/com/clickhouse/client/api/data_formats/stream/StreamTypes.java b/client-v2/src/main/java/com/clickhouse/client/api/data_formats/stream/StreamTypes.java new file mode 100644 index 000000000..603ce6d20 --- /dev/null +++ b/client-v2/src/main/java/com/clickhouse/client/api/data_formats/stream/StreamTypes.java @@ -0,0 +1,137 @@ +package com.clickhouse.client.api.data_formats.stream; + +import com.clickhouse.client.api.ClientException; +import com.clickhouse.data.ClickHouseColumn; +import com.clickhouse.data.ClickHouseDataType; + +import java.util.ArrayDeque; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; + +/** + * Shared column shapes for the streaming reader and writer. + */ +final class StreamTypes { + + static final List POINT_FIELDS = Collections.unmodifiableList(Arrays.asList( + ClickHouseColumn.of("x", "Float64"), + ClickHouseColumn.of("y", "Float64"))); + + static final ClickHouseColumn POINT = ClickHouseColumn.of("p", "Point"); + + static final ClickHouseColumn RING = ClickHouseColumn.of("r", "Ring"); + + static final ClickHouseColumn POLYGON = ClickHouseColumn.of("g", "Polygon"); + + static final ClickHouseColumn NOTHING = ClickHouseColumn.of("v", "Nothing"); + + private StreamTypes() { + } + + static boolean omitted(ClickHouseColumn column) { + return column.hasDefault() && column.getDefaultValue() != ClickHouseColumn.DefaultValue.DEFAULT; + } + + static ClickHouseColumn unwrap(ClickHouseColumn column) { + if (column.getDataType() == ClickHouseDataType.SimpleAggregateFunction) { + return column.getNestedColumns().get(0); + } + return column; + } + + static boolean isOneOf(ClickHouseDataType actual, ClickHouseDataType... allowed) { + for (int i = 0; i < allowed.length; i++) { + if (actual == allowed[i]) { + return true; + } + } + return false; + } + + static String describe(ClickHouseColumn column) { + String type = column.getOriginalTypeName(); + if (type == null || type.isEmpty()) { + type = column.getDataType().name(); + } + String name = column.getColumnName(); + if (name == null || name.isEmpty()) { + return type; + } + return name + " " + type; + } + + /** + * Element column of a length-prefixed array-like value, or null when {@code column} is not one. + */ + static ClickHouseColumn arrayElement(ClickHouseColumn column) { + ClickHouseDataType type = column.getDataType(); + if (type == ClickHouseDataType.Array || type == ClickHouseDataType.QBit) { + return column.getNestedColumns().get(0); + } + if (type == ClickHouseDataType.Ring || type == ClickHouseDataType.LineString + || type == ClickHouseDataType.MultiPoint) { + return POINT; + } + if (type == ClickHouseDataType.Polygon || type == ClickHouseDataType.MultiLineString) { + return RING; + } + if (type == ClickHouseDataType.MultiPolygon) { + return POLYGON; + } + return null; + } + + /** + * Tuple fields, or the two coordinates of a {@code Point}. Null when {@code column} is neither. + */ + static List tupleFields(ClickHouseColumn column) { + if (column.getDataType() == ClickHouseDataType.Tuple) { + return column.getNestedColumns(); + } + if (column.getDataType() == ClickHouseDataType.Point) { + return POINT_FIELDS; + } + return null; + } + + static int decimalBytes(int precision) { + if (precision <= ClickHouseDataType.Decimal32.getMaxScale()) { + return 4; + } + if (precision <= ClickHouseDataType.Decimal64.getMaxScale()) { + return 8; + } + if (precision <= ClickHouseDataType.Decimal128.getMaxScale()) { + return 16; + } + return 32; + } + + /** + * Rejects a runtime type whose dynamic tag encoding is incomplete and would desynchronize the stream. + */ + static void rejectDynamicPayload(ClickHouseColumn column) { + ArrayDeque pending = new ArrayDeque(); + pending.add(column); + while (!pending.isEmpty()) { + ClickHouseColumn current = pending.removeFirst(); + switch (current.getDataType()) { + case Tuple: + case Variant: + case Dynamic: + case JSON: + case Nested: + case QBit: + case AggregateFunction: + case SimpleAggregateFunction: + case Geometry: + throw new ClientException("Serializing a " + current.getDataType().name() + + " value inside a Dynamic column is not supported"); + default: + break; + } + pending.addAll(current.getNestedColumns()); + } + } +} diff --git a/client-v2/src/main/java/com/clickhouse/client/api/data_formats/stream/StreamWriterScope.java b/client-v2/src/main/java/com/clickhouse/client/api/data_formats/stream/StreamWriterScope.java new file mode 100644 index 000000000..44fa8d736 --- /dev/null +++ b/client-v2/src/main/java/com/clickhouse/client/api/data_formats/stream/StreamWriterScope.java @@ -0,0 +1,1228 @@ +package com.clickhouse.client.api.data_formats.stream; + +import com.clickhouse.client.api.data_formats.internal.SerializerUtils; +import com.clickhouse.data.ClickHouseAggregateFunction; +import com.clickhouse.data.ClickHouseColumn; +import com.clickhouse.data.ClickHouseDataType; +import com.clickhouse.data.format.BinaryStreamUtils; +import com.clickhouse.data.value.ClickHouseBitmap; + +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.OutputStream; +import java.math.BigDecimal; +import java.math.BigInteger; +import java.net.Inet4Address; +import java.net.Inet6Address; +import java.nio.ByteBuffer; +import java.time.Instant; +import java.time.LocalDate; +import java.time.LocalDateTime; +import java.time.ZoneId; +import java.time.ZoneOffset; +import java.time.ZonedDateTime; +import java.util.List; +import java.util.UUID; + +/** + * Scope implementation shared by a row and every nested container. + */ +final class StreamWriterScope implements ValueWriter { + + private final RowBinaryStreamWriter owner; + + private final StreamWriterScope parent; + + private final SlotCursor cursor; + + private final OutputStream out; + + private final ByteArrayOutputStream buffer; + + /** JSON column whose typed paths apply, when this scope is a JSON object. */ + private ClickHouseColumn jsonType; + + /** Dotted prefix contributed by enclosing {@code beginObject} calls. */ + private String pathPrefix = ""; + + /** Nested object whose leaf paths belong to the parent JSON value. */ + private boolean flattensIntoParent; + + private StreamWriterScope child; + + private boolean closed; + + private static final ClickHouseColumn DYN_STRING = ClickHouseColumn.of("v", "String"); + + private static final ClickHouseColumn DYN_INT32 = ClickHouseColumn.of("v", "Int32"); + + private static final ClickHouseColumn DYN_INT64 = ClickHouseColumn.of("v", "Int64"); + + private static final ClickHouseColumn DYN_FLOAT64 = ClickHouseColumn.of("v", "Float64"); + + private static final ClickHouseColumn DYN_BOOL = ClickHouseColumn.of("v", "Bool"); + + private static final ClickHouseColumn DYN_JSON = ClickHouseColumn.of("v", "JSON"); + + private StreamWriterScope(RowBinaryStreamWriter owner, StreamWriterScope parent, SlotCursor cursor, + OutputStream out, ByteArrayOutputStream buffer) { + this.owner = owner; + this.parent = parent; + this.cursor = cursor; + this.out = out; + this.buffer = buffer; + } + + static StreamWriterScope row(RowBinaryStreamWriter owner, List columns) { + return new StreamWriterScope(owner, null, SlotCursor.row(columns), owner.out(), null); + } + + boolean isClosed() { + return closed; + } + + @Override + public void writeInt8(byte value) throws IOException { + prepare(ClickHouseDataType.Int8); + body(() -> BinaryStreamUtils.writeInt8(out, value)); + } + + @Override + public void writeInt8(int value) throws IOException { + checkRange(value, Byte.MIN_VALUE, Byte.MAX_VALUE, "Int8"); + writeInt8((byte) value); + } + + @Override + public void writeInt16(short value) throws IOException { + prepare(ClickHouseDataType.Int16); + body(() -> BinaryStreamUtils.writeInt16(out, value)); + } + + @Override + public void writeInt16(int value) throws IOException { + checkRange(value, Short.MIN_VALUE, Short.MAX_VALUE, "Int16"); + writeInt16((short) value); + } + + @Override + public void writeInt32(int value) throws IOException { + prepare(ClickHouseDataType.Int32); + body(() -> BinaryStreamUtils.writeInt32(out, value)); + } + + @Override + public void writeInt32(long value) throws IOException { + if (value != (int) value) { + owner.fail(); + throw new IllegalArgumentException("Int32 overflow: " + value); + } + writeInt32((int) value); + } + + @Override + public void writeInt64(long value) throws IOException { + prepare(ClickHouseDataType.Int64); + body(() -> BinaryStreamUtils.writeInt64(out, value)); + } + + @Override + public void writeInt128(BigInteger value) throws IOException { + requireValue(value); + prepare(ClickHouseDataType.Int128); + body(() -> BinaryStreamUtils.writeInt128(out, value)); + } + + @Override + public void writeInt256(BigInteger value) throws IOException { + requireValue(value); + prepare(ClickHouseDataType.Int256); + body(() -> BinaryStreamUtils.writeInt256(out, value)); + } + + @Override + public void writeUInt8(int value) throws IOException { + checkRange(value, 0, 255, "UInt8"); + prepare(ClickHouseDataType.UInt8); + body(() -> BinaryStreamUtils.writeUnsignedInt8(out, value)); + } + + @Override + public void writeUInt16(int value) throws IOException { + checkRange(value, 0, 65535, "UInt16"); + prepare(ClickHouseDataType.UInt16); + body(() -> BinaryStreamUtils.writeUnsignedInt16(out, value)); + } + + @Override + public void writeUInt32(long value) throws IOException { + if (value < 0 || value > 0xFFFFFFFFL) { + owner.fail(); + throw new IllegalArgumentException("UInt32 overflow: " + value); + } + prepare(ClickHouseDataType.UInt32); + body(() -> BinaryStreamUtils.writeUnsignedInt32(out, value)); + } + + @Override + public void writeUInt64(long value) throws IOException { + prepare(ClickHouseDataType.UInt64); + body(() -> BinaryStreamUtils.writeUnsignedInt64(out, value)); + } + + @Override + public void writeUInt64(BigInteger value) throws IOException { + requireValue(value); + prepare(ClickHouseDataType.UInt64); + body(() -> BinaryStreamUtils.writeUnsignedInt64(out, value)); + } + + @Override + public void writeUInt128(BigInteger value) throws IOException { + requireValue(value); + prepare(ClickHouseDataType.UInt128); + body(() -> BinaryStreamUtils.writeUnsignedInt128(out, value)); + } + + @Override + public void writeUInt256(BigInteger value) throws IOException { + requireValue(value); + prepare(ClickHouseDataType.UInt256); + body(() -> BinaryStreamUtils.writeUnsignedInt256(out, value)); + } + + @Override + public void writeFloat32(float value) throws IOException { + prepare(ClickHouseDataType.Float32); + body(() -> BinaryStreamUtils.writeFloat32(out, value)); + } + + @Override + public void writeFloat64(double value) throws IOException { + prepare(ClickHouseDataType.Float64); + body(() -> BinaryStreamUtils.writeFloat64(out, value)); + } + + @Override + public void writeBFloat16(float value) throws IOException { + prepare(ClickHouseDataType.BFloat16); + body(() -> BinaryStreamUtils.writeBFloat16(out, value)); + } + + @Override + public void writeBool(boolean value) throws IOException { + prepare(ClickHouseDataType.Bool); + body(() -> BinaryStreamUtils.writeBoolean(out, value)); + } + + @Override + public void writeString(String value) throws IOException { + if (value == null) { + writeNullFor(ClickHouseDataType.String, ClickHouseDataType.JSON); + return; + } + prepare(ClickHouseDataType.String, ClickHouseDataType.JSON); + body(() -> BinaryStreamUtils.writeString(out, value)); + } + + @Override + public void writeString(byte[] value) throws IOException { + if (value == null) { + writeNullFor(ClickHouseDataType.String, ClickHouseDataType.JSON); + return; + } + prepare(ClickHouseDataType.String, ClickHouseDataType.JSON); + body(() -> BinaryStreamUtils.writeString(out, value)); + } + + @Override + public void writeString(ByteBuffer value) throws IOException { + if (value == null) { + writeString((byte[]) null); + return; + } + ByteBuffer view = value.duplicate(); + int length = view.remaining(); + prepare(ClickHouseDataType.String, ClickHouseDataType.JSON); + body(() -> { + SerializerUtils.writeVarInt(out, length); + if (view.hasArray()) { + out.write(view.array(), view.arrayOffset() + view.position(), length); + } else { + byte[] copy = new byte[length]; + view.get(copy); + out.write(copy); + } + }); + } + + @Override + public void writeFixedString(String value) throws IOException { + if (value == null) { + writeNullFor(ClickHouseDataType.FixedString); + return; + } + ClickHouseColumn column = prepare(ClickHouseDataType.FixedString); + int length = column.getPrecision(); + body(() -> BinaryStreamUtils.writeFixedString(out, value, length)); + } + + @Override + public void writeFixedString(byte[] value) throws IOException { + if (value == null) { + writeNullFor(ClickHouseDataType.FixedString); + return; + } + ClickHouseColumn column = prepare(ClickHouseDataType.FixedString); + int length = column.getPrecision(); + body(() -> SerializerUtils.writeFixedStringBytes(out, value, length)); + } + + @Override + public void writeDate(LocalDate value) throws IOException { + requireValue(value); + prepare(ClickHouseDataType.Date); + body(() -> SerializerUtils.writeDate(out, value, ZoneOffset.UTC)); + } + + @Override + public void writeDate32(LocalDate value) throws IOException { + requireValue(value); + prepare(ClickHouseDataType.Date32); + body(() -> SerializerUtils.writeDate32(out, value, ZoneOffset.UTC)); + } + + @Override + public void writeDateTime(LocalDateTime value) throws IOException { + requireValue(value); + ClickHouseColumn column = prepare(ClickHouseDataType.DateTime, ClickHouseDataType.DateTime32); + ZoneId zone = zone(column); + body(() -> SerializerUtils.writeDateTime(out, value, zone)); + } + + @Override + public void writeDateTime(ZonedDateTime value) throws IOException { + requireValue(value); + ClickHouseColumn column = prepare(ClickHouseDataType.DateTime, ClickHouseDataType.DateTime32); + ZoneId zone = zone(column); + body(() -> SerializerUtils.writeDateTime(out, value, zone)); + } + + @Override + public void writeDateTime(Instant value) throws IOException { + requireValue(value); + ClickHouseColumn column = prepare(ClickHouseDataType.DateTime, ClickHouseDataType.DateTime32); + ZoneId zone = zone(column); + body(() -> SerializerUtils.writeDateTime(out, value, zone)); + } + + @Override + public void writeDateTime64(LocalDateTime value) throws IOException { + requireValue(value); + ClickHouseColumn column = prepare(ClickHouseDataType.DateTime64); + int scale = column.getScale(); + ZoneId zone = zone(column); + body(() -> SerializerUtils.writeDateTime64(out, value, scale, zone)); + } + + @Override + public void writeDateTime64(ZonedDateTime value) throws IOException { + requireValue(value); + ClickHouseColumn column = prepare(ClickHouseDataType.DateTime64); + int scale = column.getScale(); + ZoneId zone = zone(column); + body(() -> SerializerUtils.writeDateTime64(out, value, scale, zone)); + } + + @Override + public void writeUUID(UUID value) throws IOException { + requireValue(value); + prepare(ClickHouseDataType.UUID); + body(() -> BinaryStreamUtils.writeUuid(out, value)); + } + + @Override + public void writeIPv4(Inet4Address value) throws IOException { + requireValue(value); + prepare(ClickHouseDataType.IPv4); + body(() -> BinaryStreamUtils.writeInet4Address(out, value)); + } + + @Override + public void writeIPv6(Inet6Address value) throws IOException { + requireValue(value); + prepare(ClickHouseDataType.IPv6); + body(() -> BinaryStreamUtils.writeInet6Address(out, value)); + } + + @Override + public void writeDecimal(BigDecimal value) throws IOException { + requireValue(value); + ClickHouseColumn column = prepare(ClickHouseDataType.Decimal, ClickHouseDataType.Decimal32, + ClickHouseDataType.Decimal64, ClickHouseDataType.Decimal128, ClickHouseDataType.Decimal256); + int precision = column.getPrecision(); + int scale = column.getScale(); + body(() -> BinaryStreamUtils.writeDecimal(out, value, precision, scale)); + } + + @Override + public void writeEnum8(int value) throws IOException { + checkRange(value, Byte.MIN_VALUE, Byte.MAX_VALUE, "Enum8"); + prepare(ClickHouseDataType.Enum8); + body(() -> BinaryStreamUtils.writeInt8(out, value)); + } + + @Override + public void writeEnum8(String name) throws IOException { + writeEnum(name, ClickHouseDataType.Enum8); + } + + @Override + public void writeEnum16(int value) throws IOException { + checkRange(value, Short.MIN_VALUE, Short.MAX_VALUE, "Enum16"); + prepare(ClickHouseDataType.Enum16); + body(() -> BinaryStreamUtils.writeInt16(out, value)); + } + + @Override + public void writeEnum16(String name) throws IOException { + writeEnum(name, ClickHouseDataType.Enum16); + } + + @Override + public void writeNull() throws IOException { + ensureWritable(); + if (cursor.kind == SlotCursor.Kind.MAP && !cursor.expectingValue()) { + owner.fail(); + throw new IllegalArgumentException("Map keys are not nullable"); + } + ClickHouseColumn slot = slot(); + ClickHouseColumn effective = StreamTypes.unwrap(slot); + if (effective.isNullable()) { + try { + if (cursor.kind == SlotCursor.Kind.ROW && owner.defaultsSupport()) { + SerializerUtils.writeNonNull(out); + } + SerializerUtils.writeNull(out); + } catch (IOException e) { + owner.fail(); + throw e; + } + cursor.advance(); + return; + } + if (effective.getDataType() == ClickHouseDataType.Dynamic) { + writePresent(slot); + body(() -> SerializerUtils.writeDynamicTypeTag(out, StreamTypes.NOTHING)); + return; + } + if (effective.getDataType() == ClickHouseDataType.Variant + || effective.getDataType() == ClickHouseDataType.Geometry) { + writePresent(slot); + body(() -> out.write(0xFF)); + return; + } + owner.fail(); + throw new IllegalArgumentException("Cannot write null into non-nullable " + StreamTypes.describe(effective)); + } + + @Override + public void writeDefault() throws IOException { + ensureWritable(); + if (cursor.kind != SlotCursor.Kind.ROW || !owner.defaultsSupport()) { + owner.fail(); + throw new IllegalStateException("writeDefault is only valid for a top-level column of RowBinaryWithDefaults"); + } + ClickHouseColumn slot = slot(); + if (!slot.hasDefault() || slot.getDefaultValue() != ClickHouseColumn.DefaultValue.DEFAULT) { + owner.fail(); + throw new IllegalArgumentException("Column '" + slot.getColumnName() + "' has no DEFAULT"); + } + try { + SerializerUtils.writeNull(out); + } catch (IOException e) { + owner.fail(); + throw e; + } + cursor.advance(); + } + + @Override + public void writeBitmap(ClickHouseBitmap value) throws IOException { + if (value == null) { + owner.fail(); + throw new IllegalArgumentException("groupBitmap value is null"); + } + ensureWritable(); + ClickHouseColumn slot = slot(); + ClickHouseColumn effective = StreamTypes.unwrap(slot); + if (effective.getDataType() != ClickHouseDataType.AggregateFunction + || effective.getAggregateFunction() != ClickHouseAggregateFunction.groupBitmap) { + owner.fail(); + throw new IllegalStateException("writeBitmap is only valid for AggregateFunction(groupBitmap, ...) but was " + + StreamTypes.describe(effective)); + } + writePresent(slot); + body(() -> out.write(value.toBytes())); + } + + @Override + public ValueWriter beginArray() throws IOException { + return openArray(SlotCursor.UNKNOWN); + } + + @Override + public ValueWriter beginArray(int size) throws IOException { + if (size < 0) { + owner.fail(); + throw new IllegalArgumentException("Array size is negative"); + } + return openArray(size); + } + + @Override + public ValueWriter beginTuple() throws IOException { + ensureWritable(); + ClickHouseColumn slot = slot(); + ClickHouseColumn effective = StreamTypes.unwrap(slot); + List fields = StreamTypes.tupleFields(effective); + if (fields == null) { + owner.fail(); + throw new IllegalStateException("beginTuple is not valid for " + StreamTypes.describe(effective)); + } + writePresent(slot); + return nest(new StreamWriterScope(owner, this, SlotCursor.tuple(fields), out, null)); + } + + @Override + public ValueWriter beginMap() throws IOException { + return openMap(SlotCursor.UNKNOWN); + } + + @Override + public ValueWriter beginMap(int pairs) throws IOException { + if (pairs < 0) { + owner.fail(); + throw new IllegalArgumentException("Map size is negative"); + } + return openMap(pairs); + } + + @Override + public ValueWriter beginNested() throws IOException { + return openNested(SlotCursor.UNKNOWN); + } + + @Override + public ValueWriter beginNested(int rows) throws IOException { + if (rows < 0) { + owner.fail(); + throw new IllegalArgumentException("Nested row count is negative"); + } + return openNested(rows); + } + + @Override + public ValueWriter beginVariant(int ordinal) throws IOException { + if (ordinal < 0) { + owner.fail(); + throw new IllegalArgumentException("Variant ordinal is negative"); + } + ensureWritable(); + ClickHouseColumn slot = slot(); + ClickHouseColumn effective = StreamTypes.unwrap(slot); + if (effective.getDataType() != ClickHouseDataType.Variant + && effective.getDataType() != ClickHouseDataType.Geometry) { + owner.fail(); + throw new IllegalStateException("beginVariant is not valid for " + StreamTypes.describe(effective)); + } + List alternatives = effective.getNestedColumns(); + if (ordinal >= alternatives.size()) { + owner.fail(); + throw new IllegalArgumentException("Variant ordinal " + ordinal + " is out of range for " + + StreamTypes.describe(effective)); + } + writePresent(slot); + try { + BinaryStreamUtils.writeUnsignedInt8(out, ordinal); + } catch (IOException e) { + owner.fail(); + throw e; + } + return nest(new StreamWriterScope(owner, this, SlotCursor.single(alternatives.get(ordinal)), out, null)); + } + + @Override + public ValueWriter beginDynamic(ClickHouseColumn runtimeType) throws IOException { + if (runtimeType == null) { + owner.fail(); + throw new IllegalArgumentException("Dynamic runtime type is required"); + } + ensureWritable(); + ClickHouseColumn slot = slot(); + ClickHouseColumn effective = StreamTypes.unwrap(slot); + if (effective.getDataType() != ClickHouseDataType.Dynamic) { + owner.fail(); + throw new IllegalStateException("beginDynamic is not valid for " + StreamTypes.describe(effective)); + } + try { + StreamTypes.rejectDynamicPayload(runtimeType); + } catch (RuntimeException e) { + owner.fail(); + throw e; + } + writePresent(slot); + try { + SerializerUtils.writeDynamicTypeTag(out, runtimeType); + } catch (IOException e) { + owner.fail(); + throw e; + } catch (RuntimeException e) { + owner.fail(); + throw e; + } + StreamWriterScope childScope = new StreamWriterScope(owner, this, SlotCursor.single(runtimeType), out, null); + if (StreamTypes.unwrap(runtimeType).getDataType() == ClickHouseDataType.Nothing) { + childScope.cursor.forceComplete(); + } + return nest(childScope); + } + + @Override + public ValueWriter beginJson() throws IOException { + return openDocument(SlotCursor.UNKNOWN); + } + + @Override + public ValueWriter beginJson(int paths) throws IOException { + if (paths < 0) { + owner.fail(); + throw new IllegalArgumentException("JSON path count is negative"); + } + return openDocument(paths); + } + + private ValueWriter openDocument(int paths) throws IOException { + if (cursor.kind == SlotCursor.Kind.JSON) { + owner.fail(); + throw new IllegalStateException("Inside a JSON object use beginJson(path, paths)"); + } + ensureWritable(); + ClickHouseColumn slot = slot(); + ClickHouseColumn effective = StreamTypes.unwrap(slot); + if (effective.getDataType() != ClickHouseDataType.JSON) { + owner.fail(); + throw new IllegalStateException("beginJson is not valid for " + StreamTypes.describe(effective)); + } + writePresent(slot); + return openJsonValue(effective, "", paths); + } + + @Override + public void writeString(String path, String value) throws IOException { + if (value == null) { + writeNull(path); + return; + } + writeJsonScalar(path, ClickHouseDataType.String, DYN_STRING, () -> BinaryStreamUtils.writeString(out, value)); + } + + @Override + public void writeInt32(String path, int value) throws IOException { + writeJsonScalar(path, ClickHouseDataType.Int32, DYN_INT32, () -> BinaryStreamUtils.writeInt32(out, value)); + } + + @Override + public void writeInt64(String path, long value) throws IOException { + writeJsonScalar(path, ClickHouseDataType.Int64, DYN_INT64, () -> BinaryStreamUtils.writeInt64(out, value)); + } + + @Override + public void writeFloat64(String path, double value) throws IOException { + writeJsonScalar(path, ClickHouseDataType.Float64, DYN_FLOAT64, () -> BinaryStreamUtils.writeFloat64(out, value)); + } + + @Override + public void writeBool(String path, boolean value) throws IOException { + writeJsonScalar(path, ClickHouseDataType.Bool, DYN_BOOL, () -> BinaryStreamUtils.writeBoolean(out, value)); + } + + @Override + public void writeNull(String path) throws IOException { + ensureJson(); + String full = fullPath(path); + ClickHouseColumn typed = typedPath(full); + writePathName(full); + try { + if (typed == null) { + SerializerUtils.writeDynamicTypeTag(out, StreamTypes.NOTHING); + } else if (typed.isNullable()) { + SerializerUtils.writeNull(out); + } else { + throw new IllegalArgumentException("Cannot write null into non-nullable JSON path '" + full + "'"); + } + } catch (IOException e) { + owner.fail(); + throw e; + } catch (RuntimeException e) { + owner.fail(); + throw e; + } + cursor.advance(); + } + + @Override + public ValueWriter beginObject(String path) throws IOException { + return openObject(path, SlotCursor.UNKNOWN); + } + + @Override + public ValueWriter beginObject(String path, int paths) throws IOException { + if (paths < 0) { + owner.fail(); + throw new IllegalArgumentException("JSON path count is negative"); + } + return openObject(path, paths); + } + + private ValueWriter openObject(String path, int paths) throws IOException { + ensureJson(); + StreamWriterScope nested = new StreamWriterScope(owner, this, SlotCursor.json(paths), out, null); + nested.jsonType = jsonType; + nested.pathPrefix = fullPath(path) + "."; + nested.flattensIntoParent = true; + return nest(nested); + } + + @Override + public ValueWriter beginArray(String path, ClickHouseColumn elementType) throws IOException { + return beginJsonArray(path, elementType, SlotCursor.UNKNOWN); + } + + @Override + public ValueWriter beginArray(String path, ClickHouseColumn elementType, int size) throws IOException { + if (size < 0) { + owner.fail(); + throw new IllegalArgumentException("Array size is negative"); + } + return beginJsonArray(path, elementType, size); + } + + @Override + public ValueWriter beginJson(String path) throws IOException { + return beginNestedJson(path, SlotCursor.UNKNOWN); + } + + @Override + public ValueWriter beginJson(String path, int paths) throws IOException { + if (paths < 0) { + owner.fail(); + throw new IllegalArgumentException("JSON path count is negative"); + } + return beginNestedJson(path, paths); + } + + @Override + public ValueWriter beginQBit() throws IOException { + ensureWritable(); + ClickHouseColumn effective = StreamTypes.unwrap(slot()); + if (effective.getDataType() != ClickHouseDataType.QBit) { + owner.fail(); + throw new IllegalStateException("beginQBit is not valid for " + StreamTypes.describe(effective)); + } + return beginArray(effective.getPrecision()); + } + + @Override + public void writeInt8Array(byte[] values) throws IOException { + writePrimitiveArray(values, ClickHouseDataType.Int8, (index) -> BinaryStreamUtils.writeInt8(out, values[index])); + } + + @Override + public void writeInt16Array(short[] values) throws IOException { + writePrimitiveArray(values, values == null ? 0 : values.length, ClickHouseDataType.Int16, + (index) -> BinaryStreamUtils.writeInt16(out, values[index])); + } + + @Override + public void writeInt32Array(int[] values) throws IOException { + writePrimitiveArray(values, values == null ? 0 : values.length, ClickHouseDataType.Int32, + (index) -> BinaryStreamUtils.writeInt32(out, values[index])); + } + + @Override + public void writeInt64Array(long[] values) throws IOException { + writePrimitiveArray(values, values == null ? 0 : values.length, ClickHouseDataType.Int64, + (index) -> BinaryStreamUtils.writeInt64(out, values[index])); + } + + @Override + public void writeFloat32Array(float[] values) throws IOException { + writePrimitiveArray(values, values == null ? 0 : values.length, ClickHouseDataType.Float32, + (index) -> BinaryStreamUtils.writeFloat32(out, values[index])); + } + + @Override + public void writeFloat64Array(double[] values) throws IOException { + writePrimitiveArray(values, values == null ? 0 : values.length, ClickHouseDataType.Float64, + (index) -> BinaryStreamUtils.writeFloat64(out, values[index])); + } + + @Override + public void writeBFloat16Array(float[] values) throws IOException { + writePrimitiveArray(values, values == null ? 0 : values.length, ClickHouseDataType.BFloat16, + (index) -> BinaryStreamUtils.writeBFloat16(out, values[index])); + } + + @Override + public void writeBoolArray(boolean[] values) throws IOException { + writePrimitiveArray(values, values == null ? 0 : values.length, ClickHouseDataType.Bool, + (index) -> BinaryStreamUtils.writeBoolean(out, values[index])); + } + + @Override + public void close() throws IOException { + if (closed) { + return; + } + if (owner.isFailed()) { + closed = true; + return; + } + if (child != null) { + owner.fail(); + closed = true; + throw new IllegalStateException("Unclosed nested value"); + } + if (flattensIntoParent) { + try { + cursor.checkComplete(); + } catch (RuntimeException e) { + owner.fail(); + closed = true; + throw e; + } + closed = true; + parent.child = null; + parent.cursor.credit(cursor.completed()); + return; + } + try { + cursor.checkComplete(); + if (buffer != null) { + SerializerUtils.writeVarInt(parent.out, cursor.completed()); + buffer.writeTo(parent.out); + } + } catch (RuntimeException e) { + owner.fail(); + closed = true; + throw e; + } catch (IOException e) { + owner.fail(); + closed = true; + throw e; + } + closed = true; + if (parent != null) { + parent.child = null; + parent.cursor.advance(); + } else { + owner.onRowClosed(); + } + } + + private void writeEnum(String name, ClickHouseDataType type) throws IOException { + if (name == null) { + writeNullFor(type); + return; + } + ensureWritable(); + ClickHouseColumn slot = slot(); + ClickHouseColumn effective = StreamTypes.unwrap(slot); + if (effective.getDataType() != type) { + owner.fail(); + throw new IllegalStateException("Cannot write " + type + " into " + StreamTypes.describe(effective)); + } + int wire; + try { + wire = effective.getEnumConstants().value(name); + } catch (IllegalArgumentException e) { + owner.fail(); + throw e; + } + writePresent(slot); + if (type == ClickHouseDataType.Enum8) { + body(() -> BinaryStreamUtils.writeInt8(out, wire)); + } else { + body(() -> BinaryStreamUtils.writeInt16(out, wire)); + } + } + + private void writeNullFor(ClickHouseDataType... allowed) throws IOException { + ensureWritable(); + ClickHouseColumn effective = StreamTypes.unwrap(slot()); + if (!StreamTypes.isOneOf(effective.getDataType(), allowed)) { + owner.fail(); + throw new IllegalStateException("Cannot write " + allowed[0] + " into " + StreamTypes.describe(effective)); + } + writeNull(); + } + + private ValueWriter openJsonValue(ClickHouseColumn type, String prefix, int paths) throws IOException { + OutputStream destination = out; + ByteArrayOutputStream childBuffer = null; + if (paths == SlotCursor.UNKNOWN) { + childBuffer = new ByteArrayOutputStream(); + destination = childBuffer; + } else { + try { + SerializerUtils.writeVarInt(out, paths); + } catch (IOException e) { + owner.fail(); + throw e; + } + } + StreamWriterScope nested = new StreamWriterScope(owner, this, SlotCursor.json(paths), destination, childBuffer); + nested.jsonType = type; + nested.pathPrefix = prefix; + return nest(nested); + } + + private void writeJsonScalar(String path, ClickHouseDataType type, ClickHouseColumn dynamicType, IOAction action) + throws IOException { + ensureJson(); + String full = fullPath(path); + ClickHouseColumn typed = typedPath(full); + writePathName(full); + try { + if (typed == null) { + SerializerUtils.writeDynamicTypeTag(out, dynamicType); + action.run(); + } else { + ClickHouseColumn effective = StreamTypes.unwrap(typed); + if (effective.getDataType() != type) { + throw new IllegalStateException("Cannot write " + type + " into JSON path '" + full + + "' of type " + typed.getOriginalTypeName()); + } + if (typed.isNullable()) { + SerializerUtils.writeNonNull(out); + } + action.run(); + } + } catch (IOException e) { + owner.fail(); + throw e; + } catch (RuntimeException e) { + owner.fail(); + throw e; + } + cursor.advance(); + } + + private ValueWriter beginJsonArray(String path, ClickHouseColumn elementType, int size) throws IOException { + ensureJson(); + if (elementType == null) { + owner.fail(); + throw new IllegalArgumentException("JSON array element type is required"); + } + String full = fullPath(path); + ClickHouseColumn typed = typedPath(full); + writePathName(full); + ClickHouseColumn element = elementType; + try { + if (typed == null) { + ClickHouseColumn arrayType = ClickHouseColumn.of("v", "Array(" + elementType.getOriginalTypeName() + ")"); + SerializerUtils.writeDynamicTypeTag(out, arrayType); + element = StreamTypes.arrayElement(arrayType); + } else { + ClickHouseColumn effective = StreamTypes.unwrap(typed); + if (effective.getDataType() != ClickHouseDataType.Array) { + throw new IllegalStateException("Cannot write an array into JSON path '" + full + + "' of type " + typed.getOriginalTypeName()); + } + element = StreamTypes.arrayElement(effective); + if (typed.isNullable()) { + SerializerUtils.writeNonNull(out); + } + } + } catch (IOException e) { + owner.fail(); + throw e; + } catch (RuntimeException e) { + owner.fail(); + throw e; + } + return openLengthPrefixed(SlotCursor.array(element, size), size); + } + + private ValueWriter beginNestedJson(String path, int paths) throws IOException { + ensureJson(); + String full = fullPath(path); + ClickHouseColumn typed = typedPath(full); + writePathName(full); + ClickHouseColumn nestedType = DYN_JSON; + try { + if (typed == null) { + SerializerUtils.writeDynamicTypeTag(out, DYN_JSON); + } else { + nestedType = StreamTypes.unwrap(typed); + if (nestedType.getDataType() != ClickHouseDataType.JSON) { + throw new IllegalStateException("Cannot write JSON into path '" + full + + "' of type " + typed.getOriginalTypeName()); + } + if (typed.isNullable()) { + SerializerUtils.writeNonNull(out); + } + } + } catch (IOException e) { + owner.fail(); + throw e; + } catch (RuntimeException e) { + owner.fail(); + throw e; + } + return openJsonValue(nestedType, "", paths); + } + + private void ensureJson() { + ensureWritable(); + if (cursor.kind != SlotCursor.Kind.JSON) { + owner.fail(); + throw new IllegalStateException("A JSON path is written inside beginJson or beginObject"); + } + } + + private String fullPath(String path) { + if (path == null || path.isEmpty()) { + owner.fail(); + throw new IllegalArgumentException("JSON path is empty"); + } + return pathPrefix + path; + } + + private ClickHouseColumn typedPath(String full) { + if (jsonType == null || jsonType.getJsonPredefinedPaths() == null) { + return null; + } + return jsonType.getJsonPredefinedPaths().get(full); + } + + private void writePathName(String full) throws IOException { + try { + BinaryStreamUtils.writeString(out, full); + } catch (IOException e) { + owner.fail(); + throw e; + } + } + + private ValueWriter openArray(int size) throws IOException { + ensureWritable(); + ClickHouseColumn slot = slot(); + ClickHouseColumn effective = StreamTypes.unwrap(slot); + ClickHouseColumn element = StreamTypes.arrayElement(effective); + if (element == null) { + owner.fail(); + throw new IllegalStateException("beginArray is not valid for " + StreamTypes.describe(effective)); + } + if (effective.getDataType() == ClickHouseDataType.QBit && size != effective.getPrecision()) { + owner.fail(); + throw new IllegalArgumentException("QBit column '" + effective.getColumnName() + + "' expects exactly " + effective.getPrecision() + " elements but got " + + (size == SlotCursor.UNKNOWN ? "an unknown size" : Integer.toString(size))); + } + writePresent(slot); + return openLengthPrefixed(SlotCursor.array(element, size), size); + } + + private ValueWriter openNested(int rows) throws IOException { + ensureWritable(); + ClickHouseColumn slot = slot(); + ClickHouseColumn effective = StreamTypes.unwrap(slot); + if (effective.getDataType() != ClickHouseDataType.Nested) { + owner.fail(); + throw new IllegalStateException("beginNested is not valid for " + StreamTypes.describe(effective)); + } + ClickHouseColumn tuple = ClickHouseColumn.tuple(effective.getColumnName(), false, + effective.getNestedColumns().toArray(new ClickHouseColumn[0])); + writePresent(slot); + return openLengthPrefixed(SlotCursor.array(tuple, rows), rows); + } + + private ValueWriter openMap(int pairs) throws IOException { + ensureWritable(); + ClickHouseColumn slot = slot(); + ClickHouseColumn effective = StreamTypes.unwrap(slot); + if (effective.getDataType() != ClickHouseDataType.Map) { + owner.fail(); + throw new IllegalStateException("beginMap is not valid for " + StreamTypes.describe(effective)); + } + writePresent(slot); + return openLengthPrefixed(SlotCursor.map(effective.getKeyInfo(), effective.getValueInfo(), pairs), pairs); + } + + private ValueWriter openLengthPrefixed(SlotCursor childCursor, int size) throws IOException { + OutputStream destination = out; + ByteArrayOutputStream childBuffer = null; + if (size == SlotCursor.UNKNOWN) { + childBuffer = new ByteArrayOutputStream(); + destination = childBuffer; + } else { + try { + SerializerUtils.writeVarInt(out, size); + } catch (IOException e) { + owner.fail(); + throw e; + } + } + return nest(new StreamWriterScope(owner, this, childCursor, destination, childBuffer)); + } + + private void writePrimitiveArray(byte[] values, ClickHouseDataType elementType, ElementWriter writer) throws IOException { + if (values == null) { + writeNullFor(ClickHouseDataType.Array, ClickHouseDataType.QBit); + return; + } + writePrimitiveArray(values, values.length, elementType, writer); + } + + private void writePrimitiveArray(Object values, int length, ClickHouseDataType elementType, ElementWriter writer) + throws IOException { + if (values == null) { + writeNullFor(ClickHouseDataType.Array, ClickHouseDataType.QBit); + return; + } + ensureWritable(); + ClickHouseColumn slot = slot(); + ClickHouseColumn effective = StreamTypes.unwrap(slot); + ClickHouseDataType dataType = effective.getDataType(); + if (dataType != ClickHouseDataType.Array && dataType != ClickHouseDataType.QBit) { + owner.fail(); + throw new IllegalStateException("Cannot write a primitive array into " + StreamTypes.describe(effective)); + } + if (dataType == ClickHouseDataType.QBit && length != effective.getPrecision()) { + owner.fail(); + throw new IllegalArgumentException("QBit column '" + effective.getColumnName() + + "' expects exactly " + effective.getPrecision() + " elements but got " + length); + } + ClickHouseColumn element = StreamTypes.unwrap(StreamTypes.arrayElement(effective)); + if (element.getDataType() != elementType) { + owner.fail(); + throw new IllegalStateException("Cannot write " + elementType + " array into " + StreamTypes.describe(effective)); + } + boolean nullable = element.isNullable(); + writePresent(slot); + try { + SerializerUtils.writeVarInt(out, length); + for (int i = 0; i < length; i++) { + if (nullable) { + SerializerUtils.writeNonNull(out); + } + writer.write(i); + } + } catch (IOException e) { + owner.fail(); + throw e; + } catch (RuntimeException e) { + owner.fail(); + throw e; + } + cursor.advance(); + } + + private ClickHouseColumn prepare(ClickHouseDataType... allowed) throws IOException { + ensureWritable(); + ClickHouseColumn slot = slot(); + ClickHouseColumn effective = StreamTypes.unwrap(slot); + if (!StreamTypes.isOneOf(effective.getDataType(), allowed)) { + owner.fail(); + throw new IllegalStateException("Cannot write " + allowed[0] + " into " + StreamTypes.describe(effective)); + } + writePresent(slot); + return effective; + } + + private void writePresent(ClickHouseColumn slot) throws IOException { + ClickHouseColumn effective = StreamTypes.unwrap(slot); + try { + if (cursor.kind == SlotCursor.Kind.ROW && owner.defaultsSupport()) { + SerializerUtils.writeNonNull(out); + } + if (effective.isNullable()) { + SerializerUtils.writeNonNull(out); + } + } catch (IOException e) { + owner.fail(); + throw e; + } + } + + private void body(IOAction action) throws IOException { + try { + action.run(); + } catch (IOException e) { + owner.fail(); + throw e; + } catch (RuntimeException e) { + owner.fail(); + throw e; + } + cursor.advance(); + } + + private ClickHouseColumn slot() { + try { + return cursor.current(); + } catch (RuntimeException e) { + owner.fail(); + throw e; + } + } + + private void ensureWritable() { + if (owner.isFailed()) { + throw new IllegalStateException("Writer is unusable after a previous failure"); + } + if (closed) { + throw new IllegalStateException("Scope is closed"); + } + if (child != null) { + owner.fail(); + throw new IllegalStateException("A nested value is still open"); + } + } + + private void requireValue(Object value) { + if (value == null) { + owner.fail(); + throw new IllegalArgumentException("Value is null"); + } + } + + private void checkRange(long value, long min, long max, String type) { + if (value < min || value > max) { + owner.fail(); + throw new IllegalArgumentException(type + " overflow: " + value); + } + } + + private ZoneId zone(ClickHouseColumn column) { + if (column.getTimeZone() != null) { + return column.getTimeZone().toZoneId(); + } + return owner.timeZone().toZoneId(); + } + + private StreamWriterScope nest(StreamWriterScope childScope) { + this.child = childScope; + return childScope; + } + + private interface IOAction { + void run() throws IOException; + } + + private interface ElementWriter { + void write(int index) throws IOException; + } +} diff --git a/client-v2/src/main/java/com/clickhouse/client/api/data_formats/stream/ValueReader.java b/client-v2/src/main/java/com/clickhouse/client/api/data_formats/stream/ValueReader.java new file mode 100644 index 000000000..d2b2e13f9 --- /dev/null +++ b/client-v2/src/main/java/com/clickhouse/client/api/data_formats/stream/ValueReader.java @@ -0,0 +1,126 @@ +package com.clickhouse.client.api.data_formats.stream; + +import com.clickhouse.client.api.data_formats.internal.StringValue; +import com.clickhouse.data.ClickHouseColumn; +import com.clickhouse.data.value.ClickHouseBitmap; + +import java.io.IOException; +import java.math.BigDecimal; +import java.math.BigInteger; +import java.net.Inet4Address; +import java.net.Inet6Address; +import java.time.LocalDate; +import java.time.LocalDateTime; +import java.time.ZonedDateTime; +import java.util.UUID; + +/** + * One scope of a streaming RowBinary read. The verbs match {@link ValueWriter}. + *

+ * Experimental API. Not thread-safe. The caller owns the input stream. + */ +public interface ValueReader extends AutoCloseable { + + byte readInt8() throws IOException; + + short readInt16() throws IOException; + + int readInt32() throws IOException; + + long readInt64() throws IOException; + + BigInteger readInt128() throws IOException; + + BigInteger readInt256() throws IOException; + + int readUInt8() throws IOException; + + int readUInt16() throws IOException; + + long readUInt32() throws IOException; + + BigInteger readUInt64() throws IOException; + + BigInteger readUInt128() throws IOException; + + BigInteger readUInt256() throws IOException; + + float readFloat32() throws IOException; + + double readFloat64() throws IOException; + + float readBFloat16() throws IOException; + + boolean readBool() throws IOException; + + String readString() throws IOException; + + StringValue readStringValue() throws IOException; + + String readFixedString() throws IOException; + + LocalDate readDate() throws IOException; + + LocalDate readDate32() throws IOException; + + ZonedDateTime readDateTime() throws IOException; + + ZonedDateTime readDateTime64() throws IOException; + + UUID readUUID() throws IOException; + + Inet4Address readIPv4() throws IOException; + + Inet6Address readIPv6() throws IOException; + + BigDecimal readDecimal() throws IOException; + + int readEnum8() throws IOException; + + int readEnum16() throws IOException; + + ClickHouseBitmap readBitmap() throws IOException; + + /** + * Consumes a SQL NULL and returns true, or returns false and leaves the body to be read. + * On a non-nullable {@code Variant} or {@code Geometry}, discriminator {@code 0xFF} is null. + */ + boolean isNull() throws IOException; + + boolean hasNext(); + + int size(); + + int variantOrdinal(); + + ClickHouseColumn runtimeType(); + + ValueReader beginArray() throws IOException; + + ValueReader beginTuple() throws IOException; + + ValueReader beginMap() throws IOException; + + ValueReader beginNested() throws IOException; + + ValueReader beginVariant() throws IOException; + + ValueReader beginDynamic() throws IOException; + + ValueReader beginQBit() throws IOException; + + /** + * Fills {@code dst} with up to {@code len} {@code Int32} elements of the current array. + * + * @return number of elements written into {@code dst} + */ + int readInto(int[] dst, int off, int len) throws IOException; + + /** + * Skips the current value without building a list, map, or array. + */ + void skip() throws IOException; + + @Override + void close() throws IOException; +} diff --git a/client-v2/src/main/java/com/clickhouse/client/api/data_formats/stream/ValueWriter.java b/client-v2/src/main/java/com/clickhouse/client/api/data_formats/stream/ValueWriter.java new file mode 100644 index 000000000..228669a59 --- /dev/null +++ b/client-v2/src/main/java/com/clickhouse/client/api/data_formats/stream/ValueWriter.java @@ -0,0 +1,202 @@ +package com.clickhouse.client.api.data_formats.stream; + +import com.clickhouse.data.ClickHouseColumn; +import com.clickhouse.data.value.ClickHouseBitmap; + +import java.io.IOException; +import java.math.BigDecimal; +import java.math.BigInteger; +import java.net.Inet4Address; +import java.net.Inet6Address; +import java.nio.ByteBuffer; +import java.time.Instant; +import java.time.LocalDate; +import java.time.LocalDateTime; +import java.time.ZonedDateTime; +import java.util.UUID; + +/** + * One scope of a streaming RowBinary write: a row, an array, a tuple, a map, a nested row, + * a variant arm, or a dynamic value. + *

+ * Experimental API. Not thread-safe. The caller owns the output stream. + */ +public interface ValueWriter extends AutoCloseable { + + void writeInt8(byte value) throws IOException; + + void writeInt8(int value) throws IOException; + + void writeInt16(short value) throws IOException; + + void writeInt16(int value) throws IOException; + + void writeInt32(int value) throws IOException; + + void writeInt32(long value) throws IOException; + + void writeInt64(long value) throws IOException; + + void writeInt128(BigInteger value) throws IOException; + + void writeInt256(BigInteger value) throws IOException; + + void writeUInt8(int value) throws IOException; + + void writeUInt16(int value) throws IOException; + + void writeUInt32(long value) throws IOException; + + void writeUInt64(long value) throws IOException; + + void writeUInt64(BigInteger value) throws IOException; + + void writeUInt128(BigInteger value) throws IOException; + + void writeUInt256(BigInteger value) throws IOException; + + void writeFloat32(float value) throws IOException; + + void writeFloat64(double value) throws IOException; + + void writeBFloat16(float value) throws IOException; + + void writeBool(boolean value) throws IOException; + + void writeString(String value) throws IOException; + + void writeString(byte[] value) throws IOException; + + void writeString(ByteBuffer value) throws IOException; + + void writeFixedString(String value) throws IOException; + + void writeFixedString(byte[] value) throws IOException; + + void writeDate(LocalDate value) throws IOException; + + void writeDate32(LocalDate value) throws IOException; + + void writeDateTime(LocalDateTime value) throws IOException; + + void writeDateTime(ZonedDateTime value) throws IOException; + + void writeDateTime(Instant value) throws IOException; + + void writeDateTime64(LocalDateTime value) throws IOException; + + void writeDateTime64(ZonedDateTime value) throws IOException; + + void writeUUID(UUID value) throws IOException; + + void writeIPv4(Inet4Address value) throws IOException; + + void writeIPv6(Inet6Address value) throws IOException; + + void writeDecimal(BigDecimal value) throws IOException; + + void writeEnum8(int value) throws IOException; + + void writeEnum8(String name) throws IOException; + + void writeEnum16(int value) throws IOException; + + void writeEnum16(String name) throws IOException; + + void writeNull() throws IOException; + + void writeDefault() throws IOException; + + void writeBitmap(ClickHouseBitmap value) throws IOException; + + ValueWriter beginArray() throws IOException; + + ValueWriter beginArray(int size) throws IOException; + + ValueWriter beginTuple() throws IOException; + + ValueWriter beginMap() throws IOException; + + ValueWriter beginMap(int pairs) throws IOException; + + ValueWriter beginNested() throws IOException; + + ValueWriter beginNested(int rows) throws IOException; + + ValueWriter beginVariant(int ordinal) throws IOException; + + ValueWriter beginDynamic(ClickHouseColumn runtimeType) throws IOException; + + ValueWriter beginQBit() throws IOException; + + /** + * Opens a binary {@code JSON} object. The number of leaf paths is written when the scope closes. + * A nested object opened with {@link #beginObject(String)} contributes its leaf paths to this count. + */ + ValueWriter beginJson() throws IOException; + + /** + * Opens a binary {@code JSON} object with a known number of leaf paths. The count is written immediately. + */ + ValueWriter beginJson(int paths) throws IOException; + + /** Dynamic {@code String} path. A null value is a JSON null. */ + void writeString(String path, String value) throws IOException; + + /** Dynamic {@code Int32} path. */ + void writeInt32(String path, int value) throws IOException; + + /** Dynamic {@code Int64} path. */ + void writeInt64(String path, long value) throws IOException; + + /** Dynamic {@code Float64} path. */ + void writeFloat64(String path, double value) throws IOException; + + /** Dynamic {@code Bool} path. */ + void writeBool(String path, boolean value) throws IOException; + + /** JSON null at {@code path}: the path is present and its dynamic type is {@code Nothing}. */ + void writeNull(String path) throws IOException; + + /** + * Nested object stored as dotted paths ({@code address.city}) on the enclosing JSON value. + * The leaf count is not known yet. + */ + ValueWriter beginObject(String path) throws IOException; + + /** + * Nested object stored as dotted paths. {@code paths} is the number of leaf paths it contributes. + */ + ValueWriter beginObject(String path, int paths) throws IOException; + + /** Dynamic {@code Array} path. The element type selects the array type tag. */ + ValueWriter beginArray(String path, ClickHouseColumn elementType) throws IOException; + + /** Dynamic {@code Array} path with a known element count. */ + ValueWriter beginArray(String path, ClickHouseColumn elementType, int size) throws IOException; + + /** Dynamic path whose value is itself a JSON document, for example one element of {@code Array(JSON)}. */ + ValueWriter beginJson(String path) throws IOException; + + /** Dynamic JSON path with a known number of leaf paths inside that document. */ + ValueWriter beginJson(String path, int paths) throws IOException; + + void writeInt8Array(byte[] values) throws IOException; + + void writeInt16Array(short[] values) throws IOException; + + void writeInt32Array(int[] values) throws IOException; + + void writeInt64Array(long[] values) throws IOException; + + void writeFloat32Array(float[] values) throws IOException; + + void writeFloat64Array(double[] values) throws IOException; + + void writeBFloat16Array(float[] values) throws IOException; + + void writeBoolArray(boolean[] values) throws IOException; + + @Override + void close() throws IOException; +} diff --git a/client-v2/src/test/java/com/clickhouse/client/api/data_formats/stream/RowBinaryStreamRoundTripTest.java b/client-v2/src/test/java/com/clickhouse/client/api/data_formats/stream/RowBinaryStreamRoundTripTest.java new file mode 100644 index 000000000..863450098 --- /dev/null +++ b/client-v2/src/test/java/com/clickhouse/client/api/data_formats/stream/RowBinaryStreamRoundTripTest.java @@ -0,0 +1,672 @@ +package com.clickhouse.client.api.data_formats.stream; + +import com.clickhouse.client.api.ClientException; +import com.clickhouse.client.api.data_formats.RowBinaryFormatWriter; +import com.clickhouse.client.api.data_formats.internal.BinaryStreamReader; +import com.clickhouse.client.api.data_formats.internal.StringValue; +import com.clickhouse.client.api.metadata.TableSchema; +import com.clickhouse.client.api.query.NullValueException; +import com.clickhouse.data.ClickHouseColumn; +import com.clickhouse.data.ClickHouseDataType; +import com.clickhouse.data.ClickHouseFormat; +import com.clickhouse.data.format.BinaryStreamUtils; +import com.clickhouse.data.value.ClickHouseBitmap; +import org.testng.Assert; +import org.testng.annotations.Test; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.TimeZone; + +public class RowBinaryStreamRoundTripTest { + + @Test + public void scalarsRoundTripInTheMiddleOfTheRow() throws IOException { + TableSchema schema = schema("id Int32", "name String", "tail Float64"); + byte[] streamed = streamed(schema, ClickHouseFormat.RowBinary, row -> { + row.writeInt32(7); + row.writeString("abc"); + row.writeFloat64(1.5d); + }); + byte[] buffered = buffered(schema, ClickHouseFormat.RowBinary, writer -> { + writer.setValue("id", 7); + writer.setValue("name", "abc"); + writer.setValue("tail", 1.5d); + }); + Assert.assertEquals(streamed, buffered); + + BinaryStreamReader reader = binary(streamed); + Assert.assertEquals(reader.readValue(column(schema, 0)), Integer.valueOf(7)); + Assert.assertEquals(reader.readValue(column(schema, 1)), "abc"); + Assert.assertEquals(reader.readValue(column(schema, 2)), 1.5d); + + RowBinaryStreamReader streamReader = new RowBinaryStreamReader( + new ByteArrayInputStream(streamed), schema, ClickHouseFormat.RowBinary); + Assert.assertTrue(streamReader.hasNext()); + try (ValueReader row = streamReader.beginRow()) { + Assert.assertEquals(row.readInt32(), 7); + Assert.assertEquals(row.readString(), "abc"); + Assert.assertEquals(row.readFloat64(), 1.5d); + } + Assert.assertFalse(streamReader.hasNext()); + } + + @Test + public void nullableStringAndNull() throws IOException { + TableSchema schema = schema("id Int32", "name Nullable(String)", "tail Float64"); + ByteArrayOutputStream out = new ByteArrayOutputStream(); + RowBinaryStreamWriter writer = new RowBinaryStreamWriter(out, schema, ClickHouseFormat.RowBinary); + try (ValueWriter row = writer.beginRow()) { + row.writeInt32(1); + row.writeString((String) null); + row.writeFloat64(2.5d); + } + try (ValueWriter row = writer.beginRow()) { + row.writeInt32(2); + row.writeString("ok"); + row.writeFloat64(3.5d); + } + byte[] streamed = out.toByteArray(); + BinaryStreamReader reader = binary(streamed); + Assert.assertEquals(reader.readValue(column(schema, 0)), Integer.valueOf(1)); + Assert.assertNull(reader.readValue(column(schema, 1))); + Assert.assertEquals(reader.readValue(column(schema, 2)), 2.5d); + Assert.assertEquals(reader.readValue(column(schema, 0)), Integer.valueOf(2)); + Assert.assertEquals(reader.readValue(column(schema, 1)), "ok"); + Assert.assertEquals(reader.readValue(column(schema, 2)), 3.5d); + + RowBinaryStreamReader streamReader = new RowBinaryStreamReader( + new ByteArrayInputStream(streamed), schema, ClickHouseFormat.RowBinary); + try (ValueReader row = streamReader.beginRow()) { + Assert.assertEquals(row.readInt32(), 1); + Assert.assertTrue(row.isNull()); + Assert.assertEquals(row.readFloat64(), 2.5d); + } + try (ValueReader row = streamReader.beginRow()) { + Assert.assertEquals(row.readInt32(), 2); + Assert.assertFalse(row.isNull()); + Assert.assertEquals(row.readString(), "ok"); + Assert.assertEquals(row.readFloat64(), 3.5d); + } + } + + @Test + public void arraysKnownAndUnknownSizeMatchBufferedWriter() throws IOException { + TableSchema schema = schema("id Int32", "items Array(Nullable(String))", "tail Float64"); + List items = Arrays.asList("a", null, "b"); + byte[] unknown = streamed(schema, ClickHouseFormat.RowBinary, row -> { + row.writeInt32(1); + try (ValueWriter array = row.beginArray()) { + array.writeString("a"); + array.writeString((String) null); + array.writeString("b"); + } + row.writeFloat64(1.5d); + }); + byte[] known = streamed(schema, ClickHouseFormat.RowBinary, row -> { + row.writeInt32(1); + try (ValueWriter array = row.beginArray(items.size())) { + for (String item : items) { + array.writeString(item); + } + } + row.writeFloat64(1.5d); + }); + byte[] fromBuffer = buffered(schema, ClickHouseFormat.RowBinary, writer -> { + writer.setValue("id", 1); + writer.setValue("items", items); + writer.setValue("tail", 1.5d); + }); + Assert.assertEquals(unknown, known); + Assert.assertEquals(known, fromBuffer); + + BinaryStreamReader reader = binary(known); + Assert.assertEquals(reader.readValue(column(schema, 0)), Integer.valueOf(1)); + Assert.assertEquals(asList(reader.readValue(column(schema, 1))), items); + Assert.assertEquals(reader.readValue(column(schema, 2)), 1.5d); + } + + @Test + public void nestedArraysMatchBufferedWriter() throws IOException { + TableSchema schema = schema("id Int32", "matrix Array(Array(Nullable(String)))", "tail Float64"); + List> matrix = Arrays.asList( + Arrays.asList("a", null), + Collections.emptyList()); + byte[] unknown = streamed(schema, ClickHouseFormat.RowBinary, row -> { + row.writeInt32(4); + try (ValueWriter outer = row.beginArray()) { + try (ValueWriter inner = outer.beginArray()) { + inner.writeString("a"); + inner.writeString((String) null); + } + try (ValueWriter inner = outer.beginArray(0)) { + } + } + row.writeFloat64(1.5d); + }); + byte[] known = streamed(schema, ClickHouseFormat.RowBinary, row -> { + row.writeInt32(4); + try (ValueWriter outer = row.beginArray(2)) { + try (ValueWriter inner = outer.beginArray(2)) { + inner.writeString("a"); + inner.writeString((String) null); + } + try (ValueWriter inner = outer.beginArray()) { + } + } + row.writeFloat64(1.5d); + }); + byte[] fromBuffer = buffered(schema, ClickHouseFormat.RowBinary, writer -> { + writer.setValue("id", 4); + writer.setValue("matrix", matrix); + writer.setValue("tail", 1.5d); + }); + Assert.assertEquals(unknown, known); + Assert.assertEquals(known, fromBuffer); + BinaryStreamReader reader = binary(known); + Assert.assertEquals(reader.readValue(column(schema, 0)), Integer.valueOf(4)); + Assert.assertEquals(asList(reader.readValue(column(schema, 1))), matrix); + Assert.assertEquals(reader.readValue(column(schema, 2)), 1.5d); + } + + @Test + public void tupleWithArrayInTheMiddle() throws IOException { + TableSchema schema = schema("id Int32", "t Tuple(Int32, Array(Nullable(String)), Float64)", "tail Float64"); + byte[] streamed = streamed(schema, ClickHouseFormat.RowBinary, row -> { + row.writeInt32(1); + try (ValueWriter tuple = row.beginTuple()) { + tuple.writeInt32(9); + try (ValueWriter array = tuple.beginArray(2)) { + array.writeString("x"); + array.writeString((String) null); + } + tuple.writeFloat64(0.5d); + } + row.writeFloat64(1.5d); + }); + byte[] fromBuffer = buffered(schema, ClickHouseFormat.RowBinary, writer -> { + writer.setValue("id", 1); + writer.setValue("t", Arrays.asList(9, Arrays.asList("x", null), 0.5d)); + writer.setValue("tail", 1.5d); + }); + Assert.assertEquals(streamed, fromBuffer); + BinaryStreamReader reader = binary(streamed); + Assert.assertEquals(reader.readValue(column(schema, 0)), Integer.valueOf(1)); + Object[] tuple = (Object[]) reader.readValue(column(schema, 1)); + Assert.assertEquals(tuple[0], 9); + Assert.assertEquals(asList(tuple[1]), Arrays.asList("x", null)); + Assert.assertEquals(tuple[2], 0.5d); + Assert.assertEquals(reader.readValue(column(schema, 2)), 1.5d); + } + + @Test + public void mapWithNestedArray() throws IOException { + TableSchema schema = schema("id Int32", "m Map(String, Array(Int32))", "tail Float64"); + Map> map = new LinkedHashMap>(); + map.put("a", Arrays.asList(1, 2)); + map.put("b", Collections.emptyList()); + byte[] streamed = streamed(schema, ClickHouseFormat.RowBinary, row -> { + row.writeInt32(1); + try (ValueWriter values = row.beginMap(2)) { + values.writeString("a"); + try (ValueWriter array = values.beginArray(2)) { + array.writeInt32(1); + array.writeInt32(2); + } + values.writeString("b"); + try (ValueWriter array = values.beginArray()) { + } + } + row.writeFloat64(1.5d); + }); + byte[] fromBuffer = buffered(schema, ClickHouseFormat.RowBinary, writer -> { + writer.setValue("id", 1); + writer.setValue("m", map); + writer.setValue("tail", 1.5d); + }); + Assert.assertEquals(streamed, fromBuffer); + } + + @Test + public void unflattenedNested() throws IOException { + TableSchema schema = schema("id Int32", "n Nested(n1 Int32, n2 Nullable(String))", "tail Float64"); + byte[] streamed = streamed(schema, ClickHouseFormat.RowBinary, row -> { + row.writeInt32(1); + try (ValueWriter nested = row.beginNested(2)) { + try (ValueWriter tuple = nested.beginTuple()) { + tuple.writeInt32(5); + tuple.writeString("a"); + } + try (ValueWriter tuple = nested.beginTuple()) { + tuple.writeInt32(6); + tuple.writeString((String) null); + } + } + row.writeFloat64(1.5d); + }); + byte[] fromBuffer = buffered(schema, ClickHouseFormat.RowBinary, writer -> { + writer.setValue("id", 1); + writer.setValue("n", Arrays.asList(Arrays.asList(5, "a"), Arrays.asList(6, null))); + writer.setValue("tail", 1.5d); + }); + Assert.assertEquals(streamed, fromBuffer); + BinaryStreamReader reader = binary(streamed); + Assert.assertEquals(reader.readValue(column(schema, 0)), Integer.valueOf(1)); + List rows = asList(reader.readValue(column(schema, 1))); + Assert.assertEquals(((Object[]) rows.get(0))[0], 5); + Assert.assertEquals(((Object[]) rows.get(0))[1], "a"); + Assert.assertEquals(((Object[]) rows.get(1))[0], 6); + Assert.assertNull(((Object[]) rows.get(1))[1]); + Assert.assertEquals(reader.readValue(column(schema, 2)), 1.5d); + } + + @Test + public void qbitAndBulkArrayUseTheKnownSizePath() throws IOException { + TableSchema qbit = schema("id Int32", "v QBit(Float64, 4)", "tail Float64"); + double[] vector = new double[] {1.0d, 2.0d, 3.0d, 4.0d}; + byte[] bulk = streamed(qbit, ClickHouseFormat.RowBinary, row -> { + row.writeInt32(1); + row.writeFloat64Array(vector); + row.writeFloat64(1.5d); + }); + byte[] cursor = streamed(qbit, ClickHouseFormat.RowBinary, row -> { + row.writeInt32(1); + try (ValueWriter values = row.beginQBit()) { + for (double value : vector) { + values.writeFloat64(value); + } + } + row.writeFloat64(1.5d); + }); + byte[] fromBuffer = buffered(qbit, ClickHouseFormat.RowBinary, writer -> { + writer.setValue("id", 1); + writer.setValue("v", vector); + writer.setValue("tail", 1.5d); + }); + Assert.assertEquals(bulk, cursor); + Assert.assertEquals(cursor, fromBuffer); + + TableSchema ints = schema("id Int32", "v Array(Int32)", "tail Float64"); + int[] numbers = new int[] {4, 5, 6}; + byte[] bulkInts = streamed(ints, ClickHouseFormat.RowBinary, row -> { + row.writeInt32(1); + row.writeInt32Array(numbers); + row.writeFloat64(1.5d); + }); + byte[] loopInts = streamed(ints, ClickHouseFormat.RowBinary, row -> { + row.writeInt32(1); + try (ValueWriter array = row.beginArray(numbers.length)) { + for (int number : numbers) { + array.writeInt32(number); + } + } + row.writeFloat64(1.5d); + }); + Assert.assertEquals(bulkInts, loopInts); + } + + @Test + public void defaultsPreamble() throws IOException { + ClickHouseColumn withDefault = ClickHouseColumn.of("d", "Nullable(Int32)"); + withDefault.setHasDefault(true); + withDefault.setDefaultValue(ClickHouseColumn.DefaultValue.DEFAULT); + ClickHouseColumn nullable = ClickHouseColumn.of("n", "Nullable(Int32)"); + ClickHouseColumn plain = ClickHouseColumn.of("v", "Int32"); + TableSchema schema = new TableSchema(Arrays.asList(withDefault, nullable, plain)); + + byte[] streamed = streamed(schema, ClickHouseFormat.RowBinaryWithDefaults, row -> { + row.writeDefault(); + row.writeNull(); + row.writeInt32(3); + }); + byte[] fromBuffer = buffered(schema, ClickHouseFormat.RowBinaryWithDefaults, writer -> { + writer.setValue("d", null); + writer.setValue("n", null); + writer.setValue("v", 3); + }); + Assert.assertEquals(streamed, fromBuffer); + Assert.assertEquals(streamed, new byte[] {1, 0, 1, 0, 3, 0, 0, 0}); + + byte[] value = streamed(schema, ClickHouseFormat.RowBinaryWithDefaults, row -> { + row.writeInt32(4); + row.writeInt32(5); + row.writeInt32(6); + }); + Assert.assertEquals(value[0], 0); + Assert.assertEquals(value[1], 0); + } + + @Test + public void rawStringBytesRoundTrip() throws IOException { + TableSchema schema = schema("id Int32", "raw String", "tail Float64"); + byte[] raw = new byte[] {(byte) 0xFF, 0x00, (byte) 0xFE}; + byte[] streamed = streamed(schema, ClickHouseFormat.RowBinary, row -> { + row.writeInt32(1); + row.writeString(raw); + row.writeFloat64(1.5d); + }); + BinaryStreamReader reader = new BinaryStreamReader(new ByteArrayInputStream(streamed), + TimeZone.getTimeZone("UTC"), null, new BinaryStreamReader.CachingByteBufferAllocator(), + true, null, true); + Assert.assertEquals(reader.readValue(column(schema, 0)), Integer.valueOf(1)); + Object value = reader.readValue(column(schema, 1)); + Assert.assertTrue(value instanceof StringValue); + Assert.assertEquals(((StringValue) value).toByteArray(), raw); + Assert.assertEquals(reader.readValue(column(schema, 2)), 1.5d); + + RowBinaryStreamReader streamReader = new RowBinaryStreamReader( + new ByteArrayInputStream(streamed), schema, ClickHouseFormat.RowBinary); + try (ValueReader row = streamReader.beginRow()) { + Assert.assertEquals(row.readInt32(), 1); + Assert.assertEquals(row.readStringValue().toByteArray(), raw); + Assert.assertEquals(row.readFloat64(), 1.5d); + } + } + + @Test + public void variantAndDynamic() throws IOException { + TableSchema variantSchema = schema("id Int32", "v Variant(String, Int32)", "tail Float64"); + ClickHouseColumn variant = column(variantSchema, 1); + int stringOrdinal = -1; + for (int i = 0; i < variant.getNestedColumns().size(); i++) { + if (variant.getNestedColumns().get(i).getDataType() == ClickHouseDataType.String) { + stringOrdinal = i; + } + } + Assert.assertTrue(stringOrdinal >= 0); + int ordinal = stringOrdinal; + byte[] streamed = streamed(variantSchema, ClickHouseFormat.RowBinary, row -> { + row.writeInt32(1); + try (ValueWriter value = row.beginVariant(ordinal)) { + value.writeString("hi"); + } + row.writeFloat64(1.5d); + }); + byte[] fromBuffer = buffered(variantSchema, ClickHouseFormat.RowBinary, writer -> { + writer.setValue("id", 1); + writer.setValue("v", "hi"); + writer.setValue("tail", 1.5d); + }); + Assert.assertEquals(streamed, fromBuffer); + + TableSchema dynamicSchema = schema("id Int32", "d Dynamic", "tail Float64"); + byte[] dynamicInt = streamed(dynamicSchema, ClickHouseFormat.RowBinary, row -> { + row.writeInt32(1); + try (ValueWriter value = row.beginDynamic(ClickHouseColumn.of("v", "Int32"))) { + value.writeInt32(7); + } + row.writeFloat64(1.5d); + }); + byte[] dynamicFromBuffer = buffered(dynamicSchema, ClickHouseFormat.RowBinary, writer -> { + writer.setValue("id", 1); + writer.setValue("d", 7); + writer.setValue("tail", 1.5d); + }); + Assert.assertEquals(dynamicInt, dynamicFromBuffer); + + byte[] dynamicArray = streamed(dynamicSchema, ClickHouseFormat.RowBinary, row -> { + row.writeInt32(1); + try (ValueWriter value = row.beginDynamic(ClickHouseColumn.of("v", "Array(Nullable(String))"))) { + try (ValueWriter array = value.beginArray(2)) { + array.writeString("a"); + array.writeString((String) null); + } + } + row.writeFloat64(1.5d); + }); + byte[] dynamicArrayFromBuffer = buffered(dynamicSchema, ClickHouseFormat.RowBinary, writer -> { + writer.setValue("id", 1); + writer.setValue("d", Arrays.asList("a", null)); + writer.setValue("tail", 1.5d); + }); + Assert.assertEquals(dynamicArray, dynamicArrayFromBuffer); + + RowBinaryStreamReader reader = new RowBinaryStreamReader( + new ByteArrayInputStream(dynamicInt), dynamicSchema, ClickHouseFormat.RowBinary); + try (ValueReader row = reader.beginRow()) { + Assert.assertEquals(row.readInt32(), 1); + try (ValueReader value = row.beginDynamic()) { + Assert.assertEquals(value.runtimeType().getDataType(), ClickHouseDataType.Int32); + Assert.assertEquals(value.readInt32(), 7); + } + Assert.assertEquals(row.readFloat64(), 1.5d); + } + } + + @Test + public void dynamicNullIsNothingAndIncompleteTypesAreRejected() throws IOException { + TableSchema schema = schema("d Dynamic"); + byte[] streamed = streamed(schema, ClickHouseFormat.RowBinary, ValueWriter::writeNull); + byte[] fromBuffer = buffered(schema, ClickHouseFormat.RowBinary, writer -> writer.setValue("d", null)); + Assert.assertEquals(streamed, fromBuffer); + + RowBinaryStreamReader reader = new RowBinaryStreamReader( + new ByteArrayInputStream(streamed), schema, ClickHouseFormat.RowBinary); + try (ValueReader row = reader.beginRow()) { + try (ValueReader value = row.beginDynamic()) { + Assert.assertEquals(value.runtimeType().getDataType(), ClickHouseDataType.Nothing); + } + } + + Assert.expectThrows(ClientException.class, () -> streamed(schema, ClickHouseFormat.RowBinary, row -> + row.beginDynamic(ClickHouseColumn.of("v", "Tuple(Int32, String)")))); + Assert.expectThrows(ClientException.class, () -> streamed(schema, ClickHouseFormat.RowBinary, row -> + row.beginDynamic(ClickHouseColumn.of("v", "Array(Tuple(Int32))")))); + Assert.expectThrows(ClientException.class, () -> streamed(schema, ClickHouseFormat.RowBinary, row -> + row.beginDynamic(ClickHouseColumn.of("v", "QBit(Float64, 4)")))); + } + + @Test + public void jsonIsAStringAndStructuredWriteIsRejected() throws IOException { + TableSchema schema = schema("id Int32", "j JSON", "tail Float64"); + byte[] streamed = streamed(schema, ClickHouseFormat.RowBinary, row -> { + row.writeInt32(1); + row.writeString("{\"a\":1}"); + row.writeFloat64(1.5d); + }); + byte[] fromBuffer = buffered(schema, ClickHouseFormat.RowBinary, writer -> { + writer.setValue("id", 1); + writer.setValue("j", "{\"a\":1}"); + writer.setValue("tail", 1.5d); + }); + Assert.assertEquals(streamed, fromBuffer); + Assert.expectThrows(IllegalStateException.class, () -> streamed(schema, ClickHouseFormat.RowBinary, row -> { + row.writeInt32(1); + row.beginArray(); + })); + } + + @Test + public void groupBitmapRoundTrip() throws IOException { + TableSchema schema = schema("id Int32", "b AggregateFunction(groupBitmap, UInt32)", "tail Float64"); + ClickHouseBitmap bitmap = ClickHouseBitmap.wrap(1, 2, 3); + byte[] streamed = streamed(schema, ClickHouseFormat.RowBinary, row -> { + row.writeInt32(1); + row.writeBitmap(bitmap); + row.writeFloat64(1.5d); + }); + byte[] fromBuffer = buffered(schema, ClickHouseFormat.RowBinary, writer -> { + writer.setValue("id", 1); + writer.setValue("b", bitmap); + writer.setValue("tail", 1.5d); + }); + Assert.assertEquals(streamed, fromBuffer); + BinaryStreamReader reader = binary(streamed); + Assert.assertEquals(reader.readValue(column(schema, 0)), Integer.valueOf(1)); + Assert.assertEquals(((ClickHouseBitmap) reader.readValue(column(schema, 1))).toBytes(), bitmap.toBytes()); + Assert.assertEquals(reader.readValue(column(schema, 2)), 1.5d); + } + + @Test + public void readerSkipsArrayWithoutLosingTheNextColumn() throws IOException { + TableSchema schema = schema("id Int32", "items Array(Int32)", "tail Float64"); + int[] numbers = new int[32]; + for (int i = 0; i < numbers.length; i++) { + numbers[i] = i; + } + byte[] streamed = streamed(schema, ClickHouseFormat.RowBinary, row -> { + row.writeInt32(1); + row.writeInt32Array(numbers); + row.writeFloat64(1.5d); + }); + RowBinaryStreamReader reader = new RowBinaryStreamReader( + new ByteArrayInputStream(streamed), schema, ClickHouseFormat.RowBinary); + try (ValueReader row = reader.beginRow()) { + Assert.assertEquals(row.readInt32(), 1); + row.skip(); + Assert.assertEquals(row.readFloat64(), 1.5d); + } + } + + @Test + public void readIntoFillsCallerBuffer() throws IOException { + TableSchema schema = schema("v Array(Int32)", "tail Float64"); + byte[] streamed = streamed(schema, ClickHouseFormat.RowBinary, row -> { + row.writeInt32Array(new int[] {8, 9, 10}); + row.writeFloat64(1.5d); + }); + RowBinaryStreamReader reader = new RowBinaryStreamReader( + new ByteArrayInputStream(streamed), schema, ClickHouseFormat.RowBinary); + try (ValueReader row = reader.beginRow()) { + try (ValueReader array = row.beginArray()) { + Assert.assertEquals(array.size(), 3); + int[] dst = new int[3]; + Assert.assertEquals(array.readInto(dst, 0, dst.length), 3); + Assert.assertEquals(dst, new int[] {8, 9, 10}); + Assert.assertFalse(array.hasNext()); + } + Assert.assertEquals(row.readFloat64(), 1.5d); + } + } + + @Test + public void namesAndTypesHeaderBuildsTheSchema() throws IOException { + TableSchema schema = schema("id Int32", "name String"); + ByteArrayOutputStream out = new ByteArrayOutputStream(); + BinaryStreamUtils.writeVarInt(out, 2); + BinaryStreamUtils.writeString(out, "id"); + BinaryStreamUtils.writeString(out, "name"); + BinaryStreamUtils.writeString(out, column(schema, 0).getOriginalTypeName()); + BinaryStreamUtils.writeString(out, column(schema, 1).getOriginalTypeName()); + try (ValueWriter row = new RowBinaryStreamWriter(out, schema, ClickHouseFormat.RowBinary).beginRow()) { + row.writeInt32(3); + row.writeString("z"); + } + RowBinaryStreamReader reader = new RowBinaryStreamReader( + new ByteArrayInputStream(out.toByteArray()), null, ClickHouseFormat.RowBinaryWithNamesAndTypes); + Assert.assertEquals(reader.getSchema().getColumns().size(), 2); + Assert.assertEquals(reader.getSchema().getColumns().get(0).getColumnName(), "id"); + Assert.assertEquals(reader.getSchema().getColumns().get(1).getDataType(), ClickHouseDataType.String); + try (ValueReader row = reader.beginRow()) { + Assert.assertEquals(row.readInt32(), 3); + Assert.assertEquals(row.readString(), "z"); + } + } + + @Test + public void negativePaths() throws IOException { + TableSchema arraySchema = schema("v Array(Int32)"); + Assert.expectThrows(IllegalStateException.class, () -> { + RowBinaryStreamWriter writer = new RowBinaryStreamWriter( + new ByteArrayOutputStream(), arraySchema, ClickHouseFormat.RowBinary); + ValueWriter row = writer.beginRow(); + ValueWriter array = row.beginArray(); + array.writeInt32(1); + row.close(); + }); + Assert.expectThrows(IllegalStateException.class, () -> streamed(arraySchema, ClickHouseFormat.RowBinary, row -> { + try (ValueWriter array = row.beginArray(2)) { + array.writeInt32(1); + } + })); + Assert.expectThrows(IllegalArgumentException.class, () -> + streamed(schema("v Int32"), ClickHouseFormat.RowBinary, row -> row.writeInt32(1L << 40))); + Assert.expectThrows(IllegalArgumentException.class, () -> + streamed(schema("v Int32"), ClickHouseFormat.RowBinary, row -> row.writeNull())); + Assert.expectThrows(IllegalStateException.class, () -> + streamed(schema("v Array(Int32)"), ClickHouseFormat.RowBinaryWithDefaults, row -> { + try (ValueWriter array = row.beginArray()) { + array.writeDefault(); + } + })); + Assert.expectThrows(IllegalArgumentException.class, () -> + streamed(schema("v Int32"), ClickHouseFormat.RowBinaryWithDefaults, ValueWriter::writeDefault)); + Assert.expectThrows(IllegalStateException.class, () -> streamed(schema("m Map(String, Int32)"), ClickHouseFormat.RowBinary, row -> { + try (ValueWriter map = row.beginMap(1)) { + map.writeString("a"); + } + })); + Assert.expectThrows(IllegalArgumentException.class, () -> + streamed(schema("v QBit(Float64, 4)"), ClickHouseFormat.RowBinary, + row -> row.writeFloat64Array(new double[] {1.0d}))); + Assert.expectThrows(NullValueException.class, () -> { + byte[] bytes = streamed(schema("v Nullable(Int32)"), ClickHouseFormat.RowBinary, ValueWriter::writeNull); + RowBinaryStreamReader reader = new RowBinaryStreamReader( + new ByteArrayInputStream(bytes), schema("v Nullable(Int32)"), ClickHouseFormat.RowBinary); + try (ValueReader row = reader.beginRow()) { + row.readInt32(); + } + }); + Assert.expectThrows(IllegalStateException.class, () -> + streamed(schema("a Int32", "b Int32"), ClickHouseFormat.RowBinary, row -> row.writeInt32(1))); + } + + private static TableSchema schema(String... definitions) { + List columns = new ArrayList(); + for (String definition : definitions) { + int space = definition.indexOf(' '); + columns.add(ClickHouseColumn.of(definition.substring(0, space), definition.substring(space + 1).trim())); + } + return new TableSchema(columns); + } + + private static ClickHouseColumn column(TableSchema schema, int index) { + return schema.getColumns().get(index); + } + + private static byte[] streamed(TableSchema schema, ClickHouseFormat format, StreamingRow row) throws IOException { + ByteArrayOutputStream out = new ByteArrayOutputStream(); + RowBinaryStreamWriter writer = new RowBinaryStreamWriter(out, schema, format); + try (ValueWriter value = writer.beginRow()) { + row.accept(value); + } + return out.toByteArray(); + } + + private static byte[] buffered(TableSchema schema, ClickHouseFormat format, BufferedRow row) throws IOException { + ByteArrayOutputStream out = new ByteArrayOutputStream(); + RowBinaryFormatWriter writer = new RowBinaryFormatWriter(out, schema, format); + row.accept(writer); + writer.commitRow(); + return out.toByteArray(); + } + + private static BinaryStreamReader binary(byte[] bytes) { + return new BinaryStreamReader(new ByteArrayInputStream(bytes), TimeZone.getTimeZone("UTC"), null, + new BinaryStreamReader.CachingByteBufferAllocator(), true, null, false); + } + + @SuppressWarnings("unchecked") + private static List asList(Object value) { + if (value instanceof BinaryStreamReader.ArrayValue) { + return ((BinaryStreamReader.ArrayValue) value).asList(); + } + return (List) value; + } + + private interface StreamingRow { + void accept(ValueWriter row) throws IOException; + } + + private interface BufferedRow { + void accept(RowBinaryFormatWriter writer) throws IOException; + } +} diff --git a/client-v2/src/test/java/com/clickhouse/client/api/data_formats/stream/RowBinaryStreamServerTest.java b/client-v2/src/test/java/com/clickhouse/client/api/data_formats/stream/RowBinaryStreamServerTest.java new file mode 100644 index 000000000..f206ebbd1 --- /dev/null +++ b/client-v2/src/test/java/com/clickhouse/client/api/data_formats/stream/RowBinaryStreamServerTest.java @@ -0,0 +1,297 @@ +package com.clickhouse.client.api.data_formats.stream; + +import com.clickhouse.client.BaseIntegrationTest; +import com.clickhouse.client.ClickHouseNode; +import com.clickhouse.client.ClickHouseProtocol; +import com.clickhouse.client.ClickHouseServerForTest; +import com.clickhouse.client.api.Client; +import com.clickhouse.client.api.command.CommandSettings; +import com.clickhouse.client.api.data_formats.internal.BinaryStreamReader; +import com.clickhouse.client.api.enums.Protocol; +import com.clickhouse.client.api.insert.InsertSettings; +import com.clickhouse.client.api.internal.ServerSettings; +import com.clickhouse.client.api.metadata.TableSchema; +import com.clickhouse.client.api.query.GenericRecord; +import com.clickhouse.data.ClickHouseColumn; +import com.clickhouse.data.ClickHouseFormat; +import com.clickhouse.data.ClickHouseVersion; +import org.testng.Assert; +import org.testng.annotations.AfterMethod; +import org.testng.annotations.BeforeMethod; +import org.testng.annotations.Test; + +import java.io.IOException; +import java.util.Arrays; +import java.util.List; +import java.util.Map; +import java.util.concurrent.TimeUnit; + +/** + * Inserts rows with {@link RowBinaryStreamWriter} and reads them back from a server. + */ +@Test(groups = {"integration"}) +public class RowBinaryStreamServerTest extends BaseIntegrationTest { + + private static final int TIMEOUT_SECONDS = 30; + + private Client client; + + @BeforeMethod(groups = {"integration"}) + public void setUp() { + ClickHouseNode node = getServer(ClickHouseProtocol.HTTP); + client = new Client.Builder() + .addEndpoint(Protocol.HTTP, node.getHost(), node.getPort(), isCloud()) + .setUsername("default") + .setPassword(ClickHouseServerForTest.getPassword()) + .compressServerResponse(false) + .compressClientRequest(false) + .useHttpCompression(false) + .setDefaultDatabase(ClickHouseServerForTest.getDatabase()) + .serverSetting(ServerSettings.ASYNC_INSERT, "0") + .serverSetting(ServerSettings.WAIT_END_OF_QUERY, "1") + .build(); + } + + @AfterMethod(groups = {"integration"}) + public void tearDown() { + if (client != null) { + client.close(); + } + } + + @Test(groups = {"integration"}) + public void scalarsAndNullableString() throws Exception { + List rows = insert("stream_scalars", + "id Int32, name Nullable(String), tail Float64", writer -> { + try (ValueWriter row = writer.beginRow()) { + row.writeInt32(1); + row.writeString((String) null); + row.writeFloat64(2.5d); + } + try (ValueWriter row = writer.beginRow()) { + row.writeInt32(2); + row.writeString("ok"); + row.writeFloat64(3.5d); + } + }); + Assert.assertEquals(rows.get(0).getInteger("id"), 1); + Assert.assertNull(rows.get(0).getObject("name")); + Assert.assertEquals(rows.get(0).getDouble("tail"), 2.5d); + Assert.assertEquals(rows.get(1).getInteger("id"), 2); + Assert.assertEquals(rows.get(1).getString("name"), "ok"); + Assert.assertEquals(rows.get(1).getDouble("tail"), 3.5d); + } + + @Test(groups = {"integration"}) + public void arrayTupleMapAndNested() throws Exception { + List arrays = insert("stream_array", + "id Int32, items Array(Nullable(String)), tail Float64", writer -> { + try (ValueWriter row = writer.beginRow()) { + row.writeInt32(1); + try (ValueWriter items = row.beginArray(3)) { + items.writeString("a"); + items.writeString((String) null); + items.writeString("b"); + } + row.writeFloat64(1.5d); + } + }); + Assert.assertEquals(asList(arrays.get(0).getObject("items")), Arrays.asList("a", null, "b")); + Assert.assertEquals(arrays.get(0).getDouble("tail"), 1.5d); + + List tuples = insert("stream_tuple", + "id Int32, t Tuple(Int32, Array(Nullable(String)), Float64), tail Float64", writer -> { + try (ValueWriter row = writer.beginRow()) { + row.writeInt32(1); + try (ValueWriter tuple = row.beginTuple()) { + tuple.writeInt32(9); + try (ValueWriter values = tuple.beginArray(2)) { + values.writeString("x"); + values.writeString((String) null); + } + tuple.writeFloat64(0.5d); + } + row.writeFloat64(1.5d); + } + }); + Object[] tuple = (Object[]) tuples.get(0).getObject("t"); + Assert.assertEquals(tuple[0], 9); + Assert.assertEquals(asList(tuple[1]), Arrays.asList("x", null)); + Assert.assertEquals(((Number) tuple[2]).doubleValue(), 0.5d); + + List maps = insert("stream_map", + "id Int32, m Map(String, Array(Int32)), tail Float64", writer -> { + try (ValueWriter row = writer.beginRow()) { + row.writeInt32(1); + try (ValueWriter map = row.beginMap(2)) { + map.writeString("a"); + try (ValueWriter values = map.beginArray(2)) { + values.writeInt32(1); + values.writeInt32(2); + } + map.writeString("b"); + try (ValueWriter values = map.beginArray(0)) { + } + } + row.writeFloat64(1.5d); + } + }); + @SuppressWarnings("unchecked") + Map map = (Map) maps.get(0).getObject("m"); + Assert.assertEquals(asList(map.get("a")), Arrays.asList(1, 2)); + Assert.assertEquals(asList(map.get("b")), Arrays.asList()); + + List nested = insert("stream_nested", + "id Int32, n Nested(n1 Int32, n2 Nullable(String)), tail Float64", writer -> { + try (ValueWriter row = writer.beginRow()) { + row.writeInt32(1); + try (ValueWriter rows = row.beginNested(2)) { + try (ValueWriter inner = rows.beginTuple()) { + inner.writeInt32(5); + inner.writeString("a"); + } + try (ValueWriter inner = rows.beginTuple()) { + inner.writeInt32(6); + inner.writeString((String) null); + } + } + row.writeFloat64(1.5d); + } + }); + List rows = asList(nested.get(0).getObject("n")); + Assert.assertEquals(((Object[]) rows.get(0))[0], 5); + Assert.assertEquals(((Object[]) rows.get(0))[1], "a"); + Assert.assertEquals(((Object[]) rows.get(1))[0], 6); + Assert.assertNull(((Object[]) rows.get(1))[1]); + } + + @Test(groups = {"integration"}) + public void binaryJsonWithNestedObjectsNullsAndArrays() throws Exception { + if (jsonUnsupported()) { + return; + } + List rows = insert("stream_json", "id Int32, doc JSON", writer -> { + try (ValueWriter row = writer.beginRow()) { + row.writeInt32(1); + try (ValueWriter doc = row.beginJson()) { + doc.writeString("name", "ada"); + try (ValueWriter address = doc.beginObject("address", 2)) { + address.writeString("city", "Berlin"); + address.writeNull("zip"); + } + try (ValueWriter tags = doc.beginArray("tags", ClickHouseColumn.of("e", "Nullable(String)"), 3)) { + tags.writeString("a"); + tags.writeNull(); + tags.writeString("b"); + } + try (ValueWriter items = doc.beginArray("items", ClickHouseColumn.of("e", "JSON"), 2)) { + try (ValueWriter first = items.beginJson(1)) { + first.writeInt64("a", 100L); + } + try (ValueWriter second = items.beginJson(1)) { + second.writeString("b", "name"); + } + } + } + } + try (ValueWriter row = writer.beginRow()) { + row.writeInt32(2); + try (ValueWriter doc = row.beginJson(0)) { + } + } + }); + + Map doc = jsonMap(rows.get(0).getObject("doc")); + Assert.assertEquals(doc.get("name"), "ada"); + Assert.assertEquals(doc.get("address.city"), "Berlin"); + Assert.assertEquals(asList(doc.get("tags")), Arrays.asList("a", null, "b")); + List items = asList(doc.get("items")); + Assert.assertEquals(items.size(), 2); + Assert.assertEquals(jsonMap(items.get(0)).get("a"), 100L); + Assert.assertEquals(jsonMap(items.get(1)).get("b"), "name"); + Assert.assertTrue(jsonMap(rows.get(1).getObject("doc")).isEmpty()); + + List fields = client.queryAll( + "SELECT doc.address.city AS city, doc.address.zip AS zip, toJSONString(doc) AS text " + + "FROM stream_json WHERE id = 1"); + Assert.assertEquals(fields.get(0).getString("city"), "Berlin"); + Assert.assertNull(fields.get(0).getObject("zip")); + String text = fields.get(0).getString("text"); + Assert.assertTrue(text.contains("\"city\":\"Berlin\""), text); + Assert.assertTrue(text.contains("\"tags\":[\"a\",null,\"b\"]"), text); + Assert.assertTrue(text.contains("\"items\":[{\"a\":100},{\"b\":\"name\"}]"), text); + } + + @Test(groups = {"integration"}) + public void typedJsonPathsSkipTheDynamicTag() throws Exception { + if (jsonUnsupported()) { + return; + } + List rows = insert("stream_json_typed", + "doc JSON(name String, count Int64)", writer -> { + try (ValueWriter row = writer.beginRow()) { + try (ValueWriter doc = row.beginJson(2)) { + doc.writeString("name", "ada"); + doc.writeInt64("count", 7L); + } + } + }); + Map doc = jsonMap(rows.get(0).getObject("doc")); + Assert.assertEquals(doc.get("name"), "ada"); + Assert.assertEquals(doc.get("count"), 7L); + } + + private List insert(String table, String columns, RowWriter write) throws Exception { + CommandSettings command = new CommandSettings(); + command.serverSetting("allow_experimental_json_type", "1"); + client.execute("DROP TABLE IF EXISTS " + table, command).get(TIMEOUT_SECONDS, TimeUnit.SECONDS); + String nested = columns.contains("Nested(") ? " SETTINGS flatten_nested = 0" : ""; + client.execute("CREATE TABLE " + table + " (" + columns + ") ENGINE = MergeTree ORDER BY tuple()" + nested, + command).get(TIMEOUT_SECONDS, TimeUnit.SECONDS); + TableSchema schema = client.getTableSchema(table); + client.insert(table, out -> { + RowBinaryStreamWriter writer = new RowBinaryStreamWriter(out, schema, ClickHouseFormat.RowBinary); + write.accept(writer); + }, ClickHouseFormat.RowBinary, new InsertSettings()).get(TIMEOUT_SECONDS, TimeUnit.SECONDS); + return client.queryAll("SELECT * FROM " + table + " ORDER BY " + orderBy(columns)); + } + + private static String orderBy(String columns) { + int comma = columns.indexOf(','); + String first = comma < 0 ? columns : columns.substring(0, comma); + String name = first.trim().split(" ")[0]; + if ("id".equals(name)) { + return "id"; + } + return "tuple()"; + } + + private boolean jsonUnsupported() throws Exception { + if (isCloud()) { + return true; + } + List version = client.queryAll("SELECT version()"); + return ClickHouseVersion.of(version.get(0).getString(1)).check("(,24.8]"); + } + + @SuppressWarnings("unchecked") + private static Map jsonMap(Object value) { + Assert.assertTrue(value instanceof Map, String.valueOf(value)); + return (Map) value; + } + + @SuppressWarnings("unchecked") + private static List asList(Object value) { + if (value instanceof BinaryStreamReader.ArrayValue) { + return ((BinaryStreamReader.ArrayValue) value).asList(); + } + if (value instanceof List) { + return (List) value; + } + throw new AssertionError("Expected a list but was " + (value == null ? "null" : value.getClass().getName())); + } + + private interface RowWriter { + void accept(RowBinaryStreamWriter writer) throws IOException; + } +} diff --git a/docs/features.md b/docs/features.md index a41f5f936..eafe11c39 100644 --- a/docs/features.md +++ b/docs/features.md @@ -26,6 +26,7 @@ This document lists stable, user-visible behavior in `client-v2` and `jdbc-v2` t - Geometry type support: For ClickHouse `25.11+`, where `Geometry` changed from a string alias to `Variant(Point, Ring, LineString, MultiLineString, Polygon, MultiPolygon)`, the client reads and writes `Geometry` values through generic records, binary readers, POJO binding, and SQL parameter formatting, using Java array dimensionality to represent the geometry shape. ClickHouse `26.8+` adds `MultiPoint` to that variant, appended after the existing six so their discriminators are unchanged; a `MultiPoint` value read from a `Geometry` column is returned as `double[][]`, the same shape as `Ring` and `LineString`. - MultiPoint type support: For ClickHouse `26.8+`, reads and writes the `MultiPoint` geo type, which is transmitted exactly like `Ring` and `LineString` (`Array(Point)`) and is represented as `double[][]` through generic records, binary readers, POJO binding, and SQL parameter formatting. A `MultiPoint` held in a `Dynamic` column is decoded on read; it cannot be written into a `Dynamic` column, because the client infers the type from the Java value and `double[][]` resolves to `Ring`. - Nested type support: Un-flattened `Nested(f1 T1, ..., fN TN)` columns (tables created with `flatten_nested = 0`) can be written through the insert path (`Client#insert`) using `RowBinaryFormatWriter#setValue`, and are read back through the binary readers and generic records. The column is serialized the same way it is read — identically to `Array(Tuple(T1, ..., TN))`, a var-uint row count followed by one tuple per nested row — so the value supplied for the column is a `List` (or array) of tuples, one tuple per nested row, each carrying the N field values in declaration order. +- Streaming RowBinary reader and writer: Experimental `RowBinaryStreamWriter` and `RowBinaryStreamReader` in `com.clickhouse.client.api.data_formats.stream` write and read RowBinary values as scopes, so an `Array`, `Map`, `Tuple`, unflattened `Nested`, `QBit`, `Variant`, `Dynamic`, or binary `JSON` value can be produced or consumed one element at a time without building the whole container first. A `JSON` document is a list of paths: `beginJson` opens it, `writeString(path, value)` / `writeInt64(path, value)` / `writeNull(path)` write a dynamic path, `beginObject(path)` flattens a nested object into dotted paths, and `beginArray(path, elementType)` writes an array, including an array of JSON objects. `writeString` without a path remains the string form of a JSON column. A length-prefixed container opened with a known size writes the varint immediately and sends element bytes straight to the stream; an unknown size buffers the encoded bytes and writes the length when the scope is closed. `writeString(byte[])` preserves bytes that are not valid UTF-8. The writer supports `RowBinary` and `RowBinaryWithDefaults` and writes no header — the `INSERT` statement carries names and types. The reader supports `RowBinary` when the caller supplies the schema, `RowBinaryWithNames` with the same schema, and `RowBinaryWithNamesAndTypes` with the schema taken from the header. A column left unwritten is an error. `RowBinaryFormatWriter`, POJO insert, and JDBC are unchanged. `Native` is not supported on this path. See [reader_writer_api.md](reader_writer_api.md). - Insert APIs: Supports inserting registered POJOs, raw streams, and callback-driven writers, with optional column lists and format selection. - Insert controls: Supports insert-specific settings such as deduplication token, query id, compression behavior, and request headers. - Command execution: Executes DDL or other non-result commands and exposes response summaries and operation metrics. diff --git a/docs/integration-client.md b/docs/integration-client.md index 79ff66bc2..ccfa62b00 100644 --- a/docs/integration-client.md +++ b/docs/integration-client.md @@ -812,6 +812,92 @@ public void writeEventsStream(Client client, TableSchema schema, List eve Populate each row with `setValue(column, value)` (by name or 1-based index) and finish it with `commitRow()`. Do not close the stream yourself — the client manages the lifecycle. +When a column is itself a container and you do not already hold the whole value, use the experimental streaming writer inside the same callback. `RowBinaryStreamWriter` writes each row and each nested value as a scope. Pass the element count to `beginArray` when you know it: the length goes out immediately and the elements follow with no extra copy. Close every scope, including the row. A column you do not write is an error. `RowBinaryFormatWriter` stays the API for named cells and for values you already have in memory. The matching reader is `RowBinaryStreamReader`. The full contract is in [reader_writer_api.md](reader_writer_api.md). + +The example below expects columns `id Int64`, `name String`, and `tags Array(String)`, in that order. + +```java +import com.clickhouse.client.api.Client; +import com.clickhouse.client.api.data_formats.stream.RowBinaryStreamWriter; +import com.clickhouse.client.api.data_formats.stream.ValueWriter; +import com.clickhouse.client.api.insert.InsertResponse; +import com.clickhouse.client.api.insert.InsertSettings; +import com.clickhouse.client.api.metadata.TableSchema; +import com.clickhouse.data.ClickHouseFormat; + +public static class TaggedEvent { + public long id; + public String name; + public List tags; +} + +public void writeTaggedEvents(Client client, TableSchema schema, List events) throws Exception { + ClickHouseFormat format = ClickHouseFormat.RowBinary; + + try (InsertResponse response = client.insert("events", out -> { + RowBinaryStreamWriter writer = new RowBinaryStreamWriter(out, schema, format); + for (TaggedEvent event : events) { + try (ValueWriter row = writer.beginRow()) { + row.writeInt64(event.id); + row.writeString(event.name); + List tags = event.tags == null ? Collections.emptyList() : event.tags; + try (ValueWriter tagValues = row.beginArray(tags.size())) { + for (String tag : tags) { + tagValues.writeString(tag); + } + } + } + } + }, format, new InsertSettings()).get()) { + // handle response metrics + } +} +``` + +A `JSON` column is written the same way. `beginJson` opens the document. A nested object becomes dotted paths, `writeNull` writes a JSON null, and an array of objects is an `Array(JSON)` whose elements are documents. The column in this example is `doc JSON`. + +```java +import com.clickhouse.client.api.Client; +import com.clickhouse.client.api.data_formats.stream.RowBinaryStreamWriter; +import com.clickhouse.client.api.data_formats.stream.ValueWriter; +import com.clickhouse.client.api.insert.InsertResponse; +import com.clickhouse.client.api.insert.InsertSettings; +import com.clickhouse.client.api.metadata.TableSchema; +import com.clickhouse.data.ClickHouseColumn; +import com.clickhouse.data.ClickHouseFormat; + +public void writeProfile(Client client, TableSchema schema) throws Exception { + ClickHouseFormat format = ClickHouseFormat.RowBinary; + ClickHouseColumn nullableString = ClickHouseColumn.of("e", "Nullable(String)"); + ClickHouseColumn json = ClickHouseColumn.of("e", "JSON"); + + try (InsertResponse response = client.insert("profiles", out -> { + RowBinaryStreamWriter writer = new RowBinaryStreamWriter(out, schema, format); + try (ValueWriter row = writer.beginRow()) { + try (ValueWriter doc = row.beginJson()) { + doc.writeString("name", "ada"); + try (ValueWriter address = doc.beginObject("address", 2)) { + address.writeString("city", "Berlin"); + address.writeNull("zip"); + } + try (ValueWriter tags = doc.beginArray("tags", nullableString, 3)) { + tags.writeString("a"); + tags.writeNull(); + tags.writeString("b"); + } + try (ValueWriter items = doc.beginArray("items", json, 1)) { + try (ValueWriter item = items.beginJson(1)) { + item.writeInt64("a", 100L); + } + } + } + } + }, format, new InsertSettings()).get()) { + // handle response metrics + } +} +``` + **Key classes:** | Class | Role | diff --git a/docs/reader_writer_api.md b/docs/reader_writer_api.md new file mode 100644 index 000000000..6dbfee5e8 --- /dev/null +++ b/docs/reader_writer_api.md @@ -0,0 +1,247 @@ +# Reader Writer API Spec + +## Abstract + +The client already has a reader and a writer, but the two are not the same shape, and both materialize nested values before the application can touch them. This document specifies a streaming API that is symmetrical for the types that can be both read and written. Some ClickHouse types remain read-only; those limits are listed below. + +## Data flow + +A value is encoded on one side and decoded on the other. That encoding depends on the ClickHouse type and the server version, not on the application's domain types. An `Int32` is always the same four little-endian bytes. `String`, `Array`, and `Dynamic` each have their own codec. + +The application does not have to use the same Java type as the column. It may write a Java `long` into an `Int32`, which is a conversion (and an overflow check), not a different codec. + +The format is the container around those codecs. It decides how rows are framed and how nulls and defaults are marked. `RowBinary` and `RowBinaryWithDefaults` are row-oriented: each row is column values in schema order, and each value uses the RowBinary codec. `RowBinaryWithNamesAndTypes` adds a header of names and then types. `Native` is different: values are grouped into column blocks. `Dynamic` adds a type tag in front of the codec payload. + +Three layers, in order: + +1. **Conversion.** Java value to the canonical form of the ClickHouse type (`long` to `Int32` with a range check, `LocalDate` to a date). This is the public read and write methods. +2. **Format framing.** Null markers and, for `RowBinaryWithDefaults`, the default marker. This is a property of the column (nullable, has a `DEFAULT`) and of the format. It is not part of the type codec. Nested `Nullable` elements use only the inner null byte. The default marker exists only on a top-level column. +3. **Codec.** Canonical bytes: little-endian numbers, varint lengths, dynamic type tags. The codec does not accept an `Object` and does not look at JDBC types. + +Today the public API is a buffered, JDBC-style row (`setValue` / `commitRow`, `next` / `getInteger`). That API stays. Connectors that already have a cursor need to push or pull nested values without building the whole container first. That is the streaming API in this document. + +## A scope is a sequence of values + +A row, an array, a tuple, a map, a nested row, a variant arm, and a dynamic value are the same thing: an ordered sequence of values. The type of the next value comes from the schema (or, inside `Dynamic`, from the runtime type). The same verbs work at every level. Nesting is opening a child scope. Closing the child commits it and returns to the parent. + +Types are not repeated on each call. The caller passes a `TableSchema`. For `RowBinaryWithNamesAndTypes` the reader takes the schema from the header. The current column or element type selects the codec, the null marker, and the default marker. That is how format knowledge moves through the layers. `Dynamic` is the exception: the schema only says `Dynamic`, so `beginDynamic` supplies the runtime type. + +The API is positional. Columns are written and read in schema order. Named setters stay on the buffered writer. On write, `MATERIALIZED` and `ALIAS` columns are omitted, matching `RowBinaryFormatWriter`. On read, the schema is the columns present in the stream; nothing is omitted. + +Closing a scope is explicit (`close()`, or try-with-resources). The next value does not implicitly end an open container. An open child when the parent is closed is an error. Implicit close makes a nested array ambiguous: the next call could be another element or the next column. + +```java +void writeData(MyDTO dto, OutputStream out, TableSchema schema) throws IOException { + RowBinaryStreamWriter writer = new RowBinaryStreamWriter( + out, schema, ClickHouseFormat.RowBinaryWithDefaults); + try (ValueWriter row = writer.beginRow()) { + row.writeInt32(dto.getIntValue1()); + row.writeString(dto.getStrVal()); // null is SQL NULL when the column is Nullable + + try (ValueWriter items = row.beginArray()) { // size unknown: buffer encoded bytes + for (String li : dto.getList()) { + items.writeString(li); + } + } + + try (ValueWriter matrix = row.beginArray(dto.tableData().size())) { // size known: straight to the stream + for (String[] line : dto.tableData()) { + try (ValueWriter cells = matrix.beginArray(line.length)) { + for (String col : line) { + cells.writeString(col); + } + } + } + } + row.writeDate(dto.getStartDate()); + } +} +``` + +The reader has the same shape. It does not take a destination collection. + +```java +void readData(MyDTO dto, InputStream in, TableSchema schema) throws IOException { + RowBinaryStreamReader reader = new RowBinaryStreamReader( + in, schema, ClickHouseFormat.RowBinary); + if (!reader.hasNext()) { + return; + } + try (ValueReader row = reader.beginRow()) { + dto.setIntValue1(row.readInt32()); + dto.setStrVal(row.readString()); + List list = new ArrayList(); + try (ValueReader items = row.beginArray()) { + while (items.hasNext()) { + list.add(items.readString()); + } + } + dto.setList(list); + } +} +``` + +`beginArray()` belongs outside the loop. Calling it inside the loop would open a new array for every element. + +## Streaming and zero-copy + +`Array`, `Map`, `Nested`, and `QBit` are length-prefixed: a varint count, then the elements. `Tuple` has no count; it is N fields in schema order. Because the count comes first, a container whose size is not known yet cannot be written straight to the caller stream. + +Two open modes: + +- `beginArray(int size)`, `beginMap(int pairs)`, `beginNested(int rows)` with `size >= 0`. The varint is written immediately. Element bytes go to the current stream. `close()` checks the count. This is the path that does not copy the payload. +- `beginArray()`, `beginMap()`, `beginNested()`. Element bytes go to a growable buffer. `close()` writes the varint and then the buffer. The encoded bytes are copied once. A nested unknown-size container is copied again into its parent buffer. Callers who know the size avoid that copy. + +Zero-copy rules: + +- There is no row `Object[]`. A scalar is encoded when its write method is called. +- `writeString(byte[])` and `writeString(ByteBuffer)` write a varint length and the existing bytes. There is no UTF-8 round-trip. Bytes that are not valid UTF-8 are preserved. A heap `ByteBuffer` is written from its backing array. A direct buffer is copied once into a temporary array. +- Those payload bytes are not copied again when every open ancestor was opened with a known size, or has no length prefix (`Tuple`, a variant arm, a dynamic value, a top-level column). If any ancestor is buffering, the bytes are copied into that buffer. The copy is required because the length sits in front of the bytes. +- `Tuple` never buffers on its own. Only a length-prefixed field inside it does. +- Readers do not allocate the collection. `beginArray()` reads the varint into a remaining count. `readInt32()` returns a primitive. `readStringValue()` returns a `StringValue` over the bytes just read (one `byte[]`, no `String` unless the caller asks). `readString()` decodes UTF-8 and is lossy for invalid UTF-8. `readInto(int[] dst, int off, int len)` fills a caller buffer for an `Array(Int32)`. +- `hasNext()` is the remaining count of the current array, nested rows, or map pairs. After a map key, the next read or write must be the value. `hasNext()` is true only at the start of a pair, so the loop is "key, then value". +- `size()` is the total count read from the varint, or the number of tuple fields or row columns. It is not the number of values still remaining. + +`QBit` is a fixed-length array. `beginQBit()` and `writeFloat64Array(double[])` (and the matching primitive helpers) require the length to equal the dimension. An unknown size is rejected. On the wire a `QBit` is still a varint plus elements; that layout is a RowBinary detail, not a reason to treat `QBit` as `Array`. + +Bulk helpers (`writeInt32Array(int[])`, and the same for `Int8`, `Int16`, `Int64`, `Float32`, `Float64`, `BFloat16`, and `Bool`) write a whole primitive array as the current value. They use the known-size path. The loop form stays for a cursor source. A nullable element type is written with a non-null marker per element, because a primitive array cannot hold null. A null element still uses the loop and `writeNull()`. + +On any API or IO failure the stream is desynchronized. The writer or reader is not reusable. One writer or reader, one thread, one stream. The caller owns the stream and closes it. + +`Date` and `Date32` are encoded in UTC, the same as the buffered writer. `DateTime` and `DateTime64` use the column time zone, or UTC when the column has none. A constructor `TimeZone` overrides that default. + +## Writer + +`com.clickhouse.client.api.data_formats.stream.RowBinaryStreamWriter` + +```text +RowBinaryStreamWriter(OutputStream out, TableSchema schema, ClickHouseFormat format) +RowBinaryStreamWriter(OutputStream out, TableSchema schema, ClickHouseFormat format, TimeZone timeZone) +ValueWriter beginRow() +int getRowCount() +ClickHouseFormat getFormat() +``` + +Formats: `RowBinary` and `RowBinaryWithDefaults` only. There is no header. The `INSERT` statement carries names and types. + +`ValueWriter` is implemented by every scope. `close()` commits the scope. A second `close()` does nothing. Production code is Java 8. + +Scalars, named for the ClickHouse type. The argument is the canonical Java type. A second overload exists where a wider Java type needs a range check (`writeInt32(long)`). There is no `Object` overload. + +- `writeInt8`, `writeInt16`, `writeInt32`, `writeInt64`, `writeInt128`, `writeInt256` +- `writeUInt8`, `writeUInt16`, `writeUInt32`, `writeUInt64`, `writeUInt128`, `writeUInt256` +- `writeFloat32`, `writeFloat64`, `writeBFloat16`, `writeBool` +- `writeString(String)`, `writeString(byte[])`, `writeString(ByteBuffer)` +- `writeFixedString(String)`, `writeFixedString(byte[])` — length is the column precision +- `writeDate(LocalDate)`, `writeDate32(LocalDate)` +- `writeDateTime` for `LocalDateTime`, `ZonedDateTime`, and `Instant` +- `writeDateTime64` for `LocalDateTime` and `ZonedDateTime` +- `writeUUID`, `writeIPv4`, `writeIPv6`, `writeDecimal` +- `writeEnum8` and `writeEnum16` for the numeric wire value or the enum name +- `writeBitmap` for `AggregateFunction(groupBitmap, ...)` only + +`writeString(null)` (and the other reference overloads) writes SQL NULL when the current value is a nullable string. A null passed to the wrong type is a type error. `writeNull()` writes SQL NULL for whatever nullable value is current. + +`writeDefault()` writes the `RowBinaryWithDefaults` default marker and no body. It is legal only for a top-level column whose default kind is `DEFAULT`. It is not legal inside an array, a tuple, or a map, and it is not legal for `RowBinary`. + +Containers: + +- `beginArray()`, `beginArray(int size)` +- `beginTuple()` +- `beginMap()`, `beginMap(int pairs)` — keys and values alternate; a map key is not nullable; an odd number of writes is an error +- `beginNested()`, `beginNested(int rows)` — unflattened `Nested(f1 T1, ..., fN TN)` only, wire-identical to `Array(Tuple(T1, ..., TN))`. Each row is a `beginTuple()`. Flattened Nested (`flatten_nested = 1`) is already separate `Array` columns in the schema. +- `beginVariant(int ordinal)` — one value of that alternative, which may itself be a container. Ordinal `0xFF` is `writeNull()` on a `Variant` or `Geometry` value. +- `beginDynamic(ClickHouseColumn runtimeType)` — writes the dynamic type tag, then one value of that type. +- `beginQBit()` — `beginArray(dimension)`. + +`LowCardinality(T)` and `SimpleAggregateFunction(func, T)` are written as `T`. A nullable argument of a simple aggregate function still gets the nested null marker. + +`row.close()` requires every non-omitted column to have been written. A missing column is an error. It is not silently NULL or DEFAULT. That differs from `RowBinaryFormatWriter`, which treats an unset cell as null. The buffered writer does not change. + +A negative size is an error. Unknown size is the no-argument method, not `-1`. + +## Reader + +`com.clickhouse.client.api.data_formats.stream.RowBinaryStreamReader` + +```text +RowBinaryStreamReader(InputStream in, TableSchema schema, ClickHouseFormat format) +RowBinaryStreamReader(InputStream in, TableSchema schema, ClickHouseFormat format, TimeZone timeZone) +boolean hasNext() +ValueReader beginRow() +TableSchema getSchema() +``` + +Formats: + +- `RowBinary`: schema required, no header. +- `RowBinaryWithNames`: schema required. The name header is read and checked against the schema. +- `RowBinaryWithNamesAndTypes`: schema is read from the header. A schema passed by the caller is not used. + +`Native` is out of scope. It is column blocks, and `QBit` uses a different bit-plane layout there. Streaming a Native row still means buffering the block. + +`hasNext()` peeks one byte. It is false at end of stream. It throws if a row is still open. `beginRow()` throws `EOFException` when there is no row. + +`ValueReader` mirrors the writer: `readInt32()`, `readString()`, `readStringValue()`, and the other scalar twins, plus `beginArray()`, `beginTuple()`, `beginMap()`, `beginNested()`, `beginVariant()`, `beginDynamic()`, `beginQBit()`. `beginVariant()` reads the discriminator. `variantOrdinal()` reports it. `beginDynamic()` reads the type tag. `runtimeType()` returns the concrete column. A `Dynamic` `Nothing` value has no body; the scope is already complete and `runtimeType()` is `Nothing`. + +`isNull()` consumes a SQL NULL and returns true, or returns false and leaves the body to be read. On a non-nullable `Variant` or `Geometry` it recognizes discriminator `0xFF`. Primitive reads of SQL NULL throw `NullValueException` and consume the null, same as the buffered getters. `skip()` walks the current value without building a `List`, `Map`, or `ArrayValue`. `readInto` fills a caller `int[]` from an `Array(Int32)`. + +`close()` requires every column, field, or element the caller is responsible for to have been consumed. Use `skip()` for the rest. `hasNext()` and `size()` behave as described above. + +Strings inside containers are returned as `StringValue` from `readStringValue()`, including nested strings. The buffered reader still returns nested strings as `String`. + +## Format framing + +Top-level column, `RowBinary`: + +- Nullable null: one byte `1`, no body. +- Nullable value: one byte `0`, then the body. +- Non-nullable value: body only. +- Non-nullable null: error, except `Dynamic` (encoded as `Nothing`) and `Variant` / `Geometry` (discriminator `0xFF`). + +Top-level column, `RowBinaryWithDefaults`. Every column gets a default marker, including columns that have no default expression: + +- `writeDefault()` on a `DEFAULT` column: one byte `1`, no body. +- Value: one byte `0` (not a default). If the column is nullable, another `0`, then the body. +- SQL NULL on a nullable column that is not `writeDefault()`: `0`, then `1`, no body. +- SQL NULL on a non-nullable column: error, with the same `Dynamic` and `Variant` exceptions. The "not a default" byte is still written. + +Nested `Nullable` (an array element, a tuple field, a map value): one byte, `1` for null and `0` for a value, then the body. No default marker. + +## Container encodings + +- `Array`: varint count, then elements. A nullable element is a null byte plus the body. +- `Tuple`: fields in order, no count. +- `Map`: varint pair count, then key, value, key, value. Key order is the order of the write calls. +- `Nested` (unflattened): same bytes as `Array(Tuple(...))`. +- `QBit`: same bytes as an array whose length is the dimension. +- `Variant` and `Geometry`: one discriminator byte, then one value. `0xFF` is null. `Geometry` alternatives are `Point`, `Ring`, `LineString`, `MultiLineString`, `Polygon`, `MultiPolygon`, and `MultiPoint`. `Point` is a tuple of two `Float64`. `Ring`, `LineString`, and `MultiPoint` are arrays of points. `Polygon` and `MultiLineString` are arrays of rings. `MultiPolygon` is an array of polygons. There is no separate geo API. +- `Dynamic`: type tag (the existing dynamic tag encoding), then one value. +- `JSON` (binary object, not the string form): varint path count, then each path as a string name followed by its value. A dynamic path prefixes the value with a dynamic type tag. A typed path does not. `Nothing` (`0x00`) is a JSON null. Nested objects written with `beginObject` do not have their own count; their leaves are paths of the enclosing document. A path whose value is a JSON document (an element of `Array(JSON)`) is a dynamic `JSON` type tag plus another path list. + +## Symmetry limits + +These types are not given a new structured write path: + +- `AggregateFunction` other than `groupBitmap`. `writeBitmap` / `readBitmap` cover `groupBitmap` only. +- Structured `JSON` is a map of paths, written with `beginJson`. A dynamic path is a name plus a dynamic type tag and a value. A typed path declared on the column is the value only. A nested object opened with `beginObject` is stored as dotted paths (`address.city`) on the same document. `writeNull(path)` writes a JSON null (`Nothing`), which is not the same as leaving the path out. An array of objects is `beginArray(path, JSON column)` and each element is `beginJson`. `writeString` without a path is still the string form of a `JSON` column, used when the server is reading JSON as a string. The streaming reader still reads that string form. Structured documents are read back through the existing binary reader, which returns a map of paths. +- Inside `Dynamic`, payloads whose type tag the codec does not finish: `Tuple`, `Variant`, `Dynamic`, `JSON`, `Nested`, `QBit`, `AggregateFunction`, `SimpleAggregateFunction`, and `Geometry`. `beginDynamic` rejects them, including when they are nested inside an array or a map. Calling the existing tag writer for `Tuple` would emit a partial tag and desynchronize the stream. + +`LowCardinality` is the inner type on the RowBinary value path. The low-cardinality flag is not a dictionary. + +## Errors + +- `IllegalStateException`: wrong verb for the current type, unclosed child, declared size mismatch, extra value, missing column, `writeDefault()` in the wrong place, use after a failure. +- `IllegalArgumentException`: overflow (`writeInt32(long)` and the other ranged overloads), null into a non-nullable value, negative size, `QBit` length, odd map entry, `writeDefault()` on a column without `DEFAULT`. +- `NullValueException`: primitive read of SQL NULL, or `beginVariant()` on discriminator `0xFF`. +- `ClientException`: unsupported `Dynamic` payload, malformed header. +- `IOException`: the stream failed. +- `EOFException`: `beginRow()` when `hasNext()` is false. + +After any of these the reader or writer throws `IllegalStateException` on further use. Do not retry the same stream. + +## What does not change + +`ClickHouseBinaryFormatWriter` and `RowBinaryFormatWriter` stay the buffered, name-addressable API. POJO insert and JDBC keep using them. The streaming types are a separate experimental API, used inside `DataStreamWriter.onOutput`. A retry still calls `onRetry()` and then `onOutput` again; the caller writes the rows again. + +The materialized container codecs (`serializeArrayData`, `serializeMapData`, `serializeTupleData`) stay the path that walks a finished `List`, array, or `Map`. The streaming writer does not call them. From 6c9c5a770c3626736b08501c15c358ce4aa00539 Mon Sep 17 00:00:00 2001 From: Sergey Chernov Date: Wed, 4 Nov 2026 16:45:05 -0800 Subject: [PATCH 2/2] Fixed the cluster tests --- .../api/data_formats/stream/RowBinaryStreamServerTest.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/client-v2/src/test/java/com/clickhouse/client/api/data_formats/stream/RowBinaryStreamServerTest.java b/client-v2/src/test/java/com/clickhouse/client/api/data_formats/stream/RowBinaryStreamServerTest.java index f206ebbd1..9be3605f3 100644 --- a/client-v2/src/test/java/com/clickhouse/client/api/data_formats/stream/RowBinaryStreamServerTest.java +++ b/client-v2/src/test/java/com/clickhouse/client/api/data_formats/stream/RowBinaryStreamServerTest.java @@ -246,7 +246,7 @@ private List insert(String table, String columns, RowWriter write command.serverSetting("allow_experimental_json_type", "1"); client.execute("DROP TABLE IF EXISTS " + table, command).get(TIMEOUT_SECONDS, TimeUnit.SECONDS); String nested = columns.contains("Nested(") ? " SETTINGS flatten_nested = 0" : ""; - client.execute("CREATE TABLE " + table + " (" + columns + ") ENGINE = MergeTree ORDER BY tuple()" + nested, + client.execute("CREATE TABLE " + table + " (" + columns + ") ORDER BY tuple()" + nested, command).get(TIMEOUT_SECONDS, TimeUnit.SECONDS); TableSchema schema = client.getTableSchema(table); client.insert(table, out -> {