Add aggregate function combinators: -OrNull & -OrDefault

dbms/src/AggregateFunctions/AggregateFunctionOrFill.cpp

@@ -0,0 +1,39 @@
+#include <AggregateFunctions/AggregateFunctionOrFill.h>
+
+#include <AggregateFunctions/AggregateFunctionCombinatorFactory.h>
+
+
+namespace DB
+{
+
+template <bool UseNull>
+class AggregateFunctionCombinatorOrFill final : public IAggregateFunctionCombinator
+{
+public:
+    String getName() const override
+    {
+        if constexpr (UseNull)
+            return "OrNull";
+        else
+            return "OrDefault";
+    }
+
+    AggregateFunctionPtr transformAggregateFunction(
+        const AggregateFunctionPtr & nested_function,
+        const DataTypes & arguments,
+        const Array & params) const override
+    {
+        return std::make_shared<AggregateFunctionOrFill<UseNull>>(
+            nested_function,
+            arguments,
+            params);
+    }
+};
+
+void registerAggregateFunctionCombinatorOrFill(AggregateFunctionCombinatorFactory & factory)
+{
+    factory.registerCombinator(std::make_shared<AggregateFunctionCombinatorOrFill<false>>());
+    factory.registerCombinator(std::make_shared<AggregateFunctionCombinatorOrFill<true>>());
+}
+
+}

dbms/src/AggregateFunctions/AggregateFunctionOrFill.h

@@ -0,0 +1,179 @@
+#pragma once
+
+#include <AggregateFunctions/IAggregateFunction.h>
+#include <Columns/ColumnNullable.h>
+#include <Common/typeid_cast.h>
+#include <DataTypes/DataTypeNullable.h>
+
+
+namespace DB
+{
+
+namespace ErrorCodes
+{
+    extern const int ARGUMENT_OUT_OF_BOUND;
+}
+
+/**
+  * -OrDefault and -OrNull combinators for aggregate functions.
+  * If there are no input values, return NULL or a default value, accordingly.
+  * Use a single additional byte of data after the nested function data:
+  * 0 means there was no input, 1 means there was some.
+  */
+template <bool UseNull>
+class AggregateFunctionOrFill final : public IAggregateFunctionHelper<AggregateFunctionOrFill<UseNull>>
+{
+private:
+    AggregateFunctionPtr nested_function;
+
+    size_t size_of_data;
+    DataTypePtr inner_type;
+    bool inner_nullable;
+
+public:
+    AggregateFunctionOrFill(AggregateFunctionPtr nested_function_, const DataTypes & arguments, const Array & params)
+        : IAggregateFunctionHelper<AggregateFunctionOrFill>{arguments, params}
+        , nested_function{nested_function_}
+        , size_of_data {nested_function->sizeOfData()}
+        , inner_type {nested_function->getReturnType()}
+        , inner_nullable {inner_type->isNullable()}
+    {
+        // nothing
+    }
+
+    String getName() const override
+    {
+        if constexpr (UseNull)
+            return nested_function->getName() + "OrNull";
+        else
+            return nested_function->getName() + "OrDefault";
+    }
+
+    const char * getHeaderFilePath() const override
+    {
+        return __FILE__;
+    }
+
+    bool isState() const override
+    {
+        return nested_function->isState();
+    }
+
+    bool allocatesMemoryInArena() const override
+    {
+        return nested_function->allocatesMemoryInArena();
+    }
+
+    bool hasTrivialDestructor() const override
+    {
+        return nested_function->hasTrivialDestructor();
+    }
+
+    size_t sizeOfData() const override
+    {
+        return size_of_data + sizeof(char);
+    }
+
+    size_t alignOfData() const override
+    {
+        return nested_function->alignOfData();
+    }
+
+    void create(AggregateDataPtr place) const override
+    {
+        nested_function->create(place);
+
+        place[size_of_data] = 0;
+    }
+
+    void destroy(AggregateDataPtr place) const noexcept override
+    {
+        nested_function->destroy(place);
+    }
+
+    void add(
+        AggregateDataPtr place,
+        const IColumn ** columns,
+        size_t row_num,
+        Arena * arena) const override
+    {
+        nested_function->add(place, columns, row_num, arena);
+
+        place[size_of_data] = 1;
+    }
+
+    void merge(
+        AggregateDataPtr place,
+        ConstAggregateDataPtr rhs,
+        Arena * arena) const override
+    {
+        nested_function->merge(place, rhs, arena);
+    }
+
+    void serialize(
+        ConstAggregateDataPtr place,
+        WriteBuffer & buf) const override
+    {
+        nested_function->serialize(place, buf);
+    }
+
+    void deserialize(
+        AggregateDataPtr place,
+        ReadBuffer & buf,
+        Arena * arena) const override
+    {
+        nested_function->deserialize(place, buf, arena);
+    }
+
+    DataTypePtr getReturnType() const override
+    {
+        if constexpr (UseNull)
+        {
+            // -OrNull
+
+            if (inner_nullable)
+                return inner_type;
+
+            return std::make_shared<DataTypeNullable>(inner_type);
+        }
+        else
+        {
+            // -OrDefault
+
+            return inner_type;
+        }
+    }
+
+    void insertResultInto(
+        ConstAggregateDataPtr place,
+        IColumn & to) const override
+    {
+        if (place[size_of_data])
+        {
+            if constexpr (UseNull)
+            {
+                // -OrNull
+
+                if (inner_nullable)
+                    nested_function->insertResultInto(place, to);
+                else
+                {
+                    ColumnNullable & col = typeid_cast<ColumnNullable &>(to);
+
+                    col.getNullMapColumn().insertDefault();
+                    nested_function->insertResultInto(place, col.getNestedColumn());
+                }
+            }
+            else
+            {
+                // -OrDefault
+
+                nested_function->insertResultInto(place, to);
+            }
+        }
+        else
+            to.insertDefault();
+    }
+};
+
+}

