[fix](fe) Preserve TIMESTAMPTZ type and scale in forwarded user variables

fe/fe-core/src/main/java/org/apache/doris/analysis/DateLiteralUtils.java

@@ -85,9 +85,8 @@ private static LiteralExpr createDateTimeLiteral(String s, @Nullable Type type)
             TemporalAccessor dateTime = null;
             boolean parsed = false;
             ZoneId sourceZone = null;
-            // Every pre-existing date type retains its six-digit parsing contract. A null type is
-            // used when decoding an untyped DATE_LITERAL thrift node, where 7-9 digits are the only
-            // information available to infer TIMESTAMP_NS.
+            // Parse up to nine fractional digits when the type is unknown.
+            // Explicit types other than TIMESTAMP_NS keep the six-digit parsing limit.
             boolean parseNanoseconds = type == null || type.isTimeStampNs();
 
             // parse timezone

fe/fe-core/src/main/java/org/apache/doris/datasource/mvcc/PluginDrivenPartitionView.java

@@ -262,7 +262,7 @@ List<ConnectorMvccPartition> rangePartitions() {
 
         List<Type> frozenTypes() {
             return types.stream().map(type -> {
-                // Type.fromThrift does not retain TIMESTAMPTZ scale or VARBINARY length.
+                // Type.fromThrift does not retain every scalar parameter, such as VARBINARY length.
                 // Preserve all scalar parameters when freezing the conversion's schema.
                 if (type.getTypes().get(0).isSetScalarType()) {
                     TScalarType scalar = type.getTypes().get(0).getScalarType();

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

@@ -774,9 +774,10 @@ private static LiteralExpr getLiteralExprFromThrift(TExprNode node) throws Analy
             case STRING_LITERAL: return new StringLiteral(node.string_literal.value);
             case JSON_LITERAL: return new JsonLiteral(node.json_literal.value);
             case DATE_LITERAL:
-                Type literalType = Type.fromThrift(node.type);
-                return DateLiteralUtils.createLiteral(node.date_literal.value,
-                        literalType.isTimeStampNs() ? literalType : null);
+                // Keep the sender's type and scale for all date literals.
+                // A DATETIME(6) value at midnight must not become DATE.
+                // The value .120000 must retain its declared scale of 6.
+                return DateLiteralUtils.createLiteral(node.date_literal.value, Type.fromThrift(node.type));
             case IPV4_LITERAL: return new IPv4Literal(node.ipv4_literal.value);
             case IPV6_LITERAL: return new IPv6Literal(node.ipv6_literal.value);
             case UUID_LITERAL: return new UuidLiteral(new UUID(node.uuid_literal.hi, node.uuid_literal.lo).toString());

fe/fe-core/src/test/java/org/apache/doris/catalog/TypeTest.java

@@ -60,6 +60,15 @@ public void testTimestampTzIsNotVariantSubtype() {
         Assertions.assertTrue(Type.VARIANT.supportSubType(ScalarType.createDatetimeV2Type(6)));
     }
 
+    @Test
+    public void testTimestampTzThriftKeepsScale() {
+        // A type descriptor must retain its scale without a literal value to infer it from.
+        for (int scale : new int[] {0, 3, 6}) {
+            ScalarType type = ScalarType.createTimeStampTzType(scale);
+            Assertions.assertEquals(type, Type.fromThrift(type.toThrift()));
+        }
+    }
+
     // ===================== ArrayType =====================
     @Test
     public void testArrayOfArrayExactMatch() {

fe/fe-core/src/test/java/org/apache/doris/qe/ConnectProcessorDateLiteralTest.java

@@ -0,0 +1,74 @@
+// 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.
+
+package org.apache.doris.qe;
+
+import org.apache.doris.analysis.DateLiteral;
+import org.apache.doris.analysis.ExprToThriftVisitor;
+import org.apache.doris.analysis.LiteralExpr;
+import org.apache.doris.catalog.ScalarType;
+import org.apache.doris.catalog.Type;
+import org.apache.doris.common.Config;
+import org.apache.doris.thrift.TExprNode;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import java.lang.reflect.Method;
+import java.time.LocalDateTime;
+import java.util.stream.Stream;
+
+class ConnectProcessorDateLiteralTest {
+    private static Stream<Arguments> dateLiterals() {
+        LocalDateTime midnight = LocalDateTime.of(2024, 1, 1, 0, 0);
+        LocalDateTime noon = midnight.withHour(12);
+        return Stream.of(
+                new DateLiteral(midnight, Type.DATE),
+                new DateLiteral(midnight, Type.DATEV2),
+                new DateLiteral(midnight, Type.DATETIME),
+                new DateLiteral(noon, Type.DATETIME),
+                new DateLiteral(midnight, ScalarType.createDatetimeV2Type(0)),
+                new DateLiteral(midnight, ScalarType.createDatetimeV2Type(6)),
+                new DateLiteral(noon, ScalarType.createDatetimeV2Type(6)),
+                new DateLiteral(noon.withNano(120000000), ScalarType.createDatetimeV2Type(3)),
+                new DateLiteral(noon.withNano(120000000), ScalarType.createDatetimeV2Type(6)),
+                new DateLiteral(noon.withNano(123456000), ScalarType.createDatetimeV2Type(6)))
+                .flatMap(literal -> Stream.of(Arguments.of(literal, true), Arguments.of(literal, false)));
+    }
+
+    @ParameterizedTest
+    @MethodSource("dateLiterals")
+    void testForwardedDateKeepsTypeScaleAndValue(DateLiteral literal, boolean enableDateConversion) throws Exception {
+        boolean originalDateConversion = Config.enable_date_conversion;
+        try {
+            // The receiver must use the sender's type, even when its default date type differs.
+            Config.enable_date_conversion = enableDateConversion;
+            TExprNode node = ExprToThriftVisitor.treeToThrift(literal).getNodes().get(0);
+            Method deserialize = ConnectProcessor.class.getDeclaredMethod(
+                    "getLiteralExprFromThrift", TExprNode.class);
+            deserialize.setAccessible(true);
+
+            LiteralExpr restored = (LiteralExpr) deserialize.invoke(null, node);
+            Assertions.assertEquals(literal.getType(), restored.getType());
+            Assertions.assertEquals(literal.getStringValue(), restored.getStringValue());
+        } finally {
+            Config.enable_date_conversion = originalDateConversion;
+        }
+    }
+}

fe/fe-core/src/test/java/org/apache/doris/qe/ConnectProcessorTimestampTzTest.java

@@ -0,0 +1,60 @@
+// 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.
+
+package org.apache.doris.qe;
+
+import org.apache.doris.analysis.ExprToThriftVisitor;
+import org.apache.doris.analysis.LiteralExpr;
+import org.apache.doris.nereids.trees.expressions.literal.TimestampTzLiteral;
+import org.apache.doris.nereids.types.TimeStampTzType;
+import org.apache.doris.thrift.TExprNode;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+
+import java.lang.reflect.Method;
+
+class ConnectProcessorTimestampTzTest {
+    @ParameterizedTest
+    @CsvSource({
+            "0, 2024-01-01 00:00:00+00:00",
+            "6, 2024-01-01 00:00:00.000000+00:00",
+            "3, 2024-01-01 00:00:00.120+00:00",
+            "6, 2024-01-01 00:00:00.123456+00:00"
+    })
+    void testForwardedTimestampTzKeepsTypeScaleAndValue(int scale, String value) throws Exception {
+        ConnectContext context = new ConnectContext();
+        context.getSessionVariable().setTimeZone("+08:00");
+        context.setThreadLocalInfo();
+        try {
+            LiteralExpr literal = new TimestampTzLiteral(TimeStampTzType.of(scale), value).toLegacyLiteral();
+            TExprNode node = ExprToThriftVisitor.treeToThrift(literal).getNodes().get(0);
+            Method deserialize = ConnectProcessor.class.getDeclaredMethod(
+                    "getLiteralExprFromThrift", TExprNode.class);
+            deserialize.setAccessible(true);
+
+            // A UTC midnight value must keep its time zone type, even when its fractional part is zero.
+            // The declared scale must also survive. For example, .120 has scale 3, not scale 2.
+            LiteralExpr restored = (LiteralExpr) deserialize.invoke(null, node);
+            Assertions.assertEquals(literal.getType(), restored.getType());
+            Assertions.assertEquals(value, restored.getStringValue());
+        } finally {
+            ConnectContext.remove();
+        }
+    }
+}

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

@@ -1033,6 +1033,11 @@ protected static Pair<Type, Integer> fromThrift(TTypeDesc col, int nodeIdx) {
                     Preconditions.checkState(scalarType.isSetPrecision()
                             && scalarType.isSetScale());
                     type = ScalarType.createDatetimeV2Type(scalarType.getScale());
+                } else if (scalarType.getType() == TPrimitiveType.TIMESTAMPTZ) {
+                    // Use the declared scale. For example, TIMESTAMPTZ(6) must not become TIMESTAMPTZ(0).
+                    Preconditions.checkState(scalarType.isSetPrecision()
+                            && scalarType.isSetScale());
+                    type = ScalarType.createTimeStampTzType(scalarType.getScale());
                 } else if (scalarType.getType() == TPrimitiveType.TIMESTAMP_NS) {
                     type = ScalarType.createTimeStampNsType();
                 } else if (scalarType.getType() == TPrimitiveType.TIMEV2) {

regression-test/data/datatype_p0/datetimev2/test_date_user_variable_ctas.out

@@ -0,0 +1,13 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !select --
+2024-01-01	2024-01-01T00:00	2024-01-01T12:00	2024-01-01T12:00:00.120
+
+-- !types --
+d	date
+fraction	datetime(6)
+midnight	datetime(6)
+noon	datetime(6)
+
+-- !values --
+2024-01-01	2024-01-01T00:00	2024-01-01T12:00	2024-01-01T12:00:00.120
+

regression-test/data/datatype_p0/timestamptz/test_timestamptz_user_variable_ctas.out

@@ -0,0 +1,14 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !select --
+1
+
+-- !filtered --
+1
+
+-- !type --
+hours	bigint(20)
+v	timestamp(6)
+
+-- !projected --
+2024-01-01 08:00:00.000000+08:00	0
+

regression-test/data/nereids_rules_p0/rewrite_simple_agg_to_constant/auto_partition_recycle_count_cache.out

@@ -0,0 +1,7 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !count_before_recycle --
+6
+
+-- !count_after_recycle --
+3
+

regression-test/suites/changelog_p0/test_changelog_before_column_name.groovy

@@ -25,11 +25,17 @@
 //   3. a user column ending in _COL feeding an incremental materialized view, whose before-image
 //      name then ends in _COL__ like a common hidden column.
 suite("test_changelog_before_column_name", "nonConcurrent") {
-    // Read timestamps from FE so that the window uses the same clock and timezone
-    // as SQL parsing. Leave a whole second after each group of committed writes.
+    // Changelog rows use TSO time, which can lag behind NOW().
+    // Put the boundary after prior writes, then wait before the next write.
+    // For example, the update from 10 to 11 must be inside the window.
     def nextBoundary = {
-        sleep(1200)
-        sql("SELECT DATE_FORMAT(NOW(), '%Y-%m-%d %H:%i:%s')")[0][0].toString()
+        long current = (sql "SELECT CURRENT_TSO_PHYSICAL_TIME FROM information_schema.tso_status")[0][0] as long
+        long boundary = (Math.floorDiv(current, 1000L) + 1L) * 1000L
+        awaitUntil(60, 0.1) {
+            ((sql "SELECT CURRENT_TSO_PHYSICAL_TIME FROM information_schema.tso_status")[0][0] as long) >= boundary
+        }
+        // Use the FE session time zone for both formatting and window parsing.
+        sql("SELECT DATE_FORMAT(FROM_MILLISECOND(${boundary}), '%Y-%m-%d %H:%i:%s')")[0][0].toString()
     }
 
     // A user column can not take the before-image name itself.

regression-test/suites/datatype_p0/datetimev2/test_date_user_variable_ctas.groovy

@@ -0,0 +1,41 @@
+// 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("test_date_user_variable_ctas") {
+    // Run this suite through a follower FE to test user variable forwarding.
+    // A single result row checks all values. No source table is needed.
+    sql "SET enable_sql_cache = false"
+    sql "SET @d = CAST('2024-01-01' AS DATE)"
+    sql "SET @midnight = CAST('2024-01-01 00:00:00.000000' AS DATETIME(6))"
+    sql "SET @noon = CAST('2024-01-01 12:00:00.000000' AS DATETIME(6))"
+    sql "SET @fraction = CAST('2024-01-01 12:00:00.120000' AS DATETIME(6))"
+
+    order_qt_select "SELECT @d, @midnight, @noon, @fraction"
+    sql "DROP TABLE IF EXISTS date_user_variable_ctas"
+    sql """
+        CREATE TABLE date_user_variable_ctas
+        DISTRIBUTED BY HASH(d) BUCKETS 1
+        PROPERTIES ('replication_num' = '1')
+        AS SELECT @d AS d, @midnight AS midnight, @noon AS noon, @fraction AS fraction
+    """
+    // Midnight must remain DATETIME(6). The values .000000 and .120000 must retain scale 6.
+    order_qt_types """
+        SELECT column_name, column_type FROM information_schema.columns
+        WHERE table_schema = DATABASE() AND table_name = 'date_user_variable_ctas'
+    """
+    order_qt_values "SELECT d, midnight, noon, fraction FROM date_user_variable_ctas"
+}

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

@@ -0,0 +1,58 @@
+// 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("test_timestamptz_user_variable_ctas") {
+    // Run this suite through a follower FE to test user variable forwarding.
+    // The same queries must give the same results through the master FE.
+    sql "SET time_zone = '+08:00'"
+    sql "SET enable_sql_cache = false"
+    sql "DROP TABLE IF EXISTS timestamptz_user_variable_src"
+    sql "DROP TABLE IF EXISTS timestamptz_user_variable_filtered"
+    sql "DROP TABLE IF EXISTS timestamptz_user_variable_projected"
+    sql """
+        CREATE TABLE timestamptz_user_variable_src (pk INT, ts TIMESTAMPTZ(6))
+        DUPLICATE KEY(pk)
+        DISTRIBUTED BY HASH(pk) BUCKETS 1
+        PROPERTIES ('replication_num' = '1')
+    """
+    sql """
+        INSERT INTO timestamptz_user_variable_src VALUES
+        (1, CAST('2024-01-01 00:00:00 +00:00' AS TIMESTAMPTZ(6)))
+    """
+    sql "SET @ts = CAST('2024-01-01 00:00:00 +00:00' AS TIMESTAMPTZ(6))"
+
+    order_qt_select "SELECT pk FROM timestamptz_user_variable_src WHERE ts = @ts"
+    sql """
+        CREATE TABLE timestamptz_user_variable_filtered
+        PROPERTIES ('replication_num' = '1')
+        AS SELECT pk FROM timestamptz_user_variable_src WHERE ts = @ts
+    """
+    order_qt_filtered "SELECT pk FROM timestamptz_user_variable_filtered"
+
+    // One row checks the value and the hour difference. Its type must retain scale 6.
+    sql """
+        CREATE TABLE timestamptz_user_variable_projected
+        PROPERTIES ('replication_num' = '1')
+        AS SELECT @ts AS v, timestampdiff(HOUR, @ts, ts) AS hours
+        FROM timestamptz_user_variable_src
+    """
+    order_qt_type """
+        SELECT column_name, column_type FROM information_schema.columns
+        WHERE table_schema = DATABASE() AND table_name = 'timestamptz_user_variable_projected'
+    """
+    order_qt_projected "SELECT v, hours FROM timestamptz_user_variable_projected"
+}