rust/greptimedb
1
0
Fork 0
mirror of https://github.com/GreptimeTeam/greptimedb.git synced 2026-10-21 19:32:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(partition): route NULLs like row inserts and accept signed-minimum bounds (#9697)

Browse Source 
* fix(partition): route NULLs like row inserts and accept signed-minimum bounds

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

* fix(partition): keep equality two-valued for NULL rows and accept unsigned -0

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

* fix(partition): treat NULL masks as false in batch routing and keep comparisons bare

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

* test(compat): update partition filter in analyze verbose expected output

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

---------

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>
This commit is contained in:
dennis zhuang
2026-10-19 15:22:13 +00:00
committed by GitHub
parent 7397ab6b23
commit da88ff8653
6 changed files with 261 additions and 79 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/common/sql/src/convert.rs
+67 -1
View File
@@ -128,12 +128,29 @@ pub fn sql_value_to_value(
column_schema: &ColumnSchema,
sql_val: &SqlValue,
timezone: Option<&Timezone>,
unary_op: Option<UnaryOperator>,
mut unary_op: Option<UnaryOperator>,
auto_string_to_numeric: bool,
) -> Result<Value> {
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
src/partition/src/expr.rs
+43 -76
View File
@@ -278,76 +278,43 @@ impl PartitionExpr {
}
pub fn try_as_logical_expr(&self) -> error::Result<Expr> {
// 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"
);
}
src/partition/src/multi_dim.rs
+104 -1
View File
@@ -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(&region)
.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(
tests/cases/standalone/common/partition.result
+30
View File
@@ -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
tests/cases/standalone/common/partition.sql
+16
View File
@@ -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;
tests/compatibility/cases/analyze_verbose_remote_metrics_extension/verify.result
+1 -1
View File
@@ -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_|
+-+-+-+