TCP J4: Add Dynamic support

ClickHouse.Driver.Tcp.Tests/Integration/ClickHouseTcpConnectionInsertIntegrationTests.cs

@@ -523,6 +523,66 @@ public async Task InsertAsync_DenseVariantColumnSplitAcrossBlocks_RoundTripsEver
         }
     }
 
+    [Test]
+    public async Task InsertAsync_DenseDynamicColumnSplitAcrossBlocks_RoundTripsEveryRow()
+    {
+        await using var connection = await TcpServerFixture.ConnectAsync(None);
+        string table = UniqueTableName();
+        try
+        {
+            await ExecuteAsync(connection, $"CREATE TABLE {table} (id UInt32, value Dynamic) ENGINE = Memory", DynamicSplitSettings);
+
+            // The Dynamic counterpart of the Variant case above, and the one that matters most: a Dynamic slice
+            // plan derives each type's child-column run start from the local index of that type's first in-slice
+            // row, and at start 0 every one of those is 0. Interleaving the two runtime types (and a NULL) so the
+            // maxRowsPerBlock: 2 split lands mid-run for both means every block after the first starts at a
+            // non-zero per-type offset. Discriminator 0 = String, 1 = UInt64, 2 (the type count) = NULL — unlike
+            // Variant, whose NULL is the fixed 255. String rows 1,4,6; UInt64 rows 0,3,5.
+            object[] expected = { 100UL, "a", null, 200UL, "b", 300UL, "c" };
+            var discriminators = new[] { 1, 0, 2, 1, 0, 1, 0 };
+            IColumn[] typeColumns =
+            {
+                new ArrayColumn<string>("value", "String", new[] { "a", "b", "c" }),
+                PrimitiveColumn<ulong>.FromValues("value", "UInt64", new ulong[] { 100, 200, 300 }),
+            };
+            var dense = new DynamicColumn(
+                "value",
+                "Dynamic",
+                new[] { "String", "UInt64" },
+                discriminators,
+                typeColumns,
+                expected.Length,
+                pooledDiscriminators: false,
+                ownsColumns: false);
+
+            IColumn[] columns = { PrimitiveColumn<uint>.FromValues("id", "UInt32", RowIds(expected.Length)), dense };
+            await connection.InsertAsync(
+                $"INSERT INTO {table} (id, value) VALUES", columns, maxRowsPerBlock: 2, settings: DynamicSplitSettings, cancellationToken: None);
+
+            var readBack = new List<object>(expected.Length);
+            await foreach (Block block in connection.QueryAsync($"SELECT value FROM {table} ORDER BY id", settings: DynamicSplitSettings, cancellationToken: None))
+            {
+                for (int row = 0; row < block[0].RowCount; row++)
+                {
+                    readBack.Add(block[0].GetValue(row));
+                }
+            }
+
+            Assert.That(readBack, Is.EqualTo(expected));
+            Assert.That(connection.State, Is.EqualTo(TcpConnectionState.Ready));
+        }
+        finally
+        {
+            await ExecuteAsync(connection, $"DROP TABLE IF EXISTS {table}");
+        }
+    }
+
+    private static readonly Dictionary<string, string> DynamicSplitSettings = new(StringComparer.Ordinal)
+    {
+        ["allow_experimental_dynamic_type"] = "1",
+        ["output_format_native_use_flattened_dynamic_and_json_serialization"] = "1",
+    };
+
     private static uint[] RowIds(int count)
     {
         var ids = new uint[count];

ClickHouse.Driver.Tcp.Tests/Integration/ColumnarReadSurfaceIntegrationTests.cs

@@ -712,6 +712,80 @@ FROM system.numbers LIMIT 5
         });
     }
 
