Remove null map cache from ColumnUnique.

dbms/src/Columns/ColumnUnique.h

@@ -81,7 +81,7 @@ public:
     {
         return column_holder->allocatedBytes()
                + index.allocatedBytes()
-               + (cached_null_mask ? cached_null_mask->allocatedBytes() : 0);
+               + (nested_null_mask ? nested_null_mask->allocatedBytes() : 0);
     }
     void forEachSubcolumn(IColumn::ColumnCallback callback) override
     {
@@ -100,8 +100,8 @@ private:
     ReverseIndex<UInt64, ColumnType> index;
 
     /// For DataTypeNullable, stores null map.
-    mutable ColumnPtr cached_null_mask;
+    ColumnPtr nested_null_mask;
-    mutable ColumnPtr cached_column_nullable;
+    ColumnPtr nested_column_nullable;
 
     class IncrementalHash
     {
@@ -118,6 +118,8 @@ private:
 
     mutable IncrementalHash hash;
 
+    void updateNullMask();
+
     static size_t numSpecialValues(bool is_nullable) { return is_nullable ? 2 : 1; }
     size_t numSpecialValues() const { return numSpecialValues(is_nullable); }
 
@@ -148,6 +150,7 @@ ColumnUnique<ColumnType>::ColumnUnique(const ColumnUnique & other)
     , index(numSpecialValues(is_nullable), 0)
 {
     index.setColumn(getRawColumnPtr());
+    updateNullMask();
 }
 
 template <typename ColumnType>
@@ -158,6 +161,7 @@ ColumnUnique<ColumnType>::ColumnUnique(const IDataType & type)
     const auto & holder_type = is_nullable ? *static_cast<const DataTypeNullable &>(type).getNestedType() : type;
     column_holder = holder_type.createColumn()->cloneResized(numSpecialValues());
     index.setColumn(getRawColumnPtr());
+    updateNullMask();
 }
 
 template <typename ColumnType>
@@ -172,32 +176,37 @@ ColumnUnique<ColumnType>::ColumnUnique(MutableColumnPtr && holder, bool is_nulla
         throw Exception("Holder column for ColumnUnique can't be nullable.", ErrorCodes::ILLEGAL_COLUMN);
 
     index.setColumn(getRawColumnPtr());
+    updateNullMask();
+}
+
+template <typename ColumnType>
+void ColumnUnique<ColumnType>::updateNullMask()
+{
+    if (is_nullable)
+    {
+        size_t size = getRawColumnPtr()->size();
+        if (!nested_null_mask)
+        {
+            ColumnUInt8::MutablePtr null_mask = ColumnUInt8::create(size, UInt8(0));
+            null_mask->getData()[getNullValueIndex()] = 1;
+            nested_null_mask = std::move(null_mask);
+            nested_column_nullable = ColumnNullable::create(column_holder, nested_null_mask);
+        }
+
+        if (nested_null_mask->size() != size)
+        {
+            IColumn & null_mask = nested_null_mask->assumeMutableRef();
+            static_cast<ColumnUInt8 &>(null_mask).getData().resize_fill(size);
+        }
+    }
 }
 
 template <typename ColumnType>
 const ColumnPtr & ColumnUnique<ColumnType>::getNestedColumn() const
 {
     if (is_nullable)
-    {
+        return nested_column_nullable;
-        size_t size = getRawColumnPtr()->size();
-        if (!cached_null_mask)
-        {
-            ColumnUInt8::MutablePtr null_mask = ColumnUInt8::create(size, UInt8(0));
-            null_mask->getData()[getNullValueIndex()] = 1;
-            cached_null_mask = std::move(null_mask);
-            cached_column_nullable = ColumnNullable::create(column_holder, cached_null_mask);
-        }
 
-        if (cached_null_mask->size() != size)
-        {
-            MutableColumnPtr null_mask = (*std::move(cached_null_mask)).mutate();
-            static_cast<ColumnUInt8 &>(*null_mask).getData().resize_fill(size);
-            cached_null_mask = std::move(null_mask);
-            cached_column_nullable = ColumnNullable::create(column_holder, cached_null_mask);
-        }
-
-        return cached_column_nullable;
-    }
     return column_holder;
 }
 
@@ -227,6 +236,8 @@ size_t ColumnUnique<ColumnType>::uniqueInsert(const Field & x)
     if (pos != prev_size)
         column->popBack(1);
 
+    updateNullMask();
+
     return pos;
 }
 
@@ -260,6 +271,8 @@ size_t ColumnUnique<ColumnType>::uniqueInsertData(const char * pos, size_t lengt
         index.insertFromLastRow();
     }
 
+    updateNullMask();
+
     return insertion_point;
 }
 
@@ -288,6 +301,8 @@ size_t ColumnUnique<ColumnType>::uniqueInsertDataWithTerminatingZero(const char
     if (position != prev_size)
         column->popBack(1);
 
+    updateNullMask();
+
     return static_cast<size_t>(position);
 }
 
@@ -343,6 +358,8 @@ size_t ColumnUnique<ColumnType>::uniqueDeserializeAndInsertFromArena(const char
     if (index_pos != prev_size)
         column->popBack(1);
 
+    updateNullMask();
+
     return static_cast<size_t>(index_pos);
 }
 
@@ -533,6 +550,8 @@ MutableColumnPtr ColumnUnique<ColumnType>::uniqueInsertRangeFrom(const IColumn &
     if (!positions_column)
         throw Exception("Can't find index type for ColumnUnique", ErrorCodes::LOGICAL_ERROR);
 
+    updateNullMask();
+
     return positions_column;
 }
 
@@ -577,6 +596,8 @@ IColumnUnique::IndexesWithOverflow ColumnUnique<ColumnType>::uniqueInsertRangeWi
     if (!positions_column)
         throw Exception("Can't find index type for ColumnUnique", ErrorCodes::LOGICAL_ERROR);
 
+    updateNullMask();
+
     IColumnUnique::IndexesWithOverflow indexes_with_overflow;
     indexes_with_overflow.indexes = std::move(positions_column);
     indexes_with_overflow.overflowed_keys = std::move(overflowed_keys);