[fix](arrow-flight) Return DATETIMEV2 as a timezone-naive Arrow timestamp

be/src/core/data_type_serde/data_type_datetimev2_serde.cpp

@@ -497,6 +497,11 @@ Status DataTypeDateTimeV2SerDe::read_column_from_arrow(IColumn& column,
     if (arrow_array->type()->id() == arrow::Type::TIMESTAMP) {
         const auto* concrete_array = dynamic_cast<const arrow::TimestampArray*>(arrow_array);
         const auto type = std::static_pointer_cast<arrow::TimestampType>(arrow_array->type());
+        // Mirror write_column_to_arrow: a timezone-naive Arrow timestamp (empty timezone)
+        // stores its value as an as-if-UTC epoch, so it must be read back with UTC.
+        // Interpreting it with the session timezone would shift the wall-clock value for
+        // cross-cluster (type=doris) reads. See apache/doris#65741.
+        const cctz::time_zone& real_ctz = type->timezone().empty() ? cctz::utc_time_zone() : ctz;
         switch (type->unit()) {
         case arrow::TimeUnit::type::SECOND: {
             divisor = DIVISOR_FOR_SECOND;
@@ -525,16 +530,22 @@ Status DataTypeDateTimeV2SerDe::read_column_from_arrow(IColumn& column,
         for (auto value_i = start; value_i < end; ++value_i) {
             const uint8_t* raw_byte_ptr = base_ptr + value_i * element_size;
             auto date_value = unaligned_load<int64_t>(raw_byte_ptr);
-            auto utc_epoch = static_cast<UInt64>(date_value);
+            // Keep the epoch signed: a pre-1970 DATETIMEV2 (e.g. datetime(3)
+            // '1969-12-31 23:59:59.500' -> raw -500) must not be cast to unsigned, and the
+            // sub-second remainder has to be floored toward -inf, mirroring
+            // append_datetimev2_from_utc_epoch_micros() used by the INT96 path above.
+            int64_t seconds = date_value / divisor;
+            int64_t sub_second = date_value % divisor;
+            if (sub_second < 0) {
+                sub_second += divisor;
+                --seconds;
+            }
 
             DateV2Value<DateTimeV2ValueType> v;
-            // convert second
-            v.from_unixtime(utc_epoch / divisor, ctz);
-            // get rest time
-            // add 0 on the right to make it 6 digits. DateTimeV2Value microsecond is 6 digits,
-            // the scale decides to keep the first few digits, so the valid digits should be kept at the front.
-            // "2022-01-01 11:11:11.111", utc_epoch = 1641035471111, divisor = 1000, set_microsecond(111000)
-            v.set_microsecond((utc_epoch % divisor) * DIVISOR_FOR_MICRO / divisor);
+            v.from_unixtime(seconds, real_ctz);
+            // scale the sub-second remainder into microseconds (6 digits); NANO input is truncated.
+            // "2022-01-01 11:11:11.111", raw = 1641035471111, divisor = 1000, set_microsecond(111000)
+            v.set_microsecond(static_cast<uint32_t>(sub_second * DIVISOR_FOR_MICRO / divisor));
             col_data.emplace_back(v);
         }
     } else {

be/src/exec/operator/result_sink_operator.cpp

@@ -56,8 +56,9 @@ Status ResultSinkLocalState::init(RuntimeState* state, LocalSinkStateInfo& info)
     } else {
         std::shared_ptr<arrow::Schema> arrow_schema;
         if (p._sink_type == TResultSinkType::ARROW_FLIGHT_PROTOCOL) {
-            RETURN_IF_ERROR(get_arrow_schema_from_expr_ctxs(_output_vexpr_ctxs, &arrow_schema,
-                                                            state->timezone()));
+            RETURN_IF_ERROR(get_arrow_schema_from_expr_ctxs(
+                    _output_vexpr_ctxs, &arrow_schema, state->timezone(),
+                    state->query_options().enable_arrow_flight_datetime_naive));
         }
         VLOG_DEBUG << "create sender in INIT with instance id " << fragment_instance_id;
         RETURN_IF_ERROR(state->exec_env()->result_mgr()->create_sender(
@@ -121,8 +122,9 @@ Status ResultSinkOperatorX::prepare(RuntimeState* state) {
     if (state->query_options().enable_parallel_result_sink) {
         std::shared_ptr<arrow::Schema> arrow_schema;
         if (_sink_type == TResultSinkType::ARROW_FLIGHT_PROTOCOL) {
-            RETURN_IF_ERROR(get_arrow_schema_from_expr_ctxs(_output_vexpr_ctxs, &arrow_schema,
-                                                            state->timezone()));
+            RETURN_IF_ERROR(get_arrow_schema_from_expr_ctxs(
+                    _output_vexpr_ctxs, &arrow_schema, state->timezone(),
+                    state->query_options().enable_arrow_flight_datetime_naive));
         }
         VLOG_DEBUG << "create sender in prepare with query id " << state->query_id();
         RETURN_IF_ERROR(state->exec_env()->result_mgr()->create_sender(

be/src/format/arrow/arrow_row_batch.cpp

@@ -49,8 +49,8 @@
 namespace doris {
 
 Status convert_to_arrow_type(const DataTypePtr& origin_type,
-                             std::shared_ptr<arrow::DataType>* result,
-                             const std::string& timezone) {
+                             std::shared_ptr<arrow::DataType>* result, const std::string& timezone,
+                             bool datetime_naive) {
     auto type = get_serialized_type(origin_type);
     switch (type->get_primitive_type()) {
     case TYPE_NULL:
@@ -97,17 +97,25 @@ Status convert_to_arrow_type(const DataTypePtr& origin_type,
     case TYPE_DATEV2:
         *result = std::make_shared<arrow::Date32Type>();
         break;
-    // TODO: maybe need to distinguish TYPE_DATETIME and TYPE_TIMESTAMPTZ
-    case TYPE_TIMESTAMPTZ:
+    // TIMESTAMPTZ has instant semantics and always keeps the timezone. DATETIMEV2 is a
+    // timezone-naive wall-clock type: only the Arrow Flight result path (datetime_naive) maps
+    // it WITHOUT timezone, so ADBC/pyarrow clients do not treat it as an instant and render a
+    // spurious "+00:00" (apache/doris#65741). Other callers (Parquet export, including forced
+    // INT96, and Python UDF) keep the timezone to preserve their existing round-trip semantics.
     case TYPE_DATETIMEV2:
+    case TYPE_TIMESTAMPTZ: {
+        const std::string empty_timezone;
+        const bool naive = type->get_primitive_type() == TYPE_DATETIMEV2 && datetime_naive;
+        const std::string& ts_timezone = naive ? empty_timezone : timezone;
         if (type->get_scale() > 3) {
-            *result = std::make_shared<arrow::TimestampType>(arrow::TimeUnit::MICRO, timezone);
+            *result = std::make_shared<arrow::TimestampType>(arrow::TimeUnit::MICRO, ts_timezone);
         } else if (type->get_scale() > 0) {
-            *result = std::make_shared<arrow::TimestampType>(arrow::TimeUnit::MILLI, timezone);
+            *result = std::make_shared<arrow::TimestampType>(arrow::TimeUnit::MILLI, ts_timezone);
         } else {
-            *result = std::make_shared<arrow::TimestampType>(arrow::TimeUnit::SECOND, timezone);
+            *result = std::make_shared<arrow::TimestampType>(arrow::TimeUnit::SECOND, ts_timezone);
         }
         break;
+    }
     case TYPE_DECIMALV2:
     case TYPE_DECIMAL32:
     case TYPE_DECIMAL64:
@@ -123,16 +131,19 @@ Status convert_to_arrow_type(const DataTypePtr& origin_type,
     case TYPE_ARRAY: {
         const auto* type_arr = assert_cast<const DataTypeArray*>(remove_nullable(type).get());
         std::shared_ptr<arrow::DataType> item_type;
-        RETURN_IF_ERROR(convert_to_arrow_type(type_arr->get_nested_type(), &item_type, timezone));
+        RETURN_IF_ERROR(convert_to_arrow_type(type_arr->get_nested_type(), &item_type, timezone,
+                                              datetime_naive));
         *result = std::make_shared<arrow::ListType>(item_type);
         break;
     }
     case TYPE_MAP: {
         const auto* type_map = assert_cast<const DataTypeMap*>(remove_nullable(type).get());
         std::shared_ptr<arrow::DataType> key_type;
         std::shared_ptr<arrow::DataType> val_type;
-        RETURN_IF_ERROR(convert_to_arrow_type(type_map->get_key_type(), &key_type, timezone));
-        RETURN_IF_ERROR(convert_to_arrow_type(type_map->get_value_type(), &val_type, timezone));
+        RETURN_IF_ERROR(convert_to_arrow_type(type_map->get_key_type(), &key_type, timezone,
+                                              datetime_naive));
+        RETURN_IF_ERROR(convert_to_arrow_type(type_map->get_value_type(), &val_type, timezone,
+                                              datetime_naive));
         *result = std::make_shared<arrow::MapType>(key_type, val_type);
         break;
     }
@@ -141,8 +152,8 @@ Status convert_to_arrow_type(const DataTypePtr& origin_type,
         std::vector<std::shared_ptr<arrow::Field>> fields;
         for (size_t i = 0; i < type_struct->get_elements().size(); i++) {
             std::shared_ptr<arrow::DataType> field_type;
-            RETURN_IF_ERROR(
-                    convert_to_arrow_type(type_struct->get_element(i), &field_type, timezone));
+            RETURN_IF_ERROR(convert_to_arrow_type(type_struct->get_element(i), &field_type,
+                                                  timezone, datetime_naive));
             fields.push_back(
                     std::make_shared<arrow::Field>(type_struct->get_element_name(i), field_type,
                                                    type_struct->get_element(i)->is_nullable()));
@@ -206,12 +217,16 @@ Status get_arrow_schema_from_block(const Block& block, std::shared_ptr<arrow::Sc
 
 Status get_arrow_schema_from_expr_ctxs(const VExprContextSPtrs& output_vexpr_ctxs,
                                        std::shared_ptr<arrow::Schema>* result,
-                                       const std::string& timezone) {
+                                       const std::string& timezone, bool datetime_naive) {
     std::vector<std::shared_ptr<arrow::Field>> fields;
     for (int i = 0; i < output_vexpr_ctxs.size(); i++) {
         std::shared_ptr<arrow::DataType> arrow_type;
         auto root_expr = output_vexpr_ctxs.at(i)->root();
-        RETURN_IF_ERROR(convert_to_arrow_type(root_expr->data_type(), &arrow_type, timezone));
+        // The Arrow Flight result path emits timezone-naive DATETIMEV2 when datetime_naive is set
+        // (session variable enable_arrow_flight_datetime_naive), so ADBC clients receive
+        // wall-clock values instead of timezone-aware instants (apache/doris#65741).
+        RETURN_IF_ERROR(convert_to_arrow_type(root_expr->data_type(), &arrow_type, timezone,
+                                              datetime_naive));
         auto field_name = root_expr->is_slot_ref() && !root_expr->expr_label().empty()
                                   ? root_expr->expr_label()
                                   : fmt::format("{}_{}", root_expr->data_type()->get_name(), i);

be/src/format/arrow/arrow_row_batch.h

@@ -43,8 +43,12 @@ constexpr size_t MAX_ARROW_UTF8 = (1ULL << 31); // 2G
 
 class RowDescriptor;
 
+// When datetime_naive is true, TYPE_DATETIMEV2 maps to a timezone-naive Arrow timestamp.
+// This is used only by the Arrow Flight result path; other callers (Parquet export, Python
+// UDF, ...) keep the timezone-aware mapping to preserve their existing semantics.
+// See apache/doris#65741.
 Status convert_to_arrow_type(const DataTypePtr& type, std::shared_ptr<arrow::DataType>* result,
-                             const std::string& timezone);
+                             const std::string& timezone, bool datetime_naive = false);
 
 std::shared_ptr<arrow::Field> create_arrow_field_with_metadata(
         const std::string& field_name, const std::shared_ptr<arrow::DataType>& arrow_type,
@@ -55,7 +59,7 @@ Status get_arrow_schema_from_block(const Block& block, std::shared_ptr<arrow::Sc
 
 Status get_arrow_schema_from_expr_ctxs(const VExprContextSPtrs& output_vexpr_ctxs,
                                        std::shared_ptr<arrow::Schema>* result,
-                                       const std::string& timezone);
+                                       const std::string& timezone, bool datetime_naive = false);
 
 Status serialize_record_batch(const arrow::RecordBatch& record_batch, std::string* result);
 

be/src/format/arrow/arrow_utils.cpp

@@ -17,8 +17,12 @@
 
 #include "format/arrow/arrow_utils.h"
 
+#include <arrow/array.h>
 #include <arrow/pretty_print.h>
 #include <arrow/status.h>
+#include <arrow/type.h>
+
+#include "util/timezone_utils.h"
 
 namespace doris {
 
@@ -30,6 +34,23 @@ Status to_doris_status(const arrow::Status& status) {
     }
 }
 
+cctz::time_zone resolve_arrow_reader_timezone(const arrow::Array& array,
+                                              const cctz::time_zone& default_ctz) {
+    if (array.type()->id() != arrow::Type::TIMESTAMP) {
+        return default_ctz;
+    }
+    const std::string& timezone =
+            std::static_pointer_cast<arrow::TimestampType>(array.type())->timezone();
+    if (timezone.empty()) {
+        return default_ctz;
+    }
+    cctz::time_zone parsed;
+    if (TimezoneUtils::find_cctz_time_zone(timezone, parsed)) {
+        return parsed;
+    }
+    return default_ctz;
+}
+
 arrow::Status to_arrow_status(const Status& status) {
     if (LIKELY(status.ok())) {
         return arrow::Status::OK();

be/src/format/arrow/arrow_utils.h

@@ -18,6 +18,7 @@
 #pragma once
 
 #include <arrow/result.h>
+#include <cctz/time_zone.h>
 
 #include <iostream>
 
@@ -75,6 +76,14 @@ Status arrow_pretty_print(const arrow::Array& rb, std::ostream* os);
 Status to_doris_status(const arrow::Status& status);
 arrow::Status to_arrow_status(const Status& status);
 
+// Returns the timezone to decode an Arrow timestamp `array` with. A timezone-aware Arrow timestamp
+// is decoded in its own advertised timezone, so a remote DATETIMEV2 wall-clock round-trips even
+// when the producer session timezone differs from the reader's default (apache/doris#65741). A
+// non-timestamp or timezone-naive array returns `default_ctz` (a naive timestamp is then read as
+// UTC inside the serde). Falls back to `default_ctz` when the advertised timezone cannot be parsed.
+cctz::time_zone resolve_arrow_reader_timezone(const arrow::Array& array,
+                                              const cctz::time_zone& default_ctz);
+
 template <typename T>
 inline void assign_from_result(T& output, const arrow::Result<T>& result) {
     output = *result;

be/src/format/table/remote_doris_reader.cpp

@@ -86,10 +86,14 @@ Status RemoteDorisReader::_do_get_next_block(Block* block, size_t* read_rows, bo
 
         try {
             auto block_pos = (*_col_name_to_block_idx)[column_name];
+            // Decode a timezone-aware Arrow timestamp in its own advertised timezone so a remote
+            // DATETIMEV2 wall-clock round-trips even when the producer session timezone differs
+            // from this reader's default (apache/doris#65741).
+            const cctz::time_zone col_ctz = resolve_arrow_reader_timezone(*column, _ctzz);
             RETURN_IF_ERROR(columns_guard.get_datatype_by_position(block_pos)
                                     ->get_serde()
                                     ->read_column_from_arrow(*columns[block_pos], column, 0,
-                                                             num_rows, _ctzz));
+                                                             num_rows, col_ctz));
         } catch (Exception& e) {
             return Status::InternalError(
                     "Failed to convert from arrow to block, column_name: {}, e: {}", column_name,

be/src/format_v2/table/remote_doris_reader.cpp

@@ -312,12 +312,17 @@ Status RemoteDorisFileReader::_materialize_arrow_column(const arrow::RecordBatch
     const auto column_name = batch.schema()->field(arrow_column_idx)->name();
     auto columns_guard = file_block->mutate_columns_scoped();
     auto& columns = columns_guard.mutable_columns();
+    // Decode a timezone-aware Arrow timestamp in its own advertised timezone so a remote
+    // DATETIMEV2 wall-clock round-trips even when the producer session timezone differs from this
+    // reader's default (apache/doris#65741).
+    const auto& arrow_column = batch.column(arrow_column_idx);
+    const cctz::time_zone col_ctz = resolve_arrow_reader_timezone(*arrow_column, _ctz);
     try {
         RETURN_IF_ERROR(columns_guard.get_datatype_by_position(block_position.value())
                                 ->get_serde()
                                 ->read_column_from_arrow(*columns[block_position.value()],
-                                                         batch.column(arrow_column_idx).get(), 0,
-                                                         batch.num_rows(), _ctz));
+                                                         arrow_column.get(), 0, batch.num_rows(),
+                                                         col_ctz));
     } catch (const Exception& e) {
         return Status::InternalError(
                 "Failed to convert Remote Doris Arrow column '{}' (file_column_id={}) to Doris "