dbms/src/AggregateFunctions/AggregateFunctionResample.h

@@ -29,8 +29,8 @@ class AggregateFunctionResample final : public IAggregateFunctionHelper<Aggregat
     size_t step;
 
     size_t total;
-    size_t aod;
-    size_t sod;
+    size_t align_of_data;
+    size_t size_of_data;
 
 public:
     AggregateFunctionResample(
@@ -47,8 +47,8 @@ class AggregateFunctionResample final : public IAggregateFunctionHelper<Aggregat
         , end{end_}
         , step{step_}
         , total{0}
-        , aod{nested_function->alignOfData()}
-        , sod{(nested_function->sizeOfData() + aod - 1) / aod * aod}
+        , align_of_data{nested_function->alignOfData()}
+        , size_of_data{(nested_function->sizeOfData() + align_of_data - 1) / align_of_data * align_of_data}
     {
         // notice: argument types has been checked before
         if (step == 0)
@@ -94,24 +94,24 @@ class AggregateFunctionResample final : public IAggregateFunctionHelper<Aggregat
 
     size_t sizeOfData() const override
     {
-        return total * sod;
+        return total * size_of_data;
     }
 
     size_t alignOfData() const override
     {
-        return aod;
+        return align_of_data;
     }
 
     void create(AggregateDataPtr place) const override
     {
         for (size_t i = 0; i < total; ++i)
-            nested_function->create(place + i * sod);
+            nested_function->create(place + i * size_of_data);
     }
 
     void destroy(AggregateDataPtr place) const noexcept override
     {
         for (size_t i = 0; i < total; ++i)
-            nested_function->destroy(place + i * sod);
+            nested_function->destroy(place + i * size_of_data);
     }
 
     void add(
@@ -132,7 +132,7 @@ class AggregateFunctionResample final : public IAggregateFunctionHelper<Aggregat
 
         size_t pos = (key - begin) / step;
 
-        nested_function->add(place + pos * sod, columns, row_num, arena);
+        nested_function->add(place + pos * size_of_data, columns, row_num, arena);
     }
 
     void merge(
@@ -141,15 +141,15 @@ class AggregateFunctionResample final : public IAggregateFunctionHelper<Aggregat
         Arena * arena) const override
     {
         for (size_t i = 0; i < total; ++i)
-            nested_function->merge(place + i * sod, rhs + i * sod, arena);
+            nested_function->merge(place + i * size_of_data, rhs + i * size_of_data, arena);
     }
 
     void serialize(
         ConstAggregateDataPtr place,
         WriteBuffer & buf) const override
     {
         for (size_t i = 0; i < total; ++i)
-            nested_function->serialize(place + i * sod, buf);
+            nested_function->serialize(place + i * size_of_data, buf);
     }
 
     void deserialize(
@@ -158,7 +158,7 @@ class AggregateFunctionResample final : public IAggregateFunctionHelper<Aggregat
         Arena * arena) const override
     {
         for (size_t i = 0; i < total; ++i)
-            nested_function->deserialize(place + i * sod, buf, arena);
+            nested_function->deserialize(place + i * size_of_data, buf, arena);
     }
 
     DataTypePtr getReturnType() const override
@@ -174,7 +174,7 @@ class AggregateFunctionResample final : public IAggregateFunctionHelper<Aggregat
         auto & col_offsets = assert_cast<ColumnArray::ColumnOffsets &>(col.getOffsetsColumn());
 
         for (size_t i = 0; i < total; ++i)
-            nested_function->insertResultInto(place + i * sod, col.getData());
+            nested_function->insertResultInto(place + i * size_of_data, col.getData());
 
         col_offsets.getData().push_back(col.getData().size());
     }

dbms/src/AggregateFunctions/registerAggregateFunctions.cpp

@@ -42,6 +42,7 @@ void registerAggregateFunctionCombinatorForEach(AggregateFunctionCombinatorFacto
 void registerAggregateFunctionCombinatorState(AggregateFunctionCombinatorFactory &);
 void registerAggregateFunctionCombinatorMerge(AggregateFunctionCombinatorFactory &);
 void registerAggregateFunctionCombinatorNull(AggregateFunctionCombinatorFactory &);
+void registerAggregateFunctionCombinatorOrFill(AggregateFunctionCombinatorFactory &);
 void registerAggregateFunctionCombinatorResample(AggregateFunctionCombinatorFactory &);
 
 void registerAggregateFunctions()
@@ -88,6 +89,7 @@ void registerAggregateFunctions()
         registerAggregateFunctionCombinatorState(factory);
         registerAggregateFunctionCombinatorMerge(factory);
         registerAggregateFunctionCombinatorNull(factory);
+        registerAggregateFunctionCombinatorOrFill(factory);
         registerAggregateFunctionCombinatorResample(factory);
     }
 }

dbms/tests/queries/0_stateless/01018_empty_aggregation_filling.reference

@@ -0,0 +1,48 @@
+--- Int Empty ---
+0
+\N
+0
+\N
+0
+\N
+0
+\N
+0
+\N
+0
+\N
+--- Int Non-empty ---
+1
+1
+nan
+nan
+1
+1
+1
+1
+nan
+nan
+1
+1
+--- Other Types Empty ---
+
+\N
+\N
+\N
+0.00
+\N
+0
+\N
+0.00
+\N
+--- Other Types Non-empty ---
+hello
+hello
+2011-04-05 14:19:19
+2011-04-05 14:19:19
+-123.45
+-123.45
+inf
+inf
+-123.45
+-123.45