branch-4.1: [fix](cast) Handle DST fallback in timestamp cast monotonicity #65903

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/Cast.java

@@ -17,24 +17,33 @@
 
 package org.apache.doris.nereids.trees.expressions;
 
+import org.apache.doris.common.util.TimeUtils;
 import org.apache.doris.nereids.exceptions.UnboundException;
 import org.apache.doris.nereids.trees.expressions.functions.Monotonic;
 import org.apache.doris.nereids.trees.expressions.literal.Literal;
+import org.apache.doris.nereids.trees.expressions.literal.TimestampTzLiteral;
 import org.apache.doris.nereids.trees.expressions.shape.UnaryExpression;
 import org.apache.doris.nereids.trees.expressions.visitor.ExpressionVisitor;
 import org.apache.doris.nereids.types.BigIntType;
 import org.apache.doris.nereids.types.DataType;
+import org.apache.doris.nereids.types.DateTimeV2Type;
 import org.apache.doris.nereids.types.DecimalV2Type;
 import org.apache.doris.nereids.types.DecimalV3Type;
 import org.apache.doris.nereids.types.IntegerType;
 import org.apache.doris.nereids.types.LargeIntType;
 import org.apache.doris.nereids.types.SmallIntType;
+import org.apache.doris.nereids.types.TimeStampTzType;
 import org.apache.doris.nereids.types.TinyIntType;
 import org.apache.doris.nereids.types.coercion.DateLikeType;
+import org.apache.doris.nereids.util.DateUtils;
 
 import com.google.common.base.Preconditions;
 import com.google.common.collect.ImmutableList;
 
+import java.time.DateTimeException;
+import java.time.Instant;
+import java.time.ZoneId;
+import java.time.ZoneOffset;
 import java.util.List;
 import java.util.Objects;
 
