[fix](nestedtype)fix nested type with timezone datetime

be/src/vec/data_types/serde/data_type_date64_serde.cpp

@@ -82,7 +82,8 @@ Status DataTypeDate64SerDe::deserialize_one_cell_from_json(IColumn& column, Slic
             // 1400 - 01 - 01
             val = 716833;
         }
-    } else if (ReadBuffer rb(slice.data, slice.size); !read_date_text_impl<Int64>(val, rb)) {
+    } else if (ReadBuffer rb(slice.data, slice.size);
+               !read_date_text_impl<Int64>(val, rb, *options.time_zone)) {
         return Status::InvalidArgument("parse date fail, string: '{}'",
                                        std::string(rb.position(), rb.count()).c_str());
     }
@@ -151,7 +152,8 @@ Status DataTypeDateTimeSerDe::deserialize_one_cell_from_json(IColumn& column, Sl
             // 1400 - 01 - 01
             val = 14000101000000L;
         }
-    } else if (ReadBuffer rb(slice.data, slice.size); !read_datetime_text_impl<Int64>(val, rb)) {
+    } else if (ReadBuffer rb(slice.data, slice.size);
+               !read_datetime_text_impl<Int64>(val, rb, *options.time_zone)) {
         return Status::InvalidArgument("parse datetime fail, string: '{}'",
                                        std::string(rb.position(), rb.count()).c_str());
     }

be/src/vec/data_types/serde/data_type_datetimev2_serde.cpp

@@ -92,7 +92,7 @@ Status DataTypeDateTimeV2SerDe::deserialize_one_cell_from_json(IColumn& column,
         }
 
     } else if (ReadBuffer rb(slice.data, slice.size);
-               !read_datetime_v2_text_impl<UInt64>(val, rb, scale)) {
+               !read_datetime_v2_text_impl<UInt64>(val, rb, *options.time_zone, scale)) {
         return Status::InvalidArgument("parse date fail, string: '{}'",
                                        std::string(rb.position(), rb.count()).c_str());
     }

be/src/vec/data_types/serde/data_type_datev2_serde.cpp

@@ -74,7 +74,8 @@ Status DataTypeDateV2SerDe::deserialize_one_cell_from_json(IColumn& column, Slic
         } else {
             val = MIN_DATE_V2;
         }
-    } else if (ReadBuffer rb(slice.data, slice.size); !read_date_v2_text_impl<UInt32>(val, rb)) {
+    } else if (ReadBuffer rb(slice.data, slice.size);
+               !read_date_v2_text_impl<UInt32>(val, rb, *options.time_zone)) {
         return Status::InvalidArgument("parse date fail, string: '{}'",
                                        std::string(rb.position(), rb.count()).c_str());
     }

be/src/vec/data_types/serde/data_type_serde.h

@@ -135,6 +135,10 @@ class DataTypeSerDe {
          * use this format in olap, because it is more slower, keep this option is for compatibility.
          */
         bool date_olap_format = false;
+        /**
+         * time_zone to support deserialize time with time zone
+         */
+        const cctz::time_zone* time_zone = nullptr;
         /**
          * field delimiter is used to separate fields in one row
          */

be/src/vec/functions/function_cast.h

@@ -600,6 +600,7 @@ struct ConvertImplGenericFromString {
             ColumnUInt8::Container* vec_null_map_to = &col_null_map_to->get_data();
             const bool is_complex = is_complex_type(data_type_to);
             DataTypeSerDe::FormatOptions format_options;
+            format_options.time_zone = &context->state()->timezone_obj();
             format_options.converted_from_string = true;
             format_options.escape_char = '\\';
 

regression-test/suites/load_p0/stream_load/load_nestedtype_with_timezone.groovy

@@ -0,0 +1,61 @@
+// 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("load_nestedtype_with_timezone", "p0") {
+    // create table with array<datetime>, array<date>, array<datetimev2> and array<datev2>
+    // map<datetime, datetime>, map<date, date>, map<datetimev2, datetimev2> and map<datev2, datev2>
+    // struct<f1:datetime, f2:date, f3:datetimev2, f4:datev2>
+
+    sql """ DROP TABLE IF EXISTS test_nested_type_with_timezone; """
+    sql """ CREATE TABLE IF NOT EXISTS test_nested_type_with_timezone (
+                                id INT NULL,
+                                array_datetime ARRAY<datetime> NULL ,
+                                array_date ARRAY<date> NULL ,
+                                array_datetimev2 ARRAY<datetimev2> NULL ,
+                                array_datev2 ARRAY<datev2> NULL ,
+                                map_datetime MAP<datetime, datetime> NULL ,
+                                map_date MAP<date, date> NULL ,
+                                map_datetimev2 MAP<datetimev2, datetimev2> NULL ,
+                                map_datev2 MAP<datev2, datev2> NULL ,
+                                struct_datetime STRUCT<f1:datetime, f2:date, f3:datetimev2, f4:datev2> NULL
+                            ) ENGINE=OLAP
+                             DUPLICATE KEY(`id`)
+                             DISTRIBUTED BY HASH(`id`) BUCKETS 1
+                             PROPERTIES (
+                             "replication_allocation" = "tag.location.default: 1"
+                             );
+    """
+
+    // stream_load data
+    streamLoad {
+        table "test_nested_type_with_timezone"
+        set 'column_separator', '|'
+        file "test_nested_type_with_timezone.csv"
+
+        check { result, exception, startTime, endTime ->
+                        if (exception != null) {
+                            throw exception
+                        }
+                        log.info("Stream load result: ${result}".toString())
+                        def json = parseJson(result)
+                        assertEquals("success", json.Status.toLowerCase())
+                        assertEquals(json.NumberTotalRows, json.NumberLoadedRows + json.NumberUnselectedRows + json.NumberFilteredRows)
+                    }
+    }
+
+    order_qt_sql "SELECT * FROM test_nested_type_with_timezone ORDER BY id"
+}