Add aggregate function combinator Resample

dbms/src/AggregateFunctions/AggregateFunctionFactory.cpp

@@ -77,6 +77,7 @@ AggregateFunctionPtr AggregateFunctionFactory::get(
             throw Exception("Logical error: cannot find aggregate function combinator to apply a function to Nullable arguments.", ErrorCodes::LOGICAL_ERROR);
 
         DataTypes nested_types = combinator->transformArguments(type_without_low_cardinality);
+        Array nested_parameters = combinator->transformParameters(parameters);
 
         AggregateFunctionPtr nested_function;
 
@@ -84,7 +85,7 @@ AggregateFunctionPtr AggregateFunctionFactory::get(
         /// Combinator will check if nested_function was created.
         if (name == "count" || std::none_of(argument_types.begin(), argument_types.end(),
             [](const auto & type) { return type->onlyNull(); }))
-            nested_function = getImpl(name, nested_types, parameters, recursion_level);
+            nested_function = getImpl(name, nested_types, nested_parameters, recursion_level);
 
         return combinator->transformAggregateFunction(nested_function, argument_types, parameters);
     }
@@ -126,7 +127,10 @@ AggregateFunctionPtr AggregateFunctionFactory::getImpl(
 
         String nested_name = name.substr(0, name.size() - combinator->getName().size());
         DataTypes nested_types = combinator->transformArguments(argument_types);
-        AggregateFunctionPtr nested_function = get(nested_name, nested_types, parameters, recursion_level + 1);
+        Array nested_parameters = combinator->transformParameters(parameters);
+
+        AggregateFunctionPtr nested_function = get(nested_name, nested_types, nested_parameters, recursion_level + 1);
+
         return combinator->transformAggregateFunction(nested_function, argument_types, parameters);
     }
 

dbms/src/AggregateFunctions/AggregateFunctionResample.cpp

@@ -0,0 +1,106 @@
+#include <AggregateFunctions/AggregateFunctionResample.h>
+
+#include <AggregateFunctions/AggregateFunctionCombinatorFactory.h>
+
+
+namespace DB
+{
+
+namespace ErrorCodes
+{
+    extern const int NUMBER_OF_ARGUMENTS_DOESNT_MATCH;
+}
+
+class AggregateFunctionCombinatorResample final : public
+    IAggregateFunctionCombinator
+{
+public:
+    String getName() const override {
+        return "Resample";
+    }
+
+    DataTypes transformArguments(const DataTypes & arguments) const override
+    {
+        if (arguments.empty())
+            throw Exception {
+                "Incorrect number of arguments for aggregate function with "
+                    + getName() + " suffix",
+                ErrorCodes::NUMBER_OF_ARGUMENTS_DOESNT_MATCH
+            };
+
+        return DataTypes(arguments.begin(), arguments.end() - 1);
+    }
+
+    Array transformParameters(const Array & params) const override
+    {
+        if (params.size() < 3)
+            throw Exception {
+                "Incorrect number of parameters for aggregate function with "
+                    + getName() + " suffix",
+                ErrorCodes::NUMBER_OF_ARGUMENTS_DOESNT_MATCH
+            };
+
+        return Array(params.begin(), params.end() - 3);
+    }
+
+    AggregateFunctionPtr transformAggregateFunction(
+        const AggregateFunctionPtr & nested_function,
+        const DataTypes & arguments,
+        const Array & params
+    ) const override
+    {
+        for (const Field & param : params)
+        {
+            if (
+                param.getType() != Field::Types::UInt64
+                && param.getType() != Field::Types::Int64
+            )
+                return nullptr;
+        }
+
+        WhichDataType which {
+            arguments.back()
+        };
+
+        if (
+            which.isNativeUInt()
+            || which.isDateOrDateTime()
+        )
+            return std::make_shared<AggregateFunctionResample<UInt64>>(
+                nested_function,
+                params.front().get<UInt64>(),
+                params[1].get<UInt64>(),
+                params.back().get<UInt64>(),
+                arguments,
+                params
+            );
+
+        if (
+            which.isNativeInt()
+            || which.isEnum()
+            || which.isInterval()
+        )
+            return std::make_shared<AggregateFunctionResample<Int64>>(
+                nested_function,
+                params.front().get<Int64>(),
+                params[1].get<Int64>(),
+                params.back().get<Int64>(),
+                arguments,
+                params
+            );
+
+        // TODO
+        return nullptr;
+    }
+};
+
+void registerAggregateFunctionCombinatorResample(
+    AggregateFunctionCombinatorFactory & factory
+)
+{
+    factory.registerCombinator(
+        std::make_shared<AggregateFunctionCombinatorResample>()
+    );
+}
+
+}

dbms/src/AggregateFunctions/AggregateFunctionResample.h

@@ -0,0 +1,182 @@
+#pragma once
+
+#include <AggregateFunctions/IAggregateFunction.h>
+#include <Columns/ColumnArray.h>
+#include <DataTypes/DataTypeArray.h>
+
+
+namespace DB
+{
+
+template <typename Key>
+class AggregateFunctionResample final : public IAggregateFunctionHelper<
+    AggregateFunctionResample<Key>
+>
+{
+private:
+    AggregateFunctionPtr nested_function;
+
+    size_t last_col;
+
+    Key begin;
+    Key end;
+    Key step;
+
+    size_t total;
+    size_t aod;
+    size_t sod;
+
+public:
+    AggregateFunctionResample(
+        AggregateFunctionPtr nested_function,
+        Key begin,
+        Key end,
+        Key step,
+        const DataTypes & arguments,
+        const Array & params
+    ) :
+        IAggregateFunctionHelper<
+            AggregateFunctionResample<Key>
+        > {arguments, params},
+        nested_function {nested_function},
+        last_col {arguments.size() - 1},
+        begin {begin},
+        end {end},
+        step {step},
+        total {
+            static_cast<size_t>(
+                (end - begin + step - (step >= 0 ? 1 : -1)) / step
+            )
+        },
+        aod {nested_function->alignOfData()},
+        sod {(nested_function->sizeOfData() + aod - 1) / aod * aod}
+    {
+        // notice: argument types has been checked before
+    }
+
+    String getName() const override
+    {
+        return nested_function->getName() + "Resample";
+    }
+
+    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 total * sod;
+    }
+
+    size_t alignOfData() const override
+    {
+        return aod;
+    }
+
+    void create(AggregateDataPtr place) const override
+    {
+        for (size_t i = 0; i < total; ++i)
+            nested_function->create(place + i * sod);
+    }
+
+    void destroy(AggregateDataPtr place) const noexcept override
+    {
+        for (size_t i = 0; i < total; ++i)
+            nested_function->destroy(place + i * sod);
+    }
+
+    void add(
+        AggregateDataPtr place,
+        const IColumn ** columns,
+        size_t row_num,
+        Arena * arena
+    ) const override
+    {
+        // Key key {
+        //     static_cast<const ColumnVector<Key> *>(
+        //         columns[last_col]
+        //     )->getData()[row_num]
+        // };
+        Key key;
+
+        if constexpr (static_cast<Key>(-1) < 0)
+            key = columns[last_col]->getInt(row_num);
+        else
+            key = columns[last_col]->getUInt(row_num);
+
+        size_t pos = (key - begin) / step;
+
+        if (pos >= 0 && pos < total)
+            nested_function->add(place + pos * sod, columns, row_num, arena);
+    }
+
+    void merge(
+        AggregateDataPtr place,
+        ConstAggregateDataPtr rhs,
+        Arena * arena
+    ) const override
+    {
+        for (size_t i = 0; i < total; ++i)
+            nested_function->merge(place + i * sod, rhs + i * sod, arena);
+    }
+
+    void serialize(
+        ConstAggregateDataPtr place,
+        WriteBuffer & buf
+    ) const override
+    {
+        for (size_t i = 0; i < total; ++i)
+            nested_function->serialize(place + i * sod, buf);
+    }
+
+    void deserialize(
+        AggregateDataPtr place,
+        ReadBuffer & buf,
+        Arena * arena
+    ) const override
+    {
+        for (size_t i = 0; i < total; ++i)
+            nested_function->deserialize(place + i * sod, buf, arena);
+    }
+
+    DataTypePtr getReturnType() const override
+    {
+        return std::make_shared<DataTypeArray>(
+            nested_function->getReturnType()
+        );
+    }
+
+    void insertResultInto(
+        ConstAggregateDataPtr place,
+        IColumn & to
+    ) const override
+    {
+        auto & col = static_cast<ColumnArray &>(to);
+        auto & col_offsets = static_cast<ColumnArray::ColumnOffsets &>(
+            col.getOffsetsColumn()
+        );
+
+        for (size_t i = 0; i < total; ++i)
+            nested_function->insertResultInto(place + i * sod, col.getData());
+
+        col_offsets.getData().push_back(col.getData().size());
+    }
+};
+
+}

