Merge pull request #1146 from yandex/array-functions-concat-slice-push-pop

Array functions concat slice push pop

dbms/src/Common/TypeList.h

@@ -0,0 +1,70 @@
+#pragma once
+
+namespace DB
+{
+
+template <typename... TTail>
+struct TypeList
+{
+    static constexpr size_t size = 0;
+
+    template <size_t I>
+    using At = std::nullptr_t;
+
+    template <typename Func, size_t index = 0>
+    static void forEach(Func && func)
+    {
+    }
+};
+
+
+template <typename THead, typename... TTail>
+struct TypeList<THead, TTail...>
+{
+    using Head = THead;
+    using Tail = TypeList<TTail...>;
+
+    static constexpr size_t size = 1 + sizeof...(TTail);
+
+    template <size_t I>
+    using At = typename std::template conditional<I == 0, Head, typename Tail::template At<I - 1>>::type;
+
+    template <typename Func, size_t index = 0>
+    static void ALWAYS_INLINE forEach(Func && func)
+    {
+        func.template operator()<Head, index>();
+        Tail::template forEach<Func, index + 1>(std::forward<Func>(func));
+    }
+};
+
+/// Append Type to TypeList
+/// Usage:
+///     using TypeListWithType = typename AppendToTypeList<Type, ConcreteTypeList>::Type;
+template <typename TypeToPrepend, typename List, typename ... Types>
+struct PrependToTypeList
+{
+    using Type = typename PrependToTypeList<TypeToPrepend, typename List::Tail, Types ..., typename List::Head>::Type;
+};
+
+template <typename TypeToPrepend, typename ... Types>
+struct PrependToTypeList<TypeToPrepend, TypeList<>, Types ...>
+{
+    using Type = TypeList<TypeToPrepend, Types ...>;
+};
+
+/// Apply TypeList as variadic template argument of Class.
+/// Usage:
+///     using ClassWithAppliedTypeList = typename ApplyTypeListForClass<Class, ConcreteTypeList>::Type;
+template <template <typename ...> typename Class, typename List, typename ... Types>
+struct ApplyTypeListForClass
+{
+    using Type = typename ApplyTypeListForClass<Class, typename List::Tail, Types ..., typename List::Head>::Type;
+};
+
+template <template <typename ...> typename Class, typename ... Types>
+struct ApplyTypeListForClass<Class, TypeList<>, Types ...>
+{
+    using Type = Class<Types ...>;
+};
+
+}

dbms/src/Core/TypeListNumber.h

@@ -0,0 +1,10 @@
+#pragma once
+#include <Core/Types.h>
+#include <Common/TypeList.h>
+
+namespace DB
+{
+
+using TypeListNumbers = TypeList<UInt8, UInt16, UInt32, UInt64, Int8, Int16, Int32, Int64, Float32, Float64>;
+
+}

dbms/src/Functions/CMakeLists.txt

@@ -52,6 +52,12 @@ generate_function_register(Array
   FunctionRange
   FunctionArrayReduce
   FunctionArrayReverse
+  FunctionArrayConcat
+  FunctionArraySlice
+  FunctionArrayPushBack
+  FunctionArrayPushFront
+  FunctionArrayPopBack
+  FunctionArrayPopFront
 )
 
 

dbms/src/Functions/FunctionsArray.cpp

@@ -7,6 +7,7 @@
 #include <Functions/FunctionsConversion.h>
 #include <Functions/Conditional/getArrayType.h>
 #include <Functions/Conditional/CondException.h>
+#include <Functions/GatherUtils.h>
 #include <Common/HashTable/HashMap.h>
 #include <Common/HashTable/ClearableHashMap.h>
 #include <Parsers/ExpressionListParsers.h>
@@ -2876,4 +2877,383 @@ void FunctionArrayReduce::executeImpl(Block & block, const ColumnNumbers & argum
     }
 }
 
+/// Implementation of FunctionArrayConcat.
+
+FunctionPtr FunctionArrayConcat::create(const Context & context)
+{
+    return std::make_shared<FunctionArrayConcat>();
+}
+
+String FunctionArrayConcat::getName() const
+{
+    return name;
+}
+
+DataTypePtr FunctionArrayConcat::getReturnTypeImpl(const DataTypes & arguments) const
+{
+    if (arguments.empty())
+        throw Exception{"Function array requires at least one argument.", ErrorCodes::NUMBER_OF_ARGUMENTS_DOESNT_MATCH};
+
+    DataTypes nested_types;
+    nested_types.reserve(arguments.size());
+    for (size_t i : ext::range(0, arguments.size()))
+    {
+        auto argument = arguments[i].get();
+
+        if (auto data_type_array = checkAndGetDataType<DataTypeArray>(argument))
+            nested_types.push_back(data_type_array->getNestedType());
+        else
+            throw Exception(
+                    "Argument " + toString(i) + " for function " + getName() + " must be an array but it has type "
+                    + argument->getName() + ".", ErrorCodes::ILLEGAL_TYPE_OF_ARGUMENT);
+    }
+
+    if (foundNumericType(nested_types))
+    {
+        /// Since we have found at least one numeric argument, we infer that all
+        /// the arguments are numeric up to nullity. Let's determine the least
+        /// common type.
+        auto enriched_result_type = Conditional::getArrayType(nested_types);
+        return std::make_shared<DataTypeArray>(enriched_result_type);
+    }
+    else
+    {
+        /// Otherwise all the arguments must have the same type up to nullability or nullity.
+        if (!hasArrayIdenticalTypes(nested_types))
+            throw Exception{"Arguments for function " + getName() + " must have same type or behave as number.",
+                            ErrorCodes::ILLEGAL_TYPE_OF_ARGUMENT};
+
+        return std::make_shared<DataTypeArray>(getArrayElementType(nested_types));
+    }
+}
+
+void FunctionArrayConcat::executeImpl(Block & block, const ColumnNumbers & arguments, size_t result)
+{
+    auto & result_column = block.getByPosition(result).column;
+    const DataTypePtr & return_type = block.getByPosition(result).type;
+    result_column = return_type->createColumn();
+
+    size_t size = 0;
+
+    for (auto argument : arguments)
+    {
+        const auto & argument_column = block.safeGetByPosition(argument).column;
+        size = argument_column->size();
+    }
+
+    std::vector<std::unique_ptr<IArraySource>> sources;
+
+    for (auto argument : arguments)
+    {
+        auto argument_column = block.getByPosition(argument).column.get();
+        size_t column_size = argument_column->size();
+        bool is_const = false;
+
+        if (auto argument_column_const = typeid_cast<ColumnConst *>(argument_column))
+        {
+            is_const = true;
+            argument_column = &argument_column_const->getDataColumn();
+        }
+
+        if (auto argument_column_array = typeid_cast<ColumnArray *>(argument_column))
+            sources.emplace_back(createArraySource(*argument_column_array, is_const, column_size));
+        else
+            throw Exception{"Arguments for function " + getName() + " must be arrays.", ErrorCodes::LOGICAL_ERROR};
+    }
+
+    auto sink = createArraySink(typeid_cast<ColumnArray &>(*result_column.get()), size);
+    concat(sources, *sink);
+}
+
+
+/// Implementation of FunctionArraySlice.
+
+FunctionPtr FunctionArraySlice::create(const Context & context)
+{
+    return std::make_shared<FunctionArraySlice>();
+}
+
+String FunctionArraySlice::getName() const
+{
+    return name;
+}
+
+DataTypePtr FunctionArraySlice::getReturnTypeImpl(const DataTypes & arguments) const
+{
+    size_t number_of_arguments = arguments.size();
+
+    if (number_of_arguments < 2 || number_of_arguments > 3)
+        throw Exception("Number of arguments for function " + getName() + " doesn't match: passed "
+                        + toString(number_of_arguments) + ", should be 2 or 3",
+                        ErrorCodes::NUMBER_OF_ARGUMENTS_DOESNT_MATCH);
+
+    if (arguments[0]->isNull())
+        return arguments[0];
+
+    auto array_type = typeid_cast<DataTypeArray *>(arguments[0].get());
+    if (!array_type)
+        throw Exception("First argument for function " + getName() + " must be an array but it has type "
+                        + arguments[0]->getName() + ".", ErrorCodes::ILLEGAL_TYPE_OF_ARGUMENT);
+
+    for (size_t i = 1; i < number_of_arguments; ++i)
+    {
+        if (!arguments[i]->isNumeric() && !arguments[i]->isNull())
+            throw Exception(
+                    "Argument " + toString(i) + " for function " + getName() + " must be numeric but it has type "
+                    + arguments[i]->getName() + ".", ErrorCodes::ILLEGAL_TYPE_OF_ARGUMENT);
+    }
+
+    return arguments[0];
+}
+
+void FunctionArraySlice::executeImpl(Block & block, const ColumnNumbers & arguments, size_t result)
+{
+    auto & result_colum_with_name_and_type = block.getByPosition(result);
+    auto & result_column = result_colum_with_name_and_type.column;
+    auto & return_type = result_colum_with_name_and_type.type;
+    result_column = return_type->createColumn();
+
+    auto array_column = block.getByPosition(arguments[0]).column;
+    auto offset_column = block.getByPosition(arguments[1]).column;
+    auto length_column = arguments.size() > 2 ? block.getByPosition(arguments[2]).column : nullptr;
+
+    if (return_type->isNull())
+    {
+        result_column = array_column->clone();
+        return;
+    }
+
+    std::unique_ptr<IArraySource> source;
+
+    size_t size = array_column->size();
+    bool is_const = false;
+
+    if (auto const_array_column = typeid_cast<ColumnConst *>(array_column.get()))
+    {
+        is_const = true;
+        array_column = const_array_column->getDataColumnPtr();
+    }
+
+    if (auto argument_column_array = typeid_cast<ColumnArray *>(array_column.get()))
+        source = createArraySource(*argument_column_array, is_const, size);
+    else
+        throw Exception{"First arguments for function " + getName() + " must be array.", ErrorCodes::LOGICAL_ERROR};
+
+    auto sink = createArraySink(typeid_cast<ColumnArray &>(*result_column), size);
+
+    if (offset_column->isNull())
+    {
+        if (!length_column || length_column->isNull())
+            result_column = array_column->clone();
+        else if (length_column->isConst())
+            sliceFromLeftConstantOffsetBounded(*source, *sink, 0, length_column->getInt(0));
+        else
+        {
+            auto const_offset_column = std::make_shared<ColumnConst>(std::make_shared<ColumnInt8>(1, 1), size);
+            sliceDynamicOffsetBounded(*source, *sink, *const_offset_column, *length_column);
+        }
+    }
+    else if (offset_column->isConst())
+    {
+        ssize_t offset = offset_column->getUInt(0);
+
+        if (!length_column || length_column->isNull())
+        {
+            if (offset > 0)
+                sliceFromLeftConstantOffsetUnbounded(*source, *sink, static_cast<size_t>(offset - 1));
+            else
+                sliceFromRightConstantOffsetUnbounded(*source, *sink, static_cast<size_t>(-offset));
+        }
+        else if (length_column->isConst())
+        {
+            ssize_t length = length_column->getInt(0);
+            if (offset > 0)
+                sliceFromLeftConstantOffsetBounded(*source, *sink, static_cast<size_t>(offset - 1), length);
+            else
+                sliceFromRightConstantOffsetBounded(*source, *sink, static_cast<size_t>(-offset), length);
+        }
+        else
+            sliceDynamicOffsetBounded(*source, *sink, *offset_column, *length_column);
+    }
+    else
+    {
+        if (!length_column || length_column->isNull())
+            sliceDynamicOffsetUnbounded(*source, *sink, *offset_column);
+        else
+            sliceDynamicOffsetBounded(*source, *sink, *offset_column, *length_column);
+    }
+}
+
+
+/// Implementation of FunctionArrayPush.
+
+DataTypePtr FunctionArrayPush::getReturnTypeImpl(const DataTypes & arguments) const
+{
+    if (arguments[0]->isNull())
+        return arguments[0];
+
+    auto array_type = typeid_cast<DataTypeArray *>(arguments[0].get());
+    if (!array_type)
+        throw Exception("First argument for function " + getName() + " must be an array but it has type "
+                        + arguments[0]->getName() + ".", ErrorCodes::ILLEGAL_TYPE_OF_ARGUMENT);
+
+    auto nested_type = array_type->getNestedType();
+
+    DataTypes types = {nested_type, arguments[1]};
+
+    if (foundNumericType(types))
+    {
+        /// Since we have found at least one numeric argument, we infer that all
+        /// the arguments are numeric up to nullity. Let's determine the least
+        /// common type.
+        auto enriched_result_type = Conditional::getArrayType(types);
+        return std::make_shared<DataTypeArray>(enriched_result_type);
+    }
+    else
+    {
+        /// Otherwise all the arguments must have the same type up to nullability or nullity.
+        if (!hasArrayIdenticalTypes(types))
+            throw Exception{"Arguments for function " + getName() + " must have same type or behave as number.",
+                            ErrorCodes::ILLEGAL_TYPE_OF_ARGUMENT};
+
+        return std::make_shared<DataTypeArray>(getArrayElementType(types));
+    }
+}
+
+void FunctionArrayPush::executeImpl(Block & block, const ColumnNumbers & arguments, size_t result)
+{
+    auto & result_colum_with_name_and_type = block.getByPosition(result);
+    auto & result_column = result_colum_with_name_and_type.column;
+    auto & return_type = result_colum_with_name_and_type.type;
+    result_column = return_type->createColumn();
+
+    auto array_column = block.getByPosition(arguments[0]).column;
+    auto appended_column = block.getByPosition(arguments[1]).column;
+
+    if (return_type->isNull())
+    {
+        result_column = array_column->clone();
+        return;
+    }
+
+    std::vector<std::unique_ptr<IArraySource>> sources;
+
+    size_t size = array_column->size();
+    bool is_const = false;
+
+    if (auto const_array_column = typeid_cast<ColumnConst *>(array_column.get()))
+    {
+        is_const = true;
+        array_column = const_array_column->getDataColumnPtr();
+    }
+
+    if (auto argument_column_array = typeid_cast<ColumnArray *>(array_column.get()))
+        sources.push_back(createArraySource(*argument_column_array, is_const, size));
+    else
+        throw Exception{"First arguments for function " + getName() + " must be array.", ErrorCodes::LOGICAL_ERROR};
+
+
+    bool is_appended_const = false;
+    if (auto const_appended_column = typeid_cast<ColumnConst *>(appended_column.get()))
+    {
+        is_appended_const = true;
+        appended_column = const_appended_column->getDataColumnPtr();
+    }
+
+    auto offsets = std::make_shared<ColumnArray::ColumnOffsets_t>(appended_column->size());
+    for (size_t i : ext::range(0, offsets->size()))
+        offsets->getElement(i) = i + 1;
+
+    ColumnArray appended_array_column(appended_column, offsets);
+    sources.push_back(createArraySource(appended_array_column, is_appended_const, size));
+
+    auto sink = createArraySink(typeid_cast<ColumnArray &>(*result_column), size);
+
+    if (push_front)
+        sources[0].swap(sources[1]);
+    concat(sources, *sink);
+}
+
+/// Implementation of FunctionArrayPushFront.
+
+FunctionPtr FunctionArrayPushFront::create(const Context & context)
+{
+    return std::make_shared<FunctionArrayPushFront>();
+}
+
+/// Implementation of FunctionArrayPushBack.
+
+FunctionPtr FunctionArrayPushBack::create(const Context & context)
+{
+    return std::make_shared<FunctionArrayPushBack>();
+}
+
+/// Implementation of FunctionArrayPop.
+
+DataTypePtr FunctionArrayPop::getReturnTypeImpl(const DataTypes & arguments) const
+{
+    if (arguments[0]->isNull())
+        return arguments[0];
+
+    auto array_type = typeid_cast<DataTypeArray *>(arguments[0].get());
+    if (!array_type)
+        throw Exception("First argument for function " + getName() + " must be an array but it has type "
+                        + arguments[0]->getName() + ".", ErrorCodes::ILLEGAL_TYPE_OF_ARGUMENT);
+
+    return arguments[0];
+}
+
+void FunctionArrayPop::executeImpl(Block & block, const ColumnNumbers & arguments, size_t result)
+{
+    auto & result_colum_with_name_and_type = block.getByPosition(result);
+    auto & result_column = result_colum_with_name_and_type.column;
+    auto & return_type = result_colum_with_name_and_type.type;
+    result_column = return_type->createColumn();
+
+    auto array_column = block.getByPosition(arguments[0]).column;
+
+    if (return_type->isNull())
+    {
+        result_column = array_column->clone();
+        return;
+    }
+
+    std::unique_ptr<IArraySource> source;
+
+    size_t size = array_column->size();
+    bool is_const = false;
+
+    if (auto const_array_column = typeid_cast<ColumnConst *>(array_column.get()))
+    {
+        is_const = true;
+        array_column = const_array_column->getDataColumnPtr();
+    }
+
+    if (auto argument_column_array = typeid_cast<ColumnArray *>(array_column.get()))
+        source = createArraySource(*argument_column_array, is_const, size);
+    else
+        throw Exception{"First arguments for function " + getName() + " must be array.", ErrorCodes::LOGICAL_ERROR};
+
+    auto sink = createArraySink(typeid_cast<ColumnArray &>(*result_column), size);
+
+    if (pop_front)
+        sliceFromLeftConstantOffsetUnbounded(*source, *sink, 1);
+    else
+        sliceFromLeftConstantOffsetBounded(*source, *sink, 0, -1);
+}
+
+/// Implementation of FunctionArrayPopFront.
+
+FunctionPtr FunctionArrayPopFront::create(const Context &context)
+{
+    return std::make_shared<FunctionArrayPopFront>();
+}
+
+/// Implementation of FunctionArrayPopBack.
+
+FunctionPtr FunctionArrayPopBack::create(const Context &context)
+{
+    return std::make_shared<FunctionArrayPopBack>();
+}
+
 }

