Fix the bug existed in low_cardinality(nullable) type (#43)

* fix the bug existed in low_cardinality(nullable) type

* code revise

timeplus-native-jdbc/src/main/java/com/timeplus/data/ColumnLowCardinality.java

@@ -15,6 +15,7 @@
 package com.timeplus.data;
 
 import com.timeplus.data.type.complex.DataTypeLowCardinality;
+import com.timeplus.data.type.complex.DataTypeNullable;
 import com.timeplus.serde.BinarySerializer;
 import java.io.IOException;
 import java.sql.SQLException;
@@ -27,23 +28,44 @@ public class ColumnLowCardinality extends AbstractColumn {
     private final List<Long> indexes;
     private final List<Object> dict;
     private final Long version = 1L;
+    private boolean nested_is_nullable;
 
     public ColumnLowCardinality(String name, DataTypeLowCardinality type, Object[] values) {
         super(name, type, values);
         indexes = new ArrayList<>();
         dict = new ArrayList<>();
-        data = ColumnFactory.createColumn(null, type.getNestedTypes(), null);
+        nested_is_nullable = type.getNestedTypes().nullable();
+        /// If a nested type is nullable, always add two hard dictionary keys in front: [0]: null, [1]: default value
+        if (nested_is_nullable) {
+            IDataType nested_type = ((DataTypeNullable) type.getNestedTypes()).getNestedDataType();
+            data = ColumnFactory.createColumn(null, nested_type, null);
+            dict.add(type.getNestedTypes().defaultValue());
+            dict.add(type.getNestedTypes().defaultValue());
+        }
+        else {
+            data = ColumnFactory.createColumn(null, type.getNestedTypes(), null);
+        }
     }
 
     @Override
     public void write(Object object) throws IOException, SQLException {
-        long value = dict.indexOf(object);
-        if (value != -1) { 
-            indexes.add(value);
+        if (object == null) {
+            if (nested_is_nullable) {
+                indexes.add(0l);
+            }
+            else {
+                throw new SQLException("null object appeared without nullable field");
+            }
         }
         else {
-            indexes.add((long) dict.size());
-            dict.add(object);
+            long value = dict.lastIndexOf(object);
+            if (value != -1) { 
+                indexes.add(value);
+            }
+            else {
+                indexes.add((long) dict.size());
+                dict.add(object);
+            }
         }
     }
 
@@ -53,16 +75,19 @@ public void flushToSerializer(BinarySerializer serializer, boolean immediate) th
             serializer.writeUTF8StringBinary(name);
             serializer.writeUTF8StringBinary(type.name());
         }
+
         if (immediate) {
             /// The data layout: [version][index_type][dictionary][indexes]
             serializer.writeLong(version);
             serializer.writeLong(IndexType.UInt64.getValue() | IndexType.HasAdditionalKeysBit.getValue());
+
             serializer.writeLong(dict.size());
             for (int i = 0; i < dict.size(); i++) {
                 data.write(dict.get(i));
             }
             data.flushToSerializer(serializer, true);
-            serializer.writeLong(indexes.size());
+
+            serializer.writeLong(indexes.size()); //  give index type size
             for (int i = 0; i < indexes.size(); i++) {
                 serializer.writeLong(indexes.get(i));
             }

timeplus-native-jdbc/src/main/java/com/timeplus/data/type/complex/DataTypeLowCardinality.java

@@ -46,10 +46,14 @@ public class DataTypeLowCardinality implements IDataType<Object, Object>  {
     private final Long version = 1L;
     private final Long IndexTypeMask = 0b11111111L;
     private static final Logger LOG = LoggerFactory.getLogger(DataTypeLowCardinality.class);
+    private boolean nested_is_nullable;
 
     public DataTypeLowCardinality(String name, IDataType<?, ?>  nestedDataType) {
         this.name = name;
         this.nestedDataType = nestedDataType;
+        if (nestedDataType.nullable()) {
+            nested_is_nullable = true;
+        }
     }
 
     @Override
@@ -114,25 +118,39 @@ public Object deserializeBinary(BinaryDeserializer deserializer) throws SQLExcep
 
     @Override
     public Object[] deserializeBinaryBulk(int rows, BinaryDeserializer deserializer) throws SQLException, IOException {
-        if (rows==0) {
+        if (rows == 0) {
             Object[] data = getNestedTypes().deserializeBinaryBulk(rows, deserializer);
             return data;
         }
         else {
-            Long version = deserializer.readLong(); 
+            Long version = deserializer.readLong();
+
             if (version != this.version) {
                 throw new SQLException("version error in type low_cardinality");
             }
 
             Long index_type = deserializer.readLong() & IndexTypeMask;
             Long key_nums = deserializer.readLong();
-            Object[] dictionary = getNestedTypes().deserializeBinaryBulk(key_nums.intValue(), deserializer);
+            Object[] dictionary = new Object[key_nums.intValue()];
+            IDataType inner_type;
+
+            if (nested_is_nullable) {
+                DataTypeNullable type = (DataTypeNullable) getNestedTypes();
+                inner_type = type.getNestedDataType();
+            }
+            else {
+                inner_type = getNestedTypes();
+            }
+            dictionary = inner_type.deserializeBinaryBulk(key_nums.intValue(), deserializer);
 
             Long row_nums = deserializer.readLong();
+
             if (row_nums != rows) {
                 throw new SQLException("read unexpected rows in low_cardinality, expected:" + rows + ", actual:" + row_nums);
             }
+
             IDataType type;
+
             if (index_type == IndexType.UInt8.getValue()) {
                 type = new DataTypeUInt8();
             }
@@ -148,6 +166,7 @@ else if (index_type == IndexType.UInt32.getValue()) {
 
             Object[] index_data = type.deserializeBinaryBulk(rows, deserializer);
             Object[] data = new Object[rows];
+
             if (type instanceof DataTypeUInt8) {
                 for (int i = 0; i < rows; i++) {
                     data[i] = dictionary[(short) index_data[i]];
@@ -157,7 +176,7 @@ else if (index_type == IndexType.UInt32.getValue()) {
                 for (int i = 0; i < rows; i++) {
                     data[i] = dictionary[(Integer) index_data[i]];
                 }
-            } 
+            }
             return data;
         }   
     }