@@ -257,10 +266,45 @@ public Expression withConstantArgs(Expression literal) {
 
     @Override
     public boolean isMonotonic(Literal lower, Literal upper) {
-        // Both upward and downward casting of date types satisfy monotonicity.
-        if (child().getDataType() instanceof DateLikeType && targetType instanceof DateLikeType) {
+        DataType childType = child().getDataType();
+        if (!(childType instanceof DateLikeType && targetType instanceof DateLikeType)) {
+            return false;
+        }
+
+        if (childType instanceof TimeStampTzType && targetType instanceof DateTimeV2Type) {
+            return isTimeStampTzToDateTimeV2Monotonic(
+                    (TimeStampTzType) childType, (DateTimeV2Type) targetType, lower, upper);
+        }
+        return true;
+    }
+
+    private boolean isTimeStampTzToDateTimeV2Monotonic(
+            TimeStampTzType sourceType, DateTimeV2Type destinationType, Literal lower, Literal upper) {
+        ZoneId timeZone;
+        try {
+            timeZone = TimeUtils.getDorisZoneId();
+        } catch (DateTimeException e) {
+            return false;
+        }
+        if (timeZone.getRules().isFixedOffset()) {
             return true;
         }
-        return false;
+        // Scale reduction rounds the UTC value before applying the session timezone. That rounding
+        // can move values across a fall-back transition just outside the original partition range.
+        if (destinationType.getScale() < sourceType.getScale()) {
+            return false;
+        }
+        if (!(lower instanceof TimestampTzLiteral) || !(upper instanceof TimestampTzLiteral)) {
+            return false;
+        }
+
+        // TimestampTzLiteral stores UTC civil fields. The cast renders those instants in the
+        // session timezone, which moves backward at a fall-back transition.
+        Instant lowerInstant = ((TimestampTzLiteral) lower).toJavaDateType().toInstant(ZoneOffset.UTC);
+        Instant upperInstant = ((TimestampTzLiteral) upper).toJavaDateType().toInstant(ZoneOffset.UTC);
+        if (upperInstant.isBefore(lowerInstant)) {
+            return false;
+        }
+        return !DateUtils.hasFallbackTransitionInInstantRange(timeZone, lowerInstant, upperInstant);
     }
 }

fe/fe-core/src/test/java/org/apache/doris/nereids/trees/expressions/CastMonotonicityTest.java

@@ -0,0 +1,126 @@
+// 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.nereids.trees.expressions;
+
+import org.apache.doris.nereids.trees.expressions.literal.TimestampTzLiteral;
+import org.apache.doris.nereids.types.DateTimeV2Type;
+import org.apache.doris.nereids.types.TimeStampTzType;
+import org.apache.doris.qe.ConnectContext;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+class CastMonotonicityTest {
+    private final Cast cast = new Cast(
+            new SlotReference("ts", TimeStampTzType.of(6)), DateTimeV2Type.of(6));
+    private ConnectContext previousContext;
+
+    @BeforeEach
+    void setUp() {
+        previousContext = ConnectContext.get();
+        ConnectContext connectContext = new ConnectContext();
+        connectContext.setThreadLocalInfo();
+    }
+
+    @AfterEach
+    void tearDown() {
+        ConnectContext.remove();
+        if (previousContext != null) {
+            previousContext.setThreadLocalInfo();
+        }
+    }
+
+    @Test
+    void testFixedOffsetTimeZoneIsMonotonicWithUnboundedRange() {
+        setTimeZone("+00:00");
+
+        Assertions.assertTrue(cast.isMonotonic(null, timestampTz("2024-11-03 06:00:00+00:00")));
+        Assertions.assertTrue(cast.isMonotonic(timestampTz("2024-11-03 05:00:00+00:00"), null));
+        Assertions.assertTrue(cast(6, 0).isMonotonic(
+                timestampTz("2024-11-03 05:00:00.000000+00:00"),
+                timestampTz("2024-11-03 05:59:59.900000+00:00")));
+    }
+
+    @Test
+    void testDorisTimeZoneAliasIsMonotonic() {
+        setTimeZone("CST");
+
+        Assertions.assertTrue(cast.isMonotonic(
+                timestampTz("2024-11-03 05:00:00+00:00"),
+                timestampTz("2024-11-03 06:00:00+00:00")));
+    }
+
+    @Test
+    void testFallbackAtUpperBoundaryIsNotMonotonic() {
+        setTimeZone("America/New_York");
+
+        Assertions.assertFalse(cast.isMonotonic(
+                timestampTz("2024-11-03 05:00:00+00:00"),
+                timestampTz("2024-11-03 06:00:00+00:00")));
+    }
+
+    @Test
+    void testFallbackAtLowerBoundaryIsMonotonic() {
+        setTimeZone("America/New_York");
+
+        Assertions.assertTrue(cast.isMonotonic(
+                timestampTz("2024-11-03 06:00:00+00:00"),
+                timestampTz("2024-11-03 07:00:00+00:00")));
+    }
+
+    @Test
+    void testSpringForwardIsMonotonic() {
+        setTimeZone("America/New_York");
+
+        Assertions.assertTrue(cast.isMonotonic(
+                timestampTz("2024-03-10 06:00:00+00:00"),
+                timestampTz("2024-03-10 07:00:00+00:00")));
+    }
+
+    @Test
+    void testScaleReductionInDynamicTimeZoneIsNotMonotonic() {
+        setTimeZone("America/New_York");
+
+        Assertions.assertFalse(cast(6, 0).isMonotonic(
+                timestampTz("2024-11-03 05:00:00.000000+00:00"),
+                timestampTz("2024-11-03 05:59:59.900000+00:00")));
+    }
+
+    @Test
+    void testUnboundedRangeInDynamicTimeZoneIsNotMonotonic() {
+        setTimeZone("America/New_York");
+
+        Assertions.assertFalse(cast.isMonotonic(null, timestampTz("2024-11-03 05:00:00+00:00")));
+        Assertions.assertFalse(cast.isMonotonic(timestampTz("2024-11-03 07:00:00+00:00"), null));
+    }
+
+    private TimestampTzLiteral timestampTz(String value) {
+        return new TimestampTzLiteral(TimeStampTzType.of(6), value);
+    }
+
+    private Cast cast(int sourceScale, int destinationScale) {
+        return new Cast(new SlotReference("ts", TimeStampTzType.of(sourceScale)),
+                DateTimeV2Type.of(destinationScale));
+    }
+
+    private void setTimeZone(String timeZone) {
+        ConnectContext.get().getSessionVariable().setTimeZone(timeZone);
+    }
+}

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

@@ -0,0 +1,13 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !rf_prune_off --
+2:2024-11-03 01:30:00.000000
+3:2024-11-03 01:30:00.000000
+
+-- !rf_prune_on --
+2:2024-11-03 01:30:00.000000
+3:2024-11-03 01:30:00.000000
+
+-- !static_prune --
+2:2024-11-03 01:30:00.000000
+3:2024-11-03 01:30:00.000000
+

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

@@ -0,0 +1,90 @@
+// 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_rf_partition_prune_dst") {
+    sql "SET time_zone='America/New_York'"
+    sql "SET enable_nereids_planner=true"
+    sql "SET enable_fallback_to_original_planner=false"
+    sql "SET enable_runtime_filter_prune=false"
+    sql "SET runtime_filter_wait_infinitely=true"
+    sql "SET runtime_filter_mode='GLOBAL'"
+    sql "SET runtime_filter_type='IN_OR_BLOOM_FILTER'"
+    sql "SET runtime_filter_max_in_num=1024"
+    sql "SET disable_join_reorder=true"
+    sql "SET enable_sql_cache=false"
+
+    sql "DROP TABLE IF EXISTS rfpp_tz_fall_fact"
+    sql """
+        CREATE TABLE rfpp_tz_fall_fact (
+            id INT NOT NULL,
+            part_col TIMESTAMPTZ(6) NOT NULL
+        ) DUPLICATE KEY(id)
+        PARTITION BY RANGE(part_col) (
+            PARTITION p0 VALUES LESS THAN ('2024-11-03 05:00:00.000000+00:00'),
+            PARTITION p1 VALUES LESS THAN ('2024-11-03 06:00:00.000000+00:00'),
+            PARTITION p2 VALUES LESS THAN ('2024-11-03 07:00:00.000000+00:00'),
+            PARTITION p3 VALUES LESS THAN (MAXVALUE)
+        ) DISTRIBUTED BY HASH(id) BUCKETS 1
+        PROPERTIES("replication_num"="1")
+    """
+    sql """
+        INSERT INTO rfpp_tz_fall_fact VALUES
+            (1, '2024-11-03 00:30:00.000000-04:00'),
+            (2, '2024-11-03 01:30:00.000000-04:00'),
+            (3, '2024-11-03 01:30:00.000000-05:00'),
+            (4, '2024-11-03 02:30:00.000000-05:00')
+    """
+
+    sql "DROP TABLE IF EXISTS rfpp_tz_fall_dim"
+    sql """
+        CREATE TABLE rfpp_tz_fall_dim (
+            id INT NOT NULL,
+            local_key DATETIMEV2(6) NOT NULL
+        ) DUPLICATE KEY(id)
+        DISTRIBUTED BY HASH(id) BUCKETS 1
+        PROPERTIES("replication_num"="1")
+    """
+    sql "INSERT INTO rfpp_tz_fall_dim VALUES (1, '2024-11-03 01:30:00.000000')"
+
+    sql "SET enable_runtime_filter_partition_prune=false"
+    qt_rf_prune_off """
+        SELECT CONCAT(CAST(f.id AS STRING), ':',
+                      CAST(CAST(f.part_col AS DATETIMEV2(6)) AS VARCHAR(32))) AS result
+        FROM rfpp_tz_fall_fact f JOIN [broadcast] rfpp_tz_fall_dim d
+          ON CAST(f.part_col AS DATETIMEV2(6)) = d.local_key
+        ORDER BY f.id
+    """
+
+    sql "SET enable_runtime_filter_partition_prune=true"
+    qt_rf_prune_on """
+        SELECT CONCAT(CAST(f.id AS STRING), ':',
+                      CAST(CAST(f.part_col AS DATETIMEV2(6)) AS VARCHAR(32))) AS result
+        FROM rfpp_tz_fall_fact f JOIN [broadcast] rfpp_tz_fall_dim d
+          ON CAST(f.part_col AS DATETIMEV2(6)) = d.local_key
+        ORDER BY f.id
+    """
+
+    qt_static_prune """
+        SELECT CONCAT(CAST(id AS STRING), ':',
+                      CAST(CAST(part_col AS DATETIMEV2(6)) AS VARCHAR(32))) AS result
+        FROM rfpp_tz_fall_fact
+        WHERE CAST(part_col AS DATETIMEV2(6)) = '2024-11-03 01:30:00.000000'
+        ORDER BY id
+    """
+
+    sql "SET time_zone=DEFAULT"
+}