dbms/src/Functions/FunctionsArray.h

@@ -61,6 +61,22 @@ namespace ErrorCodes
   *
   * arrayReduce('agg', arr1, ...) - apply the aggregate function `agg` to arrays `arr1...`
   *  If multiple arrays passed, then elements on corresponding positions are passed as multiple arguments to the aggregate function.
+  *
+  *  arrayConcat(arr1, ...) - concatenate arrays.
+  *
+  *  arraySlice(arr, offset, length) - make slice of array. Offsets and length may be < 0 or Null
+  *   - if offset < 0, indexation from right element
+  *   - if length < 0, length = len(array) - (positive_index(offset) - 1) + length
+  *   indexation:
+  *     [ 1,  2,  3,  4,  5,  6]
+  *     [-6, -5, -4, -3, -2, -1]
+  *   examples:
+  *     arraySlice([1, 2, 3, 4, 5, 6], -4, 2) -> [3, 4]
+  *     arraySlice([1, 2, 3, 4, 5, 6], 2, -1) -> [2, 3, 4, 5] (6 - (2 - 1) + (-1) = 4)
+  *     arraySlice([1, 2, 3, 4, 5, 6], -5, -1) = arraySlice([1, 2, 3, 4, 5, 6], 2, -1) -> [2, 3, 4, 5]
+  *
+  *  arrayPushBack(arr, x), arrayPushFront(arr, x)
+  *  arrayPopBack(arr), arrayPopFront(arr)
   */
 
 
@@ -1406,6 +1422,130 @@ private:
 };
 
 
+class FunctionArrayConcat : public IFunction
+{
+public:
+    static constexpr auto name = "arrayConcat";
+    static FunctionPtr create(const Context & context);
+
+    String getName() const override;
+
+    bool isVariadic() const override { return true; }
+    size_t getNumberOfArguments() const override { return 0; }
+
+    DataTypePtr getReturnTypeImpl(const DataTypes & arguments) const override;
+
+    void executeImpl(Block & block, const ColumnNumbers & arguments, size_t result) override;
+
+    bool useDefaultImplementationForConstants() const override { return true; }
+};
+
+
+class FunctionArraySlice : public IFunction
+{
+public:
+    static constexpr auto name = "arraySlice";
+    static FunctionPtr create(const Context & context);
+
+    String getName() const override;
+
+    bool isVariadic() const override { return true; }
+    size_t getNumberOfArguments() const override { return 0; }
+
+    DataTypePtr getReturnTypeImpl(const DataTypes & arguments) const override;
+
+    void executeImpl(Block & block, const ColumnNumbers & arguments, size_t result) override;
+
+    bool useDefaultImplementationForConstants() const override { return true; }
+    bool useDefaultImplementationForNulls() const override { return false; }
+};
+
+
+class FunctionArrayPush : public IFunction
+{
+public:
+    FunctionArrayPush(bool push_front, const char * name) : push_front(push_front), name(name) {}
+
+    String getName() const override { return name; }
+
+    bool isVariadic() const override { return false; }
+    size_t getNumberOfArguments() const override { return 2; }
+
+    DataTypePtr getReturnTypeImpl(const DataTypes & arguments) const override;
+
+    void executeImpl(Block & block, const ColumnNumbers & arguments, size_t result) override;
+
+    bool useDefaultImplementationForConstants() const override { return true; }
+    bool useDefaultImplementationForNulls() const override { return false; }
+
+private:
+    bool push_front;
+    const char * name;
+};
+
+class FunctionArrayPushFront : public FunctionArrayPush
+{
+public:
+    static constexpr auto name = "arrayPushFront";
+
+    static FunctionPtr create(const Context & context);
+
+    FunctionArrayPushFront() : FunctionArrayPush(true, name) {}
+};
+
+class FunctionArrayPushBack : public FunctionArrayPush
+{
+public:
+    static constexpr auto name = "arrayPushBack";
+
+    static FunctionPtr create(const Context & context);
+
+    FunctionArrayPushBack() : FunctionArrayPush(false, name) {}
+};
+
+class FunctionArrayPop : public IFunction
+{
+public:
+    FunctionArrayPop(bool pop_front, const char * name) : pop_front(pop_front), name(name) {}
+
+    String getName() const override { return name; }
+
+    bool isVariadic() const override { return false; }
+    size_t getNumberOfArguments() const override { return 1; }
+
+    DataTypePtr getReturnTypeImpl(const DataTypes & arguments) const override;
+
+    void executeImpl(Block & block, const ColumnNumbers & arguments, size_t result) override;
+
+    bool useDefaultImplementationForConstants() const override { return true; }
+    bool useDefaultImplementationForNulls() const override { return false; }
+
+private:
+    bool pop_front;
+    const char * name;
+};
+
+class FunctionArrayPopFront : public FunctionArrayPop
+{
+public:
+    static constexpr auto name = "arrayPopFront";
+
+    static FunctionPtr create(const Context & context);
+
+    FunctionArrayPopFront() : FunctionArrayPop(true, name) {}
+};
+
+class FunctionArrayPopBack : public FunctionArrayPop
+{
+public:
+    static constexpr auto name = "arrayPopBack";
+
+    static FunctionPtr create(const Context & context);
+
+    FunctionArrayPopBack() : FunctionArrayPop(false, name) {}
+};
+
+
 struct NameHas { static constexpr auto name = "has"; };
 struct NameIndexOf { static constexpr auto name = "indexOf"; };
 struct NameCountEqual { static constexpr auto name = "countEqual"; };

dbms/src/Functions/FunctionsString.cpp

