Merge pull request ClickHouse#62087 from ClickHouse/checkmate

Avoid crashing on column type mismatch in a few dozen places

Author: al13n321

src/Columns/ColumnArray.cpp

@@ -1283,7 +1283,7 @@ ColumnPtr ColumnArray::replicateTuple(const Offsets & replicate_offsets) const
 
 size_t ColumnArray::getNumberOfDimensions() const
 {
-    const auto * nested_array = checkAndGetColumn<ColumnArray>(*data);
+    const auto * nested_array = checkAndGetColumn<ColumnArray>(&*data);
     if (!nested_array)
         return 1;
     return 1 + nested_array->getNumberOfDimensions();   /// Every modern C++ compiler optimizes tail recursion.

src/Columns/ColumnLowCardinality.cpp

@@ -903,7 +903,7 @@ ColumnPtr ColumnLowCardinality::cloneWithDefaultOnNull() const
 
 bool isColumnLowCardinalityNullable(const IColumn & column)
 {
-    if (const auto * lc_column = checkAndGetColumn<ColumnLowCardinality>(column))
+    if (const auto * lc_column = checkAndGetColumn<ColumnLowCardinality>(&column))
         return lc_column->nestedIsNullable();
     return false;
 }

src/Columns/ColumnUnique.h

@@ -376,7 +376,7 @@ size_t ColumnUnique<ColumnType>::uniqueInsertFrom(const IColumn & src, size_t n)
     if (is_nullable && src.isNullAt(n))
         return getNullValueIndex();
 
-    if (const auto * nullable = checkAndGetColumn<ColumnNullable>(src))
+    if (const auto * nullable = checkAndGetColumn<ColumnNullable>(&src))
         return uniqueInsertFrom(nullable->getNestedColumn(), n);
 
     auto ref = src.getDataAt(n);
@@ -569,7 +569,7 @@ MutableColumnPtr ColumnUnique<ColumnType>::uniqueInsertRangeImpl(
         return nullptr;
     };
 
-    if (const auto * nullable_column = checkAndGetColumn<ColumnNullable>(src))
+    if (const auto * nullable_column = checkAndGetColumn<ColumnNullable>(&src))
     {
         src_column = typeid_cast<const ColumnType *>(&nullable_column->getNestedColumn());
         null_map = &nullable_column->getNullMapData();

src/Columns/FilterDescription.cpp

@@ -32,7 +32,7 @@ ConstantFilterDescription::ConstantFilterDescription(const IColumn & column)
 
         if (!typeid_cast<const ColumnUInt8 *>(column_nested.get()))
         {
-            const ColumnNullable * column_nested_nullable = checkAndGetColumn<ColumnNullable>(*column_nested);
+            const ColumnNullable * column_nested_nullable = checkAndGetColumn<ColumnNullable>(&*column_nested);
             if (!column_nested_nullable || !typeid_cast<const ColumnUInt8 *>(&column_nested_nullable->getNestedColumn()))
             {
                 throw Exception(ErrorCodes::ILLEGAL_TYPE_OF_COLUMN_FOR_FILTER,
@@ -66,7 +66,7 @@ FilterDescription::FilterDescription(const IColumn & column_)
         return;
     }
 
-    if (const auto * nullable_column = checkAndGetColumn<ColumnNullable>(column))
+    if (const auto * nullable_column = checkAndGetColumn<ColumnNullable>(&column))
     {
         ColumnPtr nested_column = nullable_column->getNestedColumnPtr();
         MutableColumnPtr mutable_holder = IColumn::mutate(std::move(nested_column));

src/Columns/IColumn.h

@@ -640,12 +640,16 @@ template <>
 struct IsMutableColumns<> { static const bool value = true; };
 
 
+/// Throws LOGICAL_ERROR if the type doesn't match.
 template <typename Type>
-const Type * checkAndGetColumn(const IColumn & column)
+const Type & checkAndGetColumn(const IColumn & column)
 {
-    return typeid_cast<const Type *>(&column);
+    return typeid_cast<const Type &>(column);
 }
 
+/// Returns nullptr if the type doesn't match.
+/// If you're going to dereference the returned pointer without checking for null, use the
+/// `const IColumn &` overload above instead.
 template <typename Type>
 const Type * checkAndGetColumn(const IColumn * column)
 {

src/Columns/MaskOperations.cpp

@@ -205,10 +205,10 @@ static MaskInfo extractMaskImpl(
     auto column = col->convertToFullColumnIfLowCardinality();
 
     /// Special implementation for Null and Const columns.
-    if (column->onlyNull() || checkAndGetColumn<ColumnConst>(*column))
+    if (column->onlyNull() || checkAndGetColumn<ColumnConst>(&*column))
         return extractMaskFromConstOrNull<inverted>(mask, column, null_value, nulls);
 
-    if (const auto * nullable_column = checkAndGetColumn<ColumnNullable>(*column))
+    if (const auto * nullable_column = checkAndGetColumn<ColumnNullable>(&*column))
     {
         const PaddedPODArray<UInt8> & null_map = nullable_column->getNullMapData();
         return extractMaskImpl<inverted>(mask, nullable_column->getNestedColumnPtr(), null_value, &null_map, nulls);

src/Common/ColumnsHashing.h

@@ -44,8 +44,8 @@ struct HashMethodOneNumber
     {
         if constexpr (nullable)
         {
-            const auto * null_column = checkAndGetColumn<ColumnNullable>(key_columns[0]);
-            vec = null_column->getNestedColumnPtr()->getRawData().data();
+            const auto & null_column = checkAndGetColumn<ColumnNullable>(*key_columns[0]);
+            vec = null_column.getNestedColumnPtr()->getRawData().data();
         }
         else
         {
@@ -57,8 +57,8 @@ struct HashMethodOneNumber
     {
         if constexpr (nullable)
         {
-            const auto * null_column = checkAndGetColumn<ColumnNullable>(column);
-            vec = null_column->getNestedColumnPtr()->getRawData().data();
+            const auto & null_column = checkAndGetColumn<ColumnNullable>(*column);
+            vec = null_column.getNestedColumnPtr()->getRawData().data();
         }
         else
         {
@@ -105,7 +105,7 @@ struct HashMethodString
         const IColumn * column;
         if constexpr (nullable)
         {
-            column = checkAndGetColumn<ColumnNullable>(key_columns[0])->getNestedColumnPtr().get();
+            column = checkAndGetColumn<ColumnNullable>(*key_columns[0]).getNestedColumnPtr().get();
         }
         else
         {
@@ -153,7 +153,7 @@ struct HashMethodFixedString
         const IColumn * column;
         if constexpr (nullable)
         {
-            column = checkAndGetColumn<ColumnNullable>(key_columns[0])->getNestedColumnPtr().get();
+            column = checkAndGetColumn<ColumnNullable>(*key_columns[0]).getNestedColumnPtr().get();
         }
         else
         {

src/Common/ColumnsHashingImpl.h

@@ -305,7 +305,7 @@ class HashMethodBase
         }
 
         if constexpr (nullable)
-            null_map = &checkAndGetColumn<ColumnNullable>(column)->getNullMapColumn();
+            null_map = &checkAndGetColumn<ColumnNullable>(*column).getNullMapColumn();
     }
 
     template <typename Data, typename KeyHolder>

src/Core/DecimalComparison.h

@@ -170,11 +170,11 @@ class DecimalComparison
 
             if (c0_is_const && c1_is_const)
             {
-                const ColumnConst * c0_const = checkAndGetColumnConst<ColVecA>(c0.get());
-                const ColumnConst * c1_const = checkAndGetColumnConst<ColVecB>(c1.get());
+                const ColumnConst & c0_const = checkAndGetColumnConst<ColVecA>(*c0);
+                const ColumnConst & c1_const = checkAndGetColumnConst<ColVecB>(*c1);
 
-                A a = c0_const->template getValue<A>();
-                B b = c1_const->template getValue<B>();
+                A a = c0_const.template getValue<A>();
+                B b = c1_const.template getValue<B>();
                 UInt8 res = apply<scale_left, scale_right>(a, b, scale);
                 return DataTypeUInt8().createColumnConst(c0->size(), toField(res));
             }
@@ -184,17 +184,17 @@ class DecimalComparison
 
             if (c0_is_const)
             {
-                const ColumnConst * c0_const = checkAndGetColumnConst<ColVecA>(c0.get());
-                A a = c0_const->template getValue<A>();
+                const ColumnConst & c0_const = checkAndGetColumnConst<ColVecA>(*c0);
+                A a = c0_const.template getValue<A>();
                 if (const ColVecB * c1_vec = checkAndGetColumn<ColVecB>(c1.get()))
                     constantVector<scale_left, scale_right>(a, c1_vec->getData(), vec_res, scale);
                 else
                     throw Exception(ErrorCodes::LOGICAL_ERROR, "Wrong column in Decimal comparison");
             }
             else if (c1_is_const)
             {
-                const ColumnConst * c1_const = checkAndGetColumnConst<ColVecB>(c1.get());
-                B b = c1_const->template getValue<B>();
+                const ColumnConst & c1_const = checkAndGetColumnConst<ColVecB>(*c1);
+                B b = c1_const.template getValue<B>();
                 if (const ColVecA * c0_vec = checkAndGetColumn<ColVecA>(c0.get()))
                     vectorConstant<scale_left, scale_right>(c0_vec->getData(), b, vec_res, scale);
                 else

src/DataTypes/ObjectUtils.cpp

@@ -47,7 +47,7 @@ size_t getNumberOfDimensions(const IDataType & type)
 
 size_t getNumberOfDimensions(const IColumn & column)
 {
-    if (const auto * column_array = checkAndGetColumn<ColumnArray>(column))
+    if (const auto * column_array = checkAndGetColumn<ColumnArray>(&column))
         return column_array->getNumberOfDimensions();
     return 0;
 }