[AURON #1863] Support native Flink UNIX_TIMESTAMP: converter integration - #2448
[AURON #1863] Support native Flink UNIX_TIMESTAMP: converter integration#2448weiqingy wants to merge 3 commits into
Conversation
…tegration Wire Flink's UNIX_TIMESTAMP (1-arg and 2-arg forms) to the native Flink_UnixTimestamp function. The converter matches the operator by reference identity (as TRY_CAST does), translates the supported subset of Java date-format patterns to the native format, resolves the session time zone at plan time, and builds the native scalar-function node. Formats outside the supported subset, non-literal format arguments, the 0-argument form, and format patterns whose lenient parse cannot be represented after translation all fall back to Flink's engine. The 0-argument form is rejected explicitly at both the gate and the builder, so a call that parses nothing and reads the wall clock per record can never reach a native function that expects a value operand to size its output against. Also propagate the effective node config (carrying table.local-time-zone) into the standalone-Calc converter path; the persisted config it previously used does not carry the session time zone, which would silently default the native evaluation to UTC.
There was a problem hiding this comment.
Pull request overview
Adds planner-side support for Flink UNIX_TIMESTAMP by lowering supported call shapes to the native Flink_UnixTimestamp ext scalar function, including plan-time translation of Java SimpleDateFormat patterns to the native strftime-like format and propagation of the effective session time zone into the emitted node. It also fixes config-threading so session options (notably table.local-time-zone) reach native conversion in the standalone Calc path.
Changes:
- Teach
RexCallConverterto recognizeUNIX_TIMESTAMPby operator identity and emit aFlink_UnixTimestampext-function node with args[value, chronoFormat, zoneId]. - Introduce a Java pattern translator (
FlinkDateTimeFormatConverter) with a conservative allowlist + adjacency rule to decide native eligibility vs fallback. - Thread the effective
ExecNodeConfig(not the persisted node config) into native Calc plan building to preserve session configuration such as time zone, and add targeted unit + IT coverage.
Reviewed changes
Copilot reviewed 10 out of 10 changed files in this pull request and generated 2 comments.
Show a summary per file
| File | Description |
|---|---|
| auron-flink-extension/auron-flink-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecCalcTest.java | Adds planner-level tests for 0-arg fallback and time-zone propagation into the native plan. |
| auron-flink-extension/auron-flink-planner/src/test/java/org/apache/auron/flink/table/runtime/AuronFlinkCalcITCase.java | Adds an end-to-end IT case validating native execution + non-UTC session zone behavior. |
| auron-flink-extension/auron-flink-planner/src/test/java/org/apache/auron/flink/table/planner/converter/UnixTimestampOperatorIdentityTest.java | Pins Flink operator identity invariant relied on by the converter dispatch. |
| auron-flink-extension/auron-flink-planner/src/test/java/org/apache/auron/flink/table/planner/converter/RexCallConverterTest.java | Adds unit tests for node shape, format translation, time-zone literal propagation, and fallback gates. |
| auron-flink-extension/auron-flink-planner/src/test/java/org/apache/auron/flink/table/planner/converter/FlinkDateTimeFormatConverterTest.java | Adds focused tests for accepted/rejected patterns, quote escaping, and adjacency hazard rule. |
| auron-flink-extension/auron-flink-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecCalc.java | Uses the effective ExecNodeConfig when seeding the native converter context to preserve session settings. |
| auron-flink-extension/auron-flink-planner/src/main/java/org/apache/auron/flink/table/planner/converter/RexLiteralConverter.java | Adds a helper to encode plan-time string constants into Arrow IPC literals for native arguments. |
| auron-flink-extension/auron-flink-planner/src/main/java/org/apache/auron/flink/table/planner/converter/RexCallConverter.java | Implements UNIX_TIMESTAMP support (arity gating, format translation, zone resolution, ext-function emission). |
| auron-flink-extension/auron-flink-planner/src/main/java/org/apache/auron/flink/table/planner/converter/FlinkNodeConverterUtils.java | Adds a helper to build ext scalar function nodes routed via AuronExtFunctions. |
| auron-flink-extension/auron-flink-planner/src/main/java/org/apache/auron/flink/table/planner/converter/FlinkDateTimeFormatConverter.java | New Java SimpleDateFormat → native format translator with strict fallback behavior. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| public static PhysicalExprNode stringLiteral(String value) { | ||
| RowType rowType = RowType.of(new VarCharType(VarCharType.MAX_LENGTH)); | ||
| try (BufferAllocator allocator = | ||
| FlinkArrowUtils.getRootAllocator().newChildAllocator("literal", 0, Long.MAX_VALUE); | ||
| VectorSchemaRoot root = VectorSchemaRoot.create(FlinkArrowUtils.toArrowSchema(rowType), allocator)) { | ||
|
|
||
| GenericRowData rowData = new GenericRowData(1); | ||
| rowData.setField(0, StringData.fromString(value)); | ||
|
|
There was a problem hiding this comment.
Added a requireNonNull in cccb1acb, though for a different reason than the one given here.
Null cannot reach this method today. It has two call sites, both in RexCallConverter.buildUnixTimestamp: one passes chronoFormat, which comes out of translate(...).orElseThrow(...) and so is non-null by construction, and the other passes zone.getId(), which ZoneId never returns null from.
On the diagnosis being opaque: conversion failures are already caught. FlinkNodeConverterFactory.convertRexNode wraps the convert call in catch (Exception e) and logs RexNode conversion failed for {} with the full stack trace, then returns empty so the Calc falls back to Flink. So an NPE here would surface as a logged stack trace plus a query that still returns correct results on Flink's engine, not an opaque failure.
What I did take from this: the method is public API introduced by this PR, and a null value would mean the caller failed to resolve a plan-time constant. Encoding that as a NULL literal would ship a silently wrong argument to the native side, so rejecting it at the boundary is the right contract to state. Guard plus javadoc plus testStringLiteralRejectsNull.
| public static Optional<String> translate(String javaPattern) { | ||
| List<Token> tokens = scan(javaPattern); | ||
| if (tokens == null) { |
There was a problem hiding this comment.
Added a requireNonNull in cccb1acb, but deliberately not the Optional.empty() behavior suggested here.
Optional.empty() is this converter's signal that the user wrote a pattern outside the native surface, and the whole Calc should fall back to Flink. That is a normal user-facing outcome. A null pattern is not that: it would mean our own converter never resolved a format, which is a plumbing bug. Mapping it to empty() would route the bug into the same silent fallback and hide it, so it fails fast instead. The same split already exists elsewhere in this path, where isSupported returns false for user-facing cases and buildUnixTimestamp throws IllegalArgumentException for plumbing bugs.
Null also cannot reach it today. The gate at isUnixTimestampSupported reads javaFormat != null && FlinkDateTimeFormatConverter.translate(javaFormat).isPresent(), and && short-circuits. The other call site is inside the private buildUnixTimestamp, which the factory only reaches after isSupported returned true.
Covered by testNullPatternRejectedRatherThanReportedUntranslatable.
|
Hi @Tartarus0zm, could you please help review this PR when you get a chance? Thanks! |
…lic entry points stringLiteral and translate both encode plan-time constants the caller has already resolved. A null argument means the caller never resolved one, which is a plumbing bug rather than an unsupported expression. translate rejects null instead of returning Optional.empty(): empty means the user wrote a pattern outside the native surface and the Calc should fall back, so reusing it for null would route a caller bug into the same silent fallback and hide it.
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 11 out of 11 changed files in this pull request and generated no new comments.
Suppressed comments (1)
auron-flink-extension/auron-flink-planner/src/main/java/org/apache/auron/flink/table/planner/converter/RexCallConverter.java:473
buildUnixTimestampassumes the 2nd operand is a non-nullRexLiteralstring. While the factory gates this today, callingconvert(...)directly (as some tests already do for invalid arity) would currently throwClassCastException/NullPointerExceptionrather than the documentedIllegalArgumentException. Adding explicit validation here makes failure mode deterministic and matches the method contract.
String javaFormat = operands.size() > 1
? ((RexLiteral) operands.get(1)).getValueAs(String.class)
: DEFAULT_UNIX_TIMESTAMP_FORMAT;
String chronoFormat = FlinkDateTimeFormatConverter.translate(javaFormat)
.orElseThrow(() -> new IllegalArgumentException("Unsupported UNIX_TIMESTAMP format: " + javaFormat));
Tartarus0zm
left a comment
Tartarus0zm
left a comment
There was a problem hiding this comment.
LGTM overall, just need more test cases
| * deterministic and to exercise timezone propagation into the native plan. */ | ||
| @Test | ||
| public void testUnixTimestamp() { | ||
| tableEnvironment.getConfig().setLocalTimeZone(ZoneId.of("Asia/Shanghai")); |
There was a problem hiding this comment.
The config table.local-time-zone supports multiple formats, such as GMT-08:00, I'd suggest adding more test cases for this.
There was a problem hiding this comment.
Thanks for the review.
The converter resolves the session zone at plan time and hands the id to the native function, which only knows IANA zone names. A fixed offset isn't one, so it errors during execution:
Flink_UnixTimestamp: invalid timezone GMT-08:00
By that point the plan is already native and the Calc operator has no runtime fallback, so the task dies. It also doesn't need an explicit SET. The default resolves to the JVM's default zone, so a TaskManager running with TZ=GMT-08:00 hits it with no config at all.
Fixed by checking the zone at plan time. If native can't resolve it, the Calc falls back to Flink and returns correct results, the same way the converter already handles formats it can't translate. That covers the GMT±HH:MM forms and the legacy SystemV/* ids.
Added the tests for: GMT-08:00 and UTC IT cases, plus unit tests on the gate.
Fixed-offset sessions fall back for now rather than running natively. Filed #2455 to follow up with native support for them so they stay on the native path.
…ely resolvable
The converter resolved the session time zone at plan time and passed
ZoneId.getId() to the native Flink_UnixTimestamp function, which resolves
zone ids by exact-match lookup in the IANA time zone database. Flink's
table.local-time-zone also accepts fixed-offset constructions such as
GMT-08:00, which name an offset rather than a region and have no entry in
that database. Those reached the native call and failed it:
Flink_UnixTimestamp: invalid timezone GMT-08:00
The failure lands after the plan has been converted, where the Calc
operator has no run-time fallback, so the task died rather than degrading.
Besides an explicit configuration, this was reachable through
setLocalTimeZone(ZoneId.of("+08:00")), which normalizes to GMT+08:00, and
through the default value, which resolves to ZoneId.systemDefault().
Reject such zones at plan time instead, so the Calc falls back to Flink and
returns correct results, matching how an untranslatable format literal is
already handled. The legacy SystemV/* aliases need an explicit exclusion
because getAvailableZoneIds() carries them while the native lookup does not.
Pin the session zone in the converter test setup as well: those tests
inherited the machine's default zone, which the gate now reads.
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 11 out of 11 changed files in this pull request and generated no new comments.
Suppressed comments (1)
auron-flink-extension/auron-flink-planner/src/main/java/org/apache/auron/flink/table/planner/converter/RexCallConverter.java:437
isUnixTimestampSupportedcallsTableConfigUtils.getLocalTimeZone(...)without guarding against exceptions.FlinkNodeConverterFactory.convertRexNodedoes not catch exceptions thrown fromisSupported, so a malformedtable.local-time-zone(or any parsing error thrown by Flink) would fail plan conversion instead of cleanly falling back. Catch the exception here and returnfalseso conversion remains fail-safe.
ZoneId zone = TableConfigUtils.getLocalTimeZone(context.getTableConfig());
if (!isNativelySupportedZone(zone.getId())) {
return false;
}
Which issue does this PR close?
Part of #1863. Not
Closes, because the issue description also covers the 0-input form, which this PR does not implement.This is the Flink Java side of
UNIX_TIMESTAMP. The native function it calls merged in #2409. Together they cover the 1-input and 2-input forms from the issue description.The 0-input form is still outstanding. It is a different function rather than a missing branch: Flink binds the niladic form to
DateTimeUtils.unixTimestamp(), which reads the clock per record and parses nothing, so it needs its own design pass. The converter rejects it explicitly and falls back to Flink's engine, and #1863 stays open to track it.Rationale for this change
This completes native support for Flink's
UNIX_TIMESTAMPby wiring the Flink Calc converter to emit the native function added in #2409. Without it the native function is unreachable, and any Calc containingUNIX_TIMESTAMPfalls back to Flink's engine for the whole Calc.What changes are included in this PR?
The converter recognizes
UNIX_TIMESTAMPand lowers the 1-argument and 2-argument forms to the nativeFlink_UnixTimestampnode.UNIX_TIMESTAMPresolves toSqlKind.OTHER_FUNCTION, so it is matched by reference identity on the operator before the supported-kinds switch, the same wayTRY_CASTis handled.A format scanner translates the supported subset of Java date-format letters (
yyyy MM dd HH mm ssand literals) to the native format. Anything outside that subset falls back: other pattern letters, unsupported run-lengths, a non-literal format argument, the 0-argument form, and a numeric field adjacent to another numeric field where the run-length would not survive translation (for exampleyyyyMd). Falling back keeps results correct rather than risking a silent divergence.The session time zone is resolved at plan time and passed into the node. This required completing the config threading in the standalone-Calc path, which passed a persisted config that does not carry
table.local-time-zone. That gap had no effect until now:UNIX_TIMESTAMPis the first time-zone-sensitive expression the converter supports, and the earlier ones (arithmetic, comparison, logical, cast) never read the session zone. The effective node config is threaded through instead, so the configured zone reaches the native evaluation.Are there any user-facing changes?
Yes.
UNIX_TIMESTAMP(string)andUNIX_TIMESTAMP(string, format)now execute on the native engine when the format is a supported literal pattern. Unsupported patterns and the 0-argument form continue to run on Flink's engine, with the same results as before.How was this patch tested?
Unit tests for the scanner (accept/reject, quote escaping, the adjacency rule), the converter (node shape, format translation, time-zone propagation), and the operator-identity invariant. Fallback tests assert that unsupported inputs actually fall back rather than silently producing a native plan.
An end-to-end ITCase runs
UNIX_TIMESTAMP(ts)with a non-UTC session zone and confirms the native result matches the expected epoch values. The executed native plan shows the function and the resolved zone, confirming the query runs natively rather than falling back.160 tests pass in
auron-flink-planneron the rebased branch, with spotless clean and 0 checkstyle violations.Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 5)