[feature](type)support timestamptz type (#56646) (#59399)

apache/doris-website#2962

TIMESTAMPTZ implementation does not store time zone information with
each row of data, but instead adopts the following mechanism:
1. During storage: All input time values are converted to UTC
(Coordinated Universal Time)
2. During query: Based on the session's time zone setting (specified via
the `time_zone` variable), UTC time is automatically converted to the
corresponding time zone for display

Therefore, TIMESTAMPTZ can be understood as a DATETIME type with time
zone conversion functionality, where Doris automatically handles time
zone conversions internally.

None

- Test <!-- At least one of them must be included. -->
    - [x] Regression test
    - [x] Unit Test
    - [ ] Manual test (add detailed scripts or steps below)
    - [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
- [ ] Previous test can cover this change. - [ ] No code files have been
changed. - [ ] Other reason <!-- Add your reason? -->

- Behavior changed:
    - [x] No.
    - [ ] Yes. <!-- Explain the behavior change -->

- Does this need documentation?
    - [ ] No.
- [x] Yes. <!-- Add document PR link here. eg:
apache/doris-website#1214 -->

- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label <!-- Add branch pick label that this PR
should merge into -->

---------

### What problem does this PR solve?

Issue Number: close #xxx

Related PR: #xxx

Problem Summary:

### Release note

None

### Check List (For Author)

- Test <!-- At least one of them must be included. -->
    - [ ] Regression test
    - [ ] Unit Test
    - [ ] Manual test (add detailed scripts or steps below)
    - [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
        - [ ] Previous test can cover this change.
        - [ ] No code files have been changed.
        - [ ] Other reason <!-- Add your reason?  -->

- Behavior changed:
    - [ ] No.
    - [ ] Yes. <!-- Explain the behavior change -->

- Does this need documentation?
    - [ ] No.
- [ ] Yes. <!-- Add document PR link here. eg:
apache/doris-website#1214 -->

### Check List (For Reviewer who merge this PR)

- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label <!-- Add branch pick label that this PR
should merge into -->

---------

Co-authored-by: Mryange <[email protected]>

Author: jacktengg

be/src/exec/olap_common.h

@@ -70,13 +70,15 @@ std::string cast_to_string(T value, int scale) {
     } else if constexpr (primitive_type == TYPE_LARGEINT) {
         return vectorized::int128_to_string(value);
     } else if constexpr (primitive_type == TYPE_DATETIMEV2) {
-        DateV2Value<DateTimeV2ValueType> datetimev2_val =
-                static_cast<DateV2Value<DateTimeV2ValueType>>(value);
+        auto datetimev2_val = static_cast<DateV2Value<DateTimeV2ValueType>>(value);
         char buf[30];
         datetimev2_val.to_string(buf);
         std::stringstream ss;
         ss << buf;
         return ss.str();
+    } else if constexpr (primitive_type == TYPE_TIMESTAMPTZ) {
+        auto timestamptz_val = static_cast<TimestampTzValue>(value);
+        return timestamptz_val.to_string(cctz::utc_time_zone(), scale);
     } else if constexpr (primitive_type == TYPE_TIMEV2) {
         return TimeValue::to_string(value, scale);
     } else if constexpr (primitive_type == TYPE_IPV4) {
@@ -103,16 +105,10 @@ class ColumnValueRange {
 
     ColumnValueRange(std::string col_name);
 
-    ColumnValueRange(std::string col_name, const CppType& min, const CppType& max,
-                     bool contain_null);
-
     ColumnValueRange(std::string col_name, int precision, int scale);
 
     ColumnValueRange(std::string col_name, bool is_nullable_col, int precision, int scale);
 
-    ColumnValueRange(std::string col_name, const CppType& min, const CppType& max,
-                     bool is_nullable_col, bool contain_null, int precision, int scale);
-
     // should add fixed value before add range
     Status add_fixed_value(const CppType& value);
 
@@ -368,6 +364,12 @@ class ColumnValueRange {
     bool is_in_range(const CppType& value);
 
 private:
+    ColumnValueRange(std::string col_name, const CppType& min, const CppType& max,
+                     bool contain_null);
+
+    ColumnValueRange(std::string col_name, const CppType& min, const CppType& max,
+                     bool is_nullable_col, bool contain_null, int precision, int scale);
+
     const static CppType TYPE_MIN; // Column type's min value
     const static CppType TYPE_MAX; // Column type's max value
 
@@ -396,6 +398,7 @@ class ColumnValueRange {
             primitive_type == PrimitiveType::TYPE_BOOLEAN ||
             primitive_type == PrimitiveType::TYPE_DATETIME ||
             primitive_type == PrimitiveType::TYPE_DATETIMEV2 ||
+            primitive_type == PrimitiveType::TYPE_TIMESTAMPTZ ||
             primitive_type == PrimitiveType::TYPE_DECIMAL256;
 
     int _runtime_filter_id = -1;
@@ -488,10 +491,10 @@ using ColumnValueRangeType = std::variant<
         ColumnValueRange<TYPE_IPV6>, ColumnValueRange<TYPE_CHAR>, ColumnValueRange<TYPE_VARCHAR>,
         ColumnValueRange<TYPE_STRING>, ColumnValueRange<TYPE_DATE>, ColumnValueRange<TYPE_DATEV2>,
         ColumnValueRange<TYPE_DATETIME>, ColumnValueRange<TYPE_DATETIMEV2>,
-        ColumnValueRange<TYPE_DECIMALV2>, ColumnValueRange<TYPE_BOOLEAN>,
-        ColumnValueRange<TYPE_HLL>, ColumnValueRange<TYPE_DECIMAL32>,
-        ColumnValueRange<TYPE_DECIMAL64>, ColumnValueRange<TYPE_DECIMAL128I>,
-        ColumnValueRange<TYPE_DECIMAL256>>;
+        ColumnValueRange<TYPE_TIMESTAMPTZ>, ColumnValueRange<TYPE_DECIMALV2>,
+        ColumnValueRange<TYPE_BOOLEAN>, ColumnValueRange<TYPE_HLL>,
+        ColumnValueRange<TYPE_DECIMAL32>, ColumnValueRange<TYPE_DECIMAL64>,
+        ColumnValueRange<TYPE_DECIMAL128I>, ColumnValueRange<TYPE_DECIMAL256>>;
 
 template <PrimitiveType primitive_type>
 const typename ColumnValueRange<primitive_type>::CppType

be/src/exec/schema_scanner.cpp

@@ -70,6 +70,7 @@
 #include "pipeline/dependency.h"
 #include "runtime/define_primitive_type.h"
 #include "runtime/fragment_mgr.h"
+#include "runtime/primitive_type.h"
 #include "runtime/types.h"
 #include "util/string_util.h"
 #include "util/types.h"
@@ -383,6 +384,12 @@ Status SchemaScanner::fill_dest_column_for_range(vectorized::Block* block, size_
             break;
         }
 
+        case TYPE_TIMESTAMPTZ: {
+            uint64_t num = *reinterpret_cast<uint64_t*>(data);
+            assert_cast<vectorized::ColumnTimeStampTz*>(col_ptr)->insert_value(num);
+            break;
+        }
+
         case TYPE_DECIMALV2: {
             const vectorized::Int128 num = (reinterpret_cast<PackedInt128*>(data))->value;
             assert_cast<vectorized::ColumnDecimal128V2*>(col_ptr)->insert_data(

be/src/exec/schema_scanner/schema_columns_scanner.cpp

@@ -137,6 +137,8 @@ std::string SchemaColumnsScanner::_to_mysql_data_type_string(TColumnDesc& desc)
     case TPrimitiveType::DATETIME:
     case TPrimitiveType::DATETIMEV2:
         return "datetime";
+    case TPrimitiveType::TIMESTAMPTZ:
+        return "timestamp";
     case TPrimitiveType::DECIMAL32:
     case TPrimitiveType::DECIMAL64:
     case TPrimitiveType::DECIMAL128I:
@@ -238,6 +240,16 @@ std::string SchemaColumnsScanner::_type_to_string(TColumnDesc& desc) {
         }
         return fmt::to_string(debug_string_buffer);
     }
+    case TPrimitiveType::TIMESTAMPTZ: {
+        fmt::memory_buffer debug_string_buffer;
+        if (!desc.__isset.columnScale || desc.columnScale == 0) {
+            fmt::format_to(debug_string_buffer, "timestamp");
+        } else {
+            fmt::format_to(debug_string_buffer, "timestamp({})",
+                           desc.__isset.columnScale ? std::to_string(desc.columnScale) : "UNKNOWN");
+        }
+        return fmt::to_string(debug_string_buffer);
+    }
     case TPrimitiveType::HLL: {
         return "hll";
     }
@@ -569,7 +581,8 @@ Status SchemaColumnsScanner::_fill_block_impl(vectorized::Block* block) {
         for (int i = 0; i < columns_num; ++i) {
             int data_type = _desc_result.columns[i].columnDesc.columnType;
             if (_desc_result.columns[i].columnDesc.__isset.columnScale &&
-                data_type == TPrimitiveType::DATETIMEV2) {
+                (data_type == TPrimitiveType::DATETIMEV2 ||
+                 data_type == TPrimitiveType::TIMESTAMPTZ)) {
                 srcs[i] = _desc_result.columns[i].columnDesc.columnScale;
                 datas[i] = srcs.data() + i;
             } else {

be/src/exec/table_connector.cpp

@@ -99,175 +99,5 @@ std::u16string TableConnector::utf8_to_u16string(const char* first, const char*
     return result;
 }
 
-Status TableConnector::convert_column_data(const vectorized::ColumnPtr& column_ptr,
-                                           const vectorized::DataTypePtr& type_ptr,
-                                           const vectorized::DataTypePtr& type, size_t row,
-                                           TOdbcTableType::type table_type) {
-    auto extra_convert_func = [&](const std::string_view& str, const bool& is_date) -> void {
-        if (table_type == TOdbcTableType::ORACLE || table_type == TOdbcTableType::SAP_HANA) {
-            //if is ORACLE and date type, insert into need convert
-            if (is_date) {
-                fmt::format_to(_insert_stmt_buffer, "to_date('{}','yyyy-mm-dd')", str);
-            } else {
-                fmt::format_to(_insert_stmt_buffer, "to_date('{}','yyyy-mm-dd hh24:mi:ss')", str);
-            }
-        } else if (table_type == TOdbcTableType::POSTGRESQL) {
-            fmt::format_to(_insert_stmt_buffer, "'{}'::date", str);
-        } else if (table_type == TOdbcTableType::SQLSERVER) {
-            // Values in sqlserver should be enclosed by single quotes
-            fmt::format_to(_insert_stmt_buffer, "'{}'", str);
-        } else {
-            fmt::format_to(_insert_stmt_buffer, "\"{}\"", str);
-        }
-    };
-    const vectorized::IColumn* column = column_ptr.get();
-    if (type_ptr->is_nullable()) {
-        const auto* nullable_column =
-                assert_cast<const vectorized::ColumnNullable*>(column_ptr.get());
-        if (nullable_column->is_null_at(row)) {
-            fmt::format_to(_insert_stmt_buffer, "{}", "NULL");
-            return Status::OK();
-        }
-        column = nullable_column->get_nested_column_ptr().get();
-    } else {
-        column = column_ptr.get();
-    }
-    auto [item, size] = column->get_data_at(row);
-    switch (type->get_primitive_type()) {
-    case TYPE_BOOLEAN:
-        if (table_type == TOdbcTableType::SAP_HANA) {
-            fmt::format_to(_insert_stmt_buffer, "{}", *reinterpret_cast<const bool*>(item));
-        } else {
-            fmt::format_to(_insert_stmt_buffer, "{}", *reinterpret_cast<const int8_t*>(item));
-        }
-        break;
-    case TYPE_TINYINT:
-        fmt::format_to(_insert_stmt_buffer, "{}", *reinterpret_cast<const int8_t*>(item));
-        break;
-    case TYPE_SMALLINT:
-        fmt::format_to(_insert_stmt_buffer, "{}", *reinterpret_cast<const int16_t*>(item));
-        break;
-    case TYPE_INT:
-        fmt::format_to(_insert_stmt_buffer, "{}", *reinterpret_cast<const int32_t*>(item));
-        break;
-    case TYPE_BIGINT:
-        fmt::format_to(_insert_stmt_buffer, "{}", *reinterpret_cast<const int64_t*>(item));
-        break;
-    case TYPE_FLOAT:
-        fmt::format_to(_insert_stmt_buffer, "{}", *reinterpret_cast<const float*>(item));
-        break;
-    case TYPE_DOUBLE:
-        fmt::format_to(_insert_stmt_buffer, "{}", *reinterpret_cast<const double*>(item));
-        break;
-    case TYPE_DATE: {
-        VecDateTimeValue value = binary_cast<int64_t, doris::VecDateTimeValue>(*(int64_t*)item);
-
-        char buf[64];
-        char* pos = value.to_string(buf);
-        std::string_view str(buf, pos - buf - 1);
-        extra_convert_func(str, true);
-        break;
-    }
-    case TYPE_DATETIME: {
-        VecDateTimeValue value = binary_cast<int64_t, doris::VecDateTimeValue>(*(int64_t*)item);
-
-        char buf[64];
-        char* pos = value.to_string(buf);
-        std::string_view str(buf, pos - buf - 1);
-        extra_convert_func(str, false);
-        break;
-    }
-    case TYPE_DATEV2: {
-        DateV2Value<DateV2ValueType> value =
-                binary_cast<uint32_t, DateV2Value<DateV2ValueType>>(*(int32_t*)item);
-
-        char buf[64];
-        char* pos = value.to_string(buf);
-        std::string str(buf, pos - buf - 1);
-        extra_convert_func(str, true);
-        break;
-    }
-    case TYPE_DATETIMEV2: {
-        DateV2Value<DateTimeV2ValueType> value =
-                binary_cast<uint64_t, DateV2Value<DateTimeV2ValueType>>(*(int64_t*)item);
-
-        char buf[64];
-        char* pos = value.to_string(buf, type->get_scale());
-        std::string str(buf, pos - buf - 1);
-        extra_convert_func(str, false);
-        break;
-    }
-    case TYPE_VARCHAR:
-    case TYPE_CHAR:
-    case TYPE_STRING: {
-        // for oracle/pg database string must be '
-        if (table_type == TOdbcTableType::ORACLE || table_type == TOdbcTableType::POSTGRESQL ||
-            table_type == TOdbcTableType::SAP_HANA || table_type == TOdbcTableType::MYSQL ||
-            table_type == TOdbcTableType::CLICKHOUSE || table_type == TOdbcTableType::SQLSERVER) {
-            fmt::format_to(_insert_stmt_buffer, "'{}'", fmt::basic_string_view(item, size));
-        } else {
-            fmt::format_to(_insert_stmt_buffer, "\"{}\"", fmt::basic_string_view(item, size));
-        }
-        break;
-    }
-    case TYPE_ARRAY: {
-        auto& arr_nested = reinterpret_cast<const vectorized::ColumnArray*>(column)->get_data_ptr();
-        auto& arr_offset = reinterpret_cast<const vectorized::ColumnArray*>(column)->get_offsets();
-        auto array_type = remove_nullable(type_ptr);
-        auto nested_type =
-                reinterpret_cast<const vectorized::DataTypeArray&>(*array_type).get_nested_type();
-
-        //for doris、CK insert into --->  []
-        //for PG        insert into ---> ARRAY[]
-        if (table_type == TOdbcTableType::POSTGRESQL) {
-            fmt::format_to(_insert_stmt_buffer, "{}", "ARRAY[");
-        } else if (table_type == TOdbcTableType::CLICKHOUSE ||
-                   table_type == TOdbcTableType::MYSQL) {
-            fmt::format_to(_insert_stmt_buffer, "{}", "[");
-        }
-        bool first_value = true;
-        for (auto idx = arr_offset[row - 1]; idx < arr_offset[row]; ++idx) {
-            if (first_value == false) {
-                fmt::format_to(_insert_stmt_buffer, "{}", ", ");
-            }
-            if (arr_nested->is_null_at(idx)) {
-                fmt::format_to(_insert_stmt_buffer, "{}", "NULL");
-            } else {
-                RETURN_IF_ERROR(convert_column_data(arr_nested, nested_type,
-                                                    assert_cast<const vectorized::DataTypeArray*>(
-                                                            vectorized::remove_nullable(type).get())
-                                                            ->get_nested_type(),
-                                                    idx, table_type));
-            }
-            first_value = false;
-        }
-        fmt::format_to(_insert_stmt_buffer, "{}", "]");
-        break;
-    }
-    case TYPE_DECIMALV2: {
-        DecimalV2Value value = *(DecimalV2Value*)(item);
-        fmt::format_to(_insert_stmt_buffer, "{}", value.to_string());
-        break;
-    }
-    case TYPE_DECIMAL32:
-    case TYPE_DECIMAL64:
-    case TYPE_DECIMAL128I:
-    case TYPE_DECIMAL256: {
-        auto decimal_type = remove_nullable(type_ptr);
-        auto val = decimal_type->to_string(*column, row);
-        fmt::format_to(_insert_stmt_buffer, "{}", val);
-        break;
-    }
-    case TYPE_LARGEINT: {
-        fmt::format_to(_insert_stmt_buffer, "{}", *reinterpret_cast<const __int128*>(item));
-        break;
-    }
-    default: {
-        return Status::InternalError("can't convert this type to mysql type. type = {}",
-                                     type->get_name());
-    }
-    }
-    return Status::OK();
-}
 #include "common/compile_check_end.h"
 } // namespace doris

be/src/exec/table_connector.h

@@ -71,11 +71,6 @@ class TableConnector {
 
     std::u16string utf8_to_u16string(const char* first, const char* last);
 
-    Status convert_column_data(const vectorized::ColumnPtr& column_ptr,
-                               const vectorized::DataTypePtr& type_ptr,
-                               const vectorized::DataTypePtr& type, size_t row,
-                               TOdbcTableType::type table_type);
-
     // Default max buffer size use in insert to: 50MB, normally a batch is smaller than the size
     static constexpr uint32_t INSERT_BUFFER_SIZE = 1024l * 1024 * 50;
 

be/src/exec/tablet_info.cpp

@@ -53,6 +53,7 @@
 // NOLINTNEXTLINE(unused-includes)
 #include "vec/exprs/vexpr_context.h" // IWYU pragma: keep
 #include "vec/exprs/vliteral.h"
+#include "vec/functions/cast/cast_to_timestamptz.h"
 #include "vec/runtime/vdatetime_value.h"
 
 namespace doris {
@@ -559,9 +560,10 @@ static Status _create_partition_key(const TExprNode& t_expr, BlockRow* part_key,
     //TODO: use assert_cast before insert_data
     switch (t_expr.node_type) {
     case TExprNodeType::DATE_LITERAL: {
-        if (vectorized::DataTypeFactory::instance()
-                    .create_data_type(t_expr.type)
-                    ->get_primitive_type() == TYPE_DATEV2) {
+        auto primitive_type = vectorized::DataTypeFactory::instance()
+                                      .create_data_type(t_expr.type)
+                                      ->get_primitive_type();
+        if (primitive_type == TYPE_DATEV2) {
             DateV2Value<DateV2ValueType> dt;
             if (!dt.from_date_str(t_expr.date_literal.value.c_str(),
                                   t_expr.date_literal.value.size())) {
@@ -570,9 +572,7 @@ static Status _create_partition_key(const TExprNode& t_expr, BlockRow* part_key,
                 return Status::InternalError(ss.str());
             }
             column->insert_data(reinterpret_cast<const char*>(&dt), 0);
-        } else if (vectorized::DataTypeFactory::instance()
-                           .create_data_type(t_expr.type)
-                           ->get_primitive_type() == TYPE_DATETIMEV2) {
+        } else if (primitive_type == TYPE_DATETIMEV2) {
             DateV2Value<DateTimeV2ValueType> dt;
             const int32_t scale =
                     t_expr.type.types.empty() ? -1 : t_expr.type.types.front().scalar_type.scale;
@@ -583,6 +583,21 @@ static Status _create_partition_key(const TExprNode& t_expr, BlockRow* part_key,
                 return Status::InternalError(ss.str());
             }
             column->insert_data(reinterpret_cast<const char*>(&dt), 0);
+        } else if (primitive_type == TYPE_TIMESTAMPTZ) {
+            TimestampTzValue res;
+            vectorized::CastParameters params {.status = Status::OK(), .is_strict = true};
+            const int32_t scale =
+                    t_expr.type.types.empty() ? -1 : t_expr.type.types.front().scalar_type.scale;
+            if (!vectorized::CastToTimstampTz::from_string(
+                        {t_expr.date_literal.value.c_str(), t_expr.date_literal.value.size()}, res,
+                        params, nullptr, scale)) [[unlikely]] {
+                std::stringstream ss;
+                ss << "invalid timestamptz literal in partition column, value="
+                   << t_expr.date_literal;
+                return Status::InternalError(ss.str());
+            } else {
+                column->insert_data(reinterpret_cast<const char*>(&res), 0);
+            }
         } else {
             // TYPE_DATE (DATEV1) or TYPE_DATETIME (DATETIMEV1)
             VecDateTimeValue dt;

be/src/exprs/bloom_filter_func_impl.h

@@ -44,6 +44,8 @@ struct fixed_len_to_uint32_v2 {
             } else if constexpr (std::is_same_v<T, DateV2Value<DateTimeV2ValueType>>) {
                 return uint32_t(HashCRC32<DateV2Value<DateTimeV2ValueType>::underlying_value>()(
                         value.to_date_int_val()));
+            } else if constexpr (std::is_same_v<T, TimestampTzValue>) {
+                return uint32_t(HashCRC32<typename T::underlying_value>()(value.to_date_int_val()));
             } else if constexpr (vectorized::IsDecimalNumber<T>) {
                 return uint32_t(HashCRC32<typename T::NativeType>()(value.value));
             } else {

be/src/exprs/create_predicate_function.h

@@ -103,6 +103,7 @@ class PredicateFunctionCreator {
     M(TYPE_DATETIME)          \
     M(TYPE_DATEV2)            \
     M(TYPE_DATETIMEV2)        \
+    M(TYPE_TIMESTAMPTZ)       \
     M(TYPE_CHAR)              \
     M(TYPE_VARCHAR)           \
     M(TYPE_STRING)            \