branch-4.0: [Feature](func) pick some functions (#57559)

### What problem does this PR solve?

pick: #56691, #56945 #57282 #57369 #57437

Author: linrrzqqq

be/src/vec/functions/function.h

@@ -32,6 +32,7 @@
 #include "common/logging.h"
 #include "common/status.h"
 #include "olap/rowset/segment_v2/inverted_index_iterator.h" // IWYU pragma: keep
+#include "runtime/define_primitive_type.h"
 #include "udf/udf.h"
 #include "vec/core/block.h"
 #include "vec/core/column_numbers.h"

be/src/vec/functions/function_date_or_datetime_computation.cpp

@@ -67,6 +67,8 @@ using FunctionSecToTime = FunctionCurrentDateOrDateTime<SecToTimeImpl>;
 using FunctionMicroSecToDateTime = TimestampToDateTime<MicroSec>;
 using FunctionMilliSecToDateTime = TimestampToDateTime<MilliSec>;
 using FunctionSecToDateTime = TimestampToDateTime<Sec>;
+using FunctionPeriodAdd = FunctionNeedsToHandleNull<PeriodAddImpl, PrimitiveType::TYPE_BIGINT>;
+using FunctionPeriodDiff = FunctionNeedsToHandleNull<PeriodDiffImpl, PrimitiveType::TYPE_BIGINT>;
 
 void register_function_date_time_computation(SimpleFunctionFactory& factory) {
     factory.register_function<FunctionDateDiff>();
@@ -98,6 +100,9 @@ void register_function_date_time_computation(SimpleFunctionFactory& factory) {
     factory.register_function<FunctionSecToDateTime>();
     factory.register_function<FunctionMonthsBetween>();
     factory.register_function<FunctionTime>();
+    factory.register_function<FunctionGetFormat>();
+    factory.register_function<FunctionPeriodAdd>();
+    factory.register_function<FunctionPeriodDiff>();
 
     // alias
     factory.register_alias("days_add", "date_add");

be/src/vec/functions/function_date_or_datetime_computation.h

@@ -59,6 +59,7 @@
 #include "vec/functions/datetime_errors.h"
 #include "vec/functions/function.h"
 #include "vec/functions/function_helpers.h"
+#include "vec/functions/function_needs_to_handle_null.h"
 #include "vec/runtime/time_value.h"
 #include "vec/runtime/vdatetime_value.h"
 #include "vec/utils/util.hpp"
@@ -1418,5 +1419,205 @@ class FunctionTime : public IFunction {
         return Status::OK();
     }
 };
+
+class FunctionGetFormat : public IFunction {
+public:
+    static constexpr auto name = "get_format";
+    static FunctionPtr create() { return std::make_shared<FunctionGetFormat>(); }
+    String get_name() const override { return name; }
+    size_t get_number_of_arguments() const override { return 2; }
+    DataTypePtr get_return_type_impl(const DataTypes& arguments) const override {
+        return make_nullable(std::make_shared<DataTypeString>());
+    }
+
+    Status execute_impl(FunctionContext* context, Block& block, const ColumnNumbers& arguments,
+                        uint32_t result, size_t input_rows_count) const override {
+        const auto& [left_col_ptr, left_is_const] =
+                unpack_if_const(block.get_by_position(arguments[0]).column);
+        const auto& [right_col_ptr, right_is_const] =
+                unpack_if_const(block.get_by_position(arguments[1]).column);
+
+        const auto* left_col = assert_cast<const ColumnString*>(left_col_ptr.get());
+        const auto* right_col = assert_cast<const ColumnString*>(right_col_ptr.get());
+
+        auto type_ref = left_col->get_data_at(0);
+        std::string type_str(type_ref.data, type_ref.size);
+
+        auto res_col = ColumnString::create();
+        auto res_null_map = ColumnUInt8::create(input_rows_count, 0);
+        auto& res_data = res_col->get_chars();
+        auto& res_offsets = res_col->get_offsets();
+
+        if (type_str == DATE_NAME) {
+            execute_format_type<DateFormatImpl>(res_data, res_offsets, res_null_map->get_data(),
+                                                input_rows_count, right_col);
+        } else if (type_str == DATETIME_NAME) {
+            execute_format_type<DateTimeFormatImpl>(res_data, res_offsets, res_null_map->get_data(),
+                                                    input_rows_count, right_col);
+        } else if (type_str == TIME_NAME) {
+            execute_format_type<TimeFormatImpl>(res_data, res_offsets, res_null_map->get_data(),
+                                                input_rows_count, right_col);
+        } else {
+            return Status::InvalidArgument(
+                    "Function GET_FORMAT only support DATE, DATETIME or TIME");
+        }
+
+        block.replace_by_position(
+                result, ColumnNullable::create(std::move(res_col), std::move(res_null_map)));
+        return Status::OK();
+    }
+
+private:
+    template <typename Impl>
+    static void execute_format_type(ColumnString::Chars& res_data,
+                                    ColumnString::Offsets& res_offsets,
+                                    PaddedPODArray<UInt8>& res_null_map, size_t input_rows_count,
+                                    const ColumnString* right_col) {
+        res_data.reserve(input_rows_count * Impl::ESTIMATE_SIZE);
+        res_offsets.reserve(input_rows_count);
+
+        for (int i = 0; i < input_rows_count; ++i) {
+            StringRef format_ref = right_col->get_data_at(i);
+            std::string format_str(format_ref.data, format_ref.size);
+            std::transform(format_str.begin(), format_str.end(), format_str.begin(), ::toupper);
+
+            std::string_view format_res;
+            if (format_str == "USA") {
+                format_res = Impl::USA;
+            } else if (format_str == "JIS" || format_str == "ISO") {
+                format_res = Impl::JIS_ISO;
+            } else if (format_str == "EUR") {
+                format_res = Impl::EUR;
+            } else if (format_str == "INTERNAL") {
+                format_res = Impl::INTERNAL;
+            } else {
+                res_null_map[i] = 1;
+                res_offsets.push_back(res_data.size());
+                continue;
+            }
+
+            res_data.insert(format_res.data(), format_res.data() + format_res.size());
+            res_offsets.push_back(res_data.size());
+        }
+    }
+
+    struct DateFormatImpl {
+        static constexpr auto USA = "%m.%d.%Y";
+        static constexpr auto JIS_ISO = "%Y-%m-%d";
+        static constexpr auto EUR = "%d.%m.%Y";
+        static constexpr auto INTERNAL = "%Y%m%d";
+        static constexpr size_t ESTIMATE_SIZE = 8;
+    };
+
+    struct DateTimeFormatImpl {
+        static constexpr auto USA = "%Y-%m-%d %H.%i.%s";
+        static constexpr auto JIS_ISO = "%Y-%m-%d %H:%i:%s";
+        static constexpr auto EUR = "%Y-%m-%d %H.%i.%s";
+        static constexpr auto INTERNAL = "%Y%m%d%H%i%s";
+        static constexpr size_t ESTIMATE_SIZE = 17;
+    };
+
+    struct TimeFormatImpl {
+        static constexpr auto USA = "%h:%i:%s %p";
+        static constexpr auto JIS_ISO = "%H:%i:%s";
+        static constexpr auto EUR = "%H.%i.%s";
+        static constexpr auto INTERNAL = "%H%i%s";
+        static constexpr size_t ESTIMATE_SIZE = 11;
+    };
+
+    static constexpr auto DATE_NAME = "DATE";
+    static constexpr auto DATETIME_NAME = "DATETIME";
+    static constexpr auto TIME_NAME = "TIME";
+};
+
+class PeriodHelper {
+public:
+    // For two digit year, 70-99 -> 1970-1999, 00-69 -> 2000-2069
+    // this rule is same as MySQL
+    static constexpr int YY_PART_YEAR = 70;
+    static Status valid_period(int64_t period) {
+        if (period <= 0 || (period % 100) == 0 || (period % 100) > 12) {
+            return Status::InvalidArgument("Period function got invalid period: {}", period);
+        }
+        return Status::OK();
+    }
+
+    static int64_t check_and_convert_period_to_month(uint64_t period) {
+        THROW_IF_ERROR(valid_period(period));
+        uint64_t year = period / 100;
+        if (year < 100) {
+            year += (year >= YY_PART_YEAR) ? 1900 : 2000;
+        }
+        return year * 12LL + (period % 100) - 1;
+    }
+
+    static int64_t convert_month_to_period(uint64_t month) {
+        uint64_t year = month / 12;
+        if (year < 100) {
+            year += (year >= YY_PART_YEAR) ? 1900 : 2000;
+        }
+        return year * 100 + month % 12 + 1;
+    }
+};
+
+class PeriodAddImpl {
+public:
+    static constexpr auto name = "period_add";
+    static size_t get_number_of_arguments() { return 2; }
+    static DataTypePtr get_return_type_impl(const DataTypes& arguments) {
+        return std::make_shared<DataTypeInt64>();
+    }
+
+    static void execute(const std::vector<ColumnWithConstAndNullMap>& cols_info,
+                        ColumnInt64::MutablePtr& res_col, PaddedPODArray<UInt8>& res_null_map_data,
+                        size_t input_rows_count) {
+        const auto& left_data =
+                assert_cast<const ColumnInt64*>(cols_info[0].nested_col)->get_data();
+        const auto& right_data =
+                assert_cast<const ColumnInt64*>(cols_info[1].nested_col)->get_data();
+        for (size_t i = 0; i < input_rows_count; ++i) {
+            if (cols_info[0].is_null_at(i) || cols_info[1].is_null_at(i)) {
+                res_col->insert_default();
+                res_null_map_data[i] = 1;
+                continue;
+            }
+
+            int64_t period = left_data[index_check_const(i, cols_info[0].is_const)];
+            int64_t months = right_data[index_check_const(i, cols_info[1].is_const)];
+            res_col->insert_value(PeriodHelper::convert_month_to_period(
+                    PeriodHelper::check_and_convert_period_to_month(period) + months));
+        }
+    }
+};
+class PeriodDiffImpl {
+public:
+    static constexpr auto name = "period_diff";
+    static size_t get_number_of_arguments() { return 2; }
+    static DataTypePtr get_return_type_impl(const DataTypes& arguments) {
+        return std::make_shared<DataTypeInt64>();
+    }
+
+    static void execute(const std::vector<ColumnWithConstAndNullMap>& cols_info,
+                        ColumnInt64::MutablePtr& res_col, PaddedPODArray<UInt8>& res_null_map_data,
+                        size_t input_rows_count) {
+        const auto& left_data =
+                assert_cast<const ColumnInt64*>(cols_info[0].nested_col)->get_data();
+        const auto& right_data =
+                assert_cast<const ColumnInt64*>(cols_info[1].nested_col)->get_data();
+        for (size_t i = 0; i < input_rows_count; ++i) {
+            if (cols_info[0].is_null_at(i) || cols_info[1].is_null_at(i)) {
+                res_col->insert_default();
+                res_null_map_data[i] = 1;
+                continue;
+            }
+
+            int64_t period1 = left_data[index_check_const(i, cols_info[0].is_const)];
+            int64_t period2 = right_data[index_check_const(i, cols_info[1].is_const)];
+            res_col->insert_value(PeriodHelper::check_and_convert_period_to_month(period1) -
+                                  PeriodHelper::check_and_convert_period_to_month(period2));
+        }
+    }
+};
+
 #include "common/compile_check_avoid_end.h"
 } // namespace doris::vectorized

be/src/vec/functions/function_needs_to_handle_null.h

@@ -0,0 +1,154 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+#pragma once
+#include <boost/mpl/aux_/na_fwd.hpp>
+
+#include "vec/functions/function.h"
+
+namespace doris::vectorized {
+#include "common/compile_check_begin.h"
+
+// Helper struct to store information about const+nullable columns
+struct ColumnWithConstAndNullMap {
+    const IColumn* nested_col = nullptr;
+    const NullMap* null_map = nullptr;
+    bool is_const = false;
+
+    bool is_null_at(size_t row) const { return (null_map && (*null_map)[is_const ? 0 : row]); }
+};
+
+// For functions that need to handle const+nullable column combinations
+// means that functioin `use_default_implementation_for_nulls()` returns false
+template <typename Impl, PrimitiveType ResultPrimitiveType>
+class FunctionNeedsToHandleNull : public IFunction {
+public:
+    using ResultColumnType = PrimitiveTypeTraits<ResultPrimitiveType>::ColumnType;
+
+    static constexpr auto name = Impl::name;
+    String get_name() const override { return name; }
+
+    static std::shared_ptr<IFunction> create() {
+        return std::make_shared<FunctionNeedsToHandleNull>();
+    }
+
+    size_t get_number_of_arguments() const override { return Impl::get_number_of_arguments(); }
+
+    bool is_variadic() const override {
+        if constexpr (requires { Impl::is_variadic(); }) {
+            return Impl::is_variadic();
+        }
+        return false;
+    }
+
+    bool use_default_implementation_for_nulls() const override { return false; }
+
+    DataTypePtr get_return_type_impl(const DataTypes& arguments) const override {
+        return Impl::get_return_type_impl(arguments);
+    }
+
+    Status execute_impl(FunctionContext* context, Block& block, const ColumnNumbers& arguments,
+                        uint32_t result, size_t input_rows_count) const override {
+        auto res_col = ResultColumnType::create();
+        auto null_map = ColumnUInt8::create();
+        auto& null_map_data = null_map->get_data();
+        res_col->reserve(input_rows_count);
+        null_map_data.resize_fill(input_rows_count, 0);
+
+        const size_t arg_size = arguments.size();
+
+        std::vector<ColumnWithConstAndNullMap> columns_info;
+        columns_info.resize(arg_size);
+        bool has_nullable = false;
+        collect_columns_info(columns_info, block, arguments, has_nullable);
+
+        // Check if there is a const null
+        for (size_t i = 0; i < arg_size; ++i) {
+            if (columns_info[i].is_const && columns_info[i].null_map &&
+                (*columns_info[i].null_map)[0] &&
+                execute_const_null(res_col, null_map_data, input_rows_count, i)) {
+                block.replace_by_position(
+                        result, ColumnNullable::create(std::move(res_col), std::move(null_map)));
+                return Status::OK();
+            }
+        }
+
+        Impl::execute(columns_info, res_col, null_map_data, input_rows_count);
+
+        if (is_return_nullable(has_nullable, columns_info)) {
+            block.replace_by_position(
+                    result, ColumnNullable::create(std::move(res_col), std::move(null_map)));
+        } else {
+            block.replace_by_position(result, std::move(res_col));
+        }
+
+        return Status::OK();
+    }
+
+private:
+    // Handle a NULL literal
+    // Default behavior is fill result with all NULLs
+    // return true when the res_col is ready to be written back to the block without further processing
+    bool execute_const_null(typename ResultColumnType::MutablePtr& res_col,
+                            PaddedPODArray<UInt8>& res_null_map_data, size_t input_rows_count,
+                            size_t null_index) const {
+        if constexpr (requires {
+                          Impl::execute_const_null(res_col, res_null_map_data, input_rows_count,
+                                                   null_index);
+                      }) {
+            return Impl::execute_const_null(res_col, res_null_map_data, input_rows_count,
+                                            null_index);
+        }
+
+        res_col->insert_many_defaults(input_rows_count);
+        res_null_map_data.assign(input_rows_count, (UInt8)1);
+
+        return true;
+    }
+
+    // Collect the required information for each column into columns_info
+    // Including whether it is a constant column, nested column and null map(if exists).
+    void collect_columns_info(std::vector<ColumnWithConstAndNullMap>& columns_info,
+                              const Block& block, const ColumnNumbers& arguments,
+                              bool& has_nullable) const {
+        for (size_t i = 0; i < arguments.size(); ++i) {
+            ColumnPtr col_ptr;
+            const auto& col_with_type = block.get_by_position(arguments[i]);
+            std::tie(col_ptr, columns_info[i].is_const) = unpack_if_const(col_with_type.column);
+
+            if (is_column_nullable(*col_ptr)) {
+                has_nullable = true;
+                const auto* nullable = check_and_get_column<ColumnNullable>(col_ptr.get());
+                columns_info[i].nested_col = &nullable->get_nested_column();
+                columns_info[i].null_map = &nullable->get_null_map_data();
+            } else {
+                columns_info[i].nested_col = col_ptr.get();
+            }
+        }
+    }
+
+    // Determine if the return type should be wrapped in nullable
+    // Default behavior is return nullable if any argument is nullable
+    bool is_return_nullable(bool has_nullable,
+                            const std::vector<ColumnWithConstAndNullMap>& cols_info) const {
+        if constexpr (requires { Impl::is_return_nullable(has_nullable, cols_info); }) {
+            return Impl::is_return_nullable(has_nullable, cols_info);
+        }
+        return has_nullable;
+    }
+};
+#include "common/compile_check_end.h"
+} // namespace doris::vectorized