@@ -683,6 +683,9 @@ public:
 
     DataTypePtr getReturnTypeImpl(const DataTypes & arguments) const override
     {
+        if (!is_injective && !arguments.empty() && checkDataType<DataTypeArray>(arguments[0].get()))
+            return FunctionArrayConcat().getReturnTypeImpl(arguments);
+
         if (arguments.size() < 2)
             throw Exception("Number of arguments for function " + getName() + " doesn't match: passed " + toString(arguments.size())
                 + ", should be at least 2.",
@@ -702,6 +705,9 @@ public:
 
     void executeImpl(Block & block, const ColumnNumbers & arguments, const size_t result) override
     {
+        if (!is_injective && !arguments.empty() && checkDataType<DataTypeArray>(block.getByPosition(0).type.get()))
+            return FunctionArrayConcat().executeImpl(block, arguments, result);
+
         if (arguments.size() == 2)
             executeBinary(block, arguments, result);
         else

dbms/src/Functions/GatherUtils.cpp

@@ -0,0 +1,502 @@
+#include <Functions/GatherUtils.h>
+
+namespace DB
+{
+
+/// Creates IArraySource from ColumnArray
+
+template <typename ... Types>
+struct ArraySourceCreator;
+
+template <typename Type, typename ... Types>
+struct ArraySourceCreator<Type, Types ...>
+{
+    static std::unique_ptr<IArraySource>
+    create(const ColumnArray & col, const ColumnUInt8 * null_map, bool is_const, size_t total_rows)
+    {
+        if (typeid_cast<const ColumnVector<Type> *>(&col.getData()))
+        {
+            if (null_map)
+            {
+                if (is_const)
+                    return std::make_unique<ConstSource<NullableArraySource<NumericArraySource<Type>>>>(col, *null_map, total_rows);
+                return std::make_unique<NullableArraySource<NumericArraySource<Type>>>(col, *null_map);
+            }
+            if (is_const)
+                return std::make_unique<ConstSource<NumericArraySource<Type>>>(col, total_rows);
+            return std::make_unique<NumericArraySource<Type>>(col);
+        }
+
+        return ArraySourceCreator<Types...>::create(col, null_map, is_const, total_rows);
+    }
+};
+
+template <>
+struct ArraySourceCreator<>
+{
+    static std::unique_ptr<IArraySource>
+    create(const ColumnArray & col, const ColumnUInt8 * null_map, bool is_const, size_t total_rows)
+    {
+        if (null_map)
+        {
+            if (is_const)
+                return std::make_unique<ConstSource<NullableArraySource<GenericArraySource>>>(col, *null_map, total_rows);
+            return std::make_unique<NullableArraySource<GenericArraySource>>(col, *null_map);
+        }
+        if (is_const)
+            return std::make_unique<ConstSource<GenericArraySource>>(col, total_rows);
+        return std::make_unique<GenericArraySource>(col);
+    }
+};
+
+std::unique_ptr<IArraySource> createArraySource(const ColumnArray & col, bool is_const, size_t total_rows)
+{
+    using Creator = typename ApplyTypeListForClass<ArraySourceCreator, TypeListNumbers>::Type;
+    if (auto column_nullable = typeid_cast<const ColumnNullable *>(&col.getData()))
+    {
+        ColumnArray column(column_nullable->getNestedColumn(), col.getOffsetsColumn());
+        return Creator::create(column, &column_nullable->getNullMapConcreteColumn(), is_const, total_rows);
+    }
+    return Creator::create(col, nullptr, is_const, total_rows);
+}
+
+
+/// Creates IArraySink from ColumnArray
+
+template <typename ... Types>
+struct ArraySinkCreator;
+
+template <typename Type, typename ... Types>
+struct ArraySinkCreator<Type, Types ...>
+{
+    static std::unique_ptr<IArraySink> create(ColumnArray & col, ColumnUInt8 * null_map, size_t column_size)
+    {
+        if (typeid_cast<ColumnVector<Type> *>(&col.getData()))
+        {
+            if (null_map)
+                return std::make_unique<NullableArraySink<NumericArraySink<Type>>>(col, *null_map, column_size);
+            return std::make_unique<NumericArraySink<Type>>(col, column_size);
+        }
+
+        return ArraySinkCreator<Types ...>::create(col, null_map, column_size);
+    }
+};
+
+template <>
+struct ArraySinkCreator<>
+{
+    static std::unique_ptr<IArraySink> create(ColumnArray & col, ColumnUInt8 * null_map, size_t column_size)
+    {
+        if (null_map)
+            return std::make_unique<NullableArraySink<GenericArraySink>>(col, *null_map, column_size);
+        return std::make_unique<GenericArraySink>(col, column_size);
+    }
+};
+
+std::unique_ptr<IArraySink> createArraySink(ColumnArray & col, size_t column_size)
+{
+    using Creator = ApplyTypeListForClass<ArraySinkCreator, TypeListNumbers>::Type;
+    if (auto column_nullable = typeid_cast<ColumnNullable *>(&col.getData()))
+    {
+        ColumnArray column(column_nullable->getNestedColumn(), col.getOffsetsColumn());
+        return Creator::create(column, &column_nullable->getNullMapConcreteColumn(), column_size);
+    }
+    return Creator::create(col, nullptr, column_size);
+}
+
+
+/// Base classes which selects template function implementation with concrete ArraySource or ArraySink
+/// Derived classes should implement selectImpl for ArraySourceSelector and ArraySinkSelector
+///  or selectSourceSink for ArraySinkSourceSelector
+
+template <typename Base, typename ... Types>
+struct ArraySourceSelector;
+
+template <typename Base, typename Type, typename ... Types>
+struct ArraySourceSelector<Base, Type, Types ...>
+{
+    template <typename ... Args>
+    static void select(IArraySource & source, Args && ... args)
+    {
+        if (auto array = typeid_cast<NumericArraySource<Type> *>(&source))
+            Base::selectImpl(*array, args ...);
+        else if (auto nullable_array = typeid_cast<NullableArraySource<NumericArraySource<Type>> *>(&source))
+            Base::selectImpl(*nullable_array, args ...);
+        else if (auto const_array = typeid_cast<ConstSource<NumericArraySource<Type>> *>(&source))
+            Base::selectImpl(*const_array, args ...);
+        else if (auto const_nullable_array = typeid_cast<ConstSource<NullableArraySource<NumericArraySource<Type>>> *>(&source))
+            Base::selectImpl(*const_nullable_array, args ...);
+        else
+            ArraySourceSelector<Base, Types ...>::select(source, args ...);
+    }
+};
+
+template <typename Base>
+struct ArraySourceSelector<Base>
+{
+    template <typename ... Args>
+    static void select(IArraySource & source, Args && ... args)
+    {
+        if (auto array = typeid_cast<GenericArraySource *>(&source))
+            Base::selectImpl(*array, args ...);
+        else if (auto nullable_array = typeid_cast<NullableArraySource<GenericArraySource> *>(&source))
+            Base::selectImpl(*nullable_array, args ...);
+        else if (auto const_array = typeid_cast<ConstSource<GenericArraySource> *>(&source))
+            Base::selectImpl(*const_array, args ...);
+        else if (auto const_nullable_array = typeid_cast<ConstSource<NullableArraySource<GenericArraySource>> *>(&source))
+            Base::selectImpl(*const_nullable_array, args ...);
+        else
+            throw Exception(std::string("Unknown ArraySource type: ") + typeid(source).name(), ErrorCodes::LOGICAL_ERROR);
+    }
+};
+
+template <typename Base>
+using GetArraySourceSelector = typename ApplyTypeListForClass<ArraySourceSelector,
+        typename PrependToTypeList<Base, TypeListNumbers>::Type>::Type;
+
+template <typename Base, typename ... Types>
+struct ArraySinkSelector;
+
+template <typename Base, typename Type, typename ... Types>
+struct ArraySinkSelector<Base, Type, Types ...>
+{
+    template <typename ... Args>
+    static void select(IArraySink & sink, Args && ... args)
+    {
+        if (auto nullable_numeric_sink = typeid_cast<NullableArraySink<NumericArraySink<Type>> *>(&sink))
+            Base::selectImpl(*nullable_numeric_sink, args ...);
+        else if (auto numeric_sink = typeid_cast<NumericArraySink<Type> *>(&sink))
+            Base::selectImpl(*numeric_sink, args ...);
+        else
+            ArraySinkSelector<Base, Types ...>::select(sink, args ...);
+    }
+};
+
+template <typename Base>
+struct ArraySinkSelector<Base>
+{
+    template <typename ... Args>
+    static void select(IArraySink & sink, Args && ... args)
+    {
+        if (auto nullable_generic_sink = typeid_cast<NullableArraySink<GenericArraySink> *>(&sink))
+            Base::selectImpl(*nullable_generic_sink, args ...);
+        else if (auto generic_sink = typeid_cast<GenericArraySink *>(&sink))
+            Base::selectImpl(*generic_sink, args ...);
+        else
+            throw Exception(std::string("Unknown ArraySink type: ") + typeid(sink).name(), ErrorCodes::LOGICAL_ERROR);
+    }
+};
+
+template <typename Base>
+using GetArraySinkSelector = typename ApplyTypeListForClass<ArraySinkSelector,
+        typename PrependToTypeList<Base, TypeListNumbers>::Type>::Type;
+
+template <typename Base>
+struct ArraySinkSourceSelector
+{
+    template <typename ... Args>
+    static void select(IArraySource & source, IArraySink & sink, Args && ... args)
+    {
+        GetArraySinkSelector<Base>::select(sink, source, args ...);
+    }
+
+    template <typename Sink, typename ... Args>
+    static void selectImpl(Sink && sink, IArraySource & source, Args && ... args)
+    {
+        GetArraySourceSelector<Base>::select(source, sink, args ...);
+    }
+
+    template <typename Source, typename Sink, typename ... Args>
+    static void selectImpl(Source && source, Sink && sink, Args && ... args)
+    {
+        Base::selectSourceSink(source, sink, args ...);
+    }
+};
+
+
+/// Algorithms.
+
+/// Appends slices from source to sink. Offsets for sink should be precalculated as start positions of result arrays.
+/// Only for NumericArraySource, because can't insert values in the middle of arbitary column.
+/// Used for array concat implementation.
+template <typename Source, typename Sink>
+static void append(Source && source, Sink && sink)
+{
+    sink.row_num = 0;
+    while (!source.isEnd())
+    {
+        sink.current_offset = sink.offsets[sink.row_num];
+        writeSlice(source.getWhole(), sink);
+        sink.next();
+        source.next();
+    }
+}
+
+struct ArrayAppend : public GetArraySourceSelector<ArrayAppend>
+{
+    template <typename Source, typename Sink>
+    static void selectImpl(Source && source, Sink && sink)
+    {
+        append(source, sink);
+    }
+};
+
+template <typename Sink>
+static void append(IArraySource & source, Sink && sink)
+{
+    ArrayAppend::select(source, sink);
+}
+
+/// Concat specialization for GenericArraySource. Because can't use append with arbitrary column type.
+
+template <typename SourceType, typename SinkType>
+struct ConcatGenericArrayWriteWholeImpl
+{
+    static void writeWhole(GenericArraySource * generic_source, SinkType && sink)
+    {
+        auto source = static_cast<SourceType *>(generic_source);
+        writeSlice(source->getWhole(), sink);
+        source->next();
+    }
+};
+
+template <typename Sink>
+static void NO_INLINE concatGenericArray(const std::vector<std::unique_ptr<IArraySource>> & sources, Sink && sink)
+{
+    std::vector<GenericArraySource *> generic_sources;
+    std::vector<bool> is_nullable;
+    std::vector<bool> is_const;
+
+    generic_sources.reserve(sources.size());
+    is_nullable.assign(sources.size(), false);
+    is_const.assign(sources.size(), false);
+
+    for (auto i : ext::range(0, sources.size()))
+    {
+        const auto & source = sources[i];
+        if (auto generic_source = typeid_cast<GenericArraySource *>(source.get()))
+            generic_sources.push_back(static_cast<GenericArraySource *>(generic_source));
+        else if (auto const_generic_source = typeid_cast<ConstSource<GenericArraySource> *>(source.get()))
+        {
+            generic_sources.push_back(static_cast<GenericArraySource *>(const_generic_source));
+            is_const[i] = true;
+        }
+        else if (auto nullable_source = typeid_cast<NullableArraySource<GenericArraySource> *>(source.get()))
+        {
+            generic_sources.push_back(static_cast<GenericArraySource *>(nullable_source));
+            is_nullable[i] = true;
+        }
+        else if (auto const_nullable_source = typeid_cast<ConstSource<NullableArraySource<GenericArraySource>> *>(source.get()))
+        {
+            generic_sources.push_back(static_cast<GenericArraySource *>(const_nullable_source));
+            is_nullable[i] = is_const[i] = true;
+        }
+        else
+            throw Exception(
+                    std::string("GenericArraySource expected for GenericArraySink, got: ") + typeid(source).name(),
+                    ErrorCodes::LOGICAL_ERROR);
+    }
+
+    while (!sink.isEnd())
+    {
+        for (auto i : ext::range(0, sources.size()))
+        {
+            auto source = generic_sources[i];
+            if (is_const[i])
+            {
+                if (is_nullable[i])
+                    ConcatGenericArrayWriteWholeImpl<ConstSource<NullableArraySource<GenericArraySource>>, Sink>::writeWhole(source, sink);
+                else
+                    ConcatGenericArrayWriteWholeImpl<ConstSource<GenericArraySource>, Sink>::writeWhole(source, sink);
+            }
+            else
+            {
+                if (is_nullable[i])
+                    ConcatGenericArrayWriteWholeImpl<NullableArraySource<GenericArraySource>, Sink>::writeWhole(source, sink);
+                else
+                    ConcatGenericArrayWriteWholeImpl<GenericArraySource, Sink>::writeWhole(source, sink);
+            }
+        }
+        sink.next();
+    }
+}
+
+/// Concat for array sources. Sources must be either all numeric either all generic.
+template <typename Sink>
+void NO_INLINE concat(const std::vector<std::unique_ptr<IArraySource>> & sources, Sink && sink)
+{
+    size_t elements_to_reserve = 0;
+    bool is_first = true;
+
+    /// Prepare offsets column. Offsets should point to starts of result arrays.
+    for (const auto & source : sources)
+    {
+        elements_to_reserve += source->getSizeForReserve();
+        const auto & offsets = source->getOffsets();
+
+        if (is_first)
+        {
+            sink.offsets.resize(source->getColumnSize());
+            memset(&sink.offsets[0], 0, sink.offsets.size() * sizeof(offsets[0]));
+            is_first = false;
+        }
+
+        if (source->isConst())
+        {
+            for (size_t i : ext::range(1, offsets.size()))
+            {
+                sink.offsets[i] += offsets[0];
+            }
+        }
+        else
+        {
+            for (size_t i : ext::range(1, offsets.size()))
+            {
+                sink.offsets[i] += offsets[i - 1] - (i > 1 ? offsets[i - 2] : 0);
+            }
+        }
+    }
+
+    for (auto i : ext::range(1, sink.offsets.size()))
+    {
+        sink.offsets[i] += sink.offsets[i - 1];
+    }
+
+    sink.reserve(elements_to_reserve);
+
+    for (const auto & source : sources)
+    {
+        append(*source, sink);
+    }
+}
+
+struct ArrayConcat : public GetArraySinkSelector<ArrayConcat>
+{
+    using Sources = std::vector<std::unique_ptr<IArraySource>>;
+
+    template <typename Sink>
+    static void selectImpl(Sink && sink, Sources & sources)
+    {
+        concat<Sink>(sources, sink);
+    }
+
+    static void selectImpl(GenericArraySink & sink, Sources & sources)
+    {
+        concatGenericArray(sources, sink);
+    }
+
+    static void selectImpl(NullableArraySink<GenericArraySink> & sink, Sources & sources)
+    {
+        concatGenericArray(sources, sink);
+    }
+
+    static void selectImpl(GenericArraySink && sink, Sources && sources)
+    {
+        concatGenericArray(sources, sink);
+    }
+
+    static void selectImpl(NullableArraySink<GenericArraySink> && sink, Sources & sources)
+    {
+        concatGenericArray(sources, sink);
+    }
+};
+
+void concat(std::vector<std::unique_ptr<IArraySource>> & sources, IArraySink & sink)
+{
+    return ArrayConcat::select(sink, sources);
+}
+
+
+/// Slice for array sources.
+
+struct SliceFromLeftConstantOffsetUnboundedSelectArraySource
+        : public ArraySinkSourceSelector<SliceFromLeftConstantOffsetUnboundedSelectArraySource>
+{
+    template <typename Source, typename Sink>
+    static void selectSourceSink(Source && source, Sink && sink, size_t & offset)
+    {
+        sliceFromLeftConstantOffsetUnbounded(source, sink, offset);
+    }
+};
+
+struct SliceFromLeftConstantOffsetBoundedSelectArraySource
+        : public ArraySinkSourceSelector<SliceFromLeftConstantOffsetBoundedSelectArraySource>
+{
+    template <typename Source, typename Sink>
+    static void selectSourceSink(Source && source, Sink && sink, size_t & offset, ssize_t & length)
+    {
+        sliceFromLeftConstantOffsetBounded(source, sink, offset, length);
+    }
+};
+
+struct SliceFromRightConstantOffsetUnboundedSelectArraySource
+        : public ArraySinkSourceSelector<SliceFromRightConstantOffsetUnboundedSelectArraySource>
+{
+    template <typename Source, typename Sink>
+    static void selectSourceSink(Source && source, Sink && sink, size_t & offset)
+    {
+        sliceFromRightConstantOffsetUnbounded(source, sink, offset);
+    }
+};
+
+struct SliceFromRightConstantOffsetBoundedSelectArraySource
+        : public ArraySinkSourceSelector<SliceFromRightConstantOffsetBoundedSelectArraySource>
+{
+    template <typename Source, typename Sink>
+    static void selectSourceSink(Source && source, Sink && sink, size_t & offset, ssize_t & length)
+    {
+        sliceFromRightConstantOffsetBounded(source, sink, offset, length);
+    }
+};
+
+struct SliceDynamicOffsetUnboundedSelectArraySource
+        : public ArraySinkSourceSelector<SliceDynamicOffsetUnboundedSelectArraySource>
+{
+    template <typename Source, typename Sink>
+    static void selectSourceSink(Source && source, Sink && sink, IColumn & offset_column)
+    {
+        sliceDynamicOffsetUnbounded(source, sink, offset_column);
+    }
+};
+
+struct SliceDynamicOffsetBoundedSelectArraySource
+        : public ArraySinkSourceSelector<SliceDynamicOffsetBoundedSelectArraySource>
+{
+    template <typename Source, typename Sink>
+    static void selectSourceSink(Source && source, Sink && sink, IColumn & offset_column, IColumn & length_column)
+    {
+        sliceDynamicOffsetBounded(source, sink, offset_column, length_column);
+    }
+};
+
+void sliceFromLeftConstantOffsetUnbounded(IArraySource & src, IArraySink & sink, size_t offset)
+{
+    SliceFromLeftConstantOffsetUnboundedSelectArraySource::select(src, sink, offset);
+}
+
+void sliceFromLeftConstantOffsetBounded(IArraySource & src, IArraySink & sink, size_t offset, ssize_t length)
+{
+    SliceFromLeftConstantOffsetBoundedSelectArraySource::select(src, sink, offset, length);
+}
+
+void sliceFromRightConstantOffsetUnbounded(IArraySource & src, IArraySink & sink, size_t offset)
+{
+    SliceFromRightConstantOffsetUnboundedSelectArraySource::select(src, sink, offset);
+}
+
+void sliceFromRightConstantOffsetBounded(IArraySource & src, IArraySink & sink, size_t offset, ssize_t length)
+{
+    SliceFromRightConstantOffsetBoundedSelectArraySource::select(src, sink, offset, length);
+}
+
+void sliceDynamicOffsetUnbounded(IArraySource & src, IArraySink & sink, IColumn & offset_column)
+{
+    SliceDynamicOffsetUnboundedSelectArraySource::select(src, sink, offset_column);
+}
+
+void sliceDynamicOffsetBounded(IArraySource & src, IArraySink & sink, IColumn & offset_column, IColumn & length_column)
+{
+    SliceDynamicOffsetBoundedSelectArraySource::select(src, sink, offset_column, length_column);
+}
+
+}

dbms/src/Functions/GatherUtils.h

@@ -7,13 +7,18 @@
 #include <Columns/ColumnString.h>
 #include <Columns/ColumnFixedString.h>
 #include <Columns/ColumnConst.h>
+#include <Columns/ColumnNullable.h>
 
 #include <Functions/FunctionHelpers.h>
 #include <DataTypes/NumberTraits.h>
 
 #include <Common/typeid_cast.h>
 #include <Common/memcpySmall.h>
+#include <ext/range.h>
+#include <Core/TypeListNumber.h>
+#include <Core/FieldVisitors.h>
 
+#include <DataTypes/DataTypeTraits.h>
 
 /** These methods are intended for implementation of functions, that
   *  copy ranges from one or more columns to another column.
@@ -36,6 +41,11 @@
 namespace DB
 {
 
+namespace ErrorCodes
+{
+    extern const int LOGICAL_ERROR;
+}
+
 template <typename T>
 struct NumericArraySlice
 {
@@ -44,8 +54,24 @@ struct NumericArraySlice
 };
 
 
+struct IArraySource
+{
+    virtual ~IArraySource() {}
+
+    virtual size_t getSizeForReserve() const = 0;
+    virtual const typename ColumnArray::Offsets_t & getOffsets() const = 0;
+    virtual size_t getColumnSize() const = 0;
+    virtual bool isConst() const { return false; }
+    virtual bool isNullable() const { return false; }
+};
+struct IArraySink
+{
+    virtual ~IArraySink() {}
+};
+
+
 template <typename T>
-struct NumericArraySource
+struct NumericArraySource : public IArraySource
 {
     using Slice = NumericArraySlice<T>;
     using Column = ColumnArray;
@@ -56,7 +82,7 @@ struct NumericArraySource
     size_t row_num = 0;
     ColumnArray::Offset_t prev_offset = 0;
 
-    NumericArraySource(const ColumnArray & arr)
+    explicit NumericArraySource(const ColumnArray & arr)
         : elements(typeid_cast<const ColumnVector<T> &>(arr.getData()).getData()), offsets(arr.getOffsets())
     {
     }
@@ -77,12 +103,27 @@ struct NumericArraySource
         return row_num;
     }
 
+    const typename ColumnArray::Offsets_t & getOffsets() const override
+    {
+        return offsets;
+    }
+
     /// Get size for corresponding call or Sink::reserve to reserve memory for elements.
-    size_t getSizeForReserve() const
+    size_t getSizeForReserve() const override
     {
         return elements.size();
     }
 
+    size_t getColumnSize() const override
+    {
+        return offsets.size();
+    }
+
+    size_t getElementSize() const
+    {
+        return offsets[row_num] - prev_offset;
+    }
+
     Slice getWhole() const
     {
         return {&elements[prev_offset], offsets[row_num] - prev_offset};
@@ -116,14 +157,14 @@ struct NumericArraySource
     {
         size_t elem_size = offsets[row_num] - prev_offset;
         if (offset > elem_size)
-            return {&elements[prev_offset], elem_size};
+            return {&elements[prev_offset], length + elem_size > offset ? std::min(elem_size, length + elem_size - offset) : 0};
         return {&elements[offsets[row_num] - offset], std::min(length, offset)};
     }
 };
 
 
 template <typename Base>
-struct ConstSource
+struct ConstSource : public Base
 {
     using Slice = typename Base::Slice;
     using Column = ColumnConst;
@@ -131,10 +172,20 @@ struct ConstSource
     size_t total_rows;
     size_t row_num = 0;
 
-    Base base;
+    explicit ConstSource(const ColumnConst & col)
+        : Base(static_cast<const typename Base::Column &>(col.getDataColumn())), total_rows(col.size())
+    {
+    }
 
-    ConstSource(const ColumnConst & col)
+    /// Constructors for NullableArraySource.
-        : total_rows(col.size()), base(static_cast<const typename Base::Column &>(col.getDataColumn()))
+
+    template <typename ColumnType>
+    ConstSource(const ColumnType & col, size_t total_rows) : Base(col), total_rows(total_rows)
+    {
+    }
+
+    template <typename ColumnType>
+    ConstSource(const ColumnType & col, const ColumnUInt8 & null_map, size_t total_rows) : Base(col, null_map), total_rows(total_rows)
     {
     }
 
@@ -155,32 +206,17 @@ struct ConstSource
 
     size_t getSizeForReserve() const
     {
-        return total_rows * base.getSizeForReserve();
+        return total_rows * Base::getSizeForReserve();
     }
 
-    Slice getWhole() const
+    size_t getColumnSize() const // overrides for IArraySource
     {
-        return base.getWhole();
+        return total_rows;
     }
 
-    Slice getSliceFromLeft(size_t offset) const
+    bool isConst() const // overrides for IArraySource
     {
-        return base.getSliceFromLeft(offset);
+        return true;
-    }
-
-    Slice getSliceFromLeft(size_t offset, size_t length) const
-    {
-        return base.getSliceFromLeft(offset, length);
-    }
-
-    Slice getSliceFromRight(size_t offset) const
-    {
-        return base.getSliceFromRight(offset);
-    }
-
-    Slice getSliceFromRight(size_t offset, size_t length) const
-    {
-        return base.getSliceFromRight(offset, length);
     }
 };
 
@@ -196,7 +232,7 @@ struct StringSource
     size_t row_num = 0;
     ColumnString::Offset_t prev_offset = 0;
 
-    StringSource(const ColumnString & col)
+    explicit StringSource(const ColumnString & col)
         : elements(col.getChars()), offsets(col.getOffsets())
     {
     }
@@ -222,6 +258,11 @@ struct StringSource
         return elements.size();
     }
 
+    size_t getElementSize() const
+    {
+        return offsets[row_num] - prev_offset;
+    }
+
     Slice getWhole() const
     {
         return {&elements[prev_offset], offsets[row_num] - prev_offset - 1};
@@ -255,7 +296,7 @@ struct StringSource
     {
         size_t elem_size = offsets[row_num] - prev_offset - 1;
         if (offset > elem_size)
-            return {&elements[prev_offset], elem_size};
+            return {&elements[prev_offset], length + elem_size > offset ? std::min(elem_size, length + elem_size - offset) : 0};
         return {&elements[prev_offset + elem_size - offset], std::min(length, offset)};
     }
 };
@@ -271,7 +312,7 @@ struct FixedStringSource
     size_t string_size;
     size_t row_num = 0;
 
-    FixedStringSource(const ColumnFixedString & col)
+    explicit FixedStringSource(const ColumnFixedString & col)
         : string_size(col.getN())
     {
         const auto & chars = col.getChars();
@@ -300,6 +341,11 @@ struct FixedStringSource
         return end - pos;
     }
 
+    size_t getElementSize() const
+    {
+        return string_size;
+    }
+
     Slice getWhole() const
     {
         return {pos, string_size};
@@ -329,14 +375,14 @@ struct FixedStringSource
     Slice getSliceFromRight(size_t offset, size_t length) const
     {
         if (offset > string_size)
-            return {pos, string_size};
+            return {pos, length + string_size > offset ? std::min(string_size, length + string_size - offset) : 0};
         return {pos + string_size - offset, std::min(length, offset)};
     }
 };
 
 
 template <typename T>
-struct NumericArraySink
+struct NumericArraySink : public IArraySink
 {
     typename ColumnVector<T>::Container_t & elements;
     typename ColumnArray::Offsets_t & offsets;
@@ -466,7 +512,7 @@ struct DynamicStringSource final : IStringSource
 {
     Impl impl;
 
-    DynamicStringSource(const IColumn & col) : impl(static_cast<const typename Impl::Column &>(col)) {}
+    explicit DynamicStringSource(const IColumn & col) : impl(static_cast<const typename Impl::Column &>(col)) {}
 
     void next() override { impl.next(); }
     bool isEnd() const override { return impl.isEnd(); }
@@ -490,7 +536,6 @@ inline std::unique_ptr<IStringSource> createDynamicStringSource(const IColumn &
 using StringSources = std::vector<std::unique_ptr<IStringSource>>;
 
 
-
 struct GenericArraySlice
 {
     const IColumn * elements;
@@ -499,7 +544,7 @@ struct GenericArraySlice
 };
 
 
-struct GenericArraySource
+struct GenericArraySource : public IArraySource
 {
     using Slice = GenericArraySlice;
     using Column = ColumnArray;
@@ -510,7 +555,7 @@ struct GenericArraySource
     size_t row_num = 0;
     ColumnArray::Offset_t prev_offset = 0;
 
-    GenericArraySource(const ColumnArray & arr)
+    explicit GenericArraySource(const ColumnArray & arr)
         : elements(arr.getData()), offsets(arr.getOffsets())
     {
     }
@@ -531,11 +576,26 @@ struct GenericArraySource
         return row_num;
     }
 
-    size_t getSizeForReserve() const
+    const typename ColumnArray::Offsets_t & getOffsets() const override
+    {
+        return offsets;
+    }
+
+    size_t getSizeForReserve() const override
     {
         return elements.size();
     }
 
+    size_t getColumnSize() const override
+    {
+        return elements.size();
+    }
+
+    size_t getElementSize() const
+    {
+        return offsets[row_num] - prev_offset;
+    }
+
     Slice getWhole() const
     {
         return {&elements, prev_offset, offsets[row_num] - prev_offset};
@@ -569,12 +629,12 @@ struct GenericArraySource
     {
         size_t elem_size = offsets[row_num] - prev_offset;
         if (offset > elem_size)
-            return {&elements, prev_offset, elem_size};
+            return {&elements, prev_offset, length + elem_size > offset ? std::min(elem_size, length + elem_size - offset) : 0};
         return {&elements, offsets[row_num] - offset, std::min(length, offset)};
     }
 };
 
-struct GenericArraySink
+struct GenericArraySink : public IArraySink
 {
     IColumn & elements;
     ColumnArray::Offsets_t & offsets;
@@ -611,6 +671,106 @@ struct GenericArraySink
 };
 
 
+template <typename Slice>
+struct NullableArraySlice : public Slice
+{
+    const UInt8 * null_map = nullptr;
+
+    NullableArraySlice() = default;
+    NullableArraySlice(const Slice & base) : Slice(base) {}
+};
+
+
+template <typename ArraySource>
+struct NullableArraySource : public ArraySource
+{
+    using Slice = NullableArraySlice<typename ArraySource::Slice>;
+    using ArraySource::prev_offset;
+    using ArraySource::row_num;
+    using ArraySource::offsets;
+
+    const ColumnUInt8::Container_t & null_map;
+
+    NullableArraySource(const ColumnArray & arr, const ColumnUInt8 & null_map)
+            : ArraySource(arr), null_map(null_map.getData())
+    {
+    }
+
+    Slice getWhole() const
+    {
+        Slice slice = ArraySource::getWhole();
+        slice.null_map = &null_map[prev_offset];
+        return slice;
+    }
+
+    Slice getSliceFromLeft(size_t offset) const
+    {
+        Slice slice = ArraySource::getSliceFromLeft(offset);
+        if (offsets[row_num] > prev_offset + offset)
+            slice.null_map = &null_map[prev_offset + offset];
+        else
+            slice.null_map = &null_map[prev_offset];
+        return slice;
+    }
+
+    Slice getSliceFromLeft(size_t offset, size_t length) const
+    {
+        Slice slice = ArraySource::getSliceFromLeft(offset, length);
+        if (offsets[row_num] > prev_offset + offset)
+            slice.null_map = &null_map[prev_offset + offset];
+        else
+            slice.null_map = &null_map[prev_offset];
+        return slice;
+    }
+
+    Slice getSliceFromRight(size_t offset) const
+    {
+        Slice slice = ArraySource::getSliceFromRight(offset);
+        if (offsets[row_num] > prev_offset + offset)
+            slice.null_map = &null_map[offsets[row_num] - offset];
+        else
+            slice.null_map = &null_map[prev_offset];
+        return slice;
+    }
+
+    Slice getSliceFromRight(size_t offset, size_t length) const
+    {
+        Slice slice = ArraySource::getSliceFromRight(offset, length);
+        if (offsets[row_num] > prev_offset + offset)
+            slice.null_map = &null_map[offsets[row_num] - offset];
+        else
+            slice.null_map = &null_map[prev_offset];
+        return slice;
+    }
+
+    bool isNullable() const
+    {
+        return true;
+    }
+};
+
+template <typename ArraySink>
+struct NullableArraySink : public ArraySink
+{
+    ColumnUInt8::Container_t & null_map;
+
+    NullableArraySink(ColumnArray & arr, ColumnUInt8 & null_map, size_t column_size)
+        : ArraySink(arr, column_size), null_map(null_map.getData())
+    {
+    }
+
+    void reserve(size_t num_elements)
+    {
+        ArraySink::reserve(num_elements);
+        null_map.reserve(num_elements);
+    }
+};
+
+
+std::unique_ptr<IArraySource> createArraySource(const ColumnArray & col, bool is_const, size_t total_rows);
+std::unique_ptr<IArraySink> createArraySink(ColumnArray & col, size_t column_size);
+
+
 template <typename T>
 using NumericSlice = const T *;
 
@@ -624,7 +784,7 @@ struct NumericSource
     const T * pos;
     const T * end;
 
-    NumericSource(const Column & col)
+    explicit NumericSource(const Column & col)
     {
         const auto & container = col.getData();
         begin = container.data();
@@ -700,15 +860,21 @@ struct NumericSink
 template <typename T>
 void writeSlice(const NumericArraySlice<T> & slice, NumericArraySink<T> & sink)
 {
-    sink.elements.resize(sink.current_offset + slice.size);
+    /// It's possible to write slice into the middle of numeric column. Used in numeric array concat.
-    memcpySmallAllowReadWriteOverflow15(&sink.elements[sink.current_offset], slice.data, slice.size * sizeof(T));
+    if (sink.elements.size() < sink.current_offset + slice.size)
+        sink.elements.resize(sink.current_offset + slice.size);
+    /// Can't use memcpySmallAllowReadWriteOverflow15 when need to write slice into the middle of numeric column.
+    /// TODO: Implement more efficient memcpy without overflow.
+    memcpy(&sink.elements[sink.current_offset], slice.data, slice.size * sizeof(T));
     sink.current_offset += slice.size;
 }
 
 template <typename T, typename U>
 void writeSlice(const NumericArraySlice<T> & slice, NumericArraySink<U> & sink)
 {
-    sink.elements.resize(sink.current_offset + slice.size);
+    /// It's possible to write slice into the middle of numeric column. Used in numeric array concat.
+    if (sink.elements.size() < sink.current_offset + slice.size)
+        sink.elements.resize(sink.current_offset + slice.size);
     for (size_t i = 0; i < slice.size; ++i)
     {
         sink.elements[sink.current_offset] = slice.data[i];
@@ -728,19 +894,65 @@ inline ALWAYS_INLINE void writeSlice(const StringSource::Slice & slice, FixedStr
     memcpySmallAllowReadWriteOverflow15(&sink.elements[sink.current_offset], slice.data, slice.size);
 }
 
-/// Assuming same types of underlying columns for slice and sink.
+/// Assuming same types of underlying columns for slice and sink if (ArraySlice, ArraySink) is (GenericArraySlice, GenericArraySink).
 inline ALWAYS_INLINE void writeSlice(const GenericArraySlice & slice, GenericArraySink & sink)
 {
-    sink.elements.insertRangeFrom(*slice.elements, slice.begin, slice.size);
+    if (typeid(slice.elements) == typeid(&sink.elements))
+        sink.elements.insertRangeFrom(*slice.elements, slice.begin, slice.size);
+    else
+        throw Exception("Function writeSlice expect same column types for GenericArraySlice and GenericArraySink.",
+                        ErrorCodes::LOGICAL_ERROR);
+}
+
+template <typename T>
+inline ALWAYS_INLINE void writeSlice(const GenericArraySlice & slice, NumericArraySink<T> & sink)
+{
+    /// It's possible to write slice into the middle of numeric column. Used in numeric array concat.
+    if (sink.elements.size() < sink.current_offset + slice.size)
+        sink.elements.resize(sink.current_offset + slice.size);
+    for (size_t i = 0; i < slice.size; ++i)
+    {
+        Field field;
+        slice.elements->get(slice.begin + i, field);
+        sink.elements.push_back(applyVisitor(FieldVisitorConvertToNumber<T>(), field));
+    }
     sink.current_offset += slice.size;
 }
 
-template <typename T, typename U>
+template <typename T>
-void ALWAYS_INLINE writeSlice(const NumericSlice<T> & slice, NumericSink<U> & sink)
+inline ALWAYS_INLINE void writeSlice(const NumericArraySlice<T> & slice, GenericArraySink & sink)
 {
-    *sink.pos = *slice;
+    for (size_t i = 0; i < slice.size; ++i)
+    {
+        Field field = static_cast<typename NearestFieldType<T>::Type>(slice.data[i]);
+        sink.elements.insert(field);
+    }
+    sink.current_offset += slice.size;
 }
 
+template <typename ArraySlice, typename ArraySink>
+inline ALWAYS_INLINE void writeSlice(const NullableArraySlice<ArraySlice> & slice, NullableArraySink<ArraySink> & sink)
+{
+    /// It's possible to write slice into the middle of numeric column. Used in numeric array concat.
+    if (sink.null_map.size() < sink.current_offset + slice.size)
+        sink.null_map.resize(sink.current_offset + slice.size);
+    /// Can't use memcpySmallAllowReadWriteOverflow15 when need to write slice into the middle of numeric column.
+    /// TODO: Implement more efficient memcpy without overflow.
+    memcpy(&sink.null_map[sink.current_offset], slice.null_map, slice.size * sizeof(UInt8));
+    writeSlice(static_cast<const ArraySlice &>(slice), static_cast<ArraySink &>(sink));
+}
+
+template <typename ArraySlice, typename ArraySink>
+inline ALWAYS_INLINE void writeSlice(const ArraySlice & slice, NullableArraySink<ArraySink> & sink)
+{
+    /// It's possible to write slice into the middle of numeric column. Used in numeric array concat.
+    if (sink.null_map.size() < sink.current_offset + slice.size)
+        sink.null_map.resize(sink.current_offset + slice.size);
+    /// Can't use memcpySmallAllowReadWriteOverflow15 when need to write slice into the middle of numeric column.
+    /// TODO: Implement more efficient memcpy without overflow.
+    memset(&sink.null_map[sink.current_offset], 0, slice.size * sizeof(UInt8));
+    writeSlice(slice, static_cast<ArraySink &>(sink));
+}
 
 /// Algorithms
 
@@ -774,6 +986,8 @@ void NO_INLINE concat(StringSources & sources, Sink && sink)
     }
 }
 
+void concat(std::vector<std::unique_ptr<IArraySource>> & sources, IArraySink & sink);
+
 
 template <typename Source, typename Sink>
 void NO_INLINE sliceFromLeftConstantOffsetUnbounded(Source && src, Sink && sink, size_t offset)
@@ -787,11 +1001,17 @@ void NO_INLINE sliceFromLeftConstantOffsetUnbounded(Source && src, Sink && sink,
 }
 
 template <typename Source, typename Sink>
-void NO_INLINE sliceFromLeftConstantOffsetBounded(Source && src, Sink && sink, size_t offset, size_t length)
+void NO_INLINE sliceFromLeftConstantOffsetBounded(Source && src, Sink && sink, size_t offset, ssize_t length)
 {
     while (!src.isEnd())
     {
-        writeSlice(src.getSliceFromLeft(offset, length), sink);
+        ssize_t size = length;
+        if (size < 0)
+            size += static_cast<ssize_t>(src.getElementSize()) - offset;
+
+        if (size > 0)
+            writeSlice(src.getSliceFromLeft(offset, size), sink);
+
         sink.next();
         src.next();
     }
@@ -809,11 +1029,17 @@ void NO_INLINE sliceFromRightConstantOffsetUnbounded(Source && src, Sink && sink
 }
 
 template <typename Source, typename Sink>
-void NO_INLINE sliceFromRightConstantOffsetBounded(Source && src, Sink && sink, size_t offset, size_t length)
+void NO_INLINE sliceFromRightConstantOffsetBounded(Source && src, Sink && sink, size_t offset, ssize_t length)
 {
     while (!src.isEnd())
     {
-        writeSlice(src.getSliceFromRight(offset, length), sink);
+        ssize_t size = length;
+        if (size < 0)
+            size += static_cast<ssize_t>(src.getElementSize()) - offset;
+
+        if (size > 0)
+            writeSlice(src.getSliceFromRight(offset, size), sink);
+
         sink.next();
         src.next();
     }
@@ -822,9 +1048,16 @@ void NO_INLINE sliceFromRightConstantOffsetBounded(Source && src, Sink && sink,
 template <typename Source, typename Sink>
 void NO_INLINE sliceDynamicOffsetUnbounded(Source && src, Sink && sink, IColumn & offset_column)
 {
+    const bool is_null = offset_column.isNull();
+    auto * nullable = typeid_cast<ColumnNullable *>(&offset_column);
+    ColumnUInt8::Container_t * null_map = nullable ? &nullable->getNullMapConcreteColumn().getData() : nullptr;
+    IColumn * nested_column = nullable ? nullable->getNestedColumn().get() : &offset_column;
+
     while (!src.isEnd())
     {
-        Int64 offset = offset_column.getInt(src.rowNum());
+        auto row_num = src.rowNum();
+        bool has_offset = !is_null && !(null_map && (*null_map)[row_num]);
+        Int64 offset = has_offset ? nested_column->getInt(row_num) : 1;
 
         if (offset != 0)
         {
@@ -846,13 +1079,28 @@ void NO_INLINE sliceDynamicOffsetUnbounded(Source && src, Sink && sink, IColumn
 template <typename Source, typename Sink>
 void NO_INLINE sliceDynamicOffsetBounded(Source && src, Sink && sink, IColumn & offset_column, IColumn & length_column)
 {
+    const bool is_offset_null = offset_column.isNull();
+    auto * offset_nullable = typeid_cast<ColumnNullable *>(&offset_column);
+    ColumnUInt8::Container_t * offset_null_map = offset_nullable ? &offset_nullable->getNullMapConcreteColumn().getData() : nullptr;
+    IColumn * offset_nested_column = offset_nullable ? offset_nullable->getNestedColumn().get() : &offset_column;
+
+    const bool is_length_null = length_column.isNull();
+    auto * length_nullable = typeid_cast<ColumnNullable *>(&length_column);
+    ColumnUInt8::Container_t * length_null_map = length_nullable ? &length_nullable->getNullMapConcreteColumn().getData() : nullptr;
+    IColumn * length_nested_column = length_nullable ? length_nullable->getNestedColumn().get() : &length_column;
+
     while (!src.isEnd())
     {
         size_t row_num = src.rowNum();
-        Int64 offset = offset_column.getInt(row_num);
+        bool has_offset = !is_offset_null && !(offset_null_map && (*offset_null_map)[row_num]);
-        UInt64 size = length_column.getInt(row_num);
+        bool has_length = !is_length_null && !(length_null_map && (*length_null_map)[row_num]);
+        Int64 offset = has_offset ? offset_nested_column->getInt(row_num) : 1;
+        Int64 size = has_length ? length_nested_column->getInt(row_num) : static_cast<Int64>(src.getElementSize());
 
-        if (offset != 0 && size < 0x8000000000000000ULL)
+        if (size < 0)
+            size += offset > 0 ? static_cast<Int64>(src.getElementSize()) - (offset - 1) : -offset;
+
+        if (offset != 0 && size > 0)
         {
             typename std::decay<Source>::type::Slice slice;
 
@@ -870,6 +1118,18 @@ void NO_INLINE sliceDynamicOffsetBounded(Source && src, Sink && sink, IColumn &
 }
 
 
+void sliceFromLeftConstantOffsetUnbounded(IArraySource & src, IArraySink & sink, size_t offset);
+
+void sliceFromLeftConstantOffsetBounded(IArraySource & src, IArraySink & sink, size_t offset, ssize_t length);
+
+void sliceFromRightConstantOffsetUnbounded(IArraySource & src, IArraySink & sink, size_t offset);
+
+void sliceFromRightConstantOffsetBounded(IArraySource & src, IArraySink & sink, size_t offset, ssize_t length);
+
+void sliceDynamicOffsetUnbounded(IArraySource & src, IArraySink & sink, IColumn & offset_column);
+
+void sliceDynamicOffsetBounded(IArraySource & src, IArraySink & sink, IColumn & offset_column, IColumn & length_column);
+
 template <typename SourceA, typename SourceB, typename Sink>
 void NO_INLINE conditional(SourceA && src_a, SourceB && src_b, Sink && sink, const PaddedPODArray<UInt8> & condition)
 {

dbms/src/Interpreters/SpecializedAggregator.h

@@ -1,3 +1,4 @@
+#include <Common/TypeList.h>
 #include <Interpreters/Aggregator.h>
 
 #include <AggregateFunctions/AggregateFunctionArgMinMax.h>
@@ -40,43 +41,6 @@ namespace DB
   */
 
 
-/** List of types - for convenient listing of aggregate functions.
-  */
-template <typename... TTail>
-struct TypeList
-{
-    static constexpr size_t size = 0;
-
-    template <size_t I>
-    using At = std::nullptr_t;
-
-    template <typename Func, size_t index = 0>
-    static void forEach(Func && func)
-    {
-    }
-};
-
-
-template <typename THead, typename... TTail>
-struct TypeList<THead, TTail...>
-{
-    using Head = THead;
-    using Tail = TypeList<TTail...>;
-
-    static constexpr size_t size = 1 + sizeof...(TTail);
-
-    template <size_t I>
-    using At = typename std::template conditional<I == 0, Head, typename Tail::template At<I - 1>>::type;
-
-    template <typename Func, size_t index = 0>
-    static void ALWAYS_INLINE forEach(Func && func)
-    {
-        func.template operator()<Head, index>();
-        Tail::template forEach<Func, index + 1>(std::forward<Func>(func));
-    }
-};
-
-
 struct AggregateFunctionsUpdater
 {
     AggregateFunctionsUpdater(

dbms/tests/queries/0_stateless/00498_array_functions_concat_slice_push_pop.sql

@@ -0,0 +1,368 @@
+select 'const args';
+select 'concat';
+select arrayConcat(emptyArrayUInt8());
+select arrayConcat(emptyArrayUInt8(), emptyArrayUInt8());
+select arrayConcat(emptyArrayUInt8(), emptyArrayUInt8(), emptyArrayUInt8());
+select arrayConcat([Null], emptyArrayUInt8());
+select arrayConcat([Null], emptyArrayUInt8(), [1]);
+select arrayConcat([1, 2], [-1, -2], [0.3, 0.7], [Null]);
+select arrayConcat(Null, emptyArrayUInt8());
+select arrayConcat([1], [-1], Null);
+select arrayConcat([1, 2], [3, 4]);
+select arrayConcat([1], [2, 3, 4]);
+select arrayConcat(emptyArrayUInt8(), emptyArrayUInt8());
+SELECT arrayConcat(['abc'], ['def', 'gh', 'qwe']);
+SELECT arrayConcat([1, NULL, 2], [3, NULL, 4]);
+select arrayConcat([1, Null, 2], [3, 4]);
+
+select 'slice';
+select arraySlice(Null, 1, 2);
+select arraySlice([1, 2, 3, 4, 5, 6], Null, Null);
+select arraySlice([1, 2, 3, 4, 5, 6], 2, Null);
+select arraySlice([1, 2, 3, 4, 5, 6], Null, 4);
+select arraySlice([1, 2, 3, 4, 5, 6], Null, -2);
+select arraySlice([1, 2, 3, 4, 5, 6], -3, Null);
+select arraySlice([1, 2, 3, 4, 5, 6], 2, 3);
+select arraySlice([1, 2, 3, 4, 5, 6], 2, -2);
+select arraySlice([1, 2, 3, 4, 5, 6], -4, 2);
+select arraySlice([1, 2, 3, 4, 5, 6], -4, -2);
+select arraySlice([1, 2, 3, 4, 5, 6], 2, 0);
+select arraySlice([1, 2, 3, 4, 5, 6], -10, 15);
+select arraySlice([1, 2, 3, 4, 5, 6], -15, 10);
+select arraySlice([1, 2, 3, 4, 5, 6], -15, 9);
+select arraySlice([1, 2, 3, 4, 5, 6], 10, 0);
+select arraySlice([1, 2, 3, 4, 5, 6], 10, -1);
+select arraySlice([1, 2, 3, 4, 5, 6], 10, 1);
+select arraySlice([1, 2, Null, 4, 5, 6], 2, 4);
+select arraySlice(['a', 'b', 'c', 'd', 'e'], 2, 3);
+select arraySlice([Null, 'b', Null, 'd', 'e'], 2, 3);
+
+select 'push back';
+select arrayPushBack(Null, 1);
+select arrayPushBack([1], 1);
+select arrayPushBack([Null], 1);
+select arrayPushBack([0.5, 0.7], 1);
+select arrayPushBack([1], -1);
+select arrayPushBack(['a', 'b'], 'cd');
+select arrayPushBack(emptyArrayUInt8(), 1);
+select arrayPushBack(emptyArrayUInt8(), -1);
+
+select 'push front';
+select arrayPushFront(Null, 1);
+select arrayPushFront([1], 1);
+select arrayPushFront([Null], 1);
+select arrayPushFront([0.5, 0.7], 1);
+select arrayPushFront([1], -1);
+select arrayPushFront(['a', 'b'], 'cd');
+select arrayPushFront(emptyArrayUInt8(), 1);
+select arrayPushFront(emptyArrayUInt8(), -1);
+
+select 'pop back';
+select arrayPopBack(Null);
+select arrayPopBack(emptyArrayUInt8());
+select arrayPopBack([1]);
+select arrayPopBack([1, 2, 3]);
+select arrayPopBack([0.1, 0.2, 0.3]);
+select arrayPopBack(['a', 'b', 'c']);
+
+select 'pop front';
+select arrayPopFront(Null);
+select arrayPopFront(emptyArrayUInt8());
+select arrayPopFront([1]);
+select arrayPopFront([1, 2, 3]);
+select arrayPopFront([0.1, 0.2, 0.3]);
+select arrayPopFront(['a', 'b', 'c']);
+
+DROP TABLE if exists test_array_functions;
+select '';
+select 'table';
+create table test_array_functions (arr1 Array(Int8), arr2 Array(Int8), o Int8, no Nullable(Int8), l Int8, nl Nullable(Int8)) engine = TinyLog;
+insert into test_array_functions values ([], [], 1, Null, 1, Null), ([], [1], 1, Null, 1, Null), ([1, 2, 3, 4, 5], [6, 7], 2, Null, 1, Null), ([1, 2, 3, 4, 5, 6, 7], [8], 2, 2, 3, 3), ([1, 2, 3, 4, 5, 6, 7], [], 2, Null, -3, -3), ([1, 2, 3, 4, 5, 6, 7], [], 2, Null, -3, Null), ([1, 2, 3, 4, 5, 6, 7], [], -5, -5, 4, 4), ([1, 2, 3, 4, 5, 6, 7], [], -5, -5, -3, -3);
+
+select * from test_array_functions;
+select 'concat arr1, arr2';
+select arrayConcat(arr1, arr2), arr1, arr2 from test_array_functions;
+select 'concat arr1, arr2, arr1';
+select arrayConcat(arr1, arr2, arr1), arr1, arr2 from test_array_functions;
+
+select 'arraySlice(arr1, o, l)';
+select arr1, o, l, arraySlice(arr1, o, l) from test_array_functions;
+select 'arraySlice(arr1, no, nl)';
+select arr1, no, nl, arraySlice(arr1, no, nl) from test_array_functions;
+select 'arraySlice(arr1, 2, l)';
+select arr1, 2, l, arraySlice(arr1, 2, l) from test_array_functions;
+select 'arraySlice(arr1, o, 2)';
+select arr1, o, 2, arraySlice(arr1, o, 2) from test_array_functions;
+select 'arraySlice(arr1, 2, nl)';
+select arr1, 2, nl, arraySlice(arr1, 2, nl) from test_array_functions;
+select 'arraySlice(arr1, no, 2)';
+select arr1, no, 2, arraySlice(arr1, no, 2) from test_array_functions;
+select 'arraySlice(arr1, -4, l)';
+select arr1, 2, l, arraySlice(arr1, -4, l) from test_array_functions;
+select 'arraySlice(arr1, o, -2)';
+select arr1, o, 2, arraySlice(arr1, o, -2) from test_array_functions;
+select 'arraySlice(arr1, -4, nl)';
+select arr1, 2, nl, arraySlice(arr1, -4, nl) from test_array_functions;
+select 'arraySlice(arr1, no, -2)';
+select arr1, no, 2, arraySlice(arr1, no, -2) from test_array_functions;
+select 'arraySlice(arr1, 2, 4)';
+select arr1, 2, 4, arraySlice(arr1, 2, 4) from test_array_functions;
+select 'arraySlice(arr1, 2, -4)';
+select arr1, 2, 4, arraySlice(arr1, 2, -4) from test_array_functions;
+select 'arraySlice(arr1, -4, 2)';
+select arr1, 2, 4, arraySlice(arr1, -4, 2) from test_array_functions;
+select 'arraySlice(arr1, -4, -1)';
+select arr1, 2, 4, arraySlice(arr1, -4, -1) from test_array_functions;
+
+select 'arrayPushFront(arr1, 1)';
+select arr1, arrayPushFront(arr1, 1) from test_array_functions;
+select 'arrayPushFront(arr1, 0.1)';
+select arr1, arrayPushFront(arr1, 0.1) from test_array_functions;
+select 'arrayPushFront(arr1, l)';
+select arr1, arrayPushFront(arr1, l) from test_array_functions;
+select 'arrayPushFront(arr1, nl)';
+select arr1, arrayPushFront(arr1, nl) from test_array_functions;
+select 'arrayPushFront([1, 2, 3], l)';
+select arrayPushFront([1, 2, 3], l) from test_array_functions;
+select 'arrayPushFront([1, 2, 3], nl)' from test_array_functions;
+select arrayPushFront([1, 2, 3], nl) from test_array_functions;
+
+select 'arrayPushBack(arr1, 1)';
+select arr1, arrayPushBack(arr1, 1) from test_array_functions;
+select 'arrayPushBack(arr1, 0.1)';
+select arr1, arrayPushBack(arr1, 0.1) from test_array_functions;
+select 'arrayPushBack(arr1, l)';
+select arr1, arrayPushBack(arr1, l) from test_array_functions;
+select 'arrayPushBack(arr1, nl)';
+select arr1, arrayPushBack(arr1, nl) from test_array_functions;
+select 'arrayPushBack([1, 2, 3], l)';
+select arrayPushBack([1, 2, 3], l) from test_array_functions;
+select 'arrayPushBack([1, 2, 3], nl)';
+select arrayPushBack([1, 2, 3], nl) from test_array_functions;
+
+select 'arrayPopFront(arr1)';
+select arr1, arrayPopFront(arr1) from test_array_functions;
+select 'arrayPopBack(arr1)';
+select arr1, arrayPopBack(arr1) from test_array_functions;
+
+
+DROP TABLE if exists test_array_functions;
+select '';
+select 'table';
+create table test_array_functions (arr1 Array(Nullable(Int8)), arr2 Array(Nullable(Float32)), o Int8, no Nullable(Int8), l Int8, nl Nullable(Int8)) engine = TinyLog;
+insert into test_array_functions values ([], [], 1, Null, 1, Null), ([], [1, Null], 1, Null, 1, Null), ([1, 2, 3, 4, 5], [6, Null], 2, Null, 1, Null), ([1, Null, 3, 4, Null, 6, 7], [8], 2, 2, 3, 3),([1, 2, 3, Null, 5, 6, 7], [Null, 1], 2, Null, -3, -3),([1, 2, 3, 4, 5, Null, 7], [1, Null], 2, Null, -3, Null), ([1, 2, 3, 4, 5, 6, 7], [1, 2], -5, -5, 4, 4),([1, Null, 3, Null, 5, 6, 7], [], -5, -5, -3, -3);
+
+select * from test_array_functions;
+select 'concat arr1, arr2';
+select arrayConcat(arr1, arr2), arr1, arr2 from test_array_functions;
+select 'concat arr1, arr2, arr1';
+select arrayConcat(arr1, arr2, arr1), arr1, arr2 from test_array_functions;
+
+select 'arraySlice(arr1, o, l)';
+select arr1, o, l, arraySlice(arr1, o, l) from test_array_functions;
+select 'arraySlice(arr1, no, nl)';
+select arr1, no, nl, arraySlice(arr1, no, nl) from test_array_functions;
+select 'arraySlice(arr1, 2, l)';
+select arr1, 2, l, arraySlice(arr1, 2, l) from test_array_functions;
+select 'arraySlice(arr1, o, 2)';
+select arr1, o, 2, arraySlice(arr1, o, 2) from test_array_functions;
+select 'arraySlice(arr1, 2, nl)';
+select arr1, 2, nl, arraySlice(arr1, 2, nl) from test_array_functions;
+select 'arraySlice(arr1, no, 2)';
+select arr1, no, 2, arraySlice(arr1, no, 2) from test_array_functions;
+select 'arraySlice(arr1, -4, l)';
+select arr1, 2, l, arraySlice(arr1, -4, l) from test_array_functions;
+select 'arraySlice(arr1, o, -2)';
+select arr1, o, 2, arraySlice(arr1, o, -2) from test_array_functions;
+select 'arraySlice(arr1, -4, nl)';
+select arr1, 2, nl, arraySlice(arr1, -4, nl) from test_array_functions;
+select 'arraySlice(arr1, no, -2)';
+select arr1, no, 2, arraySlice(arr1, no, -2) from test_array_functions;
+select 'arraySlice(arr1, 2, 4)';
+select arr1, 2, 4, arraySlice(arr1, 2, 4) from test_array_functions;
+select 'arraySlice(arr1, 2, -4)';
+select arr1, 2, 4, arraySlice(arr1, 2, -4) from test_array_functions;
+select 'arraySlice(arr1, -4, 2)';
+select arr1, 2, 4, arraySlice(arr1, -4, 2) from test_array_functions;
+select 'arraySlice(arr1, -4, -1)';
+select arr1, 2, 4, arraySlice(arr1, -4, -1) from test_array_functions;
+
+select 'arrayPushFront(arr1, 1)';
+select arr1, arrayPushFront(arr1, 1) from test_array_functions;
+select 'arrayPushFront(arr1, 0.1)';
+select arr1, arrayPushFront(arr1, 0.1) from test_array_functions;
+select 'arrayPushFront(arr1, l)';
+select arr1, arrayPushFront(arr1, l) from test_array_functions;
+select 'arrayPushFront(arr1, nl)';
+select arr1, arrayPushFront(arr1, nl) from test_array_functions;
+select 'arrayPushFront([1, 2, 3], l)';
+select arrayPushFront([1, 2, 3], l) from test_array_functions;
+select 'arrayPushFront([1, 2, 3], nl)' from test_array_functions;
+select arrayPushFront([1, 2, 3], nl) from test_array_functions;
+
+select 'arrayPushBack(arr1, 1)';
+select arr1, arrayPushBack(arr1, 1) from test_array_functions;
+select 'arrayPushBack(arr1, 0.1)';
+select arr1, arrayPushBack(arr1, 0.1) from test_array_functions;
+select 'arrayPushBack(arr1, l)';
+select arr1, arrayPushBack(arr1, l) from test_array_functions;
+select 'arrayPushBack(arr1, nl)';
+select arr1, arrayPushBack(arr1, nl) from test_array_functions;
+select 'arrayPushBack([1, 2, 3], l)';
+select arrayPushBack([1, 2, 3], l) from test_array_functions;
+select 'arrayPushBack([1, 2, 3], nl)';
+select arrayPushBack([1, 2, 3], nl) from test_array_functions;
+
+select 'arrayPopFront(arr1)';
+select arr1, arrayPopFront(arr1) from test_array_functions;
+select 'arrayPopBack(arr1)';
+select arr1, arrayPopBack(arr1) from test_array_functions;
+
+
+DROP TABLE if exists test_array_functions;
+select '';
+select 'table';
+create table test_array_functions (arr1 Array(Nullable(Int8)), arr2 Array(UInt8), o Int8, no Nullable(Int8), l Int8, nl Nullable(Int8)) engine = TinyLog;
+insert into test_array_functions values ([], [], 1, Null, 1, Null), ([], [1, 2], 1, Null, 1, Null), ([1, 2, 3, 4, 5], [6, 7], 2, Null, 1, Null), ([1, Null,3,4, Null, 6, 7], [8], 2, 2, 3, 3),([1, 2, 3, Null, 5, 6, 7], [0, 1], 2, Null, -3, -3),([1, 2, 3, 4, 5, Null, 7], [1, 2], 2, Null, -3, Null),([1, 2, 3,4, 5, 6, 7], [1, 2], -5, -5, 4, 4),([1, Null, 3, Null, 5, 6, 7], [], -5, -5, -3, -3);
+
+select * from test_array_functions;
+select 'concat arr1, arr2';
+select arrayConcat(arr1, arr2), arr1, arr2 from test_array_functions;
+select 'concat arr1, arr2, arr1';
+select arrayConcat(arr1, arr2, arr1), arr1, arr2 from test_array_functions;
+
+
+select * from test_array_functions;
+select 'concat arr1, arr2';
+select arrayConcat(arr1, arr2), arr1, arr2 from test_array_functions;
+select 'concat arr1, arr2, arr1';
+select arrayConcat(arr1, arr2, arr1), arr1, arr2 from test_array_functions;
+
+select 'arraySlice(arr1, o, l)';
+select arr1, o, l, arraySlice(arr1, o, l) from test_array_functions;
+select 'arraySlice(arr1, no, nl)';
+select arr1, no, nl, arraySlice(arr1, no, nl) from test_array_functions;
+select 'arraySlice(arr1, 2, l)';
+select arr1, 2, l, arraySlice(arr1, 2, l) from test_array_functions;
+select 'arraySlice(arr1, o, 2)';
+select arr1, o, 2, arraySlice(arr1, o, 2) from test_array_functions;
+select 'arraySlice(arr1, 2, nl)';
+select arr1, 2, nl, arraySlice(arr1, 2, nl) from test_array_functions;
+select 'arraySlice(arr1, no, 2)';
+select arr1, no, 2, arraySlice(arr1, no, 2) from test_array_functions;
+select 'arraySlice(arr1, -4, l)';
+select arr1, 2, l, arraySlice(arr1, -4, l) from test_array_functions;
+select 'arraySlice(arr1, o, -2)';
+select arr1, o, 2, arraySlice(arr1, o, -2) from test_array_functions;
+select 'arraySlice(arr1, -4, nl)';
+select arr1, 2, nl, arraySlice(arr1, -4, nl) from test_array_functions;
+select 'arraySlice(arr1, no, -2)';
+select arr1, no, 2, arraySlice(arr1, no, -2) from test_array_functions;
+select 'arraySlice(arr1, 2, 4)';
+select arr1, 2, 4, arraySlice(arr1, 2, 4) from test_array_functions;
+select 'arraySlice(arr1, 2, -4)';
+select arr1, 2, 4, arraySlice(arr1, 2, -4) from test_array_functions;
+select 'arraySlice(arr1, -4, 2)';
+select arr1, 2, 4, arraySlice(arr1, -4, 2) from test_array_functions;
+select 'arraySlice(arr1, -4, -1)';
+select arr1, 2, 4, arraySlice(arr1, -4, -1) from test_array_functions;
+
+select 'arrayPushFront(arr1, 1)';
+select arr1, arrayPushFront(arr1, 1) from test_array_functions;
+select 'arrayPushFront(arr1, 0.1)';
+select arr1, arrayPushFront(arr1, 0.1) from test_array_functions;
+select 'arrayPushFront(arr1, l)';
+select arr1, arrayPushFront(arr1, l) from test_array_functions;
+select 'arrayPushFront(arr1, nl)';
+select arr1, arrayPushFront(arr1, nl) from test_array_functions;
+select 'arrayPushFront([1, 2, 3], l)';
+select arrayPushFront([1, 2, 3], l) from test_array_functions;
+select 'arrayPushFront([1, 2, 3], nl)' from test_array_functions;
+select arrayPushFront([1, 2, 3], nl) from test_array_functions;
+
+select 'arrayPushBack(arr1, 1)';
+select arr1, arrayPushBack(arr1, 1) from test_array_functions;
+select 'arrayPushBack(arr1, 0.1)';
+select arr1, arrayPushBack(arr1, 0.1) from test_array_functions;
+select 'arrayPushBack(arr1, l)';
+select arr1, arrayPushBack(arr1, l) from test_array_functions;
+select 'arrayPushBack(arr1, nl)';
+select arr1, arrayPushBack(arr1, nl) from test_array_functions;
+select 'arrayPushBack([1, 2, 3], l)';
+select arrayPushBack([1, 2, 3], l) from test_array_functions;
+select 'arrayPushBack([1, 2, 3], nl)';
+select arrayPushBack([1, 2, 3], nl) from test_array_functions;
+
+select 'arrayPopFront(arr1)';
+select arr1, arrayPopFront(arr1) from test_array_functions;
+select 'arrayPopBack(arr1)';
+select arr1, arrayPopBack(arr1) from test_array_functions;
+
+DROP TABLE if exists test_array_functions;
+select '';
+select 'table';
+create table test_array_functions (arr1 Array(Nullable(String)), arr2 Array(String), val String, val2 Nullable(String), o Int8, no Nullable(Int8), l Int8, nl Nullable(Int8)) engine = TinyLog;
+insert into test_array_functions values ([], [], '', Null, 1, Null, 1, Null), ([], ['1', '2'], 'a', 'b', 1, Null, 1, Null), (['1', '2', '3', '4', '5'], ['6','7'], 'a', Null, 2, Null, 1, Null), (['1', Null, '3', '4', Null, '6', '7'], ['8'], 'a', 'b', 2, 2, 3, 3),(['1', '2', '3', Null, '5', '6', '7'], ['0','1'], 'a', Null, 2, Null, -3, -3),(['1', '2', '3', '4', '5', Null, '7'], ['1', '2'], 'a', 'b', 2, Null, -3, Null),(['1', '2', '3', '4', '5', '6', '7'],['1', '2'], 'a', Null, -5, -5, 4, 4),(['1', Null, '3', Null, '5', '6', '7'], [], 'a', 'b', -5, -5, -3, -3);
+
+
+select * from test_array_functions;
+select 'concat arr1, arr2';
+select arrayConcat(arr1, arr2), arr1, arr2 from test_array_functions;
+select 'concat arr1, arr2, arr1';
+select arrayConcat(arr1, arr2, arr1), arr1, arr2 from test_array_functions;
+
+select 'arraySlice(arr1, o, l)';
+select arr1, o, l, arraySlice(arr1, o, l) from test_array_functions;
+select 'arraySlice(arr1, no, nl)';
+select arr1, no, nl, arraySlice(arr1, no, nl) from test_array_functions;
+select 'arraySlice(arr1, 2, l)';
+select arr1, 2, l, arraySlice(arr1, 2, l) from test_array_functions;
+select 'arraySlice(arr1, o, 2)';
+select arr1, o, 2, arraySlice(arr1, o, 2) from test_array_functions;
+select 'arraySlice(arr1, 2, nl)';
+select arr1, 2, nl, arraySlice(arr1, 2, nl) from test_array_functions;
+select 'arraySlice(arr1, no, 2)';
+select arr1, no, 2, arraySlice(arr1, no, 2) from test_array_functions;
+select 'arraySlice(arr1, -4, l)';
+select arr1, 2, l, arraySlice(arr1, -4, l) from test_array_functions;
+select 'arraySlice(arr1, o, -2)';
+select arr1, o, 2, arraySlice(arr1, o, -2) from test_array_functions;
+select 'arraySlice(arr1, -4, nl)';
+select arr1, 2, nl, arraySlice(arr1, -4, nl) from test_array_functions;
+select 'arraySlice(arr1, no, -2)';
+select arr1, no, 2, arraySlice(arr1, no, -2) from test_array_functions;
+select 'arraySlice(arr1, 2, 4)';
+select arr1, 2, 4, arraySlice(arr1, 2, 4) from test_array_functions;
+select 'arraySlice(arr1, 2, -4)';
+select arr1, 2, 4, arraySlice(arr1, 2, -4) from test_array_functions;
+select 'arraySlice(arr1, -4, 2)';
+select arr1, 2, 4, arraySlice(arr1, -4, 2) from test_array_functions;
+select 'arraySlice(arr1, -4, -1)';
+select arr1, 2, 4, arraySlice(arr1, -4, -1) from test_array_functions;
+
+select 'arrayPushFront(arr1, 1)';
+select arr1, arrayPushFront(arr1, '1') from test_array_functions;
+select 'arrayPushFront(arr1, val)';
+select arr1, arrayPushFront(arr1, val) from test_array_functions;
+select 'arrayPushFront(arr1, val2)';
+select arr1, arrayPushFront(arr1, val2) from test_array_functions;
+select 'arrayPushFront([a, b, c], val)';
+select arrayPushFront(['a', 'b', 'c'], val) from test_array_functions;
+select 'arrayPushFront([a, b, c], val2)';
+select arrayPushFront(['a', 'b', 'c'], val2) from test_array_functions;
+
+select 'arrayPushBack(arr1, 1)';
+select arr1, arrayPushBack(arr1, '1') from test_array_functions;
+select 'arrayPushBack(arr1, val)';
+select arr1, arrayPushBack(arr1, val) from test_array_functions;
+select 'arrayPushBack(arr1, val2)';
+select arr1, arrayPushBack(arr1, val2) from test_array_functions;
+select 'arrayPushBack([a, b, c], val)';
+select arrayPushBack(['a', 'b', 'c'], val) from test_array_functions;
+select 'arrayPushBack([a, b, c], val2)';
+select arrayPushBack(['a', 'b', 'c'], val2) from test_array_functions;
+
+select 'arrayPopFront(arr1)';
+select arr1, arrayPopFront(arr1) from test_array_functions;
+select 'arrayPopBack(arr1)';
+select arr1, arrayPopBack(arr1) from test_array_functions;