dbms/src/AggregateFunctions/IAggregateFunctionCombinator.h

@@ -38,7 +38,19 @@ public:
       *  get the arguments for nested function (ex: UInt64 for sum).
       * If arguments are not suitable for combined function, throw an exception.
       */
-    virtual DataTypes transformArguments(const DataTypes & arguments) const = 0;
+    virtual DataTypes transformArguments(const DataTypes & arguments) const
+    {
+        return arguments;
+    }
+
+    /** From the parameters for combined function,
+      *  get the parameters for nested function.
+      * If arguments are not suitable for combined function, throw an exception.
+      */
+    virtual Array transformParameters(const Array & parameters) const
+    {
+        return parameters;
+    }
 
     /** Create combined aggregate function (ex: sumIf)
       *  from nested function (ex: sum)

dbms/src/AggregateFunctions/registerAggregateFunctions.cpp

@@ -38,6 +38,7 @@ void registerAggregateFunctionCombinatorForEach(AggregateFunctionCombinatorFacto
 void registerAggregateFunctionCombinatorState(AggregateFunctionCombinatorFactory &);
 void registerAggregateFunctionCombinatorMerge(AggregateFunctionCombinatorFactory &);
 void registerAggregateFunctionCombinatorNull(AggregateFunctionCombinatorFactory &);
+void registerAggregateFunctionCombinatorResample(AggregateFunctionCombinatorFactory &);
 
 void registerAggregateFunctionHistogram(AggregateFunctionFactory & factory);
 void registerAggregateFunctionRetention(AggregateFunctionFactory & factory);
@@ -85,6 +86,7 @@ void registerAggregateFunctions()
         registerAggregateFunctionCombinatorState(factory);
         registerAggregateFunctionCombinatorMerge(factory);
         registerAggregateFunctionCombinatorNull(factory);
+        registerAggregateFunctionCombinatorResample(factory);
     }
 }