[AURON #2455] Run UNIX_TIMESTAMP natively for fixed-offset session time zones

auron-flink-extension/auron-flink-planner/src/main/java/org/apache/auron/flink/table/planner/converter/RexCallConverter.java

@@ -23,6 +23,7 @@
 import java.util.List;
 import java.util.Optional;
 import java.util.Set;
+import java.util.regex.Pattern;
 import org.apache.auron.protobuf.ArrowType;
 import org.apache.auron.protobuf.EmptyMessage;
 import org.apache.auron.protobuf.PhysicalBinaryExprNode;
@@ -93,6 +94,15 @@ public class RexCallConverter implements FlinkRexNodeConverter {
     /** Flink's default format for the single-argument {@code UNIX_TIMESTAMP(string)} form. */
     private static final String DEFAULT_UNIX_TIMESTAMP_FORMAT = "yyyy-MM-dd HH:mm:ss";
 
+    /**
+     * A session time zone that names a constant offset rather than a region. The {@code GMT} prefix
+     * is optional because both spellings are reachable: {@link ZoneId#getId()} returns the bare form
+     * for a zone built from an offset, and the prefixed form for one built from a {@code GMT±HH:MM}
+     * id. The pattern does not re-check the numeric range, because every id tested against it came
+     * from a constructed {@link ZoneId} and is therefore already bounded to {@code ±18:00}.
+     */
+    private static final Pattern FIXED_OFFSET_ZONE = Pattern.compile("(?:GMT)?[+-]\\d{2}:\\d{2}");
+
     /** All supported SqlKinds including unary and cast. */
     private static final Set<SqlKind> SUPPORTED_KINDS = EnumSet.of(
             SqlKind.PLUS,
@@ -456,27 +466,28 @@ private static boolean isUnixTimestampSupported(RexCall call, ConverterContext c
     }
 
     /**
-     * Returns {@code true} if the native function can resolve {@code zoneId}. It resolves a zone by
-     * exact-match lookup in the IANA time zone database, so two id families that Flink's
-     * {@code table.local-time-zone} accepts have to fall back:
-     *
-     * <ul>
-     *   <li>fixed-offset constructions such as {@code GMT-08:00}, which name an offset rather than
-     *       a region and are absent from {@link ZoneId#getAvailableZoneIds()}
-     *   <li>the legacy {@code SystemV/*} aliases, which the JDK still resolves but the database
-     *       the native lookup consults does not carry
-     * </ul>
+     * Returns {@code true} if the native function can resolve {@code zoneId}. It resolves either a
+     * region id, by exact-match lookup in the IANA time zone database, or a fixed offset, by
+     * parsing it. The one family Flink's {@code table.local-time-zone} accepts that still has to
+     * fall back is the legacy {@code SystemV/*} aliases, which the JDK resolves but the database
+     * the native lookup consults does not carry.
      *
-     * <p>The check has to happen at plan time: an unresolvable id fails inside the native call,
-     * and the Calc operator has no run-time fallback to catch it.
+     * <p>The set this admits has to equal the set the native side resolves, in both directions.
+     * Admitting less costs acceleration; admitting more is worse than an error, because a zone the
+     * native function rejects surfaces as an empty result set rather than a failure — the Calc
+     * operator has no run-time fallback, and nothing downstream distinguishes "no rows" from "the
+     * zone was unresolvable".
      *
-     * <p>The membership test consults the JDK's copy of the database as a proxy for the one the
+     * <p>The region branch consults the JDK's copy of the database as a proxy for the one the
      * native side resolves against. The two are updated independently and nothing in the build
      * pins them together, so any further id family they stop agreeing on has to be excluded here
-     * the way {@code SystemV/*} is.
+     * the way {@code SystemV/*} is. The fixed-offset branch depends on neither database: an offset
+     * is parsed arithmetically on both sides.
      */
     private static boolean isNativelySupportedZone(String zoneId) {
-        return !zoneId.startsWith("SystemV/") && ZoneId.getAvailableZoneIds().contains(zoneId);
+        return !zoneId.startsWith("SystemV/")
+                && (ZoneId.getAvailableZoneIds().contains(zoneId)
+                        || FIXED_OFFSET_ZONE.matcher(zoneId).matches());
     }
 
     /**

auron-flink-extension/auron-flink-planner/src/test/java/org/apache/auron/flink/table/planner/converter/RexCallConverterTest.java

@@ -18,13 +18,17 @@
 
 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;
 
 import java.io.ByteArrayInputStream;
 import java.io.IOException;
 import java.math.BigDecimal;
+import java.time.ZoneId;
 import java.util.Arrays;
+import java.util.LinkedHashMap;
+import java.util.Locale;
+import java.util.Map;
+import java.util.TimeZone;
 import org.apache.arrow.memory.BufferAllocator;
 import org.apache.arrow.memory.RootAllocator;
 import org.apache.arrow.vector.VarCharVector;
@@ -565,16 +569,17 @@ void testUnixTimestampTimezonePropagatesToNode() throws IOException {
                 "Asia/Shanghai", decodeStringLiteral(result.getScalarFunction().getArgs(2)));
     }
 
-    /** A fixed-offset session zone names an offset rather than a region: Flink accepts it, the
-     * native lookup cannot resolve it, so the gate rejects it and the builder refuses it outright
-     * rather than emitting a node that fails at run time. */
+    /** A fixed-offset session zone names an offset rather than a region: the native function parses
+     * the offset instead of looking it up, so the gate admits it. The id reaches the node verbatim,
+     * because the two sides have to agree on the spelling and not merely on the offset it denotes. */
     @Test
-    void testUnixTimestampFixedOffsetZoneFallsBack() {
+    void testUnixTimestampFixedOffsetZoneIsSupported() throws IOException {
         ConverterContext tzContext = contextWithZone("GMT-08:00");
         RexNode call = makeCall(bigintType(), FlinkSqlOperatorTable.UNIX_TIMESTAMP, strRef(5));
 
-        assertFalse(converter.isSupported(call, tzContext));
-        assertThrows(IllegalArgumentException.class, () -> converter.convert(call, tzContext));
+        assertTrue(converter.isSupported(call, tzContext));
+        PhysicalExprNode result = converter.convert(call, tzContext);
+        assertEquals("GMT-08:00", decodeStringLiteral(result.getScalarFunction().getArgs(2)));
     }
 
     /** A legacy {@code SystemV/*} session zone still resolves in the JDK, so it reaches the gate
@@ -603,12 +608,99 @@ void testUnixTimestampZeroArgNodeShape() {
     }
 
     /** The zero-argument result is epoch seconds, which no session time zone bears on, so the gate
-     * admits it even under a zone the native lookup cannot resolve. */
+     * admits it even under a zone the native lookup cannot resolve. The zone has to be one the gate
+     * actually rejects, or the test passes without exercising the ordering it pins. */
     @Test
-    void testUnixTimestampZeroArgSupportedWithFixedOffsetZone() {
+    void testUnixTimestampZeroArgSupportedWithUnresolvableZone() {
         RexNode call = makeCall(bigintType(), FlinkSqlOperatorTable.UNIX_TIMESTAMP);
 
-        assertTrue(converter.isSupported(call, contextWithZone("GMT-08:00")));
+        assertTrue(converter.isSupported(call, contextWithZone("SystemV/PST8")));
+    }
+
+    /**
+     * The gate must admit every {@code GMT±HH:MM} id, because the native function resolves every one
+     * of them. Admitting a zone it cannot resolve is worse than rejecting one it can: the native
+     * call returns an error, the unwind handler turns that into a default-valued batch, and the
+     * query yields no rows while recording no fallback.
+     *
+     * <p>The family is swept in full rather than sampled, because where the boundary falls is the
+     * whole content of the predicate. The ids are built from the loop indices under
+     * {@link Locale#ROOT}, so a locale with non-ASCII digits cannot generate ids that fail the
+     * ASCII-only pattern for a reason unrelated to the gate.
+     */
+    @Test
+    void testUnixTimestampZoneGateAdmitsEveryFixedOffsetZone() {
+        RexNode call = makeCall(bigintType(), FlinkSqlOperatorTable.UNIX_TIMESTAMP, strRef(5));
+
+        int covered = 0;
+        for (int sign = -1; sign <= 1; sign += 2) {
+            for (int hh = 0; hh <= 18; hh++) {
+                for (int mm = 0; mm <= 59; mm++) {
+                    int magnitude = hh * 3600 + mm * 60;
+                    // ±00:00 normalizes to plain GMT and never reaches the gate spelled this way;
+                    // anything past ±18:00 is not a zone id at all.
+                    if (magnitude == 0 || magnitude > 18 * 3600) {
+                        continue;
+                    }
+                    String zoneId = String.format(Locale.ROOT, "GMT%s%02d:%02d", sign > 0 ? "+" : "-", hh, mm);
+                    assertTrue(converter.isSupported(call, contextWithZone(zoneId)), zoneId);
+                    covered++;
+                }
+            }
+        }
+        assertEquals(2160, covered, "a skipped sweep would pass vacuously");
+    }
+
+    /**
+     * The gate narrows rather than disappearing: the {@code SystemV/*} family is absent from the
+     * time zone database the native side consults, so every one of its ids must still fall back.
+     *
+     * <p>The family is swept from the JDK's own id set rather than a fixed list, so it covers
+     * whatever the runtime carries. How many ids that is comes from the bundled tzdb and can change
+     * without the gate's behavior changing, so the sweep asserts only that it was not vacuous.
+     */
+    @Test
+    void testUnixTimestampZoneGateRejectsEverySystemVZone() {
+        RexNode call = makeCall(bigintType(), FlinkSqlOperatorTable.UNIX_TIMESTAMP, strRef(5));
+
+        int covered = 0;
+        for (String zoneId : ZoneId.getAvailableZoneIds()) {
+            if (zoneId.startsWith("SystemV/")) {
+                assertFalse(converter.isSupported(call, contextWithZone(zoneId)), zoneId);
+                covered++;
+            }
+        }
+        assertTrue(covered > 0, "a skipped sweep would pass vacuously");
+    }
+
+    /**
+     * The bare {@code ±HH:MM} spelling cannot be configured — Flink's validator rejects it — but an
+     * unset session zone resolves to {@link ZoneId#systemDefault()} without passing through that
+     * validator, and a process running under {@code TZ=EST}, {@code MST} or {@code HST} lands on
+     * exactly these three ids. The native parser treats the {@code GMT} prefix as optional for this
+     * reason, so the gate has to as well.
+     *
+     * <p>Each iteration asserts the id it actually reached before asserting the verdict, so the
+     * sweep cannot silently degrade into testing the machine's own time zone three times.
+     */
+    @Test
+    void testUnixTimestampZoneGateAdmitsTheBareOffsetsReachableFromTheSystemDefault() {
+        RexNode call = makeCall(bigintType(), FlinkSqlOperatorTable.UNIX_TIMESTAMP, strRef(5));
+
+        Map<String, String> shortIdZones = new LinkedHashMap<>();
+        shortIdZones.put("EST", "-05:00");
+        shortIdZones.put("MST", "-07:00");
+        shortIdZones.put("HST", "-10:00");
+        TimeZone savedDefault = TimeZone.getDefault();
+        try {
+            for (Map.Entry<String, String> entry : shortIdZones.entrySet()) {
+                TimeZone.setDefault(TimeZone.getTimeZone(entry.getKey()));
+                assertEquals(entry.getValue(), ZoneId.systemDefault().getId());
+                assertTrue(converter.isSupported(call, contextWithUnsetZone()), entry.getValue());
+            }
+        } finally {
+            TimeZone.setDefault(savedDefault);
+        }
     }
 
     @Test
@@ -635,6 +727,15 @@ private ConverterContext contextWithZone(String zoneId) {
         return new ConverterContext(conf, null, getClass().getClassLoader(), context.getInputType());
     }
 
+    /**
+     * Returns a copy of the shared context with no session time zone configured, which resolves to
+     * {@link ZoneId#systemDefault()} without passing through Flink's zone-id validator. This is the
+     * only path by which an id the validator rejects can reach the converter.
+     */
+    private ConverterContext contextWithUnsetZone() {
+        return new ConverterContext(new Configuration(), null, getClass().getClassLoader(), context.getInputType());
+    }
+
     private static String decodeStringLiteral(PhysicalExprNode node) throws IOException {
         byte[] bytes = node.getLiteral().getIpcBytes().toByteArray();
         try (BufferAllocator alloc = new RootAllocator(Long.MAX_VALUE);

auron-flink-extension/auron-flink-planner/src/test/java/org/apache/auron/flink/table/runtime/AuronFlinkCalcITCase.java

@@ -248,14 +248,28 @@ public void testUnixTimestampUtcTimeZone() {
         assertThat(rows).isEqualTo(Arrays.asList(Row.of(1602288001L), Row.of(1602288002L), Row.of(1602288003L)));
     }
 
-    /** UNIX_TIMESTAMP yields the epoch seconds for the offset when the session timezone is a
-     * fixed-offset construction, a form Flink accepts that has no native equivalent. */
+    /**
+     * A fixed-offset session timezone names a constant offset rather than a region. The native
+     * function parses the offset instead of looking it up, so the Calc converts and applies it.
+     *
+     * <p>Flink's codegen Calc produces the same rows from the same offset, so the values alone
+     * cannot show which engine ran. The fallback counter is what discriminates: a Calc that fails
+     * to convert records either an unsupported node or a composition failure before falling back,
+     * so a count of zero means this Calc converted. The row set carries the rest — a plan that
+     * converts but hits no native registry arm fails while executing, where nothing records a
+     * fallback, and surfaces as an empty result with the counter still at zero.
+     */
     @Test
-    public void testUnixTimestampFixedOffsetTimeZoneFallsBack() {
+    public void testUnixTimestampFixedOffsetTimeZoneRunsNatively() {
+        UnsupportedFlinkNodeRecorder.resetForTest();
         tableEnvironment.getConfig().setLocalTimeZone(ZoneId.of("GMT-08:00"));
         List<Row> rows = CollectionUtil.iteratorToList(tableEnvironment
                 .executeSql("select UNIX_TIMESTAMP(`ts`) from T1")
                 .collect());
+
+        assertThat(UnsupportedFlinkNodeRecorder.peekEmitCount())
+                .as("a non-zero fallback count means the Calc did not run natively")
+                .isZero();
         rows.sort(Comparator.comparingLong(o -> (long) o.getField(0)));
         assertThat(rows).isEqualTo(Arrays.asList(Row.of(1602316801L), Row.of(1602316802L), Row.of(1602316803L)));
     }