#87. Support for LowCardinality supertype.

ClickHouse.Ado/Impl/ColumnTypes/ColumnType.cs

@@ -118,6 +118,8 @@ public static ColumnType Create(string name) {
                 switch (m.Groups["outer"].Value) {
                     case "Nullable":
                         return new NullableColumnType(Create(m.Groups["inner"].Value));
+                    case "LowCardinality":
+                        return new LowCardinalityColumnType(Create(m.Groups["inner"].Value));
                     case "Array":
                         if (m.Groups["inner"].Value == "Null")
                             return new ArrayColumnType(new NullableColumnType(new SimpleColumnType<byte>()));

ClickHouse.Ado/Impl/ColumnTypes/LowCardinalityColumnType.cs

@@ -0,0 +1,73 @@
+using System;
+using System.Collections;
+using System.Diagnostics;
+using System.Linq;
+using ClickHouse.Ado.Impl.ATG.Insert;
+using ClickHouse.Ado.Impl.Data;
+
+namespace ClickHouse.Ado.Impl.ColumnTypes {
+    internal class LowCardinalityColumnType : ColumnType {
+        public LowCardinalityColumnType(ColumnType innerType) => InnerType = innerType;
+
+        public override int Rows => Indices?.Length ?? 0;
+        internal override Type CLRType => InnerType.CLRType;
+
+        public ColumnType InnerType { get; }
+
+        private int _keySize;
+
+        private int[] Indices;
+
+        public override string AsClickHouseType(ClickHouseTypeUsageIntent usageIntent) => $"LowCardinality({InnerType.AsClickHouseType(usageIntent)})";
+
+        public override void Write(ProtocolFormatter formatter, int rows) {
+            // This is rather naive implementation of writing - without any deduplication, however who cares?
+            // Clickhouse server will re-deduplicate inserted values anyway.
+            formatter.WriteBytes(BitConverter.GetBytes(1L));
+            formatter.WriteBytes(BitConverter.GetBytes(1538L));
+            formatter.WriteBytes(BitConverter.GetBytes((long) rows));
+            InnerType.Write(formatter, rows);
+            formatter.WriteBytes(BitConverter.GetBytes((long) rows));
+            for (var i = 0; i < rows; i++)
+                formatter.WriteBytes(BitConverter.GetBytes(i));
+        }
+
+        internal override void Read(ProtocolFormatter formatter, int rows) {
+            var version = BitConverter.ToInt64(formatter.ReadBytes(8), 0);
+            if (version != 1)
+                throw new NotSupportedException("Invalid LowCardinality dictionary version");
+            var keyLength = BitConverter.ToInt64(formatter.ReadBytes(8), 0);
+            _keySize = 1 << (byte) (keyLength & 0xff);
+            if (_keySize < 0 || _keySize > 4) //LowCardinality with >4e9 keys? WTF???
+                throw new NotSupportedException("Invalid LowCardinality key size");
+            if (((keyLength >> 8) & 0xff) != 6)
+                throw new NotSupportedException("Invalid LowCardinality key flags");
+            var keyCount = BitConverter.ToInt64(formatter.ReadBytes(8), 0);
+            InnerType.Read(formatter, (int) keyCount);
+            var valueCount = BitConverter.ToInt64(formatter.ReadBytes(8), 0);
+            Indices = new int[rows];
+            for (var i = 0; i < rows; i++) {
+                Indices[i] = BitConverter.ToInt32(formatter.ReadBytes(_keySize, 4), 0);
+            }
+        }
+
+        public override void ValueFromConst(Parser.ValueType val) {
+            InnerType.ValueFromConst(val);
+            Indices = new int[InnerType.Rows];
+        }
+
+        public override void ValueFromParam(ClickHouseParameter parameter) {
+            InnerType.ValueFromParam(parameter);
+            Indices = new int[InnerType.Rows];
+        }
+
+        public override object Value(int currentRow) => InnerType.Value(Indices[currentRow]);
+
+        public override long IntValue(int currentRow) { return InnerType.IntValue(Indices[currentRow]); }
+
+        public override void ValuesFromConst(IEnumerable objects) {
+            InnerType.NullableValuesFromConst(objects);
+            Indices = new int[InnerType.Rows];
+        }
+    }
+}

ClickHouse.Ado/Impl/ProtocolFormatter.cs

@@ -305,8 +305,8 @@ internal string ReadString() {
             return rv;
         }
 
-        public byte[] ReadBytes(int i) {
-            var bytes = new byte[i];
+        public byte[] ReadBytes(int i, int size=-1) {
+            var bytes = new byte[size == -1 ? i : size];
             var read = 0;
             var cur = 0;
             var networkStream = _ioStream as NetworkStream ?? (_ioStream as UnclosableStream)?.BaseStream as NetworkStream;
@@ -411,4 +411,4 @@ private void EndDecompression() {
 
         #endregion
     }
-}
+}

ClickHouse.Test/ClickHouse.Test.csproj

@@ -62,8 +62,9 @@
     <Compile Include="Properties\AssemblyInfo.cs" />
     <Compile Include="TestDateTime64Support.cs" />
     <Compile Include="TestGuidSupport.cs" />
-    <Compile Include="Test_82_NullableFixedString.cs" />
     <Compile Include="Test_100_BulkArrayFirst.cs" />
+    <Compile Include="Test_82_NullableFixedString.cs" />
+    <Compile Include="Test_87_LowCardinality.cs" />
     <Compile Include="Test_90_CloseConnection.cs" />
     <Compile Include="Test_99_NegativeNumerics.cs" />
     <Compile Include="Test_84_DecimalSupport.cs" />

ClickHouse.Test/Test_87_LowCardinality.cs

@@ -0,0 +1,39 @@
+using System.Collections.Generic;
+using System.Data;
+using System.Linq;
+using System.Threading;
+using ClickHouse.Ado;
+using NUnit.Framework;
+
+namespace ClickHouse.Test {
+    [TestFixture]
+    public class Test_87_LowCardinality {
+        [OneTimeSetUp]
+        public void CreateStructures() {
+            using (var cnn = ConnectionHandler.GetConnection()) {
+                cnn.CreateCommand("DROP TABLE IF EXISTS test_87_lowcardinality").ExecuteNonQuery();
+                cnn.CreateCommand("CREATE TABLE test_87_lowcardinality (a LowCardinality(String)) ENGINE = Memory").ExecuteNonQuery();
+            }
+
+            Thread.Sleep(1000);
+        }
+
+        [Test]
+        public void Test() {
+            using (var cnn = ConnectionHandler.GetConnection()) {
+                var items = new List<string>();
+                for (var i = 0; i < 1000; i++)
+                    items.Add(((char) ('A' + (i % 20))).ToString());
+                var result = cnn.CreateCommand("INSERT INTO test_87_lowcardinality (a) VALUES @bulk").AddParameter("bulk", DbType.Object, items.Select(x => (object) new object[] {x}).ToArray())
+                                .ExecuteNonQuery();
+
+                var values = new List<string>();
+                using (var cmd = cnn.CreateCommand("SELECT a FROM test_87_lowcardinality"))
+                using (var reader = cmd.ExecuteReader()) {
+                    reader.ReadAll(r => { values.Add(r.GetString(0)); });
+                }
+                Assert.IsTrue(items.SequenceEqual(values));
+            }
+        }
+    }
+}