[BugFix] Drop Iceberg timestamptz file-level min/max across DST fall-back

The Iceberg datetime min/max aggregate optimization converts file-level
TIMESTAMP WITH TIME ZONE bounds (UTC micros) to session-local micros, one
endpoint at a time. UTC->local is non-monotonic across a backward UTC-offset
change (a DST fall-back or a historical rule change): the true local min/max
can lie strictly inside the file, so the endpoint conversion produces an
inexact bound and SELECT MIN/MAX(ts_tz) can return a wrong result.

Skip emitting file-level min/max for a timestamptz column when the file's
[low, high] instant range contains an offset-decreasing transition; BE then
reads that file directly (a per-file fallback). Constant-offset and
spring-forward ranges keep the optimization with exact endpoints.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Signed-off-by: Evgeniy Shishkin <eshishki@gmail.com>

Author: eshishki

fe/fe-core/src/main/java/com/starrocks/connector/iceberg/IcebergUtil.java

@@ -44,6 +44,8 @@
 import java.time.Instant;
 import java.time.ZoneId;
 import java.time.ZoneOffset;
+import java.time.zone.ZoneOffsetTransition;
+import java.time.zone.ZoneRules;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -191,15 +193,24 @@ public static Map<Integer, MinMaxValue> parseMinMaxValueBySlots(Schema schema,
                 Types.TimestampType timestampType = (Types.TimestampType) type;
                 if (timestampType.shouldAdjustToUTC() && low instanceof Long && high instanceof Long) {
                     // Iceberg TIMESTAMP WITH TIME ZONE stores instants in UTC, while StarRocks compares DATETIME
-                    // values in the session timezone, so we convert file-level bounds into session-local micros
-                    // before sending them to BE.
+                    // values in the session timezone, so file-level bounds are converted into session-local micros
+                    // before being sent to BE.
                     //
-                    // Note that this is an endpoint conversion on file-level min/max only. For timezones whose
-                    // UTC offset changes over time (for example DST or historical rule changes), UTC->local is
-                    // not strictly monotonic over an arbitrary interval, so the converted low/high remain a
-                    // best-effort approximation rather than an exact local min/max for the full file.
-                    minMaxValue.minValue = adjustTimestampMicrosToSessionTz((Long) low);
-                    minMaxValue.maxValue = adjustTimestampMicrosToSessionTz((Long) high);
+                    // This is an endpoint-only conversion: it maps just the file's min and max instants. It is exact
+                    // only while UTC->local is monotonic across the file's [low, high] range, i.e. local = utc + offset
+                    // never folds back on itself. A spring-forward transition (offset jumps up, e.g. -08:00 -> -07:00)
+                    // keeps local time strictly increasing, so the endpoints stay the true local min/max. A fall-back
+                    // transition (offset drops, e.g. -07:00 -> -08:00) rewinds local time, so the real min/max can sit
+                    // strictly inside the file and is unrecoverable from the endpoints alone. Only that case is
+                    // rejected: the file-level min/max is dropped and BE reads this file directly (a per-file
+                    // fallback) instead of trusting an inexact bound.
+                    ZoneId zoneId = TimeUtils.getTimeZone().toZoneId();
+                    if (isEndpointConversionExact(zoneId, (Long) low, (Long) high)) {
+                        minMaxValue.minValue = adjustTimestampMicrosToSessionTz((Long) low);
+                        minMaxValue.maxValue = adjustTimestampMicrosToSessionTz((Long) high);
+                    } else {
+                        minMaxValues.remove(field.fieldId());
+                    }
                 }
             }
         }
@@ -215,6 +226,34 @@ private static long adjustTimestampMicrosToSessionTz(long micros) {
         return micros + offset.getTotalSeconds() * 1_000_000L;
     }
 
