diff --git a/ClickHouse.Driver.Tcp.Tests/Integration/QBitIntegrationTests.cs b/ClickHouse.Driver.Tcp.Tests/Integration/QBitIntegrationTests.cs new file mode 100644 index 000000000..e99741480 --- /dev/null +++ b/ClickHouse.Driver.Tcp.Tests/Integration/QBitIntegrationTests.cs @@ -0,0 +1,189 @@ +using System; +using System.Globalization; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using ClickHouse.Driver.Tcp.Format; +using ClickHouse.Driver.Tcp.Tests.Utilities; +using ClickHouse.Driver.Tcp.Types; + +namespace ClickHouse.Driver.Tcp.Tests.Integration; + +[TestFixture] +[Category("Integration")] +[RequiresServerFeature(TcpFeature.QBit)] +public class QBitIntegrationTests +{ + private static readonly CancellationToken None = CancellationToken.None; + + // Spans two complete bitmap bytes and a partial third. + private const int Dimension = 17; + private const string Float32Type = "QBit(Float32, 17)"; + private const string Int8Type = "QBit(Int8, 17)"; + + private static float[] Vector() => Enumerable.Range(0, Dimension).Select(i => (i * 3f) - 20f).ToArray(); + + private static sbyte[] Int8Vector() + => Enumerable.Range(0, Dimension).Select(i => i == 16 ? sbyte.MinValue : (sbyte)((i * 7) - 60)).ToArray(); + + [Test] + public async Task InsertAsync_QBitWrittenByTheClient_IsReadBackByTheServerAsTheSameVector() + { + await using var client = TcpServerFixture.CreateClient(); + string table = UniqueTableName(); + + await Drain(client, $"CREATE TABLE {table} (v {Float32Type}) ENGINE = Memory"); + try + { + float[] vector = Vector(); + using var column = new ArrayColumn("v", Float32Type, new[] { vector }); + await client.InsertAsync($"INSERT INTO {table} (v) VALUES", new IColumn[] { column }, cancellationToken: None); + + string rendered = await ScalarStringAsync(client, $"SELECT toString(v) FROM {table}"); + + Assert.That(rendered, Is.EqualTo(Expected(vector))); + } + finally + { + await Drain(client, $"DROP TABLE IF EXISTS {table}"); + } + } + + [Test] + public async Task StreamAsync_QBitWrittenByTheServer_DecodesToTheSameVector() + { + await using var client = TcpServerFixture.CreateClient(); + string table = UniqueTableName(); + + await Drain(client, $"CREATE TABLE {table} (v {Float32Type}) ENGINE = Memory"); + try + { + float[] vector = Vector(); + string literal = string.Join(",", vector.Select(v => v.ToString("R", CultureInfo.InvariantCulture))); + await Drain(client, $"INSERT INTO {table} VALUES ([{literal}])"); + + float[] decoded = null; + await foreach (Block block in client.StreamAsync($"SELECT v FROM {table}", cancellationToken: None)) + { + decoded = (float[])block[0].GetValue(0); + } + + CollectionAssert.AreEqual(vector, decoded); + } + finally + { + await Drain(client, $"DROP TABLE IF EXISTS {table}"); + } + } + + [Test] + public async Task GetPlane_QBitWrittenByTheServer_AgreesWithTheServersOwnBitExtraction() + { + await using var client = TcpServerFixture.CreateClient(); + string table = UniqueTableName(); + + await Drain(client, $"CREATE TABLE {table} (v {Float32Type}) ENGINE = Memory"); + try + { + float[] vector = Vector(); + string literal = string.Join(",", vector.Select(v => v.ToString("R", CultureInfo.InvariantCulture))); + await Drain(client, $"INSERT INTO {table} VALUES ([{literal}])"); + + byte[] signPlane = null; + await foreach (Block block in client.StreamAsync($"SELECT v FROM {table}", cancellationToken: None)) + { + signPlane = ((IQBitColumn)block[0]).GetPlane(31).ToArray(); + } + + for (int i = 0; i < Dimension; i++) + { + int slot = signPlane.Length - 1 - (i / 8); + bool negativeInPlane = (signPlane[slot] & (1 << (i % 8))) != 0; + Assert.That(negativeInPlane, Is.EqualTo(vector[i] < 0 || float.IsNegative(vector[i])), $"element {i}"); + } + } + finally + { + await Drain(client, $"DROP TABLE IF EXISTS {table}"); + } + } + + [Test] + [RequiresServerFeature(TcpFeature.QBitInt8)] + public async Task InsertAsync_Int8QBitWrittenByTheClient_IsReadBackByTheServerAsTheSameVector() + { + await using var client = TcpServerFixture.CreateClient(); + string table = UniqueTableName(); + + await Drain(client, $"CREATE TABLE {table} (v {Int8Type}) ENGINE = Memory"); + try + { + sbyte[] vector = Int8Vector(); + using var column = new ArrayColumn("v", Int8Type, new[] { vector }); + await client.InsertAsync($"INSERT INTO {table} (v) VALUES", new IColumn[] { column }, cancellationToken: None); + + string rendered = await ScalarStringAsync(client, $"SELECT toString(v) FROM {table}"); + + Assert.That(rendered, Is.EqualTo("[" + string.Join(",", vector.Select(v => v.ToString(CultureInfo.InvariantCulture))) + "]")); + } + finally + { + await Drain(client, $"DROP TABLE IF EXISTS {table}"); + } + } + + [Test] + [RequiresServerFeature(TcpFeature.QBitInt8)] + public async Task StreamAsync_Int8QBitWrittenByTheServer_DecodesToTheSameVector() + { + await using var client = TcpServerFixture.CreateClient(); + string table = UniqueTableName(); + + await Drain(client, $"CREATE TABLE {table} (v {Int8Type}) ENGINE = Memory"); + try + { + sbyte[] vector = Int8Vector(); + string literal = string.Join(",", vector.Select(v => v.ToString(CultureInfo.InvariantCulture))); + await Drain(client, $"INSERT INTO {table} VALUES ([{literal}])"); + + sbyte[] decoded = null; + await foreach (Block block in client.StreamAsync($"SELECT v FROM {table}", cancellationToken: None)) + { + decoded = (sbyte[])block[0].GetValue(0); + } + + CollectionAssert.AreEqual(vector, decoded); + } + finally + { + await Drain(client, $"DROP TABLE IF EXISTS {table}"); + } + } + + private static string Expected(float[] vector) + => "[" + string.Join(",", vector.Select(v => v.ToString("R", CultureInfo.InvariantCulture))) + "]"; + + private static string UniqueTableName() => $"tcp_qbit_test_{Guid.NewGuid():N}"; + + private static async Task ScalarStringAsync(ClickHouseTcpClient client, string sql) + { + string value = null; + await foreach (Block block in client.StreamAsync(sql, cancellationToken: None)) + { + if (block.RowCount > 0) + { + value = (string)block[0].GetValue(0); + } + } + + return value; + } + + private static async Task Drain(ClickHouseTcpClient client, string sql) + { + await foreach (Block block in client.StreamAsync(sql, cancellationToken: None)) + { + _ = block; + } + } +} diff --git a/ClickHouse.Driver.Tcp.Tests/Types/QBitColumnCodecTests.cs b/ClickHouse.Driver.Tcp.Tests/Types/QBitColumnCodecTests.cs new file mode 100644 index 000000000..7e12d675c --- /dev/null +++ b/ClickHouse.Driver.Tcp.Tests/Types/QBitColumnCodecTests.cs @@ -0,0 +1,416 @@ +using System; +using System.IO; +using System.Threading.Tasks; +using ClickHouse.Driver.Tcp.Protocol; +using ClickHouse.Driver.Tcp.Tests.Utilities; +using ClickHouse.Driver.Tcp.Types; + +namespace ClickHouse.Driver.Tcp.Tests.Types; + +[TestFixture] +public class QBitColumnCodecTests +{ + private const string Float32X4 = "QBit(Float32, 4)"; + + // Server-produced Native body for one QBit(Float32, 4) row containing [1, 2, 3, 4]. Planes are + // ordered most-significant first. + private static readonly byte[] DocumentedBytes = + { + 0x00, // Bit 31: no elements set. + 0x0E, // Bit 30: elements 1, 2, and 3. + 0x01, // Bits 29 through 24: element 0. + 0x01, + 0x01, + 0x01, + 0x01, + 0x01, + 0x09, // Bit 23: elements 0 and 3. + 0x04, // Bit 22: element 2. + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + }; + + // Server-produced Native body for one QBit(Float32, 16) row containing [1, -2, ..., 15, -16]. Each plane + // contains a two-byte row bitmap. + private static readonly byte[] DocumentedBytes16 = + { + 0xAA, 0xAA, 0xFF, 0xFE, 0x00, 0x01, 0x00, 0x01, 0x00, 0x01, 0x00, 0x01, + 0x00, 0x01, 0xFF, 0x81, 0x80, 0x79, 0x78, 0x64, 0x66, 0x50, 0x55, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, + }; + + // Server-produced Native body for one QBit(Int8, 16) row containing [1, -2, ..., 15, -16]. The eight planes + // encode two's-complement bytes, most-significant plane first. + private static readonly byte[] DocumentedInt8Bytes16 = + { + 0xAA, 0xAA, 0xAA, 0xAA, 0xAA, 0xAA, 0xAA, 0xAA, + 0x55, 0xAA, 0x5A, 0x5A, 0x66, 0x66, 0x55, 0x55, + }; + + private static readonly sbyte[] DocumentedInt8Vector = + { + 1, -2, 3, -4, 5, -6, 7, -8, 9, -10, 11, -12, 13, -14, 15, -16, + }; + + private static IColumnCodec Codec(string type) => ColumnCodecRegistry.Default.Resolve(type, ResolveContext.ForWrite); + + [Test] + public async Task WriteColumn_DocumentedExample_ProducesTheServersOwnBytes() + { + IColumnCodec codec = Codec(Float32X4); + using var column = new ArrayColumn("v", Float32X4, new[] { new[] { 1f, 2f, 3f, 4f } }); + + byte[] bytes = await CodecTestHarness.WriteAsync(w => codec.WriteColumn(w, column)); + + CollectionAssert.AreEqual(DocumentedBytes, bytes); + } + + [Test] + public async Task ReadColumnAsync_TheServersOwnBytes_DecodesTheDocumentedVector() + { + IColumnCodec codec = Codec(Float32X4); + using ClickHouseBinaryReader reader = CodecTestHarness.ReaderOver(DocumentedBytes); + + using IColumn read = await codec.ReadColumnAsync(reader, "v", Float32X4, 1, CodecTestHarness.None); + + CollectionAssert.AreEqual(new[] { 1f, 2f, 3f, 4f }, (float[])read.GetValue(0)); + } + + [Test] + public async Task WriteColumn_DenseRowSlice_ProducesTheServersOwnBytes() + { + const string Type = "QBit(Float32, 16)"; + IColumnCodec codec = Codec(Type); + using var source = new ArrayColumn("v", Type, new[] + { + new float[16], + new[] { 1f, -2f, 3f, -4f, 5f, -6f, 7f, -8f, 9f, -10f, 11f, -12f, 13f, -14f, 15f, -16f }, + new float[16], + }); + + byte[] dense = await CodecTestHarness.WriteAsync(w => codec.WriteColumn(w, source)); + using ClickHouseBinaryReader reader = CodecTestHarness.ReaderOver(dense); + using IColumn read = await codec.ReadColumnAsync(reader, "v", Type, 3, CodecTestHarness.None); + + byte[] sliced = await CodecTestHarness.WriteSliceAsync(codec, read, start: 1, length: 1); + + CollectionAssert.AreEqual(DocumentedBytes16, sliced); + } + + [Test] + public async Task GetPlane_ReadColumn_IndexesPlanesBySignificanceNotWireOrder() + { + IColumnCodec codec = Codec(Float32X4); + using ClickHouseBinaryReader reader = CodecTestHarness.ReaderOver(DocumentedBytes); + using IColumn read = await codec.ReadColumnAsync(reader, "v", Float32X4, 1, CodecTestHarness.None); + + var qbit = (IQBitColumn)read; + + Assert.Multiple(() => + { + Assert.That(qbit.Dimension, Is.EqualTo(4)); + Assert.That(qbit.BitWidth, Is.EqualTo(32)); + Assert.That(qbit.BytesPerRow, Is.EqualTo(1)); + Assert.That(qbit.GetPlane(31)[0], Is.EqualTo(0x00), "sign plane"); + Assert.That(qbit.GetPlane(30)[0], Is.EqualTo(0x0E), "bit 30: 2.0, 3.0, 4.0"); + Assert.That(qbit.GetPlane(23)[0], Is.EqualTo(0x09), "bit 23: 1.0 and 4.0"); + Assert.That(qbit.GetPlane(22)[0], Is.EqualTo(0x04), "bit 22: 3.0"); + Assert.That(qbit.GetPlane(0)[0], Is.EqualTo(0x00), "no value has a bit that low set"); + }); + } + + [Test] + public async Task GetPlane_MultipleRows_ReturnsEveryRowsBitmapForThatPlane() + { + IColumnCodec codec = Codec("QBit(Float32, 8)"); + using var column = new ArrayColumn("v", "QBit(Float32, 8)", new[] + { + new float[8], + new[] { 1f, 1f, 1f, 1f, 1f, 1f, 1f, 1f }, + new[] { -0f, -0f, -0f, -0f, -0f, -0f, -0f, -0f }, + }); + + using IColumn read = await CodecTestHarness.RoundTripAsync(codec, column, "QBit(Float32, 8)", 3); + var qbit = (IQBitColumn)read; + + CollectionAssert.AreEqual(new byte[] { 0x00, 0x00, 0xFF }, qbit.GetPlane(31).ToArray()); + } + + [TestCase(-1)] + [TestCase(32)] + public async Task GetPlane_BitOutsideTheWidth_Throws(int bit) + { + IColumnCodec codec = Codec(Float32X4); + using ClickHouseBinaryReader reader = CodecTestHarness.ReaderOver(DocumentedBytes); + using IColumn read = await codec.ReadColumnAsync(reader, "v", Float32X4, 1, CodecTestHarness.None); + + Assert.That(() => ((IQBitColumn)read).GetPlane(bit), Throws.InstanceOf()); + } + + [Test] + public async Task Values_ReadColumn_MaterializesEveryRowThroughThePooledCache() + { + IColumnCodec codec = Codec(Float32X4); + using var column = new ArrayColumn("v", Float32X4, new[] + { + new[] { 1f, 2f, 3f, 4f }, + new[] { -1f, -2f, -3f, -4f }, + }); + + using IColumn read = await CodecTestHarness.RoundTripAsync(codec, column, Float32X4, 2); + ReadOnlySpan values = ((IColumn)read).Values; + + Assert.That(values.Length, Is.EqualTo(2)); + CollectionAssert.AreEqual(new[] { 1f, 2f, 3f, 4f }, values[0]); + CollectionAssert.AreEqual(new[] { -1f, -2f, -3f, -4f }, values[1]); + } + + [Test] + public async Task Values_ReadTwice_ReturnsTheSameCachedArrays() + { + IColumnCodec codec = Codec(Float32X4); + using var column = new ArrayColumn("v", Float32X4, new[] { new[] { 1f, 2f, 3f, 4f } }); + + using IColumn read = await CodecTestHarness.RoundTripAsync(codec, column, Float32X4, 1); + var typed = (IColumn)read; + float[] first = typed.Values[0]; + float[] second = typed.Values[0]; + + Assert.Multiple(() => + { + Assert.That(second, Is.SameAs(first)); + Assert.That(read.GetValue(0), Is.SameAs(first)); + }); + } + + [Test] + public async Task ReadColumnAsync_ZeroRows_ReadsNoBytesAndDecodesAnEmptyColumn() + { + IColumnCodec codec = Codec(Float32X4); + using ClickHouseBinaryReader reader = CodecTestHarness.ReaderOver(Array.Empty()); + + using IColumn read = await codec.ReadColumnAsync(reader, "v", Float32X4, 0, CodecTestHarness.None); + + Assert.That(read.RowCount, Is.Zero); + Assert.That(((IColumn)read).Values.Length, Is.Zero); + } + + [Test] + public async Task GetValue_RowPastTheRowCount_Throws() + { + IColumnCodec codec = Codec(Float32X4); + using ClickHouseBinaryReader reader = CodecTestHarness.ReaderOver(DocumentedBytes); + using IColumn read = await codec.ReadColumnAsync(reader, "v", Float32X4, 1, CodecTestHarness.None); + + Assert.That(() => read.GetValue(1), Throws.InstanceOf()); + } + + [Test] + public void Resolve_Float64_SurfacesDoubleVectorsAndSixtyFourPlanes() + { + IColumnCodec codec = Codec("QBit(Float64, 3)"); + + Assert.Multiple(() => + { + Assert.That(codec.ElementType, Is.EqualTo(typeof(double[]))); + Assert.That(codec.TypeName, Is.EqualTo("QBit(Float64, 3)")); + Assert.That(codec.NullPlaceholder, Is.EqualTo(new double[3])); + }); + } + + [Test] + public void Resolve_BFloat16_SurfacesWidenedFloatVectors() + { + IColumnCodec codec = Codec("QBit(BFloat16, 4)"); + + Assert.Multiple(() => + { + Assert.That(codec.ElementType, Is.EqualTo(typeof(float[]))); + Assert.That(codec.NullPlaceholder, Is.EqualTo(new float[4])); + }); + } + + [Test] + public void CanWrite_ColumnOfAnotherElementType_IsRefused() + { + IColumnCodec codec = Codec(Float32X4); + + Assert.Multiple(() => + { + Assert.That(codec.CanWrite(new ArrayColumn("v", Float32X4, new[] { new[] { 1f, 2f, 3f, 4f } })), Is.True); + Assert.That(codec.CanWrite(new ArrayColumn("v", Float32X4, new[] { new[] { 1d } })), Is.False); + Assert.That(codec.CanWrite(PrimitiveColumn.FromValues("v", "Float32", new[] { 1f })), Is.False); + Assert.That(codec.CanWriteElementType(typeof(float[])), Is.True); + Assert.That(codec.CanWriteElementType(typeof(double[])), Is.False); + }); + } + + [Test] + public void WriteColumn_VectorOfTheWrongLength_ThrowsNamingTheRow() + { + IColumnCodec codec = Codec(Float32X4); + using var column = new ArrayColumn("v", Float32X4, new[] + { + new[] { 1f, 2f, 3f, 4f }, + new[] { 1f, 2f }, + }); + + Assert.That( + async () => await CodecTestHarness.WriteAsync(w => codec.WriteColumn(w, column)), + Throws.ArgumentException.With.Message.Contains("row 1").And.Message.Contains("exactly 4")); + } + + [Test] + public void WriteColumn_NullVector_ThrowsPointingAtNullable() + { + IColumnCodec codec = Codec(Float32X4); + using var column = new ArrayColumn("v", Float32X4, new float[][] { null }); + + Assert.That( + async () => await CodecTestHarness.WriteAsync(w => codec.WriteColumn(w, column)), + Throws.ArgumentException.With.Message.Contains("Nullable")); + } + + [Test] + public async Task WriteColumn_Int8Vector_ProducesTheServersOwnBytes() + { + const string Type = "QBit(Int8, 16)"; + IColumnCodec codec = Codec(Type); + using var column = new ArrayColumn("v", Type, new[] { DocumentedInt8Vector }); + + byte[] bytes = await CodecTestHarness.WriteAsync(w => codec.WriteColumn(w, column)); + + CollectionAssert.AreEqual(DocumentedInt8Bytes16, bytes); + } + + [Test] + public async Task ReadColumnAsync_Int8ServerBytes_DecodesTheTwosComplementVector() + { + const string Type = "QBit(Int8, 16)"; + IColumnCodec codec = Codec(Type); + using ClickHouseBinaryReader reader = CodecTestHarness.ReaderOver(DocumentedInt8Bytes16); + + using IColumn read = await codec.ReadColumnAsync(reader, "v", Type, 1, CodecTestHarness.None); + + CollectionAssert.AreEqual(DocumentedInt8Vector, (sbyte[])read.GetValue(0)); + Assert.That(((IQBitColumn)read).BitWidth, Is.EqualTo(8)); + } + + [Test] + public void CanWrite_Int8Codec_AcceptsOnlySByteVectors() + { + IColumnCodec codec = Codec("QBit(Int8, 4)"); + + Assert.Multiple(() => + { + Assert.That(codec.ElementType, Is.EqualTo(typeof(sbyte[]))); + Assert.That(codec.CanWrite(new ArrayColumn("v", "QBit(Int8, 4)", new[] { new sbyte[4] })), Is.True); + Assert.That(codec.CanWrite(new ArrayColumn("v", "QBit(Int8, 4)", new[] { new float[4] })), Is.False); + Assert.That(codec.NullPlaceholder, Is.EqualTo(new sbyte[4])); + }); + } + + [Test] + public async Task GetPlane_UnstridedColumn_ReportsOneGroupAndAgreesWithTheGroupOverload() + { + IColumnCodec codec = Codec(Float32X4); + using ClickHouseBinaryReader reader = CodecTestHarness.ReaderOver(DocumentedBytes); + using IColumn read = await codec.ReadColumnAsync(reader, "v", Float32X4, 1, CodecTestHarness.None); + var qbit = (IQBitColumn)read; + + Assert.Multiple(() => + { + Assert.That(qbit.Stride, Is.EqualTo(qbit.Dimension)); + Assert.That(qbit.GroupCount, Is.EqualTo(1)); + Assert.That(qbit.GetPlane(30, 0).ToArray(), Is.EqualTo(qbit.GetPlane(30).ToArray())); + }); + } + + [Test] + public async Task GetPlane_GroupPastTheOnlyGroup_ThrowsArgumentOutOfRange() + { + IColumnCodec codec = Codec(Float32X4); + using ClickHouseBinaryReader reader = CodecTestHarness.ReaderOver(DocumentedBytes); + using IColumn read = await codec.ReadColumnAsync(reader, "v", Float32X4, 1, CodecTestHarness.None); + var qbit = (IQBitColumn)read; + + Assert.Multiple(() => + { + Assert.That(() => qbit.GetPlane(0, 1).ToArray(), Throws.InstanceOf()); + Assert.That(() => qbit.GetPlane(0, -1).ToArray(), Throws.InstanceOf()); + }); + } + + [TestCase(-1, 1, TestName = "negative start")] + [TestCase(0, -1, TestName = "negative length")] + [TestCase(0, 2, TestName = "length runs past the last row")] + public async Task WriteColumn_DenseSliceOutsideTheColumn_ThrowsArgumentOutOfRange(int start, int length) + { + IColumnCodec codec = Codec(Float32X4); + using ClickHouseBinaryReader reader = CodecTestHarness.ReaderOver(DocumentedBytes); + using IColumn dense = await codec.ReadColumnAsync(reader, "v", Float32X4, 1, CodecTestHarness.None); + + Assert.That( + async () => await CodecTestHarness.WriteSliceAsync(codec, dense, start, length), + Throws.InstanceOf()); + } + + [Test] + public async Task WriteColumn_EmptySliceAtTheEnd_WritesNoBytes() + { + IColumnCodec codec = Codec(Float32X4); + using ClickHouseBinaryReader reader = CodecTestHarness.ReaderOver(DocumentedBytes); + using IColumn dense = await codec.ReadColumnAsync(reader, "v", Float32X4, 1, CodecTestHarness.None); + + byte[] bytes = await CodecTestHarness.WriteSliceAsync(codec, dense, start: 1, length: 0); + + Assert.That(bytes, Is.Empty); + } + + [Test] + public void ReadColumnAsync_TruncatedPlaneBody_ThrowsEndOfStream() + { + IColumnCodec codec = Codec(Float32X4); + using ClickHouseBinaryReader reader = CodecTestHarness.ReaderOver(new byte[] { 0x00, 0x0E }); + + Assert.That( + async () => await codec.ReadColumnAsync(reader, "v", Float32X4, 1, CodecTestHarness.None), + Throws.InstanceOf()); + } + + [TestCase("QBit(Float32)", TestName = "one argument")] + [TestCase("QBit(Float32, 4, 2, 1)", TestName = "four arguments")] + public void Resolve_WrongArgumentCount_ThrowsFormatException(string type) + { + Assert.That(() => Codec(type), Throws.InstanceOf().With.Message.Contains("exactly two")); + } + + [Test] + public void Resolve_TheStrideFormAddedIn267_ThrowsNotSupportedException() + { + Assert.That( + () => Codec("QBit(Float32, 16, 8)"), + Throws.InstanceOf().With.Message.Contains("strided")); + } + + [TestCase("QBit(Float32, 0)")] + [TestCase("QBit(Float32, -1)")] + [TestCase("QBit(Float32, x)")] + public void Resolve_InvalidDimension_ThrowsFormatException(string type) + { + Assert.That(() => Codec(type), Throws.InstanceOf().With.Message.Contains("vector length")); + } + + [TestCase("QBit(Int16, 4)")] + [TestCase("QBit(UInt8, 4)")] + [TestCase("QBit(Float16, 4)")] + public void Resolve_ElementTypeTheServerRejects_ThrowsNotSupportedException(string type) + { + Assert.That( + () => Codec(type), + Throws.InstanceOf().With.Message.Contains("Int8, BFloat16, Float32 and Float64")); + } +} diff --git a/ClickHouse.Driver.Tcp.Tests/Utilities/InsertRoundTripCase.cs b/ClickHouse.Driver.Tcp.Tests/Utilities/InsertRoundTripCase.cs index f7ba48e74..1670be2d0 100644 --- a/ClickHouse.Driver.Tcp.Tests/Utilities/InsertRoundTripCase.cs +++ b/ClickHouse.Driver.Tcp.Tests/Utilities/InsertRoundTripCase.cs @@ -27,7 +27,12 @@ public sealed class InsertRoundTripCase private readonly Func buildInsert; private readonly Func buildExpected; - private InsertRoundTripCase(string label, string clickHouseType, Func buildInsert, Func buildExpected, IReadOnlyDictionary settings) + private InsertRoundTripCase( + string label, + string clickHouseType, + Func buildInsert, + Func buildExpected, + IReadOnlyDictionary settings) { Label = label; ClickHouseType = clickHouseType; @@ -1393,6 +1398,115 @@ public static IEnumerable Cases() yield return Same("Geometry", "Geometry", name => BuildGeometryColumn(name)); } + if (TcpServerFeatures.Has(TcpFeature.QBit)) + { + yield return Same( + "QBit(Float32, 4)", + "QBit(Float32, 4)", + name => new ArrayColumn(name, "QBit(Float32, 4)", new[] + { + new[] { 1f, 2f, 3f, 4f }, + new[] { 0f, -0f, float.MaxValue, float.MinValue }, + new[] { float.Epsilon, float.PositiveInfinity, float.NegativeInfinity, float.NaN }, + })); + + yield return Same( + "QBit(Float64, 3)", + "QBit(Float64, 3)", + name => new ArrayColumn(name, "QBit(Float64, 3)", new[] + { + new[] { 1d, -2d, 3.5d }, + new[] { double.MaxValue, double.MinValue, double.Epsilon }, + new[] { 0d, -0d, double.NaN }, + })); + + yield return Same( + "QBit(Float32, 17)", + "QBit(Float32, 17)", + name => new ArrayColumn(name, "QBit(Float32, 17)", new[] + { + Ramp(17, i => (i * 0.5f) - 4f), + Ramp(17, i => i % 2 == 0 ? float.MaxValue : float.MinValue), + })); + + yield return Same( + "QBit(Float64, 17)", + "QBit(Float64, 17)", + name => new ArrayColumn(name, "QBit(Float64, 17)", new[] + { + Ramp(17, i => (i * 0.25d) - 2d), + Ramp(17, i => i % 2 == 0 ? double.MaxValue : double.MinValue), + })); + + yield return new InsertRoundTripCase( + "QBit(BFloat16, 4)", + "QBit(BFloat16, 4)", + name => new ArrayColumn(name, "QBit(BFloat16, 4)", new[] + { + new[] { 1.0001f, 2f, -3f, 0f }, + new[] { -0f, 0.5f, -0.5f, 256f }, + }), + name => new ArrayColumn(name, "QBit(BFloat16, 4)", new[] + { + new[] { 1f, 2f, -3f, 0f }, + new[] { -0f, 0.5f, -0.5f, 256f }, + }), + settings: null); + + yield return Same( + "Nullable(QBit(Float32, 4))", + "Nullable(QBit(Float32, 4))", + name => new ArrayColumn(name, "Nullable(QBit(Float32, 4))", new[] + { + new[] { 1f, 2f, 3f, 4f }, + null, + new[] { -1f, -2f, -3f, -4f }, + })); + + yield return Same( + "Nullable(QBit(BFloat16, 4))", + "Nullable(QBit(BFloat16, 4))", + name => new ArrayColumn(name, "Nullable(QBit(BFloat16, 4))", new[] + { + new[] { 1f, 2f, -3f, 0f }, + null, + new[] { -0f, 0.5f, -0.5f, 256f }, + })); + + yield return Same( + "Nullable(QBit(Float64, 3))", + "Nullable(QBit(Float64, 3))", + name => new ArrayColumn(name, "Nullable(QBit(Float64, 3))", new[] + { + new[] { 1d, -2d, 3.5d }, + null, + new[] { 0d, -0d, double.NaN }, + })); + + if (TcpServerFeatures.Has(TcpFeature.QBitInt8)) + { + yield return Same( + "QBit(Int8, 17)", + "QBit(Int8, 17)", + name => new ArrayColumn(name, "QBit(Int8, 17)", new[] + { + Ramp(17, i => (sbyte)(i - 8)), + Ramp(17, i => i % 2 == 0 ? sbyte.MaxValue : sbyte.MinValue), + Ramp(17, i => i == 0 ? (sbyte)-1 : (sbyte)0), + })); + + yield return Same( + "Nullable(QBit(Int8, 4))", + "Nullable(QBit(Int8, 4))", + name => new ArrayColumn(name, "Nullable(QBit(Int8, 4))", new[] + { + new sbyte[] { 1, -2, sbyte.MaxValue, sbyte.MinValue }, + null, + new sbyte[] { 0, -1, 0, -1 }, + })); + } + } + // SimpleAggregateFunction(func, T) encodes as a bare T — the function only tells the server how to merge // rows — so these cases prove the alias is transparent, including when T is itself composite or nullable // and when the function carries parameters. @@ -1640,6 +1754,17 @@ private static ClickHouseDecimal ParseWide(string text) private static InsertRoundTripCase Same(string label, string clickHouseType, Func build, IReadOnlyDictionary settings = null) => new(label, clickHouseType, build, build, settings); + private static T[] Ramp(int length, Func value) + { + var values = new T[length]; + for (int i = 0; i < length; i++) + { + values[i] = value(i); + } + + return values; + } + /// Enables the experimental BFloat16 type for the round-trip. private static readonly IReadOnlyDictionary BFloat16Settings = new Dictionary(StringComparer.Ordinal) { diff --git a/ClickHouse.Driver.Tcp.Tests/Utilities/TcpFeature.cs b/ClickHouse.Driver.Tcp.Tests/Utilities/TcpFeature.cs index 2dfd68621..25e22b5b6 100644 --- a/ClickHouse.Driver.Tcp.Tests/Utilities/TcpFeature.cs +++ b/ClickHouse.Driver.Tcp.Tests/Utilities/TcpFeature.cs @@ -42,6 +42,10 @@ public enum TcpFeature [SinceVersion("26.6")] NullableTuple = 1 << 6, + /// QBit columns with Int8 elements. + [SinceVersion("26.7")] + QBitInt8 = 1 << 7, + /// Every capability. What an unrecognised or unpinned server version resolves to. All = ~None, } diff --git a/ClickHouse.Driver.Tcp.Tests/Utilities/TcpServerFeaturesTests.cs b/ClickHouse.Driver.Tcp.Tests/Utilities/TcpServerFeaturesTests.cs index 5d34ecb6c..cb725f0ae 100644 --- a/ClickHouse.Driver.Tcp.Tests/Utilities/TcpServerFeaturesTests.cs +++ b/ClickHouse.Driver.Tcp.Tests/Utilities/TcpServerFeaturesTests.cs @@ -11,6 +11,8 @@ public class TcpServerFeaturesTests [TestCase("25.8", TcpFeature.Json, ExpectedResult = true, TestName = "The oldest matrix server has Json")] [TestCase("25.11", TcpFeature.QBit, ExpectedResult = true, TestName = "QBit arrives in 25.11")] [TestCase("25.10", TcpFeature.QBit, ExpectedResult = false, TestName = "QBit is gated one release later than it shipped")] + [TestCase("26.6", TcpFeature.QBitInt8, ExpectedResult = false, TestName = "QBit Int8 is unavailable before 26.7")] + [TestCase("26.7", TcpFeature.QBitInt8, ExpectedResult = true, TestName = "QBit Int8 arrives in 26.7")] [TestCase("26.5", TcpFeature.NullableTuple, ExpectedResult = false, TestName = "Nullable Tuple Beta is unavailable before 26.6")] [TestCase("26.6", TcpFeature.NullableTuple, ExpectedResult = true, TestName = "Nullable Tuple Beta arrives in 26.6")] [TestCase("26.6", TcpFeature.Geometry, ExpectedResult = true, TestName = "A recent server has Geometry")] diff --git a/ClickHouse.Driver.Tcp/Types/Codecs/QBitColumnCodec.cs b/ClickHouse.Driver.Tcp/Types/Codecs/QBitColumnCodec.cs new file mode 100644 index 000000000..68e7ab074 --- /dev/null +++ b/ClickHouse.Driver.Tcp/Types/Codecs/QBitColumnCodec.cs @@ -0,0 +1,393 @@ +using System; +using System.Buffers; +using System.Globalization; +using System.Runtime.CompilerServices; +using System.Runtime.InteropServices; +using System.Runtime.Intrinsics; +using System.Threading; +using System.Threading.Tasks; +using ClickHouse.Driver.Tcp.Protocol; + +namespace ClickHouse.Driver.Tcp.Types.Codecs; + +/// +/// Encodes and decodes the Native layout of QBit(T, N). The body has no state prefix and contains bits(T) +/// planes in most-significant-first order. Each plane contains one big-endian ceil(N / 8)-byte bitmap per +/// row. Element i is bit i % 8 of byte ceil(N / 8) - 1 - i / 8. The body size is +/// bits(T) * rowCount * ceil(N / 8) bytes. Supported element types are Int8, BFloat16, +/// Float32, and Float64. +/// +internal abstract class QBitColumnCodec : IColumnCodec +{ + protected QBitColumnCodec(string typeName, int dimension, int bitWidth) + { + TypeName = typeName; + Dimension = dimension; + BitWidth = bitWidth; + BytesPerRow = QBitLayout.BytesPerRow(dimension); + } + + public string TypeName { get; } + + public abstract Type ElementType { get; } + + public abstract object NullPlaceholder { get; } + + protected int Dimension { get; } + + protected int BitWidth { get; } + + protected int BytesPerRow { get; } + + /// Gets the plane-group width. Strided types are rejected, so this equals . + protected int Stride => Dimension; + + /// Creates a codec for an unstrided QBit(T, N) type. + /// The argument count or dimension is invalid. + /// The element type or layout is unsupported. + public static QBitColumnCodec Create(TypeNode node) + { + // The three-argument form uses a group-major strided layout that this codec cannot decode. + if (node.Arguments.Count == 3) + { + throw new NotSupportedException( + $"QBit type '{node}' is strided; this client does not support the strided QBit layout yet."); + } + + if (node.Arguments.Count != 2) + { + throw new FormatException( + $"QBit type '{node}' must have exactly two arguments: the element type and the vector length."); + } + + string typeName = node.ToString(); + string element = node.Arguments[0].Name.Trim(); + string token = node.Arguments[1].Name.Trim(); + + if (!int.TryParse(token, NumberStyles.None, CultureInfo.InvariantCulture, out int dimension) || dimension <= 0) + { + throw new FormatException( + $"QBit type '{node}' has an invalid vector length '{token}'; expected a positive integer."); + } + + return element switch + { + "Int8" => new QBitSByteColumnCodec(typeName, dimension), + "BFloat16" => new QBitFloatColumnCodec(typeName, dimension, bitWidth: 16), + "Float32" => new QBitFloatColumnCodec(typeName, dimension, bitWidth: 32), + "Float64" => new QBitDoubleColumnCodec(typeName, dimension), + _ => throw new NotSupportedException( + $"QBit type '{node}' has element type '{element}'; only Int8, BFloat16, Float32 and Float64 are supported."), + }; + } + + public async ValueTask ReadColumnAsync(ClickHouseBinaryReader reader, string columnName, string columnType, int rowCount, CancellationToken cancellationToken) + { + if (rowCount == 0) + { + return CreateColumn(columnName, columnType, Array.Empty(), rowCount: 0, pooled: false); + } + + int byteCount = checked(BitWidth * rowCount * BytesPerRow); + byte[] blob = ArrayPool.Shared.Rent(byteCount); + try + { + await reader.ReadBytesAsync(blob.AsMemory(0, byteCount), cancellationToken).ConfigureAwait(false); + } + catch + { + // No column owns the rented buffer after a failed read. + ArrayPool.Shared.Return(blob); + throw; + } + + return CreateColumn(columnName, columnType, blob, rowCount, pooled: true); + } + + public abstract bool CanWrite(IColumn column); + + public void WriteColumn(ClickHouseBinaryWriter writer, IColumn column, int start, int length) + { + // A row range is contiguous within each plane, but planes are spaced by the source column's full row + // count. Dense copies require identical plane grouping; equal body sizes do not imply equal layouts. + if (column is QBitColumn dense && dense.Dimension == Dimension && dense.BitWidth == BitWidth && dense.Stride == Stride) + { + for (int wireIndex = 0; wireIndex < BitWidth; wireIndex++) + { + writer.WriteBytes(dense.WirePlane(wireIndex, start, length)); + } + + return; + } + + WriteTransposed(writer, column, start, length); + } + + protected abstract IColumn CreateColumn(string name, string typeName, byte[] blob, int rowCount, bool pooled); + + /// Transposes row vectors into a plane-major body. + protected abstract void WriteTransposed(ClickHouseBinaryWriter writer, IColumn column, int start, int length); + + /// + /// Rents scratch space and clears its used region because transpose implementations only set bits. + /// + protected byte[] RentScratch(int length, out int byteCount) + { + byteCount = checked(BitWidth * length * BytesPerRow); + byte[] scratch = ArrayPool.Shared.Rent(byteCount); + Array.Clear(scratch, 0, byteCount); + return scratch; + } + + /// + /// Returns a non-null vector with exactly elements. + /// + /// The vector is null or its length differs from . + protected T[] Validate(T[] vector, int row) + { + if (vector is null) + { + throw new ArgumentException( + $"A {TypeName} column cannot hold a null vector (at row {row}); wrap the type in Nullable to write nulls.", + nameof(vector)); + } + + if (vector.Length != Dimension) + { + throw new ArgumentException( + $"A {TypeName} vector at row {row} has {vector.Length} element(s); every vector must have exactly {Dimension}.", + nameof(vector)); + } + + return vector; + } +} + +/// +/// Handles QBit(Int8, N) as eight most-significant-first planes over each element's two's-complement byte. +/// +internal sealed class QBitSByteColumnCodec : QBitColumnCodec +{ + private sbyte[] nullPlaceholder; + + public QBitSByteColumnCodec(string typeName, int dimension) + : base(typeName, dimension, bitWidth: 8) + { + } + + public override Type ElementType => typeof(sbyte[]); + + /// Gets the all-zero vector used for null positions in a nullable column. + public override object NullPlaceholder => nullPlaceholder ??= new sbyte[Dimension]; + + public override bool CanWrite(IColumn column) => column is IColumn; + + protected override IColumn CreateColumn(string name, string typeName, byte[] blob, int rowCount, bool pooled) + => new QBitSByteColumn(name, typeName, Dimension, blob, rowCount, pooled); + + protected override void WriteTransposed(ClickHouseBinaryWriter writer, IColumn column, int start, int length) + { + var typed = (IColumn)column; + byte[] scratch = RentScratch(length, out int byteCount); + try + { + int planeStride = length * BytesPerRow; + for (int r = 0; r < length; r++) + { + sbyte[] vector = Validate(typed[start + r], start + r); + int rowBase = r * BytesPerRow; + for (int i = 0; i < vector.Length; i++) + { + // Preserve the element's two's-complement bit pattern. + uint raw = unchecked((byte)vector[i]); + int slot = QBitLayout.ByteOfGroup(i >> 3, BytesPerRow); + byte bit = (byte)(1 << (i & 7)); + for (int wireIndex = 0; wireIndex < BitWidth; wireIndex++) + { + if (((raw >> (7 - wireIndex)) & 1) != 0) + { + scratch[(wireIndex * planeStride) + rowBase + slot] |= bit; + } + } + } + } + + writer.WriteBytes(scratch.AsSpan(0, byteCount)); + } + finally + { + ArrayPool.Shared.Return(scratch); + } + } +} + +/// +/// Handles QBit(Float32, N) and QBit(BFloat16, N) as arrays. BFloat16 +/// retains only the high 16 bits of each value. +/// +internal sealed class QBitFloatColumnCodec : QBitColumnCodec +{ + private float[] nullPlaceholder; + + public QBitFloatColumnCodec(string typeName, int dimension, int bitWidth) + : base(typeName, dimension, bitWidth) + { + } + + public override Type ElementType => typeof(float[]); + + /// Gets the lazily allocated all-zero vector used for null positions in a nullable column. + public override object NullPlaceholder => nullPlaceholder ??= new float[Dimension]; + + public override bool CanWrite(IColumn column) => column is IColumn; + + protected override IColumn CreateColumn(string name, string typeName, byte[] blob, int rowCount, bool pooled) + => new QBitFloatColumn(name, typeName, Dimension, BitWidth, blob, rowCount, pooled); + + protected override void WriteTransposed(ClickHouseBinaryWriter writer, IColumn column, int start, int length) + { + var typed = (IColumn)column; + byte[] scratch = RentScratch(length, out int byteCount); + try + { + bool simd = Vector256.IsHardwareAccelerated; + int planeStride = length * BytesPerRow; + for (int r = 0; r < length; r++) + { + float[] vector = Validate(typed[start + r], start + r); + int rowBase = (r * BytesPerRow); + int whole = simd ? Dimension >> 3 : 0; + + if (whole != 0) + { + TransposeGroups(scratch, vector, whole, rowBase, planeStride); + } + + // Handle the tail, or the entire vector when SIMD is unavailable. + for (int i = whole << 3; i < vector.Length; i++) + { + uint raw = BitConverter.SingleToUInt32Bits(vector[i]); + int slot = QBitLayout.ByteOfGroup(i >> 3, BytesPerRow); + byte bit = (byte)(1 << (i & 7)); + for (int wireIndex = 0; wireIndex < BitWidth; wireIndex++) + { + // BFloat16 planes are the high 16 bits of the widened float. + if (((raw >> (31 - wireIndex)) & 1) != 0) + { + scratch[(wireIndex * planeStride) + rowBase + slot] |= bit; + } + } + } + } + + writer.WriteBytes(scratch.AsSpan(0, byteCount)); + } + finally + { + ArrayPool.Shared.Return(scratch); + } + } + + /// + /// Transposes complete eight-element groups. Each call + /// produces one plane byte; shifting the lanes left exposes the next plane. + /// + private void TransposeGroups(byte[] scratch, float[] vector, int whole, int rowBase, int planeStride) + { + ref uint source = ref Unsafe.As(ref MemoryMarshal.GetArrayDataReference(vector)); + for (int group = 0; group < whole; group++) + { + Vector256 lanes = Vector256.LoadUnsafe(ref source, (nuint)(group << 3)); + int slot = rowBase + QBitLayout.ByteOfGroup(group, BytesPerRow); + for (int wireIndex = 0; wireIndex < BitWidth; wireIndex++) + { + scratch[(wireIndex * planeStride) + slot] = (byte)lanes.ExtractMostSignificantBits(); + lanes <<= 1; + } + } + } +} + +/// Handles QBit(Float64, N) as 64 planes over each element's IEEE-754 bit pattern. +internal sealed class QBitDoubleColumnCodec : QBitColumnCodec +{ + private double[] nullPlaceholder; + + public QBitDoubleColumnCodec(string typeName, int dimension) + : base(typeName, dimension, bitWidth: 64) + { + } + + public override Type ElementType => typeof(double[]); + + /// Gets the all-zero vector used for null positions in a nullable column. + public override object NullPlaceholder => nullPlaceholder ??= new double[Dimension]; + + public override bool CanWrite(IColumn column) => column is IColumn; + + protected override IColumn CreateColumn(string name, string typeName, byte[] blob, int rowCount, bool pooled) + => new QBitDoubleColumn(name, typeName, Dimension, blob, rowCount, pooled); + + protected override void WriteTransposed(ClickHouseBinaryWriter writer, IColumn column, int start, int length) + { + var typed = (IColumn)column; + byte[] scratch = RentScratch(length, out int byteCount); + try + { + bool simd = Vector256.IsHardwareAccelerated; + int planeStride = length * BytesPerRow; + for (int r = 0; r < length; r++) + { + double[] vector = Validate(typed[start + r], start + r); + int rowBase = r * BytesPerRow; + int whole = simd ? Dimension >> 3 : 0; + + if (whole != 0) + { + TransposeGroups(scratch, vector, whole, rowBase, planeStride); + } + + for (int i = whole << 3; i < vector.Length; i++) + { + ulong raw = BitConverter.DoubleToUInt64Bits(vector[i]); + int slot = QBitLayout.ByteOfGroup(i >> 3, BytesPerRow); + byte bit = (byte)(1 << (i & 7)); + for (int wireIndex = 0; wireIndex < BitWidth; wireIndex++) + { + if (((raw >> (63 - wireIndex)) & 1) != 0) + { + scratch[(wireIndex * planeStride) + rowBase + slot] |= bit; + } + } + } + } + + writer.WriteBytes(scratch.AsSpan(0, byteCount)); + } + finally + { + ArrayPool.Shared.Return(scratch); + } + } + + /// + /// Transposes complete eight-element groups. Each four-lane vector contributes one nibble to a plane byte. + /// + private void TransposeGroups(byte[] scratch, double[] vector, int whole, int rowBase, int planeStride) + { + ref ulong source = ref Unsafe.As(ref MemoryMarshal.GetArrayDataReference(vector)); + for (int group = 0; group < whole; group++) + { + Vector256 low = Vector256.LoadUnsafe(ref source, (nuint)(group << 3)); + Vector256 high = Vector256.LoadUnsafe(ref source, (nuint)((group << 3) + 4)); + int slot = rowBase + QBitLayout.ByteOfGroup(group, BytesPerRow); + for (int wireIndex = 0; wireIndex < BitWidth; wireIndex++) + { + uint bits = low.ExtractMostSignificantBits() | (high.ExtractMostSignificantBits() << 4); + scratch[(wireIndex * planeStride) + slot] = (byte)bits; + low <<= 1; + high <<= 1; + } + } + } +} diff --git a/ClickHouse.Driver.Tcp/Types/ColumnCodecRegistry.cs b/ClickHouse.Driver.Tcp/Types/ColumnCodecRegistry.cs index 69f0cf236..fea7c6a97 100644 --- a/ClickHouse.Driver.Tcp/Types/ColumnCodecRegistry.cs +++ b/ClickHouse.Driver.Tcp/Types/ColumnCodecRegistry.cs @@ -96,6 +96,9 @@ private static ColumnCodecRegistry CreateDefault() // FixedString(N): N contiguous bytes per row, the length parsed from the type argument. AddFactory("FixedString", static (TypeNode node, in ResolveContext _, ColumnCodecRegistry _) => FixedStringColumnCodec.Create(node)); + // QBit(T, N): no state prefix; most-significant-first planes containing one big-endian bitmap per row. + AddFactory("QBit", static (TypeNode node, in ResolveContext _, ColumnCodecRegistry _) => QBitColumnCodec.Create(node)); + // Dates and times. AddConstant(DateColumnCodec.Instance); AddConstant(Date32ColumnCodec.Instance); diff --git a/ClickHouse.Driver.Tcp/Types/IQBitColumn.cs b/ClickHouse.Driver.Tcp/Types/IQBitColumn.cs new file mode 100644 index 000000000..4b44c261f --- /dev/null +++ b/ClickHouse.Driver.Tcp/Types/IQBitColumn.cs @@ -0,0 +1,61 @@ +using System; + +namespace ClickHouse.Driver.Tcp.Types; + +/// +/// Exposes the transposed bit planes of a decoded QBit(T, N) column without materializing row vectors. +/// The corresponding typed column exposes [] for BFloat16 and Float32, +/// [] for Float64, or [] for Int8. +/// +public interface IQBitColumn : IColumn +{ + /// Gets the number of elements in each row. + int Dimension { get; } + + /// + /// Gets the number of planes: 8 for Int8, 16 for BFloat16, 32 for Float32, or 64 for + /// Float64. + /// + int BitWidth { get; } + + /// + /// Gets the number of elements represented by each plane group. Columns decoded by this driver report + /// because strided QBit(T, N, stride) columns are unsupported. + /// + int Stride { get; } + + /// Gets Dimension / Stride, the number of plane groups in each row. + int GroupCount { get; } + + /// + /// Gets the size of one row bitmap, ceil(Stride / 8) bytes. Element i within a group is bit + /// i % 8 of byte BytesPerRow - 1 - i / 8. Unused bits are the high bits of the first byte. + /// + int BytesPerRow { get; } + + /// + /// Gets one plane as consecutive row bitmaps. Row r occupies + /// [r * BytesPerRow, (r + 1) * BytesPerRow). The bit number denotes significance, with 0 the least + /// significant stored bit and BitWidth - 1 the sign bit. For BFloat16, it refers to the 16-bit + /// stored value rather than the widened . The returned span borrows the column's storage + /// and is valid only for the owning block's lifetime. + /// + /// The bit position in the stored element. + /// RowCount * BytesPerRow bytes. + /// is outside + /// [0, ). + /// The column contains more than one plane group. + ReadOnlySpan GetPlane(int bit); + + /// + /// Gets one plane for one group as consecutive row bitmaps. Group g covers elements + /// [g * Stride, (g + 1) * Stride). Bit numbering matches . The returned span + /// borrows the column's storage and is valid only for the owning block's lifetime. + /// + /// The bit position in the stored element. + /// The zero-based group index. + /// RowCount * BytesPerRow bytes. + /// or is out + /// of range. + ReadOnlySpan GetPlane(int bit, int group); +} diff --git a/ClickHouse.Driver.Tcp/Types/QBitColumn.cs b/ClickHouse.Driver.Tcp/Types/QBitColumn.cs new file mode 100644 index 000000000..839f4d205 --- /dev/null +++ b/ClickHouse.Driver.Tcp/Types/QBitColumn.cs @@ -0,0 +1,298 @@ +using System; +using System.Buffers; +using System.Runtime.InteropServices; + +namespace ClickHouse.Driver.Tcp.Types; + +/// +/// Maps elements to bytes within a QBit row bitmap. +/// +internal static class QBitLayout +{ + /// + /// Returns the byte containing an eight-element group. Row bitmaps are big-endian, so group 0 occupies the + /// last byte. + /// + public static int ByteOfGroup(int group, int bytesPerRow) => bytesPerRow - 1 - group; + + /// + /// Returns ceil(span / 8) without overflowing when is positive. + /// + public static int BytesPerRow(int span) => ((span - 1) / 8) + 1; +} + +/// +/// Stores a decoded Native QBit(T, N) body in its transposed form. Planes are ordered most-significant +/// first, and each plane contains one fixed-width bitmap per row. A rented body is returned on +/// ; plane spans borrow the body's storage. Typed row vectors are materialized lazily. +/// +internal abstract class QBitColumn : IQBitColumn +{ + private readonly int rowCount; + private readonly bool pooled; + private byte[] blob; + + protected QBitColumn(string name, string typeName, int dimension, int bitWidth, byte[] blob, int rowCount, bool pooled) + { + Name = name; + TypeName = typeName; + Dimension = dimension; + BitWidth = bitWidth; + BytesPerRow = QBitLayout.BytesPerRow(dimension); + this.blob = blob ?? throw new ArgumentNullException(nameof(blob)); + this.rowCount = rowCount; + this.pooled = pooled; + } + + public string Name { get; } + + public string TypeName { get; } + + public int RowCount => rowCount; + + public int Dimension { get; } + + public int BitWidth { get; } + + public int BytesPerRow { get; } + + public int Stride => Dimension; + + public int GroupCount => 1; + + public ReadOnlySpan GetPlane(int bit) + { + if (GroupCount != 1) + { + throw new InvalidOperationException( + $"Column '{Name}' ({TypeName}) has {GroupCount} plane groups, so a plane is not one contiguous run; use GetPlane(bit, group)."); + } + + return GetPlane(bit, group: 0); + } + + public ReadOnlySpan GetPlane(int bit, int group) + { + if ((uint)bit >= (uint)BitWidth) + { + throw new ArgumentOutOfRangeException( + nameof(bit), + $"Bit {bit} is outside the {BitWidth} plane(s) of column '{Name}' ({TypeName})."); + } + + if ((uint)group >= (uint)GroupCount) + { + throw new ArgumentOutOfRangeException( + nameof(group), + $"Group {group} is outside the {GroupCount} group(s) of column '{Name}' ({TypeName})."); + } + + return WirePlane(((group * BitWidth) + BitWidth - 1 - bit), 0, rowCount); + } + + public abstract object GetValue(int row); + + public virtual void Dispose() + { + if (pooled && blob.Length != 0) + { + ArrayPool.Shared.Return(blob); + } + + blob = Array.Empty(); + } + + /// + /// Returns a zero-copy row range from a plane indexed in wire order, most-significant plane first. + /// + /// The requested range is outside the logical row count. + internal ReadOnlySpan WirePlane(int wireIndex, int start, int length) + { + // Validate against the logical row count because a rented blob can contain unused trailing bytes. + if (start < 0 || length < 0 || start + (long)length > rowCount) + { + throw new ArgumentOutOfRangeException( + length < 0 ? nameof(length) : nameof(start), + $"Rows [{start}, {start + (long)length}) lie outside the {rowCount} row(s) of column '{Name}'."); + } + + return blob.AsSpan(((wireIndex * rowCount) + start) * BytesPerRow, length * BytesPerRow); + } + + protected ReadOnlySpan WirePlaneRow(int wireIndex, int row) => WirePlane(wireIndex, row, 1); + + protected void CheckRow(int row) + { + if ((uint)row >= (uint)rowCount) + { + throw new IndexOutOfRangeException(); + } + } +} + +/// +/// Provides lazily materialized [] rows over a transposed . +/// +internal abstract class QBitColumnBase : QBitColumn, IColumn + where T : struct +{ + private T[][] cache; + + protected QBitColumnBase(string name, string typeName, int dimension, int bitWidth, byte[] blob, int rowCount, bool pooled) + : base(name, typeName, dimension, bitWidth, blob, rowCount, pooled) + { + } + + /// + /// Gets row vectors, materializing and caching all rows on first access. Use to + /// inspect transposed data without this allocation. + /// + public ReadOnlySpan Values + { + get + { + if (cache is null) + { + T[][] decoded = ArrayPool.Shared.Rent(RowCount); + for (int i = 0; i < RowCount; i++) + { + decoded[i] = DetransposeRow(i); + } + + cache = decoded; + } + + return cache.AsSpan(0, RowCount); + } + } + + // Slice to RowCount so indexing cannot reach unused entries in the pooled array. + public T[] this[int row] => cache is not null ? cache.AsSpan(0, RowCount)[row] : DetransposeRow(row); + + public override object GetValue(int row) => this[row]; + + public override void Dispose() + { + base.Dispose(); + + if (cache is not null) + { + // Clear references so the pool does not retain decoded row arrays. + ArrayPool.Shared.Return(cache, clearArray: true); + cache = null; + } + } + + protected abstract T[] DetransposeRow(int row); +} + +/// +/// Decodes QBit(Float32, N) and QBit(BFloat16, N) rows as arrays. A +/// BFloat16 value fills the high 16 bits of the widened result. +/// +internal sealed class QBitFloatColumn : QBitColumnBase +{ + public QBitFloatColumn(string name, string typeName, int dimension, int bitWidth, byte[] blob, int rowCount, bool pooled) + : base(name, typeName, dimension, bitWidth, blob, rowCount, pooled) + { + } + + protected override float[] DetransposeRow(int row) + { + CheckRow(row); + + var vector = new float[Dimension]; + + // Build each IEEE-754 pattern directly in the destination array. + Span bits = MemoryMarshal.Cast(vector.AsSpan()); + for (int wireIndex = 0; wireIndex < BitWidth; wireIndex++) + { + ReadOnlySpan plane = WirePlaneRow(wireIndex, row); + uint mask = 1u << (BitWidth - 1 - wireIndex); + for (int i = 0; i < bits.Length; i++) + { + if ((plane[QBitLayout.ByteOfGroup(i >> 3, BytesPerRow)] & (1 << (i & 7))) != 0) + { + bits[i] |= mask; + } + } + } + + // BFloat16 occupies the high half of the widened float. + if (BitWidth == 16) + { + for (int i = 0; i < bits.Length; i++) + { + bits[i] <<= 16; + } + } + + return vector; + } +} + +/// +/// Decodes QBit(Int8, N) rows from each element's two's-complement bit pattern. +/// +internal sealed class QBitSByteColumn : QBitColumnBase +{ + public QBitSByteColumn(string name, string typeName, int dimension, byte[] blob, int rowCount, bool pooled) + : base(name, typeName, dimension, bitWidth: 8, blob, rowCount, pooled) + { + } + + protected override sbyte[] DetransposeRow(int row) + { + CheckRow(row); + + var vector = new sbyte[Dimension]; + + // Use an unsigned view so the sign plane sets bit 7 without numeric conversion. + Span bits = MemoryMarshal.Cast(vector.AsSpan()); + for (int wireIndex = 0; wireIndex < BitWidth; wireIndex++) + { + ReadOnlySpan plane = WirePlaneRow(wireIndex, row); + byte mask = (byte)(1 << (BitWidth - 1 - wireIndex)); + for (int i = 0; i < bits.Length; i++) + { + if ((plane[QBitLayout.ByteOfGroup(i >> 3, BytesPerRow)] & (1 << (i & 7))) != 0) + { + bits[i] |= mask; + } + } + } + + return vector; + } +} + +/// Decodes QBit(Float64, N) rows from each element's IEEE-754 bit pattern. +internal sealed class QBitDoubleColumn : QBitColumnBase +{ + public QBitDoubleColumn(string name, string typeName, int dimension, byte[] blob, int rowCount, bool pooled) + : base(name, typeName, dimension, bitWidth: 64, blob, rowCount, pooled) + { + } + + protected override double[] DetransposeRow(int row) + { + CheckRow(row); + + var vector = new double[Dimension]; + Span bits = MemoryMarshal.Cast(vector.AsSpan()); + for (int wireIndex = 0; wireIndex < BitWidth; wireIndex++) + { + ReadOnlySpan plane = WirePlaneRow(wireIndex, row); + ulong mask = 1UL << (BitWidth - 1 - wireIndex); + for (int i = 0; i < bits.Length; i++) + { + if ((plane[QBitLayout.ByteOfGroup(i >> 3, BytesPerRow)] & (1 << (i & 7))) != 0) + { + bits[i] |= mask; + } + } + } + + return vector; + } +}