Copilot commented on code in PR #2409: URL: https://github.com/apache/auron/pull/2409#discussion_r3625166899
########## native-engine/datafusion-ext-functions/src/flink_datetime.rs: ########## @@ -0,0 +1,637 @@ +// 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. +/// +/// 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; + while i < bytes.len() { + if bytes[i] == b'%' { + let spec = bytes.get(i + 1).ok_or_else(|| { + DataFusionError::Execution("Flink_UnixTimestamp: dangling '%' in format".into()) + })?; + let token = match spec { + b'Y' => Token::Field(FieldKind::Year), + b'm' => Token::Field(FieldKind::Month), + b'd' => Token::Field(FieldKind::Day), + b'H' => Token::Field(FieldKind::Hour), + b'M' => Token::Field(FieldKind::Minute), + b'S' => Token::Field(FieldKind::Second), + b'%' => Token::Literal(b'%'), + other => { + return Err(DataFusionError::Execution(format!( + "Flink_UnixTimestamp: unsupported format specifier %{}", + *other as char + ))); + } + }; + tokens.push(token); + i += 2; + } else { + tokens.push(Token::Literal(bytes[i])); + i += 1; + } + } + Ok(tokens) +} + +struct Fields { + year: i32, + month: i32, + day: i32, + hour: i32, + minute: i32, + second: i32, +} + +impl Default for Fields { + fn default() -> Self { + Fields { + year: 1970, + month: 1, + day: 1, + hour: 0, + minute: 0, + second: 0, + } + } +} + +impl Fields { + fn set(&mut self, kind: FieldKind, value: i32) { + match kind { + FieldKind::Year => self.year = value, + FieldKind::Month => self.month = value, + FieldKind::Day => self.day = value, + FieldKind::Hour => self.hour = value, + FieldKind::Minute => self.minute = value, + FieldKind::Second => self.second = value, + } + } +} + +/// Walk the tokens over `input`. Returns the Unix timestamp in seconds, or +/// `None` on any parse failure (mapped to `i64::MIN` by the caller). +fn parse_datetime(input: &str, tokens: &[Token], tz: Tz) -> Option<i64> { + let bytes = input.as_bytes(); + let mut pos = 0usize; + let mut fields = Fields::default(); + + for (idx, token) in tokens.iter().enumerate() { + match token { + Token::Literal(lit) => { + if bytes.get(pos) != Some(lit) { + return None; + } + pos += 1; + } + Token::Field(kind) => { + // Skip ' ' and '\t' before the field, but anchor the obeyCount + // window at the pre-skip position so leading whitespace eats into + // the field's digit budget (a genuine SimpleDateFormat quirk). + let start0 = pos; + while matches!(bytes.get(pos), Some(b' ') | Some(b'\t')) { + pos += 1; + } + if pos >= bytes.len() { + return None; + } + + // obeyCount holds when the next token is another numeric field; + // then the scan is bounded to `width` chars, otherwise it is greedy. + let obey_count = matches!(tokens.get(idx + 1), Some(Token::Field(_))); + let window_end = if obey_count { + let end = start0 + kind.width(); + if end > bytes.len() { + return None; + } + end + } else { + bytes.len() + }; + + let (value, next) = scan_number(bytes, pos, window_end)?; + fields.set(*kind, value); + pos = next; + } + } + } + + let local_sec = normalize(&fields); + let offset = resolve_offset_secs(local_sec, tz); + Some(local_sec - offset) +} + +/// Scan an optional leading `-` then one or more ASCII digits within `[start, +/// end)`. A `+` is not a sign and terminates the scan before any digit. Digits +/// accumulate with wrapping and narrow to `i32` via low-32-bit truncation, +/// mirroring Java's `Number.intValue()`. Returns `None` when no digit is +/// consumed. +fn scan_number(bytes: &[u8], start: usize, end: usize) -> Option<(i32, usize)> { + let mut pos = start; + let negative = bytes.get(pos) == Some(&b'-'); + if negative { + pos += 1; + } + + let digits_start = pos; + let mut magnitude: i64 = 0; + while pos < end { + let b = bytes[pos]; + if !b.is_ascii_digit() { + break; + } + magnitude = magnitude.wrapping_mul(10).wrapping_add((b - b'0') as i64); + pos += 1; + } + if pos == digits_start { + return None; + } + + let signed = if negative { + magnitude.wrapping_neg() + } else { + magnitude + }; + Some((signed as i32, pos)) +} + +/// Normalize the (possibly out-of-range, possibly negative) fields into a local +/// wall-clock time in seconds since the Unix epoch. Rollover falls out of a +/// single month-then-day computation, so hour/minute/second overflow needs no +/// special case. Dates before the 1582-10-15 Gregorian cutover use the Julian +/// calendar, matching `GregorianCalendar`'s hybrid behavior (chrono is +/// proleptic Gregorian). +fn normalize(fields: &Fields) -> i64 { + let year = fields.year as i64; + let month = fields.month as i64; + let day = fields.day as i64; + let hour = fields.hour as i64; + let minute = fields.minute as i64; + let second = fields.second as i64; + + let year_adj = year + (month - 1).div_euclid(12); + let month_idx = (month - 1).rem_euclid(12) + 1; + + let jdn_gregorian = gregorian_jdn(year_adj, month_idx) + (day - 1); + let jdn = if jdn_gregorian >= 2_299_161 { + jdn_gregorian + } else { + julian_jdn(year_adj, month_idx) + (day - 1) + }; + + let epoch_day = jdn - 2_440_588; + epoch_day * 86_400 + hour * 3_600 + minute * 60 + second +} + +/// Julian Day Number of the first of month `(year, month)` in the proleptic +/// Gregorian calendar. +fn gregorian_jdn(year: i64, month: i64) -> i64 { + let a = (14 - month).div_euclid(12); + let y = year + 4800 - a; + let m = month + 12 * a - 3; + 1 + (153 * m + 2).div_euclid(5) + 365 * y + y.div_euclid(4) - y.div_euclid(100) + + y.div_euclid(400) + - 32045 +} + +/// Julian Day Number of the first of month `(year, month)` in the Julian +/// calendar. +fn julian_jdn(year: i64, month: i64) -> i64 { + let a = (14 - month).div_euclid(12); + let y = year + 4800 - a; + let m = month + 12 * a - 3; + 1 + (153 * m + 2).div_euclid(5) + 365 * y + y.div_euclid(4) - 32083 +} + +/// UTC offset in seconds for a local wall-clock instant. When the local time is +/// ambiguous (fall-back overlap) or nonexistent (spring-forward gap), the +/// zone's standard (non-DST) offset is used, matching Flink. +/// Out-of-representable-range inputs (only reachable from far-future garbage) +/// fall back to UTC. +fn resolve_offset_secs(local_sec: i64, tz: Tz) -> i64 { + let naive: NaiveDateTime = match DateTime::from_timestamp(local_sec, 0) { + Some(dt) => dt.naive_utc(), + None => return 0, + }; Review Comment: `DateTime::from_timestamp(local_sec, 0)` is not callable on the generic `chrono::DateTime` type (it’s an associated function on `DateTime<Utc>`). This will fail to compile with chrono 0.4.45. Use `NaiveDateTime::from_timestamp_opt` directly (or specify `DateTime::<Utc>`). -- 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]
