[fix](timestamptz) Report TIMESTAMPTZ as string to MySQL clients

fe/fe-core/src/main/java/org/apache/doris/mysql/MysqlSerializer.java

@@ -288,9 +288,11 @@ private int getMysqlTypeLength(Type type) {
             case DATEV2:
             case DATE:
                 return 10;
+            case TIMESTAMPTZ:
+                // yyyy-MM-dd HH:mm:ss[.ffffff]+HH:mm
+                return 32;
             case DATETIME:
-            case DATETIMEV2:
-            case TIMESTAMPTZ: {
+            case DATETIMEV2: {
                 if (type.getPrimitiveType().isTimeType()) {
                     return 10;
                 } else {
@@ -338,7 +340,6 @@ public int getMysqlDecimals(Type type) {
             case DECIMAL256:
             case TIMEV2:
             case DATETIMEV2:
-            case TIMESTAMPTZ:
                 return ((ScalarType) type).decimalScale();
             case FLOAT:
             case DOUBLE:

fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java

@@ -1870,7 +1870,6 @@ protected void sendBinaryResultRow(ResultSet resultSet) throws IOException {
                             break;
                         case DATETIME:
                         case DATETIMEV2:
-                        case TIMESTAMPTZ:
                             DateTimeV2Literal datetime = new DateTimeV2Literal(item);
                             long microSecond = datetime.getMicroSecond();
                             // https://dev.mysql.com/doc/dev/mysql-server/latest/page_protocol_com_query_response_text_resultset.html

fe/fe-core/src/test/java/org/apache/doris/mysql/MysqlSerializerVarbinaryTest.java

@@ -17,6 +17,7 @@
 
 package org.apache.doris.mysql;
 
+import org.apache.doris.catalog.MysqlColType;
 import org.apache.doris.catalog.ScalarType;
 import org.apache.doris.catalog.Type;
 
@@ -118,6 +119,35 @@ public void testFieldPacketForVarcharUsesUtf8Collation() {
         Assertions.assertEquals(0, flags); // not BINARY
     }
 
+    @Test
+    public void testFieldPacketForTimestampTzUsesStringMetadata() {
+        MysqlSerializer ser = MysqlSerializer.newInstance();
+        Type type = ScalarType.createTimeStampTzType(6);
+        ser.writeField("ts", type);
+        byte[] out = ser.toArray();
+
+        Assertions.assertEquals(MysqlColType.MYSQL_TYPE_STRING, type.getPrimitiveType().toMysqlType());
+
+        int off = skipFieldHeaderStrings(out);
+        int charset = leUInt2(out, off);
+        Assertions.assertEquals(33, charset); // utf8_general_ci
+        off += 2;
+
+        long displayLen = leUInt4(out, off);
+        Assertions.assertEquals(32L, displayLen);
+        off += 4;
+
+        int colType = out[off] & 0xFF;
+        Assertions.assertEquals(MysqlColType.MYSQL_TYPE_STRING.getCode(), colType);
+        off += 1;
+
+        int flags = leUInt2(out, off);
+        Assertions.assertEquals(0, flags);
+        off += 2;
+
+        Assertions.assertEquals(0, out[off] & 0xFF);
+    }
+
     @Test
     public void testWriteLenEncodedBytesPreservesNullByte() {
         MysqlSerializer ser = MysqlSerializer.newInstance();

fe/fe-type/src/main/java/org/apache/doris/catalog/PrimitiveType.java

@@ -427,9 +427,10 @@ public MysqlColType toMysqlType() {
             case DATE:
             case DATEV2:
                 return MysqlColType.MYSQL_TYPE_DATE;
+            case TIMESTAMPTZ:
+                return MysqlColType.MYSQL_TYPE_STRING;
             case DATETIME:
-            case DATETIMEV2:
-            case TIMESTAMPTZ: {
+            case DATETIMEV2: {
                 if (isTimeType) {
                     return MysqlColType.MYSQL_TYPE_TIME;
                 } else {

regression-test/suites/datatype_p0/timestamptz/test_timestamptz_jdbc_binary_protocol.groovy

@@ -0,0 +1,85 @@
+// 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.
+
+import com.mysql.cj.jdbc.ServerPreparedStatement
+
+import java.sql.PreparedStatement
+import java.sql.ResultSet
+import java.sql.SQLException
+
+suite("test_timestamptz_jdbc_binary_protocol") {
+    String tableName = "test_timestamptz_jdbc_binary_protocol"
+    String dbName = "regression_test_datatype_p0_timestamptz"
+    def user = context.config.jdbcUser
+    def password = context.config.jdbcPassword
+
+    sql "SET time_zone = '+00:00'"
+    sql "DROP TABLE IF EXISTS ${tableName}"
+    sql """
+        CREATE TABLE ${tableName} (
+            id INT,
+            ts TIMESTAMPTZ(6),
+            note VARCHAR(16)
+        )
+        DUPLICATE KEY(id)
+        DISTRIBUTED BY HASH(id) BUCKETS 1
+        PROPERTIES("replication_num" = "1")
+    """
+    sql """
+        INSERT INTO ${tableName} VALUES
+            (1, NULL, 'null'),
+            (2, '2024-01-01 08:00:00.123456 +08:00', 'equiv_utc'),
+            (3, '2024-01-01 00:00:01.654321 +00:00', 'micro')
+    """
+
+    String url = getServerPrepareJdbcUrl(context.config.jdbcUrl, dbName) +
+            "&emulateUnsupportedPstmts=true&useLocalSessionState=true"
+    logger.info("jdbc prepare statement url: ${url}")
+
+    connect(user, password, url) {
+        sql "SET time_zone = '+00:00'"
+
+        PreparedStatement stmt = prepareStatement("""
+            SELECT id, ts, CAST(ts AS STRING) AS ts_text, note
+            FROM ${tableName}
+            ORDER BY id
+        """)
+        assertEquals(ServerPreparedStatement, stmt.class)
+
+        ResultSet rs = stmt.executeQuery()
+        int rowCount = 0
+        try {
+            while (rs.next()) {
+                rowCount++
+                String direct
+                try {
+                    direct = rs.getString(2)
+                } catch (SQLException e) {
+                    logger.info("failed to read TIMESTAMPTZ directly with ResultSet.getString", e)
+                    throw e
+                }
+
+                String castText = rs.getString(3)
+                assertEquals(castText, direct)
+            }
+            assertEquals(3, rowCount)
+        } finally {
+            rs.close()
+            stmt.close()
+        }
+    }
+}