[AURON #1863] Support the zero-argument Flink UNIX_TIMESTAMP natively - #2465
Conversation
…ction Flink's UNIX_TIMESTAMP() with no arguments returns the current epoch seconds. Add Flink_UnixTimestampNow, a zero-argument ext function that returns it. The function takes no operand. A ColumnarValue::Scalar return bypasses the row-count check in ScalarFunctionExpr::evaluate, which applies only to an Array return, and the projection broadcasts the scalar to the batch width. So no argument is needed to size the output. The clock is read as timestamp_millis() / 1000, mirroring Flink's System.currentTimeMillis() / 1000 operator for operator. The Java converter still rejects the zero-argument form, so this arm is dormant until that gate opens, in the same way the arity-3 function shipped ahead of its converter.
…tively Admit UNIX_TIMESTAMP() in the converter and emit the native Flink_UnixTimestampNow ext function with no operands. The zero-argument arm sits above the session time zone check. Its result is epoch seconds, which is zone-independent, so a session zone the native side cannot resolve is no reason to fall back a query that never consults a zone. The class and method javadoc described the zero-argument form as falling back and described UNIX_TIMESTAMP as always mapping to Flink_UnixTimestamp with [value, chronoFormat, zoneId]. Both are now scoped to the string-parsing forms, with the wall-clock form named alongside them. The clock is read once per evaluation and broadcast across the batch, whereas Flink evaluates its niladic form per record.
Assert that SELECT UNIX_TIMESTAMP() returns one clock-bracketed row per input row and that the Calc records no fallback. Both assertions are load-bearing and neither subsumes the other. A zero fallback count establishes that the Calc converted, not that it ran: a native library holding no registry arm for the function still converts at plan time and fails only during execution, where nothing records a fallback. The row count is what establishes that the native plan executed, since a vacuous pass over an empty result set would otherwise read as success.
There was a problem hiding this comment.
Pull request overview
Adds native support for Flink SQL UNIX_TIMESTAMP() (zero-argument form) by introducing a dedicated ext-function that returns epoch seconds as a scalar (broadcast per batch) and updating the Flink Calc converter to emit it, with unit/integration coverage to ensure the Calc both converts and executes natively.
Changes:
- Added and registered a new native ext-function
Flink_UnixTimestampNow(0 args) returningColumnarValue::Scalar(Int64)epoch seconds. - Updated
RexCallConverterto accept the 0-argUNIX_TIMESTAMP()and lower it toFlink_UnixTimestampNowwithout requiring a session time zone. - Expanded planner/unit tests and IT coverage to validate node shape, gating behavior, and native execution (non-fallback + non-empty results).
Reviewed changes
Copilot reviewed 6 out of 6 changed files in this pull request and generated 2 comments.
Show a summary per file
| File | Description |
|---|---|
| native-engine/datafusion-ext-functions/src/lib.rs | Registers the new Flink_UnixTimestampNow ext-function in the native function factory. |
| native-engine/datafusion-ext-functions/src/flink_datetime.rs | Implements flink_unix_timestamp_now and adds unit tests for scalar broadcasting and seconds-level output. |
| auron-flink-extension/auron-flink-planner/src/main/java/org/apache/auron/flink/table/planner/converter/RexCallConverter.java | Lowers 0-arg UNIX_TIMESTAMP() to Flink_UnixTimestampNow and adjusts gating/docs around timezone requirements. |
| auron-flink-extension/auron-flink-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecCalcTest.java | Updates the converter contract test to assert native conversion (no fallback) for 0-arg UNIX_TIMESTAMP(). |
| auron-flink-extension/auron-flink-planner/src/test/java/org/apache/auron/flink/table/runtime/AuronFlinkCalcITCase.java | Adds an end-to-end test that asserts no fallback and verifies results are non-vacuous and clock-bracketed. |
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
| /// The clock is read as `timestamp_millis() / 1000` to mirror Flink's | ||
| /// `System.currentTimeMillis() / 1000` operator for operator, so the two | ||
| /// implementations can be compared side by side. |
There was a problem hiding this comment.
Not a duplicated word. operator for operator was meant the way word for word is, that the arithmetic mirrors Flink's term by term. It does read as a typo though, and the sentence says the same thing without it, so I dropped the phrasing in 4b115097:
/// The clock is read as `timestamp_millis() / 1000` so the truncation matches
/// Flink's `System.currentTimeMillis() / 1000` exactly.
| pub fn flink_unix_timestamp_now(_args: &[ColumnarValue]) -> Result<ColumnarValue> { | ||
| let secs = Utc::now().timestamp_millis() / 1000; | ||
| Ok(ColumnarValue::Scalar(ScalarValue::Int64(Some(secs)))) | ||
| } |
There was a problem hiding this comment.
Added in 4b115097. Nothing rejects a wrong-arity call before this point: planner.rs:1028 builds each ext function's UDF signature from the argument list the plan node already carries, so DataFusion never checks arity itself. The in-body check is the only gate, which is why flink_unix_timestamp checks its own at line 56.
One qualifier on the reasoning. Extra operands can't make the result wrong here, since the value is a clock read and does not depend on the arguments. So this buys early detection of a converter or plumbing mistake, not correctness. arity_mismatch_errors now covers the zero-argument form too.
Flink_UnixTimestampNow ignored any operands it was handed. Nothing rejects a wrong-arity call before it: the planner builds each ext function's UDF signature from the argument list the plan node already carries, so DataFusion never checks arity, and the in-body check is the only gate. That is the same reason Flink_UnixTimestamp checks its own arity. Extra operands cannot make the result wrong here, since the value is a clock read, so this buys early detection of a converter or plumbing mistake rather than correctness. Also drop the "operator for operator" phrasing from the clock doc, which read as a duplicated word.
|
Hi @Tartarus0zm, could you please help review this PR when you get a chance? Thanks! |
Tartarus0zm
left a comment
Tartarus0zm
left a comment
There was a problem hiding this comment.
@weiqingy thanks for your contribution! LGTM
Which issue does this PR close?
Part of #1863.
#1863 enumerates all three argument forms of
UNIX_TIMESTAMP. The 1-argument and 2-argument string-parsing forms landed in #2409 and #2448, and this PR adds the remaining 0-argument form. #1863 stays open after this one merges, for the timezone configuration work tracked as sub-issue #2455.Rationale for this change
UNIX_TIMESTAMP()with no arguments returns the current epoch seconds. The converter rejected it, so any Calc containing it fell back to Flink wholesale, taking every other expression in that Calc down with it.Two things were believed to block it. Both turned out to be false, and I stated one of them myself earlier in this issue.
"A zero-argument ext function cannot size its output." The ext-function ABI drops the batch row count, so a 0-argument function appears to have no way to know how many values to return. That is true only for a function returning an
Array. The length check inScalarFunctionExpr::evaluatesits entirely inside theColumnarValue::Arraybranch, so a function returningColumnarValue::Scalarnever reaches it, and the projection broadcasts the scalar to the batch width. No operand is needed, and no carrier column either."The 0-argument form needs the session time zone." It does not. The generated Flink operator holds a
timeZonefield, passes it to the 1-argument and 2-argument calls, and emitsresult$1 = DateTimeUtils.unixTimestamp()with no argument at all. The result is epoch seconds, which is zone-independent.An approach was proposed on #1863 that works around the first premise: pass any input column to the Rust function purely to convey the batch length, ignore its data, and return that many timestamps. That would have worked, and it was a sound answer to the constraint as I had described it. This PR does not use it, because once the constraint turns out not to exist the carrier column has no job left. The difference is visible in the diff, so it is worth stating plainly: no column is passed, the Rust function reads no length, and the row count comes from the projection broadcasting a scalar. The observable behavior is the same either way.
What changes are included in this PR?
Three commits.
Flink_UnixTimestampNowreturningColumnarValue::Scalar(Int64). The clock is read astimestamp_millis() / 1000, mirroring Flink'sSystem.currentTimeMillis() / 1000operator for operator.UNIX_TIMESTAMPas always mapping toFlink_UnixTimestampwith[value, chronoFormat, zoneId].AuronFlinkCalcITCase.No protobuf change, no
Cargo.tomlchange, nopom.xmlchange. It is a separate native function rather than a fourth arity onFlink_UnixTimestamp, whose contract is "parse this string with this format in this zone" and which the 0-argument form shares none of.Are there any user-facing changes?
Yes, and one of them is a deliberate semantic divergence worth reviewing on its own merits.
A query using
UNIX_TIMESTAMP()that previously fell back to Flink now runs natively.Flink evaluates the niladic form per record. This implementation reads the clock once per call site per evaluation and broadcasts it across the batch, so rows in one batch share a timestamp, and that timestamp is taken when the batch is evaluated rather than when each row arrived. Two
UNIX_TIMESTAMP()calls in one projection are read independently, which matches Flink.How large the skew can get depends on what closes the batch:
WATERMARK FORautoWatermarkInterval8192 / rateseconds. About 82 s at 100 rec/sThe unwatermarked case is the common shape for a processing-time query, so this is not a corner case. It was discussed and settled on the issue (#1863 (comment)): per-batch is acceptable, since the zero-argument form is non-idempotent anyway and what users generally want is to know roughly when a record was processed. It is also the finest granularity among comparable engines: StreamFusion stamps
PROCTIME()once per operator compile, Flink's own batch mode foldsCURRENT_TIMESTAMPto a query-start literal, DataFusion, ClickHouse and Velox are per query, DuckDB is per transaction, and ClickHouse shipsnowInBlock()as a documented per-block clock. Flink also declares the niladic form onlySqlMonotonicity.INCREASING, which a per-batch constant satisfies.Gating it conditionally on a declared watermark was considered and rejected: the converter has no access to the ExecNode graph, the predicate misses DataStream-level watermarks,
auto-watermark-interval=0and idle partitions, it keys native support off an unrelated DDL clause, and it only shrinks the skew rather than removing it.How was this patch tested?
Native unit tests for the scalar return and its broadcast to batch width, and for the value being a plausible epoch second rather than milliseconds.
Converter tests for the emitted node shape, and
testUnixTimestampZeroArgSupportedWithFixedOffsetZone, which exists to catch the 0-argument arm being placed below the time zone check. It is a real discriminator: moving the arm below the check fails that test and no other. Two tests that asserted the form falls back are inverted.An ITCase asserting one clock-bracketed row per input row and no recorded fallback. Both assertions are load-bearing and neither subsumes the other. A zero fallback count establishes that the Calc converted, not that it ran: a native library holding no registry arm for the function still converts at plan time and fails only during execution, where nothing records a fallback, leaving the counter at zero and the result set empty. The row count is what establishes the native plan executed.
The ITCase was checked to be non-vacuous by running rather than by inspection: reverting the converter gate fails it on the fallback count, and running against a library without the registry arm fails it on the row count.
Full module build green, 0 checkstyle violations, spotless clean.
Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 5)