Support the modern JSON type (#142)

Object('json') was removed from the server in 24.8, so JSON was not merely
legacy on current ClickHouse - it was unusable: reading a JSON column raised
UnknownTypeError, and the one format asynch implemented no longer existed
server-side.

Read path composes the format out of column readers rather than hand-rolling
it. dynamiccolumn.pyx implements SerializationDynamic/SerializationVariant:
the variant list arrives in the state prefix, global discriminators are
recovered by sorting the type names (SharedVariant sorts among them, so its
index has to be derived, not assumed), and values that match no declared
variant are decoded from their encodeDataType + serializeBinary blob. That
decoder builds the value's column reader from the decoded type string and
drives it over an in-memory reader, so every type asynch already supports is
reachable inside a shared JSON value for free.

jsoncolumn.pyx folds the per-path dynamic columns and the
Array(Tuple(String, String)) shared sub-column back into nested dicts,
including dotted-path denormalisation. Writing infers a spec per value,
groups rows per path and variant, and emits V2 framing; the block's items now
reach write_state_prefix, since the prefix lists the paths about to be
written. Dynamic is registered as a column type too - it shows up as a
variant spec for heterogeneous arrays.

Object('json') keeps its own reader: its framing (UInt8 version then a type
spec) is incompatible, and only old servers speak it.

Verified against ClickHouse 26.7: nested objects, heterogeneous and nested
arrays, arrays of objects, nulls, big integers, unicode, empty documents, a
2000-row multi-path batch, dict and text input, and server-side path access
(doc.a.b) on what the driver wrote. Round-trip equality holds for every
document in the test set.

Known gap: a JSON column nested inside Array/Tuple can be read but not
written - that needs the container columns to thread items through their
prefixes.

Author: long2ice

CHANGELOG.md

@@ -32,6 +32,12 @@ unchanged.
 
 #### API
 
+- Support for the modern `JSON` type (ClickHouse 24.8+). Reading composes a
+  `Dynamic`/`Variant` reader per path with the shared-data sub-column and
+  returns nested dicts; writing accepts dicts or JSON text and emits the V2
+  object framing. `Dynamic` is readable as a column type in its own right.
+  The pre-24.8 `Object('json')` spelling keeps its own reader, since modern
+  servers reject the type outright (#142)
 - `Connection.cancel()` / `Cursor.cancel()` stop a running query from another
   task. Only the cancel packet is sent - the task awaiting the query owns the
   read side and drains the stream, so the connection stays usable (#104)

README.md

@@ -210,6 +210,29 @@ async def stream_rows(conn: Connection):
             process(row)
 ```
 
+### JSON columns
+
+`JSON` columns (ClickHouse 24.8+) read as nested dicts and accept dicts or
+JSON text on insert.
+
+```python
+async def use_json(conn: Connection):
+    async with conn.cursor() as cursor:
+        await cursor.execute(
+            "CREATE TABLE test.events (id UInt32, doc JSON) ENGINE = MergeTree ORDER BY id"
+        )
+        await cursor.execute(
+            "INSERT INTO test.events (id, doc) VALUES",
+            [
+                (1, {"user": {"name": "ada"}, "tags": ["a", "b"]}),
+                (2, '{"user": {"name": "bob"}}'),   # JSON text works too
+            ],
+        )
+
+        await cursor.execute("SELECT doc FROM test.events ORDER BY id")
+        assert await cursor.fetchone() == ({"user": {"name": "ada"}, "tags": ["a", "b"]},)
+```
+
 ### Cancelling a query
 
 A long-running query can be stopped from another task; the connection is left

asynch/proto/columns/__init__.pyx

@@ -41,6 +41,7 @@ from .intervalcolumn import (
     IntervalYearColumn,
 )
 from .ipcolumn import IPv4Column, IPv6Column
+from .dynamiccolumn import create_dynamic_column
 from .jsoncolumn import create_json_column
 from .lowcardinalitycolumn import create_low_cardinality_column
 from .mapcolumn import create_map_column
@@ -135,7 +136,12 @@ def get_column_by_spec(spec, column_options):
 
     elif spec.startswith("Map"):
         return create_map_column(spec, create_column_with_options, column_options)
-    elif spec.startswith("Object('json')"):
+    elif spec == "Dynamic" or spec.startswith("Dynamic("):
+        return create_dynamic_column(spec, create_column_with_options, column_options)
+
+    elif spec.startswith("Object('json')") or spec == "JSON" or spec.startswith("JSON("):
+        # `Object('json')` is the pre-24.8 spelling; modern servers only know
+        # `JSON`, optionally with parameters that do not change the layout.
         return create_json_column(spec, create_column_with_options, column_options)
     else:
         for alias, primitive in aliases:
@@ -183,7 +189,12 @@ async def write_column(
     column = get_column_by_spec(column_spec, column_options)
 
     try:
-        await column.write_state_prefix()
+        if getattr(column, "prefix_needs_items", False):
+            # A JSON column's prefix lists the paths it is about to write, so
+            # it can only be produced from the block's items.
+            await column.write_state_prefix(items)
+        else:
+            await column.write_state_prefix()
         await column.write_data(items)
 
     except ColumnTypeMismatchException as e:

asynch/proto/columns/dynamiccolumn.pyi

@@ -0,0 +1,51 @@
+# Generated by stubgen-pyx from asynch/proto/columns/dynamiccolumn.pyx
+
+"""Reader for ClickHouse's Dynamic / Variant column families.
+
+These are not user-facing types yet: they exist so the JSON column can be
+composed out of them instead of hand-rolling Variant deserialization. The
+byte layout mirrors `SerializationDynamic` and `SerializationVariant`.
+"""
+
+from .base import Column
+
+DYNAMIC_V1 = 1
+DYNAMIC_V2 = 2
+VARIANT_MODE_BASIC = 0
+VARIANT_MODE_COMPACT = 1
+NULL_DISCRIMINATOR = 255
+SHARED_VARIANT_NAME = "SharedVariant"
+
+class DynamicColumn(Column):
+    """A ClickHouse `Dynamic` column.
+
+    The variant types are not known at construction: they arrive on the wire
+    in `read_state_prefix`. The list always ends with an implicit
+    `SharedVariant` - a byte string carrying `encodeDataType +
+    serializeBinary` blobs - which catches values whose type matches none of
+    the declared variants.
+    """
+
+    py_types: tuple[type, ...] | None
+
+    def __init__(self, column_by_spec_getter, shared_value_decoder=None, **kwargs): ...
+    async def read_state_prefix(self): ...
+    async def read_items(self, n_items): ...
+
+class SharedValueDecoder:
+    """Decoder for the `encodeDataType + serializeBinary` payloads carried by
+    a SharedVariant's underlying byte-string column.
+
+    One instance per block: the column cache then accumulates across every
+    overflow value, both the JSON column's own shared paths and those of
+    every nested Dynamic column.
+    """
+    def __init__(self, column_by_spec_getter): ...
+    async def decode(self, blob):
+        """Decode one shared value; returns the Python object it encodes.
+
+        Async because the column readers are: nothing here touches the
+        socket, the reader is the in-memory blob.
+        """
+
+def create_dynamic_column(spec, column_by_spec_getter, column_options): ...