+    /**
+     * Returns true when the endpoint UTC-&gt;local conversion in {@link #adjustTimestampMicrosToSessionTz(long)}
+     * produces the exact local min/max for a file covering the instant range {@code [lowMicros, highMicros]} in
+     * {@code zoneId}.
+     *
+     * <p>{@code local = utc + offset(utc)} preserves order as long as the offset never decreases across the range, so
+     * the file's earliest/latest instants stay its earliest/latest local times. A spring-forward jump (offset
+     * increases) only widens the gap and keeps that order; a fall-back jump (offset decreases) rewinds local time and
+     * can move the true min/max onto an interior row that the endpoints no longer represent. We therefore reject the
+     * range only if it contains an offset-decreasing transition; constant-offset and spring-forward ranges stay exact.
+     */
+    private static boolean isEndpointConversionExact(ZoneId zoneId, long lowMicros, long highMicros) {
+        Instant low = Instant.ofEpochSecond(Math.floorDiv(lowMicros, 1_000_000L));
+        Instant high = Instant.ofEpochSecond(Math.floorDiv(highMicros, 1_000_000L));
+        ZoneRules rules = zoneId.getRules();
+        // nextTransition returns the first transition strictly after its argument; zone transitions never land on
+        // sub-second boundaries, so truncating micros to whole seconds cannot skip one. Walk every transition in
+        // (low, high] and reject as soon as one decreases the UTC offset (a fall-back).
+        for (ZoneOffsetTransition t = rules.nextTransition(low);
+                t != null && !t.getInstant().isAfter(high);
+                t = rules.nextTransition(t.getInstant())) {
+            if (t.getOffsetAfter().getTotalSeconds() < t.getOffsetBefore().getTotalSeconds()) {
+                return false;
+            }
+        }
+        return true;
+    }
+
     public static Map<Integer, TExprMinMaxValue> toThriftMinMaxValueBySlots(Schema schema,
                                                                             Map<Integer, ByteBuffer> lowerBounds,
                                                                             Map<Integer, ByteBuffer> upperBounds,

fe/fe-core/src/test/java/com/starrocks/connector/iceberg/IcebergUtilTest.java

@@ -32,14 +32,17 @@
 import org.junit.jupiter.api.Test;
 
 import java.nio.ByteBuffer;
+import java.time.Instant;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.TimeZone;
 
 import static org.apache.iceberg.types.Types.NestedField.required;
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 
 public class IcebergUtilTest {
 
@@ -258,6 +261,123 @@ public void testParseMinMaxValueBySlotsWithIcebergTimestampTypes() {
         }
     }
 
+    @Test
+    public void testTimestamptzMinMaxDroppedWhenFileSpansDstTransition() {
+        // America/Los_Angeles DST fall-back at 2023-11-05T09:00:00Z: PDT (-07:00) drops to PST (-08:00). A file whose
+        // timestamptz values straddle that instant covers a backward (offset-decreasing) transition, so UTC->local
+        // rewinds and the endpoint conversion is no longer exact; the file-level min/max must be dropped.
+        Schema schema = new Schema(
+                required(3, "ts_ntz", Types.TimestampType.withoutZone()),
+                required(5, "ts_tz", Types.TimestampType.withZone()));
+        List<SlotDescriptor> slots = List.of(
+                new SlotDescriptor(new SlotId(3), "ts_ntz", DateType.DATETIME, true),
+                new SlotDescriptor(new SlotId(5), "ts_tz", DateType.DATETIME, true));
+        slots.get(0).setColumn(new Column("ts_ntz", DateType.DATETIME, true));
+        slots.get(1).setColumn(new Column("ts_tz", DateType.DATETIME, true));
+
+        long lowMicros = Instant.parse("2023-11-05T08:30:00Z").getEpochSecond() * 1_000_000L;
+        long highMicros = Instant.parse("2023-11-05T09:30:00Z").getEpochSecond() * 1_000_000L;
+        Map<Integer, ByteBuffer> lowerBounds = Map.of(
+                3, org.apache.iceberg.types.Conversions.toByteBuffer(Types.TimestampType.withoutZone(), lowMicros),
+                5, org.apache.iceberg.types.Conversions.toByteBuffer(Types.TimestampType.withZone(), lowMicros));
+        Map<Integer, ByteBuffer> upperBounds = Map.of(
+                3, org.apache.iceberg.types.Conversions.toByteBuffer(Types.TimestampType.withoutZone(), highMicros),
+                5, org.apache.iceberg.types.Conversions.toByteBuffer(Types.TimestampType.withZone(), highMicros));
+        Map<Integer, Long> nullValueCounts = Map.of(3, 0L, 5, 0L);
+        Map<Integer, Long> valueCounts = Map.of(3, 2L, 5, 2L);
+
+        ConnectContext ctx = new ConnectContext();
+        ctx.getSessionVariable().setTimeZone("America/Los_Angeles");
+        ctx.setThreadLocalInfo();
+        try {
+            Map<Integer, IcebergUtil.MinMaxValue> result = IcebergUtil.parseMinMaxValueBySlots(
+                    schema, lowerBounds, upperBounds, nullValueCounts, valueCounts, slots);
+            // timestamp without time zone is unaffected and keeps its raw wall-clock micros.
+            assertTrue(result.containsKey(3));
+            assertEquals(lowMicros, result.get(3).minValue);
+            // timestamptz spanning the DST transition is dropped, so BE falls back to reading the file.
+            assertFalse(result.containsKey(5));
+
+            Map<Integer, TExprMinMaxValue> thriftValues = IcebergUtil.toThriftMinMaxValueBySlots(
+                    schema, lowerBounds, upperBounds, nullValueCounts, valueCounts, slots);
+            assertTrue(thriftValues.containsKey(3));
+            assertFalse(thriftValues.containsKey(5));
+        } finally {
+            ConnectContext.remove();
+        }
+    }
+
+    @Test
+    public void testTimestamptzMinMaxKeptWhenFileWithinSingleOffset() {
+        // A file whose timestamptz values stay within one offset period (here PDT, summer 2023) has no offset
+        // change inside its range, so endpoint UTC->local conversion is exact and the bounds are converted and kept.
+        Schema schema = new Schema(required(5, "ts_tz", Types.TimestampType.withZone()));
+        List<SlotDescriptor> slots = List.of(
+                new SlotDescriptor(new SlotId(5), "ts_tz", DateType.DATETIME, true));
+        slots.get(0).setColumn(new Column("ts_tz", DateType.DATETIME, true));
+
+        long lowMicros = Instant.parse("2023-07-01T10:00:00Z").getEpochSecond() * 1_000_000L;
+        long highMicros = Instant.parse("2023-07-01T11:00:00Z").getEpochSecond() * 1_000_000L;
+        Map<Integer, ByteBuffer> lowerBounds = Map.of(
+                5, org.apache.iceberg.types.Conversions.toByteBuffer(Types.TimestampType.withZone(), lowMicros));
+        Map<Integer, ByteBuffer> upperBounds = Map.of(
+                5, org.apache.iceberg.types.Conversions.toByteBuffer(Types.TimestampType.withZone(), highMicros));
+        Map<Integer, Long> nullValueCounts = Map.of(5, 0L);
+        Map<Integer, Long> valueCounts = Map.of(5, 2L);
+
+        ConnectContext ctx = new ConnectContext();
+        ctx.getSessionVariable().setTimeZone("America/Los_Angeles");
+        ctx.setThreadLocalInfo();
+        try {
+            Map<Integer, IcebergUtil.MinMaxValue> result = IcebergUtil.parseMinMaxValueBySlots(
+                    schema, lowerBounds, upperBounds, nullValueCounts, valueCounts, slots);
+            assertTrue(result.containsKey(5));
+            // PDT offset is -07:00 across the whole range; endpoint conversion adds that offset to each bound.
+            assertEquals(lowMicros + TimeZone.getTimeZone("America/Los_Angeles").getOffset(
+                    Instant.parse("2023-07-01T10:00:00Z").toEpochMilli()) * 1000L, result.get(5).minValue);
+            assertEquals(highMicros + TimeZone.getTimeZone("America/Los_Angeles").getOffset(
+                    Instant.parse("2023-07-01T11:00:00Z").toEpochMilli()) * 1000L, result.get(5).maxValue);
+        } finally {
+            ConnectContext.remove();
+        }
+    }
+
+    @Test
+    public void testTimestamptzMinMaxKeptAcrossSpringForward() {
+        // America/Los_Angeles spring-forward at 2023-03-12T10:00:00Z: PST (-08:00) jumps to PDT (-07:00). The offset
+        // increases, so UTC->local stays monotonic and the file's endpoints remain its true local min/max even though
+        // it straddles the transition; the bounds are converted (each with its own instant's offset) and kept.
+        Schema schema = new Schema(required(5, "ts_tz", Types.TimestampType.withZone()));
+        List<SlotDescriptor> slots = List.of(
+                new SlotDescriptor(new SlotId(5), "ts_tz", DateType.DATETIME, true));
+        slots.get(0).setColumn(new Column("ts_tz", DateType.DATETIME, true));
+
+        long lowMicros = Instant.parse("2023-03-12T09:30:00Z").getEpochSecond() * 1_000_000L;
+        long highMicros = Instant.parse("2023-03-12T10:30:00Z").getEpochSecond() * 1_000_000L;
+        Map<Integer, ByteBuffer> lowerBounds = Map.of(
+                5, org.apache.iceberg.types.Conversions.toByteBuffer(Types.TimestampType.withZone(), lowMicros));
+        Map<Integer, ByteBuffer> upperBounds = Map.of(
+                5, org.apache.iceberg.types.Conversions.toByteBuffer(Types.TimestampType.withZone(), highMicros));
+        Map<Integer, Long> nullValueCounts = Map.of(5, 0L);
+        Map<Integer, Long> valueCounts = Map.of(5, 2L);
+
+        ConnectContext ctx = new ConnectContext();
+        ctx.getSessionVariable().setTimeZone("America/Los_Angeles");
+        ctx.setThreadLocalInfo();
+        try {
+            Map<Integer, IcebergUtil.MinMaxValue> result = IcebergUtil.parseMinMaxValueBySlots(
+                    schema, lowerBounds, upperBounds, nullValueCounts, valueCounts, slots);
+            assertTrue(result.containsKey(5));
+            // low uses PST (-08:00), high uses PDT (-07:00); each endpoint converts with its own instant's offset.
+            assertEquals(lowMicros + TimeZone.getTimeZone("America/Los_Angeles").getOffset(
+                    Instant.parse("2023-03-12T09:30:00Z").toEpochMilli()) * 1000L, result.get(5).minValue);
+            assertEquals(highMicros + TimeZone.getTimeZone("America/Los_Angeles").getOffset(
+                    Instant.parse("2023-03-12T10:30:00Z").toEpochMilli()) * 1000L, result.get(5).maxValue);
+        } finally {
+            ConnectContext.remove();
+        }
+    }
+
     private static org.apache.iceberg.Table tableWithProperty(String key, String value) {
         org.apache.iceberg.Table table = org.mockito.Mockito.mock(org.apache.iceberg.Table.class);
         Map<String, String> props = new HashMap<>();