weiqingy commented on code in PR #2409: URL: https://github.com/apache/auron/pull/2409#discussion_r3661552790
########## native-engine/datafusion-ext-functions/src/flink_datetime.rs: ########## @@ -0,0 +1,648 @@ +// Licensed to the Apache Software Foundation (ASF) under one or more +// contributor license agreements. See the NOTICE file distributed with +// this work for additional information regarding copyright ownership. +// The ASF licenses this file to You under the Apache License, Version 2.0 +// (the "License"); you may not use this file except in compliance with +// the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use std::sync::Arc; + +use arrow::{ + array::{Int64Array, StringArray}, + datatypes::DataType, +}; +use chrono::{DateTime, LocalResult, NaiveDateTime, Offset, TimeZone}; +use chrono_tz::{OffsetComponents, Tz}; +use datafusion::{ + common::{DataFusionError, Result, ScalarValue}, + physical_plan::ColumnarValue, +}; +use datafusion_ext_commons::arrow::cast::cast; + +/// Native implementation of Flink SQL `UNIX_TIMESTAMP(value, format)`: parse a +/// formatted date-time string to a Unix timestamp in seconds, replicating +/// `java.text.SimpleDateFormat` lenient semantics. +/// +/// Arguments are always `[value, chronoFormat, zoneId]` (arity 3). `value` is +/// the column of strings to parse; `chronoFormat` and `zoneId` are literal +/// scalars bound at plan time. The format is a `%`-specifier string built by +/// the JVM-side converter from `%Y %m %d %H %M %S` plus literal characters +/// (`%%` for a literal percent) — it is not the original Java pattern. +/// +/// Arity 3 is the normalized form of Flink's two string-parsing arities: the +/// JVM-side converter supplies Flink's default `yyyy-MM-dd HH:mm:ss` for the +/// single-argument call, translates the literal pattern for the two-argument +/// call, and resolves the session time zone at plan time. A non-literal or +/// untranslatable format is rejected there and evaluated by Flink instead, so +/// every call reaching this function carries all three arguments already bound. +/// +/// Flink's no-argument `UNIX_TIMESTAMP()` parses nothing and yields the current +/// wall-clock time per record. It is not routed here: it has no input column to +/// size the output against, and it shares none of the parsing contract below. +/// +/// An unparseable value yields `i64::MIN` (Flink's `Long.MIN_VALUE`), never +/// NULL and never an error; a NULL value yields NULL. Arity, format and +/// timezone problems are hard errors rather than silent defaults, so a plumbing +/// bug cannot surface as silently wrong data. +pub fn flink_unix_timestamp(args: &[ColumnarValue]) -> Result<ColumnarValue> { + if args.len() != 3 { + return Err(DataFusionError::Execution(format!( + "Flink_UnixTimestamp requires 3 arguments [value, chronoFormat, zoneId], got {}", + args.len() + ))); + } + + let format = utf8_scalar(&args[1]).ok_or_else(|| { + DataFusionError::Execution("Flink_UnixTimestamp: format must be a non-null string".into()) + })?; + 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 tokens = parse_format(&format)?; + + let num_rows = match &args[0] { + ColumnarValue::Array(array) => array.len(), + ColumnarValue::Scalar(_) => 1, + }; + let value = cast(&args[0].clone().into_array(num_rows)?, &DataType::Utf8)?; + let value = value + .as_any() + .downcast_ref::<StringArray>() + .expect("internal cast to Utf8 must succeed"); + + let result = Int64Array::from_iter( + value + .iter() + .map(|opt_s| opt_s.map(|s| parse_datetime(s, &tokens, tz).unwrap_or(i64::MIN))), + ); + + Ok(ColumnarValue::Array(Arc::new(result))) +} + +fn utf8_scalar(arg: &ColumnarValue) -> Option<String> { + match arg { + ColumnarValue::Scalar(ScalarValue::Utf8(Some(s))) + | ColumnarValue::Scalar(ScalarValue::LargeUtf8(Some(s))) => Some(s.clone()), + _ => None, + } +} + +/// 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. +#[derive(Clone, Copy)] +enum FieldKind { + Year, + Month, + Day, + Hour, + Minute, + Second, +} + +impl FieldKind { + /// The `obeyCount` window width. Run length is erased by the Java→chrono + /// translation (both `M` and `MM` become `%m`), so a canonical width per + /// field is used: 4 for the year, 2 for the rest — matching the zero-padded + /// widths of Flink's default `yyyy-MM-dd HH:mm:ss`. + fn width(self) -> usize { + match self { + FieldKind::Year => 4, + _ => 2, + } + } +} + +enum Token { + Field(FieldKind), + Literal(u8), +} + +/// Tokenize the `%`-specifier format. Errors on an unknown specifier or a +/// dangling `%`, since the format is a plan-time constant and such a format is +/// a wiring bug, not bad row data. +fn parse_format(format: &str) -> Result<Vec<Token>> { + let bytes = format.as_bytes(); + let mut tokens = Vec::new(); + let mut i = 0; Review Comment: Fixed in `7cd8c8cf`, though as a per-row `Long.MIN_VALUE` rather than a hard error. It's reachable, which makes it worse than a doc-contract inconsistency. `FlinkDateTimeFormatConverter.scan("")` returns an empty token list rather than null, and the adjacency check passes vacuously, so `translate("")` returns `Optional.of("")`. A user writing `UNIX_TIMESTAMP(ts, '')` gets the native path with an empty format and silently reads back 1970-01-01. On rejecting it up front: I went the other way, because Flink doesn't fail the query there. Running Flink's own code: ``` DateTimeUtils.unixTimestamp("2020-10-10 00:00:01", "", tz) = -9223372036854775808 DateTimeUtils.unixTimestamp("", "", tz) = -9223372036854775808 ``` It returns `Long.MIN_VALUE` for all 30 (input, timezone) pairs I checked. So a hard error would abort a query that runs fine on Flink, trading a wrong value for a crash. The "format problems are hard errors" line rests on the format being a converter-supplied constant, where a bad one is a plumbing bug. An empty format isn't a plumbing bug, it's legal SQL, so that contract doesn't cover this case. Your `SimpleDateFormat` reasoning is what pointed at the right fix, so I anchored the guard on it. "Fails if it consumes 0 chars" is `DateFormat.parse`'s `pos.index == 0` rule, so the condition is a zero-consumption parse rather than an empty format string. Those coincide here: a literal consumes one byte and a field needs at least one digit, so an empty token list is the only way to finish having consumed nothing. Keeping it phrased that way means the guard doesn't over-reject the neighbouring case. A format of `%%` against input `%` still yields 0, matching `unixTimestamp("%", "%", UTC) = 0`. Covered by `empty_format_never_matches`, which I confirmed fails without the guard (returns 0 instead of `MIN`). -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
