branch-4.1: [fix](be) Reject oversized Unix timestamps before year narrowing #68115

be/src/core/value/vdatetime_value.h

@@ -209,6 +209,10 @@ static constexpr uint64_t MAX_DATETIME_V2 = ((uint64_t)MAX_DATE_V2 << TIME_PART_
 static constexpr uint64_t MIN_DATETIME_V2 = (uint64_t)MIN_DATE_V2 << TIME_PART_LENGTH;
 
 static constexpr uint32_t MAX_YEAR = 9999;
+// UTC 9999-12-31 23:59:59 plus one day for time zone offsets. Reject larger inputs
+// before narrowing the civil year; the converted date still needs its exact range check.
+static constexpr int64_t MAX_UNIX_TIMESTAMP_WITH_TIMEZONE =
+        253402300799LL + HOUR_PER_DAY * SECOND_PER_HOUR;
 static constexpr uint32_t MAX_MONTH = 12;
 static constexpr uint32_t MAX_HOUR = 23;
 static constexpr uint32_t MAX_MINUTE = 59;

be/src/exprs/function/date_time_transforms.h

@@ -410,7 +410,7 @@ struct FromUnixTimeImpl {
 
     [[nodiscard]] static bool check_valid(const ArgType& val) {
         if constexpr (NewVersion) {
-            if (val < 0) [[unlikely]] {
+            if (val < 0 || val > MAX_UNIX_TIMESTAMP_WITH_TIMEZONE) [[unlikely]] {
                 return false;
             }
         } else {

be/src/exprs/function/function_date_or_datetime_computation.h

@@ -1292,12 +1292,13 @@ struct TimestampToDateTime : IFunction {
                 continue;
             }
             Int64 value = column_data.get_element(i);
-            if (value < 0) [[unlikely]] {
+            const Int64 seconds = value / Impl::ratio;
+            if (value < 0 || seconds > MAX_UNIX_TIMESTAMP_WITH_TIMEZONE) [[unlikely]] {
                 throw_out_of_bound_int(name, value);
             }
 
             auto& dt = reinterpret_cast<DateV2Value<DateTimeV2ValueType>&>(res_data[i]);
-            dt.from_unixtime(value / Impl::ratio, time_zone);
+            dt.from_unixtime(seconds, time_zone);
 
             if (!dt.is_valid_date()) [[unlikely]] {
                 throw_out_of_bound_int(name, value);

be/test/exprs/function/function_time_test.cpp

@@ -17,6 +17,7 @@
 
 #include <gtest/gtest.h>
 
+#include <limits>
 #include <string>
 
 #include "core/data_type/data_type_date.h"
@@ -234,6 +235,42 @@ TEST(VTimestampFunctionsTest, from_unix_test) {
     }
 }
 
+TEST(VTimestampFunctionsTest, from_unixtime_rejects_year_wrap) {
+    TimezoneUtils::load_timezones_to_cache();
+    // The first two values produce civil years 67506 and 133042 in UTC. Narrowing
+    // either year to uint16_t produces 1970, which used to pass date validation.
+    const int64_t timestamps[] = {2068116364800LL, 4136232816000LL, 1789000000000000000LL,
+                                  std::numeric_limits<int64_t>::max()};
+    for (const int64_t seconds : timestamps) {
+        SCOPED_TRACE(seconds);
+        const DataSet data_set = {{{seconds}, std::string("unused")}};
+        EXPECT_FALSE(
+                (check_function<DataTypeString, true>(
+                         "from_unixtime_new", {PrimitiveType::TYPE_BIGINT}, data_set, -1, -1, true)
+                         .ok()));
+        EXPECT_FALSE((check_function<DataTypeString>("from_unixtime_new",
+                                                     {ConstedNotnull {PrimitiveType::TYPE_BIGINT}},
+                                                     data_set, -1, -1, true)
+                              .ok()));
+    }
+}
+
+TEST(VTimestampFunctionsTest, timestamp_units_reject_year_wrap) {
+    TimezoneUtils::load_timezones_to_cache();
+    for (const int64_t seconds : {2068116364800LL, 4136232816000LL}) {
+        SCOPED_TRACE(seconds);
+        for (const auto& [name, ratio] : {std::pair {"from_second", int64_t {1}},
+                                          std::pair {"from_millisecond", int64_t {1000}},
+                                          std::pair {"from_microsecond", int64_t {1000000}}}) {
+            SCOPED_TRACE(name);
+            const DataSet data_set = {{{seconds * ratio}, std::string("unused")}};
+            EXPECT_FALSE((check_function<DataTypeDateTimeV2, true>(
+                                  name, {PrimitiveType::TYPE_BIGINT}, data_set, -1, -1, true)
+                                  .ok()));
+        }
+    }
+}
+
 TEST(VTimestampFunctionsTest, unix_timestamp_test) {
     std::string func_name = "unix_timestamp_new";
     TimezoneUtils::load_timezones_to_cache();

regression-test/data/datatype_p0/date/test_unix_timestamp_range.out

@@ -0,0 +1,28 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !upper_0_false --
+9999-12-31 23:59:59	9999-12-31 23:59:59.999999	9999-12-31T23:59:59	9999-12-31T23:59:59.999	9999-12-31T23:59:59.999999
+
+-- !upper_0_true --
+9999-12-31 23:59:59	9999-12-31 23:59:59.999999	9999-12-31T23:59:59	9999-12-31T23:59:59.999	9999-12-31T23:59:59.999999
+
+-- !null_0 --
+\N	\N	\N	\N
+
+-- !upper_1_false --
+9999-12-31 23:59:59	9999-12-31 23:59:59.999999	9999-12-31T23:59:59	9999-12-31T23:59:59.999	9999-12-31T23:59:59.999999
+
+-- !upper_1_true --
+9999-12-31 23:59:59	9999-12-31 23:59:59.999999	9999-12-31T23:59:59	9999-12-31T23:59:59.999	9999-12-31T23:59:59.999999
+
+-- !null_1 --
+\N	\N	\N	\N
+
+-- !upper_2_false --
+9999-12-31 23:59:59	9999-12-31 23:59:59.999999	9999-12-31T23:59:59	9999-12-31T23:59:59.999	9999-12-31T23:59:59.999999
+
+-- !upper_2_true --
+9999-12-31 23:59:59	9999-12-31 23:59:59.999999	9999-12-31T23:59:59	9999-12-31T23:59:59.999	9999-12-31T23:59:59.999999
+
+-- !null_2 --
+\N	\N	\N	\N
+

regression-test/suites/datatype_p0/date/test_unix_timestamp_range.groovy

@@ -0,0 +1,102 @@
+// 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.
+
+suite("test_unix_timestamp_range") {
+    sql "DROP TABLE IF EXISTS test_unix_timestamp_range_values"
+    sql """
+        CREATE TABLE test_unix_timestamp_range_values (
+            id INT,
+            seconds BIGINT,
+            milliseconds BIGINT,
+            microseconds BIGINT
+        )
+        DISTRIBUTED BY HASH(id) BUCKETS 1
+        PROPERTIES("replication_num" = "1")
+    """
+    // The first two inputs reach civil years 67506 and 133042. Both years used
+    // to wrap to 1970 when passed to the uint16_t date setter.
+    def invalidValues = [
+        [2068116364800L, 2068116364800000L, 2068116364800000000L],
+        [4136232816000L, 4136232816000000L, 4136232816000000000L],
+        [Long.MAX_VALUE, Long.MAX_VALUE, Long.MAX_VALUE]
+    ]
+    invalidValues.eachWithIndex { values, index ->
+        sql """INSERT INTO test_unix_timestamp_range_values
+               VALUES (${index}, ${values[0]}, ${values[1]}, ${values[2]})"""
+    }
+    sql "INSERT INTO test_unix_timestamp_range_values VALUES (3, NULL, NULL, NULL)"
+
+    def functions = [
+        ["from_unixtime", "seconds", 0, ""],
+        ["from_unixtime", "seconds", 0, ", '%Y-%m-%d %H:%i:%s'"],
+        ["from_second", "seconds", 0, ""],
+        ["from_millisecond", "milliseconds", 1, ""],
+        ["from_microsecond", "microseconds", 2, ""]
+    ]
+    def zones = [
+        ["+00:00", 253402300799L],
+        ["+14:00", 253402250399L],
+        ["-12:00", 253402343999L]
+    ]
+    zones.eachWithIndex { zone, zoneIndex ->
+        sql "SET time_zone = '${zone[0]}'"
+        [false, true].each { skipFold ->
+            sql "SET debug_skip_fold_constant = ${skipFold}"
+            invalidValues.eachWithIndex { values, index ->
+                functions.each { function ->
+                    def name = function[0]
+                    def argument = function[1]
+                    def value = values[function[2]]
+                    def format = function[3]
+                    test {
+                        sql "SELECT ${name}(${value}${format})"
+                        exception "Operation ${name}"
+                    }
+                    test {
+                        sql """SELECT ${name}(${argument}${format})
+                               FROM test_unix_timestamp_range_values WHERE id = ${index}"""
+                        exception "Operation ${name}"
+                    }
+                }
+            }
+
+            // The exact upper bound depends on the session time zone. In -12:00,
+            // a valid local date can have a UTC timestamp beyond the UTC year boundary.
+            def lastSecond = zone[1]
+            "order_qt_upper_${zoneIndex}_${skipFold}" """
+                SELECT from_unixtime(${lastSecond}),
+                       from_unixtime(CAST('${lastSecond}.999999' AS DECIMAL(18,6))),
+                       from_second(${lastSecond}),
+                       from_millisecond(${lastSecond * 1000L + 999L}),
+                       from_microsecond(${lastSecond * 1000000L + 999999L})
+            """
+            functions.each { function ->
+                def nextSecond = lastSecond + 1L
+                def ratios = [1L, 1000L, 1000000L]
+                test {
+                    sql "SELECT ${function[0]}(${nextSecond * ratios[function[2]]}${function[3]})"
+                    exception "Operation ${function[0]}"
+                }
+            }
+        }
+        "qt_null_${zoneIndex}" """
+            SELECT from_unixtime(seconds), from_second(seconds),
+                   from_millisecond(milliseconds), from_microsecond(microseconds)
+            FROM test_unix_timestamp_range_values WHERE id = 3 ORDER BY id
+        """
+    }
+}