branch-4.1: [fix](exec) Return Thrift DATETIME as naive Arrow timestamp #67232 (#67299)

Cherry-picked from #67232

Co-authored-by: Mryange <yanxuecheng@selectdb.com>

Author: github-actions[bot]

be/src/exec/operator/memory_scratch_sink_operator.cpp

@@ -107,7 +107,8 @@ Status MemoryScratchSinkOperatorX::sink_impl(RuntimeState* state, Block* input_b
     {
         SCOPED_TIMER(local_state._get_arrow_schema_timer);
         // After expr executed, use recaculated schema as final schema
-        RETURN_IF_ERROR(get_arrow_schema_from_block(block, &block_arrow_schema, state->timezone()));
+        RETURN_IF_ERROR(get_arrow_schema_from_block(block, &block_arrow_schema, state->timezone(),
+                                                    /*datetime_naive=*/true));
     }
     {
         SCOPED_TIMER(local_state._convert_block_to_arrow_batch_timer);

be/src/format/arrow/arrow_row_batch.cpp

@@ -203,11 +203,12 @@ std::shared_ptr<arrow::Field> create_arrow_field_with_metadata(
 }
 
 Status get_arrow_schema_from_block(const Block& block, 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 (const auto& type_and_name : block) {
         std::shared_ptr<arrow::DataType> arrow_type;
-        RETURN_IF_ERROR(convert_to_arrow_type(type_and_name.type, &arrow_type, timezone));
+        RETURN_IF_ERROR(
+                convert_to_arrow_type(type_and_name.type, &arrow_type, timezone, datetime_naive));
         auto field = create_arrow_field_with_metadata(type_and_name.name, arrow_type,
                                                       type_and_name.type->is_nullable(),
                                                       type_and_name.type->get_primitive_type());

be/src/format/arrow/arrow_row_batch.h

@@ -54,7 +54,7 @@ std::shared_ptr<arrow::Field> create_arrow_field_with_metadata(
         bool is_nullable, PrimitiveType primitive_type);
 
 Status get_arrow_schema_from_block(const Block& block, std::shared_ptr<arrow::Schema>* result,
-                                   const std::string& timezone);
+                                   const std::string& timezone, bool datetime_naive = false);
 
 Status get_arrow_schema_from_expr_ctxs(const VExprContextSPtrs& output_vexpr_ctxs,
                                        std::shared_ptr<arrow::Schema>* result,