+    [Test]
+    public async Task QueryAsync_DynamicColumn_ExposesRuntimeTypeNamesAndPerTypeChildColumnsThroughIDynamicColumn()
+    {
+        // Dynamic differs from Variant in two ways the columnar view has to expose. Its type list is discovered per
+        // block rather than declared, so TypeNames carries the wire's own spelling of each runtime type — that is how
+        // a caller knows which typed column to cast a child to. And because the list is discovered, NULL cannot use a
+        // fixed sentinel: it is encoded as TypeCount, one past the last type, rather than Variant's 255.
+        var settings = new Dictionary<string, string>(StringComparer.Ordinal)
+        {
+            ["allow_experimental_dynamic_type"] = "1",
+
+            // The client reads only the flattened serialization (version 3); without this the server sends version 1
+            // and the codec refuses the block rather than guessing at the layout.
+            ["output_format_native_use_flattened_dynamic_and_json_serialization"] = "1",
+        };
+        await using var connection = await TcpServerFixture.ConnectAsync(None);
+
+        bool matched = false;
+        int typeCount = 0;
+        int rowCount = 0;
+        string[] typeNames = null;
+        int[] discriminators = null;
+        int discriminatorLength = 0;
+        int[] localIndices = null;
+        var materialized = new List<object>();
+        var childRowCounts = new List<int>();
+
+        // Rows: 'a', NULL, 100, 'b' — two runtime types plus a NULL.
+        await foreach (Block block in connection.QueryAsync(
+            """
+            SELECT CAST(multiIf(number = 1, CAST(NULL, 'Dynamic'),
+                                number = 2, CAST(toInt64(100), 'Dynamic'),
+                                CAST(concat('s', toString(number)), 'Dynamic')), 'Dynamic')
+            FROM system.numbers LIMIT 4
+            """,
+            settings: settings,
+            cancellationToken: None))
+        {
+            IColumn column = block[0];
+            matched = column is IDynamicColumn;
+
+            var dynamicColumn = (IDynamicColumn)column;
+            typeCount = dynamicColumn.TypeCount;
+            rowCount = dynamicColumn.RowCount;
+            typeNames = dynamicColumn.TypeNames.ToArray();
+            discriminators = dynamicColumn.Discriminators.ToArray();
+            discriminatorLength = dynamicColumn.Discriminators.Length;
+            localIndices = dynamicColumn.LocalIndices.ToArray();
+
+            for (int i = 0; i < dynamicColumn.TypeCount; i++)
+            {
+                childRowCounts.Add(dynamicColumn.GetTypeColumn(i).RowCount);
+            }
+
+            for (int row = 0; row < dynamicColumn.RowCount; row++)
+            {
+                materialized.Add(column.GetValue(row));
+            }
+        }
+
+        Assert.Multiple(() =>
+        {
+            Assert.That(matched, Is.True);
+            Assert.That(rowCount, Is.EqualTo(4));
+            Assert.That(typeCount, Is.EqualTo(2), "two runtime types appeared in this block");
+            Assert.That(typeNames, Is.EquivalentTo(new[] { "Int64", "String" }), "the wire's own spelling, so a caller knows how to read each child");
+            Assert.That(discriminatorLength, Is.EqualTo(rowCount), "sliced to the row count, not the pooled buffer length");
+            Assert.That(discriminators[1], Is.EqualTo(typeCount), "NULL is TypeCount — one past the last type, not a fixed sentinel");
+            Assert.That(localIndices[1], Is.EqualTo(-1), "a NULL row addresses no child");
+            Assert.That(childRowCounts.Sum(), Is.EqualTo(rowCount - 1), "the children together hold every non-NULL row exactly once");
+            Assert.That(materialized, Is.EqualTo(new object[] { "s0", null, 100L, "s3" }));
+        });
+    }
+
     private static async Task ExecuteAsync(ClickHouseTcpConnection connection, string sql)
     {
         await foreach (Block block in connection.QueryAsync(sql, cancellationToken: None))

ClickHouse.Driver.Tcp.Tests/Types/DynamicColumnCodecTests.cs

@@ -0,0 +1,234 @@
+using System;
+using System.Threading.Tasks;
+using ClickHouse.Driver.Tcp.Protocol;
+using ClickHouse.Driver.Tcp.Tests.Utilities;
+using ClickHouse.Driver.Tcp.Types;
+using ClickHouse.Driver.Tcp.Types.Codecs;
+
+namespace ClickHouse.Driver.Tcp.Tests.Types;
+
+[TestFixture]
+public class DynamicColumnCodecTests
+{
+    private static IColumnCodec Resolve(string type) => ColumnCodecRegistry.Default.Resolve(type, default);
+
+    // Captured verbatim from a ClickHouse server (FORMAT Native, flattened serialization) for a Dynamic column
+    // holding [42::UInt64, 'hi'::String, NULL]. String sorts before UInt64, so discriminator 0 = String, 1 =
+    // UInt64, and NULL is the discriminator equal to the type count (2). The version (3) and type list are the
+    // state prefix; the discriminators and per-type runs are the body.
+    private static readonly byte[] DocumentedBytes =
+    {
+        0x03, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, // state prefix: serialization version = 3 (flattened)
+        0x02,                                           // num_types = 2
+        0x06, 0x53, 0x74, 0x72, 0x69, 0x6E, 0x67,       // type[0] = "String"
+        0x06, 0x55, 0x49, 0x6E, 0x74, 0x36, 0x34,       // type[1] = "UInt64"
+        0x01, 0x00, 0x02,                               // discriminators: 1 (UInt64), 0 (String), 2 (NULL)
+        0x02, 0x68, 0x69,                               // String run (1 value): len = 2, "hi"
+        0x2A, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, // UInt64 run (1 value): 42
+    };
+
+    [Test]
+    public async Task WriteFull_ErgonomicColumn_ProducesTheDocumentedBytes()
+    {
+        IColumnCodec codec = Resolve("Dynamic");
+        var column = new ArrayColumn<object>("d", "Dynamic", new object[] { 42UL, "hi", null });
+
+        // The inferred type list is name-sorted (String before UInt64), matching the server's canonicalization.
+        byte[] bytes = await CodecTestHarness.WriteAsync(w => codec.WriteFull(w, column));
+
+        CollectionAssert.AreEqual(DocumentedBytes, bytes);
+    }
+
+    [Test]
+    public async Task WriteFull_DenseColumnReadBack_RoundTripsToIdenticalBytes()
+    {
+        IColumnCodec codec = Resolve("Dynamic");
+
+        using ClickHouseBinaryReader reader = CodecTestHarness.ReaderOver(DocumentedBytes);
+        await codec.ReadStatePrefixAsync(reader, CodecTestHarness.None);
+        using IColumn dense = await codec.ReadColumnAsync(reader, "d", "Dynamic", 3, CodecTestHarness.None);
+
+        // The read-back DynamicColumn is the zero-copy write source: writing it reproduces the exact bytes.
+        byte[] bytes = await CodecTestHarness.WriteAsync(w => codec.WriteFull(w, dense));
+
+        CollectionAssert.AreEqual(DocumentedBytes, bytes);
+    }
+
+    [Test]
+    public async Task WriteStatePrefixThenColumn_SeparateStateFreeCalls_ProducesTheDocumentedBytes()
+    {
+        IColumnCodec codec = Resolve("Dynamic");
+        var column = new ArrayColumn<object>("d", "Dynamic", new object[] { 42UL, "hi", null });
+
+        // The state-free prefix and body calls each recompute the (deterministic) type list independently; the
+        // combined output must still match the shared-state path.
+        byte[] bytes = await CodecTestHarness.WriteAsync(w =>
+        {
+            codec.WriteStatePrefix(w, column);
+            codec.WriteColumn(w, column);
+        });
+
+        CollectionAssert.AreEqual(DocumentedBytes, bytes);
+    }
+
+    // The same layout as DocumentedBytes but with two values per type, so a slice can start a run part-way through
+    // it: [42::UInt64, 'hi'::String, NULL, 7::UInt64, 'yo'::String].
+    private static readonly byte[] DocumentedBytesTwoPerType =
+    {
+        0x03, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, // state prefix: serialization version = 3 (flattened)
+        0x02,                                           // num_types = 2
+        0x06, 0x53, 0x74, 0x72, 0x69, 0x6E, 0x67,       // type[0] = "String"
+        0x06, 0x55, 0x49, 0x6E, 0x74, 0x36, 0x34,       // type[1] = "UInt64"
+        0x01, 0x00, 0x02, 0x01, 0x00,                   // discriminators: UInt64, String, NULL, UInt64, String
+        0x02, 0x68, 0x69,                               // String run[0] = "hi"
+        0x02, 0x79, 0x6F,                               // String run[1] = "yo"
+        0x2A, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, // UInt64 run[0] = 42
+        0x07, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, // UInt64 run[1] = 7
+    };
+
+    // The dense planner derives each type's child-column run start from the local index of that type's first
+    // in-slice row. At start 0 every one of those is 0, so a slice beginning at row 0 cannot tell a correct planner
+    // from one that ignores the offset entirely — this is the only test that can.
+    [Test]
+    public async Task WriteColumn_DenseColumnSliceAfterEarlierValues_StartsEachRunAtItsSliceOffset()
+    {
+        IColumnCodec codec = Resolve("Dynamic");
+
+        using ClickHouseBinaryReader reader = CodecTestHarness.ReaderOver(DocumentedBytesTwoPerType);
+        await codec.ReadStatePrefixAsync(reader, CodecTestHarness.None);
+        using IColumn dense = await codec.ReadColumnAsync(reader, "d", "Dynamic", 5, CodecTestHarness.None);
+
+        // Slice rows [3, 5): 7 (UInt64) and "yo" (String). Each is the *second* value of its run, so each run must
+        // be written from offset 1.
+        byte[] bytes = await CodecTestHarness.WriteAsync(w => codec.WriteColumn(w, dense, 3, 2));
+
+        byte[] expected =
+        {
+            0x01, 0x00,                                     // discriminators: UInt64, String
+            0x02, 0x79, 0x6F,                               // String run from offset 1: "yo"
+            0x07, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, // UInt64 run from offset 1: 7
+        };
+        CollectionAssert.AreEqual(expected, bytes);
+    }
+
+    [Test]
+    public async Task ReadColumn_DocumentedBytes_ReconstructsValuesAndSurfacesTheRuntimeTypeList()
+    {
+        // Decoding the golden vector: the type list and discriminators are the dynamic-structure surface no
+        // integration test can see (it only calls GetValue), and keeping the value assertions alongside them
+        // guards the decoder against drifting even if a future server emits a different type ordering. The
+        // values on their own are covered by the "Dynamic [scalars + null]" case in InsertRoundTripCase.
+        IColumnCodec codec = Resolve("Dynamic");
+
+        using ClickHouseBinaryReader reader = CodecTestHarness.ReaderOver(DocumentedBytes);
+        await codec.ReadStatePrefixAsync(reader, CodecTestHarness.None);
+        using IColumn column = await codec.ReadColumnAsync(reader, "d", "Dynamic", 3, CodecTestHarness.None);
+
+        var dynamic = (IDynamicColumn)column;
+        Assert.Multiple(() =>
+        {
+            Assert.That(dynamic.TypeCount, Is.EqualTo(2));
+            Assert.That(dynamic.TypeNames, Is.EqualTo(new[] { "String", "UInt64" }));
+            Assert.That(dynamic.Discriminators.ToArray(), Is.EqualTo(new[] { 1, 0, 2 }));
+            Assert.That(column.RowCount, Is.EqualTo(3));
+            Assert.That(column.GetValue(0), Is.EqualTo(42UL));
+            Assert.That(column.GetValue(1), Is.EqualTo("hi"));
+            Assert.That(column.GetValue(2), Is.Null);
+        });
+    }
+
+    [Test]
+    public void ReadStatePrefix_VersionNotFlattened_Throws()
+    {
+        IColumnCodec codec = Resolve("Dynamic");
+
+        // Version 2 is the non-flat native default, which this client does not decode.
+        byte[] bytes = { 0x02, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00 };
+        using ClickHouseBinaryReader reader = CodecTestHarness.ReaderOver(bytes);
+
+        Assert.ThrowsAsync<ClickHouseProtocolException>(async () => await codec.ReadStatePrefixAsync(reader, CodecTestHarness.None));
+    }
+
+    [Test]
+    public async Task ReadColumn_DiscriminatorPastTypeCount_Throws()
+    {
+        IColumnCodec codec = Resolve("Dynamic");
+
+        // Prefix declares two types; a discriminator of 5 selects neither a type (0, 1) nor NULL (2).
+        byte[] prefix =
+        {
+            0x03, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+            0x02,
+            0x06, 0x53, 0x74, 0x72, 0x69, 0x6E, 0x67,
+            0x06, 0x55, 0x49, 0x6E, 0x74, 0x36, 0x34,
+            0x05, // discriminator out of range
+        };
+        using ClickHouseBinaryReader reader = CodecTestHarness.ReaderOver(prefix);
+        await codec.ReadStatePrefixAsync(reader, CodecTestHarness.None);
+
+        Assert.ThrowsAsync<FormatException>(async () => await codec.ReadColumnAsync(reader, "d", "Dynamic", 1, CodecTestHarness.None));
+    }
+
+    [Test]
+    public async Task ReadColumn_ZeroRows_ReturnsEmptyColumn()
+    {
+        IColumnCodec codec = Resolve("Dynamic");
+
+        // A zero-row block carries no prefix and no body, so read straight from an empty buffer.
+        using ClickHouseBinaryReader reader = CodecTestHarness.ReaderOver(Array.Empty<byte>());
+        using IColumn column = await codec.ReadColumnAsync(reader, "d", "Dynamic", 0, CodecTestHarness.None);
+
+        Assert.That(column.RowCount, Is.Zero);
+    }
+
+    [Test]
+    public void Create_MaxTypesArgument_IsAccepted()
+    {
+        IColumnCodec codec = Resolve("Dynamic(max_types=5)");
+        Assert.That(codec.TypeName, Is.EqualTo("Dynamic(max_types=5)"));
+    }
+
+    [Test]
+    public void Indexer_RowPastRowCount_ThrowsRatherThanReadingStaleDiscriminators()
+    {
+        // The discriminator buffer is normally a pooled array longer than the column, so a row past RowCount read a
+        // stale value from its tail — and the outcome depended on the leftover: a stale NULL discriminator (which is
+        // just typeColumns.Length, here 1) reported the row as an existing NULL, while any other value fell through to
+        // the exactly-sized local-index array and threw. Both spellings must be a bounds failure.
+        using var alternative = new ArrayColumn<long>("d", "Int64", new[] { 7L });
+        var discriminators = new[] { 0, 1, 0 };
+        using var column = new DynamicColumn(
+            "d", "Dynamic", new[] { "Int64" }, discriminators, new IColumn[] { alternative }, rowCount: 1, pooledDiscriminators: false, ownsColumns: false);
+
+        Assert.Multiple(() =>
+        {
+            Assert.That(column[0], Is.EqualTo(7L));
+            Assert.That(() => column[1], Throws.InstanceOf<IndexOutOfRangeException>(), "a stale NULL discriminator must not read as an existing NULL row");
+            Assert.That(() => column[2], Throws.InstanceOf<IndexOutOfRangeException>());
+        });
+    }
+
+    [Test]
+    public void Constructor_DiscriminatorsShorterThanRowCount_Throws()
+    {
+        // rowCount is load-bearing here — each child holds only the rows that selected it, and a NULL row takes a slot
+        // in none of them — so it is validated rather than derived.
+        using var alternative = new ArrayColumn<long>("d", "Int64", new[] { 7L });
+
+        Assert.That(
+            () => new DynamicColumn("d", "Dynamic", new[] { "Int64" }, new[] { 0 }, new IColumn[] { alternative }, rowCount: 2, pooledDiscriminators: false, ownsColumns: false),
+            Throws.ArgumentException.With.Message.Contains("fewer than"));
+    }
+
+    [Test]
+    public void Create_UnknownArgument_Throws()
+        => Assert.Throws<FormatException>(() => Resolve("Dynamic(max_sizes=5)"));
+
+    [TestCase(1, 1)]
+    [TestCase(255, 1)]
+    [TestCase(256, 2)]
+    [TestCase(65535, 2)]
+    [TestCase(65536, 4)]
+    public void DiscriminatorWidth_GrowsWithTypeCount(int typeCount, int expectedWidth)
+        => Assert.That(DynamicColumnCodec.DiscriminatorWidth(typeCount), Is.EqualTo(expectedWidth));
+}