branch-4.1: [Fix](ttz) Fix TIMESTAMPTZ elapsed-time semantics to use UTC #63161

be/src/core/value/timestamptz_value.h

@@ -96,6 +96,10 @@ class TimestampTzValue {
         return _utc_dt.datetime_diff_in_seconds(other._utc_dt);
     }
 
+    int64_t datetime_diff_in_microseconds(const TimestampTzValue& other) const {
+        return _utc_dt.datetime_diff_in_microseconds(other._utc_dt);
+    }
+
     template <TimeUnit unit>
     bool date_set_interval(const TimeInterval& interval) {
         return _utc_dt.date_set_interval<unit>(interval);
@@ -143,6 +147,8 @@ class TimestampTzValue {
         _utc_dt.unix_timestamp(timestamp, ctz);
     }
 
+    std::string debug_string() const { return _utc_dt.debug_string(); }
+
     // Convert UTC time to local time based on the given timezone
     void convert_utc_to_local(const cctz::time_zone& local_time_zone,
                               DateV2Value<DateTimeV2ValueType>& dt) const;

be/src/exprs/aggregate/aggregate_function_sequence_match.cpp

@@ -52,6 +52,9 @@ AggregateFunctionPtr create_aggregate_function_sequence_base(const std::string&
     case TYPE_DATEV2:
         return creator_without_type::create<AggregateFunction<TYPE_DATEV2>>(
                 argument_types, result_is_nullable, attr);
+    case TYPE_TIMESTAMPTZ:
+        return creator_without_type::create<AggregateFunction<TYPE_TIMESTAMPTZ>>(
+                argument_types, result_is_nullable, attr);
     default:
         return nullptr;
     }

be/src/exprs/aggregate/aggregate_function_sequence_match.h

@@ -74,10 +74,6 @@ constexpr auto sequence_match_max_iterations = 1000000l;
 template <PrimitiveType T, typename Derived>
 struct AggregateFunctionSequenceMatchData final {
     using Timestamp = typename PrimitiveTypeTraits<T>::CppType;
-    using NativeType =
-            std::conditional_t<T == TYPE_DATEV2, uint32_t,
-                               std::conditional_t<T == TYPE_DATETIMEV2, uint64_t,
-                                                  typename PrimitiveTypeTraits<T>::CppType>>;
     using Events = std::bitset<MAX_EVENTS>;
     using TimestampEvents = std::pair<Timestamp, Events>;
     using Comparator = ComparePairFirst<std::less>;
@@ -280,7 +276,7 @@ struct AggregateFunctionSequenceMatchData final {
                         return;
                     }
 
-                    NativeType duration = 0;
+                    uint64_t duration = 0;
                     if (!parse_uint(duration)) {
                         throw_exception("Could not parse number");
                         return;
@@ -618,7 +614,6 @@ class AggregateFunctionSequenceBase
         : public IAggregateFunctionDataHelper<AggregateFunctionSequenceMatchData<T, Derived>,
                                               Derived> {
 public:
-    using NativeType = typename PrimitiveTypeTraits<T>::CppType;
     AggregateFunctionSequenceBase(const DataTypes& arguments)
             : IAggregateFunctionDataHelper<AggregateFunctionSequenceMatchData<T, Derived>, Derived>(
                       arguments) {

be/src/exprs/aggregate/aggregate_function_window_funnel.cpp

@@ -38,10 +38,13 @@ AggregateFunctionPtr create_aggregate_function_window_funnel(const std::string&
         return nullptr;
     }
     if (argument_types[2]->get_primitive_type() == TYPE_DATETIMEV2) {
-        return creator_without_type::create<AggregateFunctionWindowFunnel>(
+        return creator_without_type::create<AggregateFunctionWindowFunnel<TYPE_DATETIMEV2>>(
+                argument_types, result_is_nullable, attr);
+    } else if (argument_types[2]->get_primitive_type() == TYPE_TIMESTAMPTZ) {
+        return creator_without_type::create<AggregateFunctionWindowFunnel<TYPE_TIMESTAMPTZ>>(
                 argument_types, result_is_nullable, attr);
     } else {
-        LOG(WARNING) << "Only support DateTime type as window argument!";
+        LOG(WARNING) << "Only support DateTime or TimeStampTz type as window argument!";
         return nullptr;
     }
 }

be/src/exprs/aggregate/aggregate_function_window_funnel.h

@@ -70,9 +70,11 @@ inline WindowFunnelMode string_to_window_funnel_mode(const String& string) {
     }
 }
 
+template <PrimitiveType T>
 struct DataValue {
     using TimestampEvent = std::vector<ColumnUInt8::Container>;
-    std::vector<DateV2Value<DateTimeV2ValueType>> dt;
+    using DateValueType = typename PrimitiveTypeTraits<T>::CppType;
+    std::vector<DateValueType> dt;
     TimestampEvent event_columns_data;
     bool operator<(const DataValue& other) const { return dt < other.dt; }
     void clear() {
@@ -97,15 +99,16 @@ struct DataValue {
     }
 };
 
+template <PrimitiveType T>
 struct WindowFunnelState {
-    static constexpr PrimitiveType PType = PrimitiveType::TYPE_DATETIMEV2;
-    using NativeType = UInt64;
-    using DateValueType = DateV2Value<DateTimeV2ValueType>;
+    static constexpr PrimitiveType PType = T;
+    using NativeType = typename PrimitiveTypeTraits<T>::StorageFieldType;
+    using DateValueType = typename PrimitiveTypeTraits<T>::CppType;
     int event_count = 0;
     int64_t window;
     bool enable_mode;
     WindowFunnelMode window_funnel_mode;
-    DataValue events_list;
+    DataValue<T> events_list;
 
     WindowFunnelState() {
         event_count = 0;
@@ -123,7 +126,8 @@ struct WindowFunnelState {
         window = win;
         window_funnel_mode = enable_mode ? mode : WindowFunnelMode::DEFAULT;
         events_list.dt.emplace_back(
-                assert_cast<const ColumnVector<PType>&>(*arg_columns[2]).get_data()[row_num]);
+                assert_cast<const typename PrimitiveTypeTraits<PType>::ColumnType&>(*arg_columns[2])
+                        .get_data()[row_num]);
         for (int i = 0; i < event_count; i++) {
             events_list.event_columns_data[i].emplace_back(
                     assert_cast<const ColumnUInt8&>(*arg_columns[3 + i]).get_data()[row_num]);
@@ -268,7 +272,7 @@ struct WindowFunnelState {
         }
     }
 
-    void merge(const WindowFunnelState& other) {
+    void merge(const WindowFunnelState<T>& other) {
         if (other.events_list.empty()) {
             return;
         }
@@ -327,7 +331,9 @@ struct WindowFunnelState {
         events_list.clear();
         events_list.dt.resize(size);
         for (auto i = 0; i < size; i++) {
-            read_var_int(*reinterpret_cast<Int64*>(&events_list.dt[i]), in);
+            Int64 timestamp = 0;
+            read_var_int(timestamp, in);
+            events_list.dt[i] = DateValueType(static_cast<UInt64>(timestamp));
         }
         events_list.event_columns_data.resize(event_count);
         for (int64_t i = 0; i < event_count; i++) {
@@ -342,17 +348,19 @@ struct WindowFunnelState {
     }
 };
 
+template <PrimitiveType T>
 class AggregateFunctionWindowFunnel final
-        : public IAggregateFunctionDataHelper<WindowFunnelState, AggregateFunctionWindowFunnel>,
+        : public IAggregateFunctionDataHelper<WindowFunnelState<T>,
+                                              AggregateFunctionWindowFunnel<T>>,
           MultiExpression,
           NullableAggregateFunction {
 public:
     AggregateFunctionWindowFunnel(const DataTypes& argument_types_)
-            : IAggregateFunctionDataHelper<WindowFunnelState, AggregateFunctionWindowFunnel>(
+            : IAggregateFunctionDataHelper<WindowFunnelState<T>, AggregateFunctionWindowFunnel<T>>(
                       argument_types_) {}
 
     void create(AggregateDataPtr __restrict place) const override {
-        auto data = new (place) WindowFunnelState(
+        auto data = new (place) WindowFunnelState<T>(
                 cast_set<int>(IAggregateFunction::get_argument_types().size() - 3));
         /// support window funnel mode from 2.0. See `BeExecVersionManager::max_be_exec_version`
         data->enable_mode = IAggregateFunction::version >= 3;
@@ -390,8 +398,8 @@ class AggregateFunctionWindowFunnel final
         // place is essentially an AggregateDataPtr, passed as a ConstAggregateDataPtr.
         this->data(const_cast<AggregateDataPtr>(place)).sort();
         assert_cast<ColumnInt32&>(to).get_data().push_back(
-                IAggregateFunctionDataHelper<WindowFunnelState,
-                                             AggregateFunctionWindowFunnel>::data(place)
+                IAggregateFunctionDataHelper<WindowFunnelState<T>,
+                                             AggregateFunctionWindowFunnel<T>>::data(place)
                         .get());
     }
 

be/src/exprs/aggregate/aggregate_function_window_funnel_v2.cpp

@@ -40,10 +40,13 @@ AggregateFunctionPtr create_aggregate_function_window_funnel_v2(const std::strin
         return nullptr;
     }
     if (argument_types[2]->get_primitive_type() == TYPE_DATETIMEV2) {
-        return creator_without_type::create<AggregateFunctionWindowFunnelV2>(
+        return creator_without_type::create<AggregateFunctionWindowFunnelV2<TYPE_DATETIMEV2>>(
+                argument_types, result_is_nullable, attr);
+    } else if (argument_types[2]->get_primitive_type() == TYPE_TIMESTAMPTZ) {
+        return creator_without_type::create<AggregateFunctionWindowFunnelV2<TYPE_TIMESTAMPTZ>>(
                 argument_types, result_is_nullable, attr);
     } else {
-        LOG(WARNING) << "Only support DateTime type as window argument!";
+        LOG(WARNING) << "Only support DateTime or TimeStampTz type as window argument!";
         return nullptr;
     }
 }

be/src/exprs/aggregate/aggregate_function_window_funnel_v2.h

@@ -77,6 +77,7 @@ void merge_events_list(T& events_list, size_t prefix_size, bool prefix_sorted, b
 /// The algorithm uses this to ensure each funnel step comes from a different row.
 ///
 /// This approach adds ZERO storage overhead — each event remains 9 bytes (UInt64 + UInt8).
+template <PrimitiveType T>
 struct WindowFunnelStateV2 {
     /// (timestamp_int_val, 1-based event_index with continuation flag in bit 7)
     ///
@@ -136,10 +137,10 @@ struct WindowFunnelStateV2 {
         window = win;
         window_funnel_mode = mode;
 
-        // get_data() returns DateV2Value<DateTimeV2ValueType>; convert to packed UInt64
-        auto timestamp = assert_cast<const ColumnVector<TYPE_DATETIMEV2>&>(*arg_columns[2])
-                                 .get_data()[row_num]
-                                 .to_date_int_val();
+        auto timestamp =
+                assert_cast<const typename PrimitiveTypeTraits<T>::ColumnType&>(*arg_columns[2])
+                        .get_data()[row_num]
+                        .to_date_int_val();
 
         // Iterate from last event to first (reverse order).
         // This ensures that after stable_sort, events with the same timestamp
@@ -174,7 +175,7 @@ struct WindowFunnelStateV2 {
         }
     }
 
-    void merge(const WindowFunnelStateV2& other) {
+    void merge(const WindowFunnelStateV2<T>& other) {
         if (other.events_list.empty()) {
             return;
         }
@@ -617,20 +618,22 @@ struct WindowFunnelStateV2 {
     }
 };
 
+template <PrimitiveType T>
 class AggregateFunctionWindowFunnelV2 final
-        : public IAggregateFunctionDataHelper<WindowFunnelStateV2, AggregateFunctionWindowFunnelV2>,
+        : public IAggregateFunctionDataHelper<WindowFunnelStateV2<T>,
+                                              AggregateFunctionWindowFunnelV2<T>>,
           MultiExpression,
           NullableAggregateFunction {
 public:
     AggregateFunctionWindowFunnelV2(const DataTypes& argument_types_)
-            : IAggregateFunctionDataHelper<WindowFunnelStateV2, AggregateFunctionWindowFunnelV2>(
-                      argument_types_) {
-        WindowFunnelStateV2::validate_event_count(
+            : IAggregateFunctionDataHelper<WindowFunnelStateV2<T>,
+                                           AggregateFunctionWindowFunnelV2<T>>(argument_types_) {
+        WindowFunnelStateV2<T>::validate_event_count(
                 cast_set<int>(IAggregateFunction::get_argument_types().size() - 3));
     }
 
     void create(AggregateDataPtr __restrict place) const override {
-        new (place) WindowFunnelStateV2(
+        new (place) WindowFunnelStateV2<T>(
                 cast_set<int>(IAggregateFunction::get_argument_types().size() - 3));
     }
 
@@ -665,8 +668,8 @@ class AggregateFunctionWindowFunnelV2 final
     void insert_result_into(ConstAggregateDataPtr __restrict place, IColumn& to) const override {
         this->data(const_cast<AggregateDataPtr>(place)).sort();
         assert_cast<ColumnInt32&>(to).get_data().push_back(
-                IAggregateFunctionDataHelper<WindowFunnelStateV2,
-                                             AggregateFunctionWindowFunnelV2>::data(place)
+                IAggregateFunctionDataHelper<WindowFunnelStateV2<T>,
+                                             AggregateFunctionWindowFunnelV2<T>>::data(place)
                         .get());
     }
 

be/src/exprs/function/function_date_or_datetime_computation.cpp

@@ -186,7 +186,8 @@ using FunctionSubTimeTimestampTz =
 
 #define ALL_FUNCTION_TIME_DIFF(NAME, IMPL)          \
     FUNCTION_TIME_DIFF(NAME, IMPL, TYPE_DATETIMEV2) \
-    FUNCTION_TIME_DIFF(NAME, IMPL, TYPE_DATEV2)
+    FUNCTION_TIME_DIFF(NAME, IMPL, TYPE_DATEV2)     \
+    FUNCTION_TIME_DIFF(NAME, IMPL, TYPE_TIMESTAMPTZ)
 // these diff functions accept all v2 types. but for v1 only datetime.
 ALL_FUNCTION_TIME_DIFF(FunctionDatetimeDateDiff, DateDiffImpl)
 ALL_FUNCTION_TIME_DIFF(FunctionDatetimeTimeDiff, TimeDiffImpl)
@@ -324,7 +325,8 @@ void register_function_date_time_computation(SimpleFunctionFactory& factory) {
 
 #define REGISTER_ALL_DATEV2_FUNCTIONS_DIFF(NAME)          \
     REGISTER_DATEV2_FUNCTIONS_DIFF(NAME, TYPE_DATETIMEV2) \
-    REGISTER_DATEV2_FUNCTIONS_DIFF(NAME, TYPE_DATEV2)
+    REGISTER_DATEV2_FUNCTIONS_DIFF(NAME, TYPE_DATEV2)     \
+    REGISTER_DATEV2_FUNCTIONS_DIFF(NAME, TYPE_TIMESTAMPTZ)
 
     REGISTER_ALL_DATEV2_FUNCTIONS_DIFF(FunctionDatetimeDateDiff)
     REGISTER_ALL_DATEV2_FUNCTIONS_DIFF(FunctionDatetimeTimeDiff)

be/src/exprs/function/function_date_or_datetime_computation.h

@@ -573,7 +573,8 @@ struct TimeDiffImpl {
     using ValueType = typename PrimitiveTypeTraits<DateType>::CppType;
     using ArgType = typename PrimitiveTypeTraits<DateType>::DataType::FieldType;
     //TODO: remove V1 since FE already removed it.
-    static constexpr bool UsingTimev2 = is_date_v2_or_datetime_v2(DateType);
+    static constexpr bool UsingTimev2 =
+            is_date_v2_or_datetime_v2(DateType) || DateType == TYPE_TIMESTAMPTZ;
     static constexpr PrimitiveType ReturnType = TYPE_TIMEV2;
 
     static constexpr auto name = "timediff";
@@ -601,8 +602,19 @@ struct TimeDiffImpl {
         return std::make_shared<DataTypeTimeV2>(arguments[0].type->get_scale());
     }
 };
+
+template <TimeUnit UNIT, typename T0, typename T1>
+int64_t diff_on_utc_datetime(const T0& ts1, const T1& ts0) {
+    return datetime_diff<UNIT>(ts1, ts0);
+}
+
+template <TimeUnit UNIT>
+int64_t diff_on_utc_datetime(const TimestampTzValue& ts1, const TimestampTzValue& ts0) {
+    return datetime_diff<UNIT>(ts1.utc_dt(), ts0.utc_dt());
+}
+
 #define TIME_DIFF_FUNCTION_IMPL(CLASS, NAME, UNIT) \
-    DECLARE_DATE_FUNCTIONS(CLASS, NAME, TYPE_BIGINT, datetime_diff<TimeUnit::UNIT>(ts1, ts0))
+    DECLARE_DATE_FUNCTIONS(CLASS, NAME, TYPE_BIGINT, diff_on_utc_datetime<TimeUnit::UNIT>(ts1, ts0))
 
 // all these functions implemented by datediff
 TIME_DIFF_FUNCTION_IMPL(YearsDiffImpl, years_diff, YEAR);

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/agg/SequenceCount.java

@@ -27,6 +27,7 @@
 import org.apache.doris.nereids.types.DateTimeV2Type;
 import org.apache.doris.nereids.types.DateV2Type;
 import org.apache.doris.nereids.types.StringType;
+import org.apache.doris.nereids.types.TimeStampTzType;
 import org.apache.doris.nereids.util.ExpressionUtils;
 
 import com.google.common.base.Preconditions;
@@ -43,6 +44,8 @@ public class SequenceCount extends NotNullableAggregateFunction
     public static final List<FunctionSignature> SIGNATURES = ImmutableList.of(
             FunctionSignature.ret(BigIntType.INSTANCE)
                     .varArgs(StringType.INSTANCE, DateV2Type.INSTANCE, BooleanType.INSTANCE),
+            FunctionSignature.ret(BigIntType.INSTANCE)
+                    .varArgs(StringType.INSTANCE, TimeStampTzType.WILDCARD, BooleanType.INSTANCE),
             FunctionSignature.ret(BigIntType.INSTANCE)
                     .varArgs(StringType.INSTANCE, DateTimeV2Type.WILDCARD, BooleanType.INSTANCE)
     );

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/agg/SequenceFunction.java

@@ -36,7 +36,7 @@ default void checkLegalityBeforeTypeCoercion() {
         }
         if (!getArgumentType(1).isDateLikeType()) {
             throw new AnalysisException("The timestamp params of " + functionName
-                    + " function must be DATE or DATETIME, but it is " + getArgumentType(1));
+                    + " function must be DATE, DATETIME or TIMESTAMPTZ, but it is " + getArgumentType(1));
         }
         String pattern = ((StringLikeLiteral) firstArg).getStringValue();
         if (!FunctionCallExpr.parsePattern(pattern)) {

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/agg/SequenceMatch.java

@@ -25,6 +25,7 @@
 import org.apache.doris.nereids.types.DateTimeV2Type;
 import org.apache.doris.nereids.types.DateV2Type;
 import org.apache.doris.nereids.types.StringType;
+import org.apache.doris.nereids.types.TimeStampTzType;
 import org.apache.doris.nereids.util.ExpressionUtils;
 
 import com.google.common.base.Preconditions;
@@ -41,6 +42,8 @@ public class SequenceMatch extends NullableAggregateFunction
     public static final List<FunctionSignature> SIGNATURES = ImmutableList.of(
             FunctionSignature.ret(BooleanType.INSTANCE)
                     .varArgs(StringType.INSTANCE, DateV2Type.INSTANCE, BooleanType.INSTANCE),
+            FunctionSignature.ret(BooleanType.INSTANCE)
+                    .varArgs(StringType.INSTANCE, TimeStampTzType.WILDCARD, BooleanType.INSTANCE),
             FunctionSignature.ret(BooleanType.INSTANCE)
                     .varArgs(StringType.INSTANCE, DateTimeV2Type.WILDCARD, BooleanType.INSTANCE)
     );

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/agg/TopNWeighted.java

@@ -35,6 +35,7 @@
 import org.apache.doris.nereids.types.LargeIntType;
 import org.apache.doris.nereids.types.SmallIntType;
 import org.apache.doris.nereids.types.StringType;
+import org.apache.doris.nereids.types.TimeStampTzType;
 import org.apache.doris.nereids.types.TinyIntType;
 import org.apache.doris.nereids.types.VarcharType;
 
@@ -71,6 +72,8 @@ public class TopNWeighted extends NullableAggregateFunction
                                     .args(FloatType.INSTANCE, BigIntType.INSTANCE, IntegerType.INSTANCE),
                             FunctionSignature.ret(ArrayType.of(DateV2Type.INSTANCE))
                     .args(DateV2Type.INSTANCE, BigIntType.INSTANCE, IntegerType.INSTANCE),
+            FunctionSignature.ret(ArrayType.of(TimeStampTzType.WILDCARD))
+                    .args(TimeStampTzType.WILDCARD, BigIntType.INSTANCE, IntegerType.INSTANCE),
             FunctionSignature.ret(ArrayType.of(DateTimeV2Type.WILDCARD))
                     .args(DateTimeV2Type.WILDCARD, BigIntType.INSTANCE, IntegerType.INSTANCE),
             FunctionSignature.ret(ArrayType.of(StringType.INSTANCE))
@@ -105,6 +108,11 @@ public class TopNWeighted extends NullableAggregateFunction
                                                     IntegerType.INSTANCE),
                             FunctionSignature.ret(ArrayType.of(DateV2Type.INSTANCE))
                     .args(DateV2Type.INSTANCE, BigIntType.INSTANCE, IntegerType.INSTANCE, IntegerType.INSTANCE),
+            FunctionSignature.ret(ArrayType.of(TimeStampTzType.WILDCARD))
+                    .args(TimeStampTzType.WILDCARD,
+                            BigIntType.INSTANCE,
+                            IntegerType.INSTANCE,
+                            IntegerType.INSTANCE),
             FunctionSignature.ret(VarcharType.SYSTEM_DEFAULT)
                     .args(DateTimeV2Type.WILDCARD,
                             BigIntType.INSTANCE,