From 45ac0b623f797fbd89ba28921be31c10b11bb10d Mon Sep 17 00:00:00 2001
From: daoready <info@daoready.com>
Date: Thu, 23 Aug 2018 21:43:17 +0300
Subject: [PATCH] ArrayDifferenceImpl

---
 dbms/src/Functions/FunctionsHigherOrder.cpp |   1 +
 dbms/src/Functions/FunctionsHigherOrder.h   | 113 ++++++++++++++++++++
 2 files changed, 114 insertions(+)

diff --git a/dbms/src/Functions/FunctionsHigherOrder.cpp b/dbms/src/Functions/FunctionsHigherOrder.cpp
index 555181f0ebc..fbeaeedbcaa 100644
--- a/dbms/src/Functions/FunctionsHigherOrder.cpp
+++ b/dbms/src/Functions/FunctionsHigherOrder.cpp
@@ -18,6 +18,7 @@ void registerFunctionsHigherOrder(FunctionFactory & factory)
     factory.registerFunction<FunctionArrayReverseSort>();
     factory.registerFunction<FunctionArrayCumSum>();
     factory.registerFunction<FunctionArrayCumSumLimited>();
+    factory.registerFunction<FunctionArrayDifference>();
 }
 
 }
diff --git a/dbms/src/Functions/FunctionsHigherOrder.h b/dbms/src/Functions/FunctionsHigherOrder.h
index 39d88517dc6..fbe13dd5ec7 100644
--- a/dbms/src/Functions/FunctionsHigherOrder.h
+++ b/dbms/src/Functions/FunctionsHigherOrder.h
@@ -705,6 +705,117 @@ struct ArrayCumSumImpl
 
 };
 
+struct ArrayDifferenceImpl
+{
+    static bool needBoolean() { return false; }
+    static bool needExpression() { return false; }
+    static bool needOneArray() { return false; }
+
+    static DataTypePtr getReturnType(const DataTypePtr & expression_return, const DataTypePtr & /*array_element*/)
+    {
+        if (checkDataType<DataTypeUInt8>(&*expression_return) ||
+            checkDataType<DataTypeUInt16>(&*expression_return) ||
+            checkDataType<DataTypeUInt32>(&*expression_return) ||
+            checkDataType<DataTypeUInt64>(&*expression_return))
+            return std::make_shared<DataTypeArray>(std::make_shared<DataTypeUInt64>());
+
+        if (checkDataType<DataTypeInt8>(&*expression_return) ||
+            checkDataType<DataTypeInt16>(&*expression_return) ||
+            checkDataType<DataTypeInt32>(&*expression_return) ||
+            checkDataType<DataTypeInt64>(&*expression_return))
+            return std::make_shared<DataTypeArray>(std::make_shared<DataTypeInt64>());
+
+        if (checkDataType<DataTypeFloat32>(&*expression_return) ||
+            checkDataType<DataTypeFloat64>(&*expression_return))
+            return std::make_shared<DataTypeArray>(std::make_shared<DataTypeFloat64>());
+
+        throw Exception("arrayDifference cannot process values of type " + expression_return->getName(), ErrorCodes::ILLEGAL_TYPE_OF_ARGUMENT);
+    }
+
+
+    template <typename Element, typename Result>
+    static bool executeType(const ColumnPtr & mapped, const ColumnArray & array, ColumnPtr & res_ptr)
+    {
+        const ColumnVector<Element> * column = checkAndGetColumn<ColumnVector<Element>>(&*mapped);
+
+        if (!column)
+        {
+            const ColumnConst * column_const = checkAndGetColumnConst<ColumnVector<Element>>(&*mapped);
+
+            if (!column_const)
+                return false;
+
+            const Element x = column_const->template getValue<Element>();
+            const IColumn::Offsets & offsets = array.getOffsets();
+
+            auto res_nested = ColumnVector<Result>::create();
+            typename ColumnVector<Result>::Container & res_values = res_nested->getData();
+            res_values.resize(column_const->size());
+
+            size_t pos = 0;
+            for (size_t i = 0; i < offsets.size(); ++i)
+            {
+                // skip empty arrays
+                if (pos < offsets[i])
+                {
+                    res_values[pos++] = 0;
+                    for (; pos < offsets[i]; ++pos)
+                    {
+                        res_values[pos] = 0;
+                    }
+                }
+            }
+
+            res_ptr = ColumnArray::create(std::move(res_nested), array.getOffsetsPtr());
+            return true;
+        }
+
+        const IColumn::Offsets & offsets = array.getOffsets();
+        const typename ColumnVector<Element>::Container & data = column->getData();
+
+        auto res_nested = ColumnVector<Result>::create();
+        typename ColumnVector<Result>::Container & res_values = res_nested->getData();
+        res_values.resize(data.size());
+
+        size_t pos = 0;
+        for (size_t i = 0; i < offsets.size(); ++i)
+        {
+            // skip empty arrays
+            if (pos < offsets[i])
+            {
+                res_values[pos] = 0;
+                for (++pos; pos < offsets[i]; ++pos)
+                {
+                    res_values[pos] = data[pos] - data[pos - 1];
+                }
+            }
+        }
+        res_ptr = ColumnArray::create(std::move(res_nested), array.getOffsetsPtr());
+        return true;
+
+    }
+
+    static ColumnPtr execute(const ColumnArray & array, ColumnPtr mapped)
+    {
+        ColumnPtr res;
+
+        if (executeType< UInt8 , UInt64>(mapped, array, res) ||
+            executeType< UInt16, UInt64>(mapped, array, res) ||
+            executeType< UInt32, UInt64>(mapped, array, res) ||
+            executeType< UInt64, UInt64>(mapped, array, res) ||
+            executeType<  Int8 ,  Int64>(mapped, array, res) ||
+            executeType<  Int16,  Int64>(mapped, array, res) ||
+            executeType<  Int32,  Int64>(mapped, array, res) ||
+            executeType<  Int64,  Int64>(mapped, array, res) ||
+            executeType<Float32,Float64>(mapped, array, res) ||
+            executeType<Float64,Float64>(mapped, array, res))
+            return res;
+        else
+            throw Exception("Unexpected column for arrayDifference: " + mapped->getName());
+    }
+
+};
+
 
 struct ArrayCumSumLimitedImpl
 {
@@ -1073,6 +1184,7 @@ struct NameArraySort        { static constexpr auto name = "arraySort"; };
 struct NameArrayReverseSort { static constexpr auto name = "arrayReverseSort"; };
 struct NameArrayCumSum      { static constexpr auto name = "arrayCumSum"; };
 struct NameArrayCumSumLimited      { static constexpr auto name = "arrayCumSumLimited"; };
+struct NameArrayDifference     { static constexpr auto name = "arrayDifference"; };
 
 using FunctionArrayMap = FunctionArrayMapped<ArrayMapImpl, NameArrayMap>;
 using FunctionArrayFilter = FunctionArrayMapped<ArrayFilterImpl, NameArrayFilter>;
@@ -1086,5 +1198,6 @@ using FunctionArraySort = FunctionArrayMapped<ArraySortImpl<true>, NameArraySort
 using FunctionArrayReverseSort = FunctionArrayMapped<ArraySortImpl<false>, NameArrayReverseSort>;
 using FunctionArrayCumSum = FunctionArrayMapped<ArrayCumSumImpl, NameArrayCumSum>;
 using FunctionArrayCumSumLimited = FunctionArrayMapped<ArrayCumSumLimitedImpl, NameArrayCumSumLimited>;
+using FunctionArrayDifference = FunctionArrayMapped<ArrayDifferenceImpl, NameArrayDifference>;
 
 }