| // 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 crate::planner::{ContextProvider, PlannerContext, SqlToRel}; |
| use arrow::compute::kernels::cast_utils::{ |
| parse_interval_month_day_nano_config, IntervalParseConfig, IntervalUnit, |
| }; |
| use arrow::datatypes::{ |
| i256, FieldRef, DECIMAL128_MAX_PRECISION, DECIMAL256_MAX_PRECISION, |
| }; |
| use bigdecimal::num_bigint::BigInt; |
| use bigdecimal::{BigDecimal, Signed, ToPrimitive}; |
| use datafusion_common::{ |
| internal_datafusion_err, not_impl_err, plan_err, DFSchema, DataFusionError, Result, |
| ScalarValue, |
| }; |
| use datafusion_expr::expr::{BinaryExpr, Placeholder}; |
| use datafusion_expr::planner::PlannerResult; |
| use datafusion_expr::{lit, Expr, Operator}; |
| use log::debug; |
| use sqlparser::ast::{ |
| BinaryOperator, Expr as SQLExpr, Interval, UnaryOperator, Value, ValueWithSpan, |
| }; |
| use sqlparser::parser::ParserError::ParserError; |
| use std::borrow::Cow; |
| use std::cmp::Ordering; |
| use std::ops::Neg; |
| use std::str::FromStr; |
| |
| impl<S: ContextProvider> SqlToRel<'_, S> { |
| pub(crate) fn parse_value( |
| &self, |
| value: Value, |
| param_data_types: &[FieldRef], |
| ) -> Result<Expr> { |
| match value { |
| Value::Number(n, _) => self.parse_sql_number(&n, false), |
| Value::SingleQuotedString(s) | Value::DoubleQuotedString(s) => Ok(lit(s)), |
| Value::Null => Ok(Expr::Literal(ScalarValue::Null, None)), |
| Value::Boolean(n) => Ok(lit(n)), |
| Value::Placeholder(param) => { |
| Self::create_placeholder_expr(param, param_data_types) |
| } |
| Value::HexStringLiteral(s) => { |
| if let Some(v) = try_decode_hex_literal(&s) { |
| Ok(lit(v)) |
| } else { |
| plan_err!("Invalid HexStringLiteral '{s}'") |
| } |
| } |
| Value::DollarQuotedString(s) => Ok(lit(s.value)), |
| Value::EscapedStringLiteral(s) => Ok(lit(s)), |
| _ => plan_err!("Unsupported Value '{value:?}'"), |
| } |
| } |
| |
| /// Parse number in sql string, convert to Expr::Literal |
| pub(super) fn parse_sql_number( |
| &self, |
| unsigned_number: &str, |
| negative: bool, |
| ) -> Result<Expr> { |
| let signed_number: Cow<str> = if negative { |
| Cow::Owned(format!("-{unsigned_number}")) |
| } else { |
| Cow::Borrowed(unsigned_number) |
| }; |
| |
| // Try to parse as i64 first, then u64 if negative is false, then decimal or f64 |
| |
| if let Ok(n) = signed_number.parse::<i64>() { |
| return Ok(lit(n)); |
| } |
| |
| if !negative { |
| if let Ok(n) = unsigned_number.parse::<u64>() { |
| return Ok(lit(n)); |
| } |
| } |
| |
| if self.options.parse_float_as_decimal { |
| parse_decimal(unsigned_number, negative) |
| } else { |
| signed_number.parse::<f64>().map(lit).map_err(|_| { |
| DataFusionError::from(ParserError(format!( |
| "Cannot parse {signed_number} as f64" |
| ))) |
| }) |
| } |
| } |
| |
| /// Create a placeholder expression |
| /// This is the same as Postgres's prepare statement syntax in which a placeholder starts with `$` sign and then |
| /// number 1, 2, ... etc. For example, `$1` is the first placeholder; $2 is the second one and so on. |
| fn create_placeholder_expr( |
| param: String, |
| param_data_types: &[FieldRef], |
| ) -> Result<Expr> { |
| // Parse the placeholder as a number because it is the only support from sqlparser and postgres |
| let index = param[1..].parse::<usize>(); |
| let idx = match index { |
| Ok(0) => { |
| return plan_err!( |
| "Invalid placeholder, zero is not a valid index: {param}" |
| ); |
| } |
| Ok(index) => index - 1, |
| Err(_) => { |
| return if param_data_types.is_empty() { |
| Ok(Expr::Placeholder(Placeholder::new_with_field(param, None))) |
| } else { |
| // when PREPARE Statement, param_data_types length is always 0 |
| plan_err!("Invalid placeholder, not a number: {param}") |
| }; |
| } |
| }; |
| // Check if the placeholder is in the parameter list |
| let param_type = param_data_types.get(idx); |
| // Data type of the parameter |
| debug!("type of param {param} param_data_types[idx]: {param_type:?}"); |
| |
| Ok(Expr::Placeholder(Placeholder::new_with_field( |
| param, |
| param_type.cloned(), |
| ))) |
| } |
| |
| pub(super) fn sql_array_literal( |
| &self, |
| elements: Vec<SQLExpr>, |
| schema: &DFSchema, |
| ) -> Result<Expr> { |
| let values = elements |
| .into_iter() |
| .map(|element| { |
| self.sql_expr_to_logical_expr(element, schema, &mut PlannerContext::new()) |
| }) |
| .collect::<Result<Vec<_>>>()?; |
| |
| self.try_plan_array_literal(values, schema) |
| } |
| |
| fn try_plan_array_literal( |
| &self, |
| values: Vec<Expr>, |
| schema: &DFSchema, |
| ) -> Result<Expr> { |
| let mut exprs = values; |
| for planner in self.context_provider.get_expr_planners() { |
| match planner.plan_array_literal(exprs, schema)? { |
| PlannerResult::Planned(expr) => { |
| return Ok(expr); |
| } |
| PlannerResult::Original(values) => exprs = values, |
| } |
| } |
| |
| not_impl_err!("Could not plan array literal. Hint: Please try with `nested_expressions` DataFusion feature enabled") |
| } |
| |
| /// Convert a SQL interval expression to a DataFusion logical plan |
| /// expression |
| #[allow(clippy::only_used_in_recursion)] |
| pub(super) fn sql_interval_to_expr( |
| &self, |
| negative: bool, |
| interval: Interval, |
| ) -> Result<Expr> { |
| if interval.leading_precision.is_some() { |
| return not_impl_err!( |
| "Unsupported Interval Expression with leading_precision {:?}", |
| interval.leading_precision |
| ); |
| } |
| |
| if interval.last_field.is_some() { |
| return not_impl_err!( |
| "Unsupported Interval Expression with last_field {:?}", |
| interval.last_field |
| ); |
| } |
| |
| if interval.fractional_seconds_precision.is_some() { |
| return not_impl_err!( |
| "Unsupported Interval Expression with fractional_seconds_precision {:?}", |
| interval.fractional_seconds_precision |
| ); |
| } |
| |
| if let SQLExpr::BinaryOp { left, op, right } = *interval.value { |
| let df_op = match op { |
| BinaryOperator::Plus => Operator::Plus, |
| BinaryOperator::Minus => Operator::Minus, |
| _ => { |
| return not_impl_err!("Unsupported interval operator: {op:?}"); |
| } |
| }; |
| let left_expr = self.sql_interval_to_expr( |
| negative, |
| Interval { |
| value: left, |
| leading_field: interval.leading_field.clone(), |
| leading_precision: None, |
| last_field: None, |
| fractional_seconds_precision: None, |
| }, |
| )?; |
| let right_expr = self.sql_interval_to_expr( |
| false, |
| Interval { |
| value: right, |
| leading_field: interval.leading_field, |
| leading_precision: None, |
| last_field: None, |
| fractional_seconds_precision: None, |
| }, |
| )?; |
| return Ok(Expr::BinaryExpr(BinaryExpr::new( |
| Box::new(left_expr), |
| df_op, |
| Box::new(right_expr), |
| ))); |
| } |
| |
| let value = interval_literal(*interval.value, negative)?; |
| |
| // leading_field really means the unit if specified |
| // For example, "month" in `INTERVAL '5' month` |
| let value = match interval.leading_field.as_ref() { |
| Some(leading_field) => format!("{value} {leading_field}"), |
| None => value, |
| }; |
| |
| let config = IntervalParseConfig::new(IntervalUnit::Second); |
| let val = parse_interval_month_day_nano_config(&value, config)?; |
| Ok(lit(ScalarValue::IntervalMonthDayNano(Some(val)))) |
| } |
| } |
| |
| fn interval_literal(interval_value: SQLExpr, negative: bool) -> Result<String> { |
| let s = match interval_value { |
| SQLExpr::Value(ValueWithSpan { |
| value: Value::SingleQuotedString(s) | Value::DoubleQuotedString(s), |
| span: _, |
| }) => s, |
| SQLExpr::Value(ValueWithSpan { |
| value: Value::Number(ref v, long), |
| span: _, |
| }) => { |
| if long { |
| return not_impl_err!( |
| "Unsupported interval argument. Long number not supported: {interval_value:?}" |
| ); |
| } else { |
| v.to_string() |
| } |
| } |
| SQLExpr::UnaryOp { op, expr } => { |
| let negative = match op { |
| UnaryOperator::Minus => !negative, |
| UnaryOperator::Plus => negative, |
| _ => { |
| return not_impl_err!( |
| "Unsupported SQL unary operator in interval {op:?}" |
| ); |
| } |
| }; |
| interval_literal(*expr, negative)? |
| } |
| _ => { |
| return not_impl_err!("Unsupported interval argument. Expected string literal or number, got: {interval_value:?}"); |
| } |
| }; |
| if negative { |
| Ok(format!("-{s}")) |
| } else { |
| Ok(s) |
| } |
| } |
| |
| /// Try to decode bytes from hex literal string. |
| /// |
| /// None will be returned if the input literal is hex-invalid. |
| fn try_decode_hex_literal(s: &str) -> Option<Vec<u8>> { |
| let hex_bytes = s.as_bytes(); |
| |
| let mut decoded_bytes = Vec::with_capacity(hex_bytes.len().div_ceil(2)); |
| |
| let start_idx = hex_bytes.len() % 2; |
| if start_idx > 0 { |
| // The first byte is formed of only one char. |
| decoded_bytes.push(try_decode_hex_char(hex_bytes[0])?); |
| } |
| |
| for i in (start_idx..hex_bytes.len()).step_by(2) { |
| let high = try_decode_hex_char(hex_bytes[i])?; |
| let low = try_decode_hex_char(hex_bytes[i + 1])?; |
| decoded_bytes.push((high << 4) | low); |
| } |
| |
| Some(decoded_bytes) |
| } |
| |
| /// Try to decode a byte from a hex char. |
| /// |
| /// None will be returned if the input char is hex-invalid. |
| const fn try_decode_hex_char(c: u8) -> Option<u8> { |
| match c { |
| b'A'..=b'F' => Some(c - b'A' + 10), |
| b'a'..=b'f' => Some(c - b'a' + 10), |
| b'0'..=b'9' => Some(c - b'0'), |
| _ => None, |
| } |
| } |
| |
| /// Returns None if the value can't be converted to i256. |
| /// Modified from <https://github.com/apache/arrow-rs/blob/c4dbf0d8af6ca5a19b8b2ea777da3c276807fc5e/arrow-buffer/src/bigint/mod.rs#L303> |
| fn bigint_to_i256(v: &BigInt) -> Option<i256> { |
| let v_bytes = v.to_signed_bytes_le(); |
| match v_bytes.len().cmp(&32) { |
| Ordering::Less => { |
| let mut bytes = if v.is_negative() { |
| [255_u8; 32] |
| } else { |
| [0; 32] |
| }; |
| bytes[0..v_bytes.len()].copy_from_slice(&v_bytes[..v_bytes.len()]); |
| Some(i256::from_le_bytes(bytes)) |
| } |
| Ordering::Equal => Some(i256::from_le_bytes(v_bytes.try_into().unwrap())), |
| Ordering::Greater => None, |
| } |
| } |
| |
| fn parse_decimal(unsigned_number: &str, negative: bool) -> Result<Expr> { |
| let mut dec = BigDecimal::from_str(unsigned_number).map_err(|e| { |
| DataFusionError::from(ParserError(format!( |
| "Cannot parse {unsigned_number} as BigDecimal: {e}" |
| ))) |
| })?; |
| if negative { |
| dec = dec.neg(); |
| } |
| |
| let digits = dec.digits(); |
| let (int_val, scale) = dec.into_bigint_and_exponent(); |
| if scale < i8::MIN as i64 { |
| return not_impl_err!( |
| "Decimal scale {} exceeds the minimum supported scale: {}", |
| scale, |
| i8::MIN |
| ); |
| } |
| let precision = if scale > 0 { |
| // arrow-rs requires the precision to include the positive scale. |
| // See <https://github.com/apache/arrow-rs/blob/123045cc766d42d1eb06ee8bb3f09e39ea995ddc/arrow-array/src/types.rs#L1230> |
| std::cmp::max(digits, scale.unsigned_abs()) |
| } else { |
| digits |
| }; |
| if precision <= DECIMAL128_MAX_PRECISION as u64 { |
| let val = int_val.to_i128().ok_or_else(|| { |
| // Failures are unexpected here as we have already checked the precision |
| internal_datafusion_err!( |
| "Unexpected overflow when converting {} to i128", |
| int_val |
| ) |
| })?; |
| Ok(Expr::Literal( |
| ScalarValue::Decimal128(Some(val), precision as u8, scale as i8), |
| None, |
| )) |
| } else if precision <= DECIMAL256_MAX_PRECISION as u64 { |
| let val = bigint_to_i256(&int_val).ok_or_else(|| { |
| // Failures are unexpected here as we have already checked the precision |
| internal_datafusion_err!( |
| "Unexpected overflow when converting {} to i256", |
| int_val |
| ) |
| })?; |
| Ok(Expr::Literal( |
| ScalarValue::Decimal256(Some(val), precision as u8, scale as i8), |
| None, |
| )) |
| } else { |
| not_impl_err!( |
| "Decimal precision {} exceeds the maximum supported precision: {}", |
| precision, |
| DECIMAL256_MAX_PRECISION |
| ) |
| } |
| } |
| |
| #[cfg(test)] |
| mod tests { |
| use super::*; |
| |
| #[test] |
| fn test_decode_hex_literal() { |
| let cases = [ |
| ("", Some(vec![])), |
| ("FF00", Some(vec![255, 0])), |
| ("a00a", Some(vec![160, 10])), |
| ("FF0", Some(vec![15, 240])), |
| ("f", Some(vec![15])), |
| ("FF0X", None), |
| ("X0", None), |
| ("XX", None), |
| ("x", None), |
| ]; |
| |
| for (input, expect) in cases { |
| let output = try_decode_hex_literal(input); |
| assert_eq!(output, expect); |
| } |
| } |
| |
| #[test] |
| fn test_bigint_to_i256() { |
| let cases = [ |
| (BigInt::from(0), Some(i256::from(0))), |
| (BigInt::from(1), Some(i256::from(1))), |
| (BigInt::from(-1), Some(i256::from(-1))), |
| ( |
| BigInt::from_str(i256::MAX.to_string().as_str()).unwrap(), |
| Some(i256::MAX), |
| ), |
| ( |
| BigInt::from_str(i256::MIN.to_string().as_str()).unwrap(), |
| Some(i256::MIN), |
| ), |
| ( |
| // Can't fit into i256 |
| BigInt::from_str((i256::MAX.to_string() + "1").as_str()).unwrap(), |
| None, |
| ), |
| ]; |
| |
| for (input, expect) in cases { |
| let output = bigint_to_i256(&input); |
| assert_eq!(output, expect); |
| } |
| } |
| |
| #[test] |
| fn test_parse_decimal() { |
| // Supported cases |
| let cases = [ |
| ("0", ScalarValue::Decimal128(Some(0), 1, 0)), |
| ("1", ScalarValue::Decimal128(Some(1), 1, 0)), |
| ("123.45", ScalarValue::Decimal128(Some(12345), 5, 2)), |
| // Digit count is less than scale |
| ("0.001", ScalarValue::Decimal128(Some(1), 3, 3)), |
| // Scientific notation |
| ("123.456e-2", ScalarValue::Decimal128(Some(123456), 6, 5)), |
| // Negative scale |
| ("123456e128", ScalarValue::Decimal128(Some(123456), 6, -128)), |
| // Decimal256 |
| ( |
| &("9".repeat(39) + "." + "99999"), |
| ScalarValue::Decimal256( |
| Some(i256::from_string(&"9".repeat(44)).unwrap()), |
| 44, |
| 5, |
| ), |
| ), |
| ]; |
| for (input, expect) in cases { |
| let output = parse_decimal(input, true).unwrap(); |
| assert_eq!( |
| output, |
| Expr::Literal(expect.arithmetic_negate().unwrap(), None) |
| ); |
| |
| let output = parse_decimal(input, false).unwrap(); |
| assert_eq!(output, Expr::Literal(expect, None)); |
| } |
| |
| // scale < i8::MIN |
| assert_eq!( |
| parse_decimal("1e129", false) |
| .unwrap_err() |
| .strip_backtrace(), |
| "This feature is not implemented: Decimal scale -129 exceeds the minimum supported scale: -128" |
| ); |
| |
| // Unsupported precision |
| assert_eq!( |
| parse_decimal(&"1".repeat(77), false) |
| .unwrap_err() |
| .strip_backtrace(), |
| "This feature is not implemented: Decimal precision 77 exceeds the maximum supported precision: 76" |
| ); |
| } |
| } |