[opt](exec) skip result serialization for dry run queries (#63356)

### What problem does this PR solve?

Issue Number: N/A

Related PR: None

Problem Summary:
When dry_run_query is enabled, FE only needs the returned row count, but
BE still spends most of PhysicalResultSink time serializing MySQL result
rows In a local dry-run case against numbers("number"="1000000"), the
profile showed AppendBatchTime = 77.689ms, TupleConvertTime = 68.650ms,
and ResultSendTime = 2.702us, which means the dry-run path was still
paying almost the full result sink conversion cost.

This change keeps output expr evaluation intact, but returns early in
the MySQL result writers once the output block is produced in dry-run
mode. That preserves returned row accounting while skipping result
serialization, block copy, and sink enqueue work that dry-run queries
never consume.

Author: Mryange

be/src/exec/sink/writer/vmysql_result_writer.cpp

@@ -296,6 +296,12 @@ Status VMysqlResultWriter::write(RuntimeState* state, Block& input_block) {
     Block block;
     RETURN_IF_ERROR(VExprContext::get_output_block_after_execute_exprs(_output_vexpr_ctxs,
                                                                        input_block, &block));
+
+    if (_is_dry_run) {
+        _written_rows += cast_set<int64_t>(block.rows());
+        return Status::OK();
+    }
+
     const auto total_bytes = block.bytes();
 
     if (total_bytes > config::thrift_max_message_size) [[unlikely]] {

be/test/core/data_type_serde/data_type_serde_mysql_test.cpp

@@ -77,6 +77,10 @@ class TestBlockSerializer final : public MySQLResultBlockBuffer {
 public:
     TestBlockSerializer(RuntimeState* state) : MySQLResultBlockBuffer(state) {}
     ~TestBlockSerializer() override = default;
+    size_t queue_size() {
+        std::lock_guard<std::mutex> l(_lock);
+        return _result_batch_queue.size();
+    }
     std::shared_ptr<TFetchDataResult> get_block() {
         std::lock_guard<std::mutex> l(_lock);
         DCHECK_EQ(_result_batch_queue.size(), 1);
@@ -86,7 +90,7 @@ class TestBlockSerializer final : public MySQLResultBlockBuffer {
     }
 };
 
-void serialize_and_deserialize_mysql_test() {
+void serialize_and_deserialize_mysql_test(bool dry_run) {
     Block block;
     //    create_descriptor_tablet();
     std::vector<std::tuple<std::string, FieldType, int, PrimitiveType, bool>> cols {
@@ -317,12 +321,25 @@ void serialize_and_deserialize_mysql_test() {
     auto serializer = std::make_shared<TestBlockSerializer>(&state);
     VMysqlResultWriter mysql_writer(serializer, _output_vexpr_ctxs, nullptr, false);
 
-    Status st = mysql_writer.write(&runtime_stat, block);
+    TQueryOptions query_options;
+    query_options.__set_dry_run_query(dry_run);
+    runtime_stat.set_query_options(query_options);
+
+    Status st = mysql_writer.init(&runtime_stat);
     EXPECT_TRUE(st.ok());
+
+    st = mysql_writer.write(&runtime_stat, block);
+    EXPECT_TRUE(st.ok());
+    EXPECT_EQ(mysql_writer.get_written_rows(), row_num);
+    EXPECT_EQ(serializer->queue_size(), dry_run ? 0 : 1);
 }
 
 TEST(DataTypeSerDeMysqlTest, ScalaSerDeTest) {
-    serialize_and_deserialize_mysql_test();
+    serialize_and_deserialize_mysql_test(false);
+}
+
+TEST(DataTypeSerDeMysqlTest, DryRunSkipsSerialization) {
+    serialize_and_deserialize_mysql_test(true);
 }
 
 } // namespace doris