diff --git a/src/common/sql/src/convert.rs b/src/common/sql/src/convert.rs index fdd90ebe3df..8b84d591ef0 100644 --- a/src/common/sql/src/convert.rs +++ b/src/common/sql/src/convert.rs @@ -128,12 +128,29 @@ pub fn sql_value_to_value( column_schema: &ColumnSchema, sql_val: &SqlValue, timezone: Option<&Timezone>, - unary_op: Option, + mut unary_op: Option, auto_string_to_numeric: bool, ) -> Result { let column_name = &column_schema.name; let data_type = &column_schema.data_type; let mut value = match sql_val { + // Parse the sign with the digits, otherwise the minimum of a signed integer + // or timestamp overflows before it is negated. Other types keep the negation + // below: it accepts unsigned `-0`, and the decimal parser depends on length. + SqlValue::Number(n, _) + if unary_op == Some(UnaryOperator::Minus) + && matches!( + data_type, + ConcreteDataType::Int8(_) + | ConcreteDataType::Int16(_) + | ConcreteDataType::Int32(_) + | ConcreteDataType::Int64(_) + | ConcreteDataType::Timestamp(_) + ) => + { + unary_op = None; + sql_number_to_value(data_type, &format!("-{n}"))? + } SqlValue::Number(n, _) => sql_number_to_value(data_type, n)?, SqlValue::Null => Value::Null, SqlValue::Boolean(b) => { @@ -946,6 +963,55 @@ mod test { ); } + #[test] + fn test_negative_number_at_type_minimum() { + for (data_type, digits, expected) in [ + ( + ConcreteDataType::int8_datatype(), + "128", + Value::Int8(i8::MIN), + ), + ( + ConcreteDataType::int32_datatype(), + "2147483648", + Value::Int32(i32::MIN), + ), + ( + ConcreteDataType::int64_datatype(), + "9223372036854775808", + Value::Int64(i64::MIN), + ), + ] { + let value = call_sql_value_to_value!( + "test", + data_type, + &SqlValue::Number(digits.into(), false), + unary_op = UnaryOperator::Minus + ) + .unwrap(); + assert_eq!(value, expected); + } + let value = call_sql_value_to_value!( + "test", + ConcreteDataType::uint64_datatype(), + &SqlValue::Number("0".into(), false), + unary_op = UnaryOperator::Minus + ) + .unwrap(); + assert_eq!(value, Value::UInt64(0)); + + // 28 characters without the sign, 29 with it: the sign must not move the + // literal to the other decimal parser. + let value = call_sql_value_to_value!( + "test", + ConcreteDataType::decimal128_datatype(38, 30), + &SqlValue::Number("0.01234567890123456789012345".into(), false), + unary_op = UnaryOperator::Minus + ) + .unwrap(); + assert_eq!(value.to_string(), "-0.012345678901234567890123450000"); + } + #[test] fn test_auto_string_to_numeric() { // Test with auto_string_to_numeric=true diff --git a/src/partition/src/expr.rs b/src/partition/src/expr.rs index 121533acc60..560dc517060 100644 --- a/src/partition/src/expr.rs +++ b/src/partition/src/expr.rs @@ -278,76 +278,43 @@ impl PartitionExpr { } pub fn try_as_logical_expr(&self) -> error::Result { - // Special handling for null equality. - // `col = NULL` -> `col IS NULL` to match SQL (DataFusion) semantics. - let lhs_is_null = matches!(self.lhs.as_ref(), Operand::Value(Value::Null)); - let rhs_is_null = matches!(self.rhs.as_ref(), Operand::Value(Value::Null)); - - match (self.op.clone(), lhs_is_null, rhs_is_null) { - (RestrictedOp::Eq, _, true) => { - return Ok(self.lhs.try_as_logical_expr()?.is_null()); + // Row routing compares `Value`s, which order NULL before every other value. + // Batch routing and region filters treat a NULL result as false, so only the + // comparisons that hold for NULL need an explicit `IS NULL`. + let comparison = match (self.lhs.as_ref(), self.rhs.as_ref()) { + (column @ Operand::Column(_), Operand::Value(value)) => { + Some((column, value, self.op.clone())) } - (RestrictedOp::Eq, true, _) => { - return Ok(self.rhs.try_as_logical_expr()?.is_null()); + (Operand::Value(value), column @ Operand::Column(_)) => { + Some((column, value, self.op.invert_for_swap())) } - (RestrictedOp::NotEq, _, true) => { - return Ok(self.lhs.try_as_logical_expr()?.is_not_null()); - } - (RestrictedOp::NotEq, true, _) => { - return Ok(self.rhs.try_as_logical_expr()?.is_not_null()); - } - _ => {} - } - - if matches!( - self.op, - RestrictedOp::Lt | RestrictedOp::LtEq | RestrictedOp::Gt | RestrictedOp::GtEq - ) { - // Keep filtering semantics aligned with direct PartitionExpr evaluation (null-first ordering). - // In DataFusion SQL semantics, range comparisons with NULL yield NULL, so we inject - // `OR col IS NULL` on the null-first side of the comparison. - if matches!(self.lhs.as_ref(), Operand::Column(_)) { - let column_expr = self.lhs.try_as_logical_expr()?; - let other_expr = self.rhs.try_as_logical_expr()?; - let base = match self.op { - RestrictedOp::Lt => { - column_expr.clone().lt(other_expr).or(column_expr.is_null()) + _ => None, + }; + if let Some((column, value, op)) = comparison { + let column = column.try_as_logical_expr()?; + if matches!(value, Value::Null) { + return match op { + RestrictedOp::Eq | RestrictedOp::LtEq => Ok(column.is_null()), + RestrictedOp::NotEq | RestrictedOp::Gt => Ok(column.is_not_null()), + RestrictedOp::Lt => Ok(datafusion_expr::lit(false)), + RestrictedOp::GtEq => Ok(datafusion_expr::lit(true)), + RestrictedOp::And | RestrictedOp::Or => { + error::InvalidExprSnafu { expr: self.clone() }.fail() } - RestrictedOp::LtEq => column_expr - .clone() - .lt_eq(other_expr) - .or(column_expr.is_null()), - RestrictedOp::Gt => column_expr - .clone() - .gt(other_expr) - .and(column_expr.is_not_null()), - RestrictedOp::GtEq => column_expr - .clone() - .gt_eq(other_expr) - .and(column_expr.is_not_null()), - _ => unreachable!(), }; - return Ok(base); - } else if matches!(self.rhs.as_ref(), Operand::Column(_)) { - let other_expr = self.lhs.try_as_logical_expr()?; - let column_expr = self.rhs.try_as_logical_expr()?; - let base = match self.op { - RestrictedOp::Lt => other_expr - .lt(column_expr.clone()) - .and(column_expr.is_not_null()), - RestrictedOp::LtEq => other_expr - .lt_eq(column_expr.clone()) - .and(column_expr.is_not_null()), - RestrictedOp::Gt => { - other_expr.gt(column_expr.clone()).or(column_expr.is_null()) - } - RestrictedOp::GtEq => other_expr - .gt_eq(column_expr.clone()) - .or(column_expr.is_null()), - _ => unreachable!(), - }; - return Ok(base); } + let bound = Operand::Value(value.clone()).try_as_logical_expr()?; + return match op { + RestrictedOp::Eq => Ok(column.eq(bound)), + RestrictedOp::NotEq => Ok(column.clone().not_eq(bound).or(column.is_null())), + RestrictedOp::Lt => Ok(column.clone().lt(bound).or(column.is_null())), + RestrictedOp::LtEq => Ok(column.clone().lt_eq(bound).or(column.is_null())), + RestrictedOp::Gt => Ok(column.gt(bound)), + RestrictedOp::GtEq => Ok(column.gt_eq(bound)), + RestrictedOp::And | RestrictedOp::Or => { + error::InvalidExprSnafu { expr: self.clone() }.fail() + } + }; } // Normal cases handling, without NULL @@ -589,7 +556,7 @@ mod tests { .try_as_logical_expr() .unwrap() .to_string(), - "a > Int64(10) AND a IS NOT NULL" + "a > Int64(10)" ); // Test Gt with column on LHS @@ -600,7 +567,7 @@ mod tests { ); assert_eq!( gt_expr.try_as_logical_expr().unwrap().to_string(), - "a > Int64(10) AND a IS NOT NULL" + "a > Int64(10)" ); // Test Gt with column on RHS @@ -625,7 +592,7 @@ mod tests { ); assert_eq!( gteq_expr.try_as_logical_expr().unwrap().to_string(), - "a >= Int64(10) AND a IS NOT NULL" + "a >= Int64(10)" ); // Test LtEq with column on LHS @@ -662,7 +629,7 @@ mod tests { .try_as_logical_expr() .unwrap() .to_string(), - "a >= Int64(10) AND a IS NOT NULL" + "a >= Int64(10)" ); let and_expr = PartitionExpr::new( @@ -680,7 +647,7 @@ mod tests { ); assert_eq!( and_expr.try_as_logical_expr().unwrap().to_string(), - "(a <= Int64(10) OR a IS NULL) AND b > Int64(5) AND b IS NOT NULL" + "(a <= Int64(10) OR a IS NULL) AND b > Int64(5)" ); let and_expr = PartitionExpr::new( @@ -698,7 +665,7 @@ mod tests { ); assert_eq!( and_expr.try_as_logical_expr().unwrap().to_string(), - "(a <= Int64(10) OR a IS NULL) AND a > Int64(5) AND a IS NOT NULL" + "(a <= Int64(10) OR a IS NULL) AND a > Int64(5)" ); let and_expr_strict_lower = PartitionExpr::new( @@ -719,7 +686,7 @@ mod tests { .try_as_logical_expr() .unwrap() .to_string(), - "(a < Int64(10) OR a IS NULL) AND a >= Int64(5) AND a IS NOT NULL" + "(a < Int64(10) OR a IS NULL) AND a >= Int64(5)" ); let and_expr_rhs_column = PartitionExpr::new( @@ -740,7 +707,7 @@ mod tests { .try_as_logical_expr() .unwrap() .to_string(), - "(a <= Int64(10) OR a IS NULL) AND a > Int64(5) AND a IS NOT NULL" + "(a <= Int64(10) OR a IS NULL) AND a > Int64(5)" ); let or_expr_same_column = PartitionExpr::new( @@ -761,7 +728,7 @@ mod tests { .try_as_logical_expr() .unwrap() .to_string(), - "a <= Int64(10) OR a IS NULL OR a > Int64(5) AND a IS NOT NULL" + "a <= Int64(10) OR a IS NULL OR a > Int64(5)" ); } @@ -777,7 +744,7 @@ mod tests { .try_as_logical_expr() .unwrap() .to_string(), - "Int64(10) > a OR a IS NULL" + "a < Int64(10) OR a IS NULL" ); let gteq_expr_rhs_column = PartitionExpr { @@ -790,7 +757,7 @@ mod tests { .try_as_logical_expr() .unwrap() .to_string(), - "Int64(10) >= a OR a IS NULL" + "a <= Int64(10) OR a IS NULL" ); } diff --git a/src/partition/src/multi_dim.rs b/src/partition/src/multi_dim.rs index df7b5291e90..29e45126343 100644 --- a/src/partition/src/multi_dim.rs +++ b/src/partition/src/multi_dim.rs @@ -21,8 +21,9 @@ use std::sync::{Arc, RwLock}; use datafusion_expr::ColumnarValue; use datafusion_physical_expr::PhysicalExpr; use datatypes::arrow; -use datatypes::arrow::array::{BooleanArray, BooleanBufferBuilder, RecordBatch}; +use datatypes::arrow::array::{Array, BooleanArray, BooleanBufferBuilder, RecordBatch}; use datatypes::arrow::buffer::BooleanBuffer; +use datatypes::arrow::compute::prep_null_mask_filter; use datatypes::arrow::datatypes::Schema; use datatypes::prelude::Value; use datatypes::vectors::{Helper, VectorRef}; @@ -272,6 +273,9 @@ impl MultiDimPartitionRule { } }; let array = match columnar_value_to_boolean_array(col_val, num_rows) { + // A NULL result means the row is not in this region. Masks are merged + // with non-Kleene `or` below, where a NULL would drop the row. + Ok(array) if array.null_count() > 0 => prep_null_mask_filter(&array), Ok(array) => array, Err(e) => { return Some(Err(e)); @@ -977,6 +981,105 @@ mod test_split_record_batch { assert_eq!(total_rows, batch.num_rows()); } + #[test] + fn test_comparison_batch_matches_row() { + use datatypes::arrow::array::Array; + use datatypes::arrow::datatypes::{DataType, Field}; + + let inputs = [None, Some(4), Some(5), Some(6)]; + let batch = RecordBatch::try_new( + Arc::new(Schema::new(vec![Field::new("a", DataType::Int64, true)])), + vec![Arc::new(Int64Array::from(inputs.to_vec()))], + ) + .unwrap(); + let mut mismatches = vec![]; + for op in [ + RestrictedOp::Eq, + RestrictedOp::NotEq, + RestrictedOp::Lt, + RestrictedOp::LtEq, + RestrictedOp::Gt, + RestrictedOp::GtEq, + ] { + for bound in [Value::Null, Value::Int64(5)] { + // Both operand orders, since a rule restored without canonicalization may + // keep the literal on the left. + for expr in [ + PartitionExpr::new(col("a"), op.clone(), bound.clone().into()), + PartitionExpr { + lhs: Box::new(bound.clone().into()), + op: op.clone(), + rhs: Box::new(col("a")), + }, + ] { + let rule = MultiDimPartitionRule::try_new( + vec!["a".to_string()], + vec![1], + vec![expr.clone()], + false, + ) + .unwrap(); + let by_batch = match expr + .try_as_physical_expr(batch.schema_ref()) + .unwrap() + .evaluate(&batch) + { + Ok(by_batch) => { + columnar_value_to_boolean_array(by_batch, batch.num_rows()).unwrap() + } + Err(err) => { + mismatches.push(format!("{expr}: batch error {err}")); + continue; + } + }; + for (row, input) in inputs.iter().enumerate() { + let value = input.map_or(Value::Null, Value::Int64); + let by_row = rule.evaluate_expr(&expr, &[value]).unwrap(); + let by_batch = by_batch.is_valid(row) && by_batch.value(row); + if by_row != by_batch { + mismatches.push(format!( + "{expr} with a = {input:?}: row {by_row}, batch {by_batch}" + )); + } + } + } + } + } + assert!(mismatches.is_empty(), "{mismatches:#?}"); + } + + #[test] + fn test_split_null_rows_match_row_routing() { + use datatypes::arrow::datatypes::{DataType, Field}; + + let rule = MultiDimPartitionRule::try_new( + vec!["host".to_string()], + vec![0, 1], + vec![ + col("host").not_eq(Value::from("a")), + col("host").eq(Value::from("a")), + ], + true, + ) + .unwrap(); + for hosts in [vec![Some("a"), None, Some("z")], vec![Some("a"), None]] { + let batch = RecordBatch::try_new( + Arc::new(Schema::new(vec![Field::new("host", DataType::Utf8, true)])), + vec![Arc::new(StringArray::from(hosts.clone()))], + ) + .unwrap(); + let by_row = rule.split_record_batch_naive(&batch).unwrap(); + let by_batch = rule.split_record_batch(&batch).unwrap(); + for (region, expected) in by_row { + let actual = by_batch + .get(®ion) + .map(|mask| mask.array().clone()) + .unwrap_or_else(|| BooleanArray::from(vec![false; batch.num_rows()])); + assert_eq!(actual, expected, "{hosts:?}, region {region}"); + } + } + } + #[test] fn test_columnar_value_to_boolean_array_scalar_false() { let result = columnar_value_to_boolean_array( diff --git a/tests/cases/standalone/common/partition.result b/tests/cases/standalone/common/partition.result index fe6f99df591..132f2a325f6 100644 --- a/tests/cases/standalone/common/partition.result +++ b/tests/cases/standalone/common/partition.result @@ -463,3 +463,33 @@ DROP TABLE `molestiAe`; Affected Rows: 0 +-- The minimum of a signed type is a valid bound. +CREATE TABLE int_min_bound ( + a INT PRIMARY KEY, + ts TIMESTAMP TIME INDEX, +) PARTITION ON COLUMNS (a) ( + a < -2147483648, + a >= -2147483648 AND a < 0, + a >= 0 +); + +Affected Rows: 0 + +INSERT INTO int_min_bound VALUES (-2147483648, 1), (NULL, 2), (0, 3); + +Affected Rows: 3 + +SELECT * FROM int_min_bound ORDER BY ts; + ++-------------+-------------------------+ +| a | ts | ++-------------+-------------------------+ +| -2147483648 | 1970-01-01T00:00:00.001 | +| | 1970-01-01T00:00:00.002 | +| 0 | 1970-01-01T00:00:00.003 | ++-------------+-------------------------+ + +DROP TABLE int_min_bound; + +Affected Rows: 0 + diff --git a/tests/cases/standalone/common/partition.sql b/tests/cases/standalone/common/partition.sql index 7b5e8ea7750..41627bf4d09 100644 --- a/tests/cases/standalone/common/partition.sql +++ b/tests/cases/standalone/common/partition.sql @@ -204,3 +204,19 @@ INSERT INTO `molestiAe` VALUES SELECT * FROM `molestiAe`; DROP TABLE `molestiAe`; + +-- The minimum of a signed type is a valid bound. +CREATE TABLE int_min_bound ( + a INT PRIMARY KEY, + ts TIMESTAMP TIME INDEX, +) PARTITION ON COLUMNS (a) ( + a < -2147483648, + a >= -2147483648 AND a < 0, + a >= 0 +); + +INSERT INTO int_min_bound VALUES (-2147483648, 1), (NULL, 2), (0, 3); + +SELECT * FROM int_min_bound ORDER BY ts; + +DROP TABLE int_min_bound; diff --git a/tests/compatibility/cases/analyze_verbose_remote_metrics_extension/verify.result b/tests/compatibility/cases/analyze_verbose_remote_metrics_extension/verify.result index 7394ca8faee..2bdcb907ebf 100644 --- a/tests/compatibility/cases/analyze_verbose_remote_metrics_extension/verify.result +++ b/tests/compatibility/cases/analyze_verbose_remote_metrics_extension/verify.result @@ -28,7 +28,7 @@ EXPLAIN ANALYZE VERBOSE SELECT count(*) FROM t_analyze_verbose_remote_metrics_ex | 1_| 1_|_AggregateExec: mode=Final, gby=[], aggr=[__count_state(t_analyze_verbose_remote_REDACTED |_|_|_CoalescePartitionsExec REDACTED |_|_|_AggregateExec: mode=Partial, gby=[], aggr=[__count_state(t_analyze_verbose_remote_REDACTED -|_|_|_SeqScan: region=REDACTED, {"partition_count":REDACTED, "projection": ["ts"], "filters": ["host >= Utf8(\"m\") AND host IS NOT NULL"], "flat_format": REDACTED, "REDACTED +|_|_|_SeqScan: region=REDACTED, {"partition_count":REDACTED, "projection": ["ts"], "filters": ["host >= Utf8(\"m\")"], "flat_format": REDACTED, "REDACTED |_|_|_| |_|_| Total rows: 1_| +-+-+-+