You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@auron.apache.org by sl...@apache.org on 2026/08/20 06:41:17 UTC
(auron) branch master updated: [AURON #2455] Run UNIX_TIMESTAMP natively for fixed-offset session time zones (#2466)
This is an automated email from the ASF dual-hosted git repository.
slfan1989 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/auron.git
The following commit(s) were added to refs/heads/master by this push:
new 2412b016f [AURON #2455] Run UNIX_TIMESTAMP natively for fixed-offset session time zones (#2466)
2412b016f is described below
commit 2412b016fc45e7d075d76b2509f6fd14ea13cb10
Author: Weiqing Yang <ya...@gmail.com>
AuthorDate: Wed Aug 19 23:41:12 2026 -0700
[AURON #2455] Run UNIX_TIMESTAMP natively for fixed-offset session time zones (#2466)
# Which issue does this PR close?
Closes #2455
# Rationale for this change
`table.local-time-zone` accepts fixed-offset ids such as `GMT-08:00`,
and Flink's own validation message recommends that form. The native
`Flink_UnixTimestamp` resolved its session zone by exact-match lookup
against the IANA time zone database, which carries no `GMT±HH:MM`
entries, so #2448 rejects those ids at plan time and the whole Calc
falls back to Flink.
That gate was the right call, since before it the id reached the native
call and failed past the point where any fallback remained. But it costs
native execution for a large family of zones, and it is reachable with
no explicit configuration at all: the default value resolves to
`ZoneId.systemDefault()`, so a TaskManager running with `TZ=GMT-08:00`
lands there silently.
One correction to the issue body while I am here. I wrote that 141 of
745 accepted ids are in this family. Enumerating them gives **2160 of
2764**. The family is every minute, not quarter-hours: `GMT+05:07` is
accepted and normalizes to itself. My original count assumed
quarter-hour granularity and also dropped the negative sub-hour ids. The
design does not change, since 2160 is still small enough to enumerate
exhaustively, but the number is wrong in the issue and I would rather
correct it than leave it.
Two spellings in the issue body also turn out not to need support.
Flink's validator rejects `UTC+8` and a bare `+05:30` as configured
values, so only the `GMT±HH:MM` form is reachable that way.
# What changes are included in this PR?
**Native.** A `ZoneSpec` that is either an IANA region or a constant UTC
offset. The region arm delegates to `resolve_offset_secs` **unchanged**,
so the ambiguous and gap handling validated for that path is preserved
by construction rather than by retesting. The fixed arm returns the
constant, which is what Flink computes for these zones at every instant.
The offset parser is hand-rolled rather than delegating to
`FixedOffset`'s `FromStr`, which discards unconsumed input and would
read `+08:00:30` as `+08:00` with no error.
**Plan-time gate.** An id is admitted when it is not `SystemV/*` and is
either a database region or a fixed offset. `SystemV/*` still falls
back: those ids are genuinely absent from the database the native lookup
consults, so the gate narrows rather than disappears. Both call sites
route through the one predicate method, so there is no second site to
keep in sync.
Bare `±HH:MM` is admitted as well. The default configuration path
returns `ZoneId.systemDefault()` without validating it, so a TaskManager
on `TZ=EST` delivers `-05:00`, a spelling the validator itself would
reject. Three are reachable this way.
A note on why the offset is resolved natively rather than mapped on the
JVM side. Mapping onto the database's `Etc/GMT∓N` names would invert the
sign, since `Etc/GMT+8` is UTC−8, and an error there ships timestamps
wrong by twice the offset rather than merely unaccelerated. Those names
are also whole-hours-only across UTC−12 to +14, so `+05:45` and
everything past ±12 has no equivalent at any sign.
`arrow::array::timezone::Tz` is worth mentioning because it is exactly
this abstraction and is already on the classpath. It does not fit here:
its grammar accepts `+08:00` but rejects `GMT-08:00`, which is the
entire family in question, and its offset type does not expose the
base-UTC-offset accessor that the ambiguous and gap branches depend on,
so adopting it would have meant rewriting the semantics this change is
trying to leave alone.
# Are there any user-facing changes?
Yes. A session on a fixed-offset time zone now runs `UNIX_TIMESTAMP`
natively instead of falling back to Flink. Results are unchanged; only
the execution path differs. Sessions on `SystemV/*` continue to fall
back.
# How was this patch tested?
The gate and the native parser have to admit exactly the same set.
Admitting more would not surface as an error: a failure inside the
native call is turned into a default value, so the query would return no
rows rather than fail. That agreement was therefore checked over every
reachable id, and independently three times, each derivation building
both sides differently. The strongest of the three compiled the shipped
Rust source directly rather than transcribing it, and derived the
reachable set by calling Flink's own config accessor over 92,476
candidate strings rather than modelling it. All three agree: 2754
admitted, 2754 resolvable, zero admitted-but-unresolvable and zero
resolvable-but-rejected. The 13-id gap is exactly `SystemV/*`.
Native tests cover the whole family exhaustively rather than by
sampling, with expected offsets recomputed from the loop indices so the
assertion is not circular. Because that loop cannot falsify the closed
form it is built on, a separate fixture pins 13 spot values taken from a
`SimpleDateFormat` oracle, covering `GMT`, `GMT±00:01`, `GMT±18:00`,
`GMT+05:07`, `GMT+05:45`, an epoch day crossing, 1900, and 1582. The
differential sweep behind those values ran 155,520 comparisons across
four JDKs with no mismatches.
The negative cases are the ones that discriminate between a strict
parser and a lenient one, so they cover near-misses rather than garbage:
second precision, an embedded space, repeated colons, a Unicode minus
sign, and a single-digit hour. Out-of-range offsets are covered too, and
both guards were mutation-tested: deleting either one admits bad offsets
while leaving the other's cases rejected.
On the JVM side the family is enumerated against the gate, `SystemV/*`
is asserted to still fall back, and the integration test now asserts on
the fallback counter rather than only on the answer, since native and
fallback both produce the right answer and an answer-only assertion
cannot tell them apart.
`RexCallConverterTest` 54 green, `AuronFlinkCalcITCase` 23 green, native
suite 25 green, Checkstyle clean.
# Was this patch authored or co-authored using generative AI tooling?
- [x] Yes
- [ ] No
`Generated-by: Claude Code (Claude Opus 5)`
---
.../table/planner/converter/RexCallConverter.java | 41 ++--
.../planner/converter/RexCallConverterTest.java | 121 ++++++++++-
.../flink/table/runtime/AuronFlinkCalcITCase.java | 20 +-
.../datafusion-ext-functions/src/flink_datetime.rs | 221 +++++++++++++++++++--
4 files changed, 362 insertions(+), 41 deletions(-)
diff --git a/auron-flink-extension/auron-flink-planner/src/main/java/org/apache/auron/flink/table/planner/converter/RexCallConverter.java b/auron-flink-extension/auron-flink-planner/src/main/java/org/apache/auron/flink/table/planner/converter/RexCallConverter.java
index d776f76dc..2cc446c6f 100644
--- a/auron-flink-extension/auron-flink-planner/src/main/java/org/apache/auron/flink/table/planner/converter/RexCallConverter.java
+++ b/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.EnumSet;
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 @@ public class RexCallConverter implements FlinkRexNodeConverter {
}
/**
- * 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());
}
/**
diff --git a/auron-flink-extension/auron-flink-planner/src/test/java/org/apache/auron/flink/table/planner/converter/RexCallConverterTest.java b/auron-flink-extension/auron-flink-planner/src/test/java/org/apache/auron/flink/table/planner/converter/RexCallConverterTest.java
index de0599785..9f7ec3c92 100644
--- a/auron-flink-extension/auron-flink-planner/src/test/java/org/apache/auron/flink/table/planner/converter/RexCallConverterTest.java
+++ b/auron-flink-extension/auron-flink-planner/src/test/java/org/apache/auron/flink/table/planner/converter/RexCallConverterTest.java
@@ -18,13 +18,17 @@ package org.apache.auron.flink.table.planner.converter;
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 @@ class RexCallConverterTest {
"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 @@ class RexCallConverterTest {
}
/** 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 @@ class RexCallConverterTest {
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);
diff --git a/auron-flink-extension/auron-flink-planner/src/test/java/org/apache/auron/flink/table/runtime/AuronFlinkCalcITCase.java b/auron-flink-extension/auron-flink-planner/src/test/java/org/apache/auron/flink/table/runtime/AuronFlinkCalcITCase.java
index 89f6cf7ed..bf466937b 100644
--- a/auron-flink-extension/auron-flink-planner/src/test/java/org/apache/auron/flink/table/runtime/AuronFlinkCalcITCase.java
+++ b/auron-flink-extension/auron-flink-planner/src/test/java/org/apache/auron/flink/table/runtime/AuronFlinkCalcITCase.java
@@ -248,14 +248,28 @@ public class AuronFlinkCalcITCase extends AuronFlinkTableTestBase {
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)));
}
diff --git a/native-engine/datafusion-ext-functions/src/flink_datetime.rs b/native-engine/datafusion-ext-functions/src/flink_datetime.rs
index f89d50264..d6ed2508d 100644
--- a/native-engine/datafusion-ext-functions/src/flink_datetime.rs
+++ b/native-engine/datafusion-ext-functions/src/flink_datetime.rs
@@ -19,7 +19,7 @@ use arrow::{
array::{Int64Array, StringArray},
datatypes::DataType,
};
-use chrono::{DateTime, LocalResult, NaiveDateTime, Offset, TimeZone, Utc};
+use chrono::{DateTime, FixedOffset, LocalResult, NaiveDateTime, Offset, TimeZone, Utc};
use chrono_tz::{OffsetComponents, Tz};
use datafusion::{
common::{DataFusionError, Result, ScalarValue},
@@ -66,9 +66,7 @@ pub fn flink_unix_timestamp(args: &[ColumnarValue]) -> Result<ColumnarValue> {
let zone_id = utf8_scalar(&args[2]).ok_or_else(|| {
DataFusionError::Execution("Flink_UnixTimestamp: zoneId must be a non-null string".into())
})?;
- let tz: Tz = zone_id.parse().map_err(|_| {
- DataFusionError::Execution(format!("Flink_UnixTimestamp: invalid timezone {zone_id}"))
- })?;
+ let zone = parse_zone_spec(&zone_id)?;
let tokens = parse_format(&format)?;
@@ -85,7 +83,7 @@ pub fn flink_unix_timestamp(args: &[ColumnarValue]) -> Result<ColumnarValue> {
let result = Int64Array::from_iter(
value
.iter()
- .map(|opt_s| opt_s.map(|s| parse_datetime(s, &tokens, tz).unwrap_or(i64::MIN))),
+ .map(|opt_s| opt_s.map(|s| parse_datetime(s, &tokens, &zone).unwrap_or(i64::MIN))),
);
Ok(ColumnarValue::Array(Arc::new(result)))
@@ -125,6 +123,54 @@ fn utf8_scalar(arg: &ColumnarValue) -> Option<String> {
}
}
+/// Resolve the session time zone id to a zone specification.
+///
+/// The id arrives already normalized by `ZoneId.getId()`, so the accepted forms
+/// are an IANA region id, `GMT±HH:MM`, or a bare `±HH:MM`. Anything else is an
+/// error rather than a default: the plan-time gate admits only what this
+/// function resolves, so a disagreement between the two is a plumbing bug, and
+/// returning an error keeps it from surfacing as silently wrong data.
+fn parse_zone_spec(zone_id: &str) -> Result<ZoneSpec> {
+ if let Ok(tz) = zone_id.parse::<Tz>() {
+ return Ok(ZoneSpec::Iana(tz));
+ }
+ if let Some(secs) = parse_fixed_offset_secs(zone_id.strip_prefix("GMT").unwrap_or(zone_id)) {
+ if let Some(offset) = FixedOffset::east_opt(secs) {
+ return Ok(ZoneSpec::Fixed(offset));
+ }
+ }
+ Err(DataFusionError::Execution(format!(
+ "Flink_UnixTimestamp: invalid timezone {zone_id}"
+ )))
+}
+
+/// Parse exactly `±HH:MM`, rejecting any trailing input so that a
+/// second-precision id cannot silently lose its seconds field.
+fn parse_fixed_offset_secs(s: &str) -> Option<i32> {
+ let b = s.as_bytes();
+ if b.len() != 6 || b[3] != b':' {
+ return None;
+ }
+ let sign = match b[0] {
+ b'+' => 1,
+ b'-' => -1,
+ _ => return None,
+ };
+ if !b[1..3].iter().chain(&b[4..6]).all(u8::is_ascii_digit) {
+ return None;
+ }
+ let hours = i32::from(b[1] - b'0') * 10 + i32::from(b[2] - b'0');
+ let minutes = i32::from(b[4] - b'0') * 10 + i32::from(b[5] - b'0');
+ if minutes > 59 {
+ return None;
+ }
+ let secs = sign * (hours * 3600 + minutes * 60);
+ if secs.abs() > 18 * 3600 {
+ return None;
+ }
+ Some(secs)
+}
+
/// A field is one of the six numeric SimpleDateFormat components; every
/// supported specifier maps to exactly one. `width` is the `obeyCount` scan
/// window used when the field is immediately followed by another numeric field.
@@ -237,7 +283,7 @@ impl Fields {
/// format. Every other token consumes at least one byte — a literal matches
/// one, a field needs at least one digit — so an empty token list is the only
/// way to finish having consumed nothing.
-fn parse_datetime(input: &str, tokens: &[Token], tz: Tz) -> Option<i64> {
+fn parse_datetime(input: &str, tokens: &[Token], zone: &ZoneSpec) -> Option<i64> {
if tokens.is_empty() {
return None;
}
@@ -287,7 +333,7 @@ fn parse_datetime(input: &str, tokens: &[Token], tz: Tz) -> Option<i64> {
}
let local_sec = normalize(&fields);
- let offset = resolve_offset_secs(local_sec, tz);
+ let offset = zone.offset_secs(local_sec);
Some(local_sec - offset)
}
@@ -394,6 +440,25 @@ fn resolve_offset_secs(local_sec: i64, tz: Tz) -> i64 {
}
}
+/// Session time zone as resolved at plan time: an IANA region from the time
+/// zone database, or a constant UTC offset.
+#[derive(Clone, Copy)]
+enum ZoneSpec {
+ Iana(Tz),
+ Fixed(FixedOffset),
+}
+
+impl ZoneSpec {
+ /// UTC offset in seconds for a local wall-clock instant. A fixed offset is
+ /// constant by construction, so it ignores the instant entirely.
+ fn offset_secs(&self, local_sec: i64) -> i64 {
+ match self {
+ ZoneSpec::Iana(tz) => resolve_offset_secs(local_sec, *tz),
+ ZoneSpec::Fixed(offset) => offset.local_minus_utc() as i64,
+ }
+ }
+}
+
#[cfg(test)]
mod tests {
use arrow::array::Array;
@@ -420,6 +485,18 @@ mod tests {
out.value(0)
}
+ /// Whether the zone id is rejected outright. A rejected zone is an error
+ /// from the function itself, distinct from the `i64::MIN` sentinel that
+ /// a row-level parse failure produces.
+ fn zone_rejected(tz: &str) -> bool {
+ let args = vec![
+ ColumnarValue::Scalar(ScalarValue::Utf8(Some("2020-10-10 00:00:01".to_string()))),
+ ColumnarValue::Scalar(ScalarValue::Utf8(Some(DEFAULT.to_string()))),
+ ColumnarValue::Scalar(ScalarValue::Utf8(Some(tz.to_string()))),
+ ];
+ flink_unix_timestamp(&args).is_err()
+ }
+
const MIN: i64 = i64::MIN;
const DEFAULT: &str = "%Y-%m-%d %H:%M:%S";
@@ -557,6 +634,84 @@ mod tests {
);
}
+ // Every `GMT±HH:MM` zone applies the constant offset `sign × (hh·3600 +
+ // mm·60)`, across the whole family. The expected offset is recomputed from
+ // the loop indices that built the id, never re-parsed from the id string,
+ // so the assertion cannot degenerate into re-running the code under test.
+ //
+ // `GMT±00:00` is excluded because it normalizes to plain `GMT` and so never
+ // reaches this function in that spelling; anything past `±18:00` is not a zone
+ // id at all. The counter guards against a skip condition that silently empties
+ // the loop.
+ #[test]
+ fn oracle_fixed_offset_family_exhaustive() {
+ const VALUE: &str = "2020-10-10 00:00:01";
+ let local_sec = run(VALUE, DEFAULT, "UTC");
+
+ let mut covered = 0;
+ for sign in [1i64, -1i64] {
+ for hh in 0..=18i64 {
+ for mm in 0..=59i64 {
+ let magnitude = hh * 3600 + mm * 60;
+ if magnitude == 0 || magnitude > 18 * 3600 {
+ continue;
+ }
+ let sign_char = if sign > 0 { '+' } else { '-' };
+ let id = format!("GMT{sign_char}{hh:02}:{mm:02}");
+ let expected_offset = sign * magnitude;
+ assert_eq!(
+ run(VALUE, DEFAULT, &id),
+ local_sec - expected_offset,
+ "zone {id}"
+ );
+ covered += 1;
+ }
+ }
+ }
+ assert_eq!(covered, 2160);
+ }
+
+ // Spot values for the fixed-offset family, taken from the Java oracle rather
+ // than from the closed form, so the closed form itself is falsifiable: the
+ // exhaustive sweep above recomputes the same arithmetic and therefore cannot
+ // detect a sign or unit error shared by both sides.
+ //
+ // Covers plain `GMT` (which resolves as a region, not an offset), the smallest
+ // and largest offsets in both directions, a non-quarter-hour offset, a
+ // day-crossing at the epoch, a pre-1900 date, and the 1582 hybrid-calendar
+ // boundary.
+ //
+ // The last three rows are the bare `±HH:MM` spelling, and are the complete
+ // reachable set of it: the session zone is only unvalidated when it comes from
+ // `ZoneId.systemDefault()`, where `TZ=EST`/`MST`/`HST` map through
+ // `ZoneId.SHORT_IDS` to exactly these three. They assert the resolved offset
+ // rather than merely the absence of an error, so a resolver that accepted the
+ // bare form but mapped it to UTC would still fail here.
+ #[test]
+ fn oracle_fixed_offset() {
+ assert_eq!(run("2020-10-10 00:00:01", DEFAULT, "GMT"), 1602288001);
+ assert_eq!(run("2020-10-10 00:00:01", DEFAULT, "GMT+00:01"), 1602287941);
+ assert_eq!(run("2020-10-10 00:00:01", DEFAULT, "GMT-00:01"), 1602288061);
+ assert_eq!(run("2020-10-10 00:00:01", DEFAULT, "GMT+18:00"), 1602223201);
+ assert_eq!(run("2020-10-10 00:00:01", DEFAULT, "GMT-18:00"), 1602352801);
+ assert_eq!(run("2020-10-10 00:00:01", DEFAULT, "GMT+05:07"), 1602269581);
+ assert_eq!(run("2020-10-10 00:00:01", DEFAULT, "GMT+05:45"), 1602267301);
+ assert_eq!(run("2020-10-10 00:00:01", DEFAULT, "GMT+08:00"), 1602259201);
+ assert_eq!(run("2020-10-10 00:00:01", DEFAULT, "GMT-08:00"), 1602316801);
+ assert_eq!(run("1970-01-01 00:00:00", DEFAULT, "GMT+08:00"), -28800);
+ assert_eq!(
+ run("1900-01-01 00:00:00", DEFAULT, "GMT-08:00"),
+ -2208960000
+ );
+ assert_eq!(
+ run("1582-10-15 00:00:00", DEFAULT, "GMT-08:00"),
+ -12219264000
+ );
+ assert_eq!(run("2020-10-10 00:00:01", DEFAULT, "-05:00"), 1602306001);
+ assert_eq!(run("2020-10-10 00:00:01", DEFAULT, "-07:00"), 1602313201);
+ assert_eq!(run("2020-10-10 00:00:01", DEFAULT, "-10:00"), 1602324001);
+ }
+
// Literals from quoted patterns: a case-sensitive letter, a literal percent
// (`%%`), and a literal single quote.
#[test]
@@ -668,14 +823,54 @@ mod tests {
assert!(flink_unix_timestamp_now(&one).is_err());
}
+ // A zone id that is not resolvable is an error, not a silent default. Beyond
+ // outright nonsense, the rows here are spellings close enough to a real id that
+ // a permissive resolver would accept them: wrong case, surrounding whitespace,
+ // and the `SystemV/*` family, which is absent from the time zone database and
+ // must keep being rejected rather than being swept in with the fixed offsets.
#[test]
fn invalid_timezone_errors() {
- let args = vec![
- ColumnarValue::Scalar(ScalarValue::Utf8(Some("2020-10-10 00:00:01".to_string()))),
- ColumnarValue::Scalar(ScalarValue::Utf8(Some(DEFAULT.to_string()))),
- ColumnarValue::Scalar(ScalarValue::Utf8(Some("Mars/Olympus".to_string()))),
- ];
- assert!(flink_unix_timestamp(&args).is_err());
+ assert!(zone_rejected("Mars/Olympus"));
+ assert!(zone_rejected("gmt+08:00"));
+ assert!(zone_rejected(" GMT+08:00"));
+ assert!(zone_rejected("GMT+08:00 "));
+ assert!(zone_rejected("SystemV/PST8"));
+ }
+
+ // The fixed-offset grammar is exactly `±HH:MM`, optionally prefixed by `GMT`,
+ // with nothing before or after it. Each row is a spelling a permissive offset
+ // parser would accept: a trailing seconds field (which a truncating parser
+ // silently drops, turning `+08:00:30` into `+08:00`), a separator that is not a
+ // colon, a repeated separator, a Unicode minus sign in place of the ASCII
+ // hyphen, a single-digit hour, and a single-digit minute.
+ #[test]
+ fn invalid_fixed_offset_zone_errors() {
+ assert!(zone_rejected("GMT+08:00:30"));
+ assert!(zone_rejected("GMT+08 00"));
+ assert!(zone_rejected("GMT+08::::00"));
+ assert!(zone_rejected("GMT\u{2212}08:00")); // U+2212 MINUS SIGN, not '-'
+ assert!(zone_rejected("GMT+8"));
+ assert!(zone_rejected("+08:0"));
+ }
+
+ // The two numeric bounds on a fixed offset: minutes are a sixtieth of an hour,
+ // not a free two-digit field, and the magnitude stops at ±18:00.
+ //
+ // Both are unreachable through the plan-time gate, which only ever emits ids
+ // built from a constructed `ZoneId` — `ZoneId.of` throws on all of these. They
+ // are asserted because the bounds are the parser's own guarantee rather than an
+ // inherited one, and a caller that does not hold the `ZoneId` invariant would
+ // otherwise resolve `GMT+08:60` to a well-formed offset an hour past where it
+ // should be. Each row fails for exactly one of the two checks, so neither can
+ // be deleted while the suite stays green.
+ #[test]
+ fn out_of_range_fixed_offset_zone_errors() {
+ assert!(zone_rejected("GMT+08:60"));
+ assert!(zone_rejected("GMT-08:60"));
+
+ assert!(zone_rejected("GMT+18:01"));
+ assert!(zone_rejected("GMT-18:01"));
+ assert!(zone_rejected("GMT+19:00"));
}
// An empty format consumes nothing, so it matches no input at all rather