fix(query): guard deep expressions during query planning (#9564)

* fix(query): avoid speculative clones in constant normalization

Signed-off-by: evenyag <realevenyag@gmail.com>

* fix(query): reject deeply nested expressions before optimization

Signed-off-by: evenyag <realevenyag@gmail.com>

---------

Signed-off-by: evenyag <realevenyag@gmail.com>
This commit is contained in:
Yingwen
2026-10-14 06:49:35 +00:00
committed by GitHub
parent 4ff3296f97
commit 00aa65de7a
14 changed files with 433 additions and 95 deletions
@@ -109,6 +109,11 @@ impl EmptyMetric {
})
}
/// Returns the expression used to generate metric values.
pub fn expr(&self) -> Option<&Expr> {
self.expr.as_ref()
}
pub const fn name() -> &'static str {
"EmptyMetric"
}
+5
View File
@@ -68,6 +68,7 @@ use crate::metrics::{
OnDone, QUERY_STAGE_ELAPSED, maybe_attach_region_watermark_metrics,
should_collect_region_watermark_from_query_ctx,
};
use crate::optimizer::expr_depth;
use crate::options::ScheduledTimeExtension;
use crate::physical_wrapper::PhysicalPlanWrapperRef;
use crate::planner::{DfLogicalPlanner, LogicalPlanner};
@@ -381,6 +382,8 @@ impl DatafusionQueryEngine {
}
}
expr_depth::validate_plan(logical_plan)?;
let mut logger = PanicLogger {
input_logical_plan: logical_plan,
after_analyze: None,
@@ -408,6 +411,7 @@ impl DatafusionQueryEngine {
|_, _| {},
)?;
expr_depth::validate_plan(&analyzed_plan)?;
logger.after_analyze = Some(analyzed_plan.clone());
common_telemetry::debug!("Create physical plan, analyzed plan: {analyzed_plan}");
@@ -423,6 +427,7 @@ impl DatafusionQueryEngine {
.optimize(analyzed_plan, state, |_, _| {})?
};
expr_depth::validate_plan(&optimized_plan)?;
common_telemetry::debug!("Create physical plan, optimized plan: {optimized_plan}");
logger.after_optimize = Some(optimized_plan.clone());
+1 -1
View File
@@ -27,7 +27,7 @@ mod remote_dyn_filter_registry;
pub use analyzer::{DistPlannerAnalyzer, DistPlannerOptions};
pub use filter_id::{FilterFingerprint, FilterId, ParseFilterIdError, RemoteDynFilterProducerId};
pub use merge_scan::{MergeScanExec, MergeScanLogicalPlan};
pub(crate) use merge_sort::MergeSortExec;
pub(crate) use merge_sort::{MergeSortExec, MergeSortLogicalPlan};
pub use planner::{DistExtensionPlanner, MergeSortExtensionPlanner};
pub use predicate_extractor::PredicateExtractor;
pub use region_pruner::ConstraintPruner;
+3
View File
@@ -53,6 +53,7 @@ use crate::dist_plan::commutativity::{
use crate::dist_plan::merge_scan::MergeScanLogicalPlan;
use crate::dist_plan::merge_sort::MergeSortLogicalPlan;
use crate::metrics::PUSH_DOWN_FALLBACK_ERRORS_TOTAL;
use crate::optimizer::expr_depth;
use crate::options::ScheduledTimeExtension;
use crate::plan::ExtractExpr;
use crate::query_engine::DefaultSerializer;
@@ -118,6 +119,7 @@ impl AnalyzerRule for DistPlannerAnalyzer {
plan: LogicalPlan,
config: &ConfigOptions,
) -> datafusion_common::Result<LogicalPlan> {
expr_depth::validate_plan(&plan)?;
let mut config = config.clone();
// Aligned with the behavior in `datafusion_optimizer::OptimizerContext::new()`.
config.optimizer.filter_null_join_keys = true;
@@ -142,6 +144,7 @@ impl AnalyzerRule for DistPlannerAnalyzer {
let plan = plan
.rewrite_with_subqueries(&mut PlanTreeExpressionSimplifier::new(optimizer_context))?
.data;
expr_depth::validate_plan(&plan)?;
let fallback_plan = plan.clone();
// Run a filter-focused optimizer subset before MergeScan wraps remote
+2
View File
@@ -50,6 +50,7 @@ use crate::dist_plan::merge_scan::{MergeScanExec, MergeScanLogicalPlan};
use crate::dist_plan::merge_sort::{MergeSortExec, MergeSortLogicalPlan};
use crate::dist_plan::region_pruner::ConstraintPruner;
use crate::error::{CatalogSnafu, PartitionRuleManagerSnafu, TableNotFoundSnafu};
use crate::optimizer::expr_depth;
use crate::region_query::RegionQueryHandlerRef;
/// Planner for converting merge sort logical plan to physical plan.
@@ -447,6 +448,7 @@ impl DistExtensionPlanner {
session: &dyn Session,
plan: &LogicalPlan,
) -> Result<LogicalPlan> {
expr_depth::validate_plan(plan)?;
let session_state = session
.as_any()
.downcast_ref::<SessionState>()
+1
View File
@@ -17,6 +17,7 @@ pub mod constant_term;
pub mod count_nest_aggr;
pub mod count_wildcard;
pub mod enforce_sorting;
pub(crate) mod expr_depth;
pub mod global_limit;
pub(crate) mod insert_assignment;
pub(crate) mod json_schema_concretize;
+96 -93
View File
@@ -23,8 +23,6 @@ use datafusion_expr::{Between, BinaryExpr, Expr, ExprSchemable, LogicalPlan, Ope
use datafusion_expr_common::casts::try_cast_literal_to_type;
use datafusion_optimizer::analyzer::AnalyzerRule;
use crate::plan::ExtractExpr;
/// ConstNormalizationRule rewrites castable constants against their
/// non-constant comparison operand ahead of filter pushdown.
#[derive(Debug)]
@@ -52,26 +50,12 @@ impl AnalyzerRule for ConstNormalizationRule {
}
fn rewrite_plan_exprs(plan: LogicalPlan, schema: DFSchemaRef) -> Result<Transformed<LogicalPlan>> {
let mut rewriter = ConstNormalizationRewriter {
schema,
transformed: false,
};
let exprs = plan
.expressions_consider_join()
.into_iter()
.map(|expr| expr.rewrite(&mut rewriter).map(|rewritten| rewritten.data))
.collect::<Result<Vec<_>>>()?;
if !rewriter.transformed {
return Ok(Transformed::no(plan));
}
let inputs = plan.inputs().into_iter().cloned().collect::<Vec<_>>();
plan.with_new_exprs(exprs, inputs).map(Transformed::yes)
let mut rewriter = ConstNormalizationRewriter { schema };
plan.map_expressions(|expr| expr.rewrite(&mut rewriter))
}
struct ConstNormalizationRewriter {
schema: DFSchemaRef,
transformed: bool,
}
impl TreeNodeRewriter for ConstNormalizationRewriter {
@@ -91,44 +75,36 @@ impl TreeNodeRewriter for ConstNormalizationRewriter {
}
fn f_up(&mut self, expr: Expr) -> Result<Transformed<Expr>> {
let rewritten = rewrite_expr_node(expr, &self.schema)?;
self.transformed |= rewritten.transformed;
Ok(rewritten)
rewrite_expr_node(expr, &self.schema)
}
}
fn rewrite_expr_node(expr: Expr, schema: &DFSchemaRef) -> Result<Transformed<Expr>> {
match expr {
Expr::BinaryExpr(binary) => match rewrite_binary_expr(binary.clone(), schema)? {
Some(expr) => Ok(Transformed::yes(expr)),
None => Ok(Transformed::no(Expr::BinaryExpr(binary))),
},
Expr::Between(between) => match rewrite_between_expr(between.clone(), schema)? {
Some(expr) => Ok(Transformed::yes(expr)),
None => Ok(Transformed::no(Expr::Between(between))),
},
Expr::InList(in_list) => match rewrite_in_list_expr(in_list.clone(), schema)? {
Some(expr) => Ok(Transformed::yes(expr)),
None => Ok(Transformed::no(Expr::InList(in_list))),
},
Expr::Like(like) => rewrite_like_expr(like, PatternMatchKind::Like, schema),
Expr::SimilarTo(like) => rewrite_like_expr(like, PatternMatchKind::SimilarTo, schema),
expr => Ok(Transformed::no(expr)),
}
// Borrow while checking eligibility: cloning unchanged subtrees here can overflow
// the stack even though DataFusion protects the surrounding rewrite traversal.
let rewritten = match &expr {
Expr::BinaryExpr(binary) => rewrite_binary_expr(binary, schema)?,
Expr::Between(between) => rewrite_between_expr(between, schema)?,
Expr::InList(in_list) => rewrite_in_list_expr(in_list, schema)?,
Expr::Like(like) => rewrite_like_expr(like, PatternMatchKind::Like, schema)?,
Expr::SimilarTo(like) => rewrite_like_expr(like, PatternMatchKind::SimilarTo, schema)?,
_ => None,
};
Ok(match rewritten {
Some(rewritten) => Transformed::yes(rewritten),
None => Transformed::no(expr),
})
}
fn rewrite_between_expr(between: Between, schema: &DFSchemaRef) -> Result<Option<Expr>> {
fn rewrite_between_expr(between: &Between, schema: &DFSchemaRef) -> Result<Option<Expr>> {
let Between {
expr,
negated,
low,
high,
} = between;
let expr = *expr;
let low_expr = *low;
let high_expr = *high;
let Some((target, constants)) =
extract_rewrite_operands(&expr, &[low_expr.clone(), high_expr.clone()], schema)?
extract_rewrite_operands(expr, [low.as_ref(), high.as_ref()], schema)?
else {
return Ok(None);
};
@@ -142,7 +118,7 @@ fn rewrite_between_expr(between: Between, schema: &DFSchemaRef) -> Result<Option
.expect("between normalization expects low constant");
return Ok(Some(Expr::Between(Between {
expr: Box::new(target.expr.clone()),
negated,
negated: *negated,
low: Box::new(lit(low)),
high: Box::new(lit(high)),
})));
@@ -153,14 +129,13 @@ fn rewrite_between_expr(between: Between, schema: &DFSchemaRef) -> Result<Option
.flatten())
}
fn rewrite_in_list_expr(in_list: InList, schema: &DFSchemaRef) -> Result<Option<Expr>> {
fn rewrite_in_list_expr(in_list: &InList, schema: &DFSchemaRef) -> Result<Option<Expr>> {
let InList {
expr,
list,
negated,
} = in_list;
let expr = *expr;
let Some((target, constants)) = extract_rewrite_operands(&expr, &list, schema)? else {
let Some((target, constants)) = extract_rewrite_operands(expr, list, schema)? else {
return Ok(None);
};
@@ -168,19 +143,15 @@ fn rewrite_in_list_expr(in_list: InList, schema: &DFSchemaRef) -> Result<Option<
target
.expr
.clone()
.in_list(constants.into_iter().map(lit).collect(), negated)
.in_list(constants.into_iter().map(lit).collect(), *negated)
}))
}
fn rewrite_like_expr(
like: Like,
like: &Like,
kind: PatternMatchKind,
schema: &DFSchemaRef,
) -> Result<Transformed<Expr>> {
let original = match kind {
PatternMatchKind::Like => Expr::Like(like.clone()),
PatternMatchKind::SimilarTo => Expr::SimilarTo(like.clone()),
};
) -> Result<Option<Expr>> {
let Like {
negated,
expr,
@@ -188,36 +159,33 @@ fn rewrite_like_expr(
escape_char,
case_insensitive,
} = like;
let expr = *expr;
let pattern = *pattern;
let Some((target, constants)) =
extract_rewrite_operands(&expr, std::slice::from_ref(&pattern), schema)?
let Some((target, constants)) = extract_rewrite_operands(expr, [pattern.as_ref()], schema)?
else {
return Ok(Transformed::no(original));
return Ok(None);
};
let Some(mut constants) = target.normalize_constants(&constants) else {
return Ok(Transformed::no(original));
return Ok(None);
};
let pattern = lit(constants
.pop()
.expect("pattern normalization expects one constant"));
let like = Like::new(
negated,
*negated,
Box::new(target.expr.clone()),
Box::new(pattern),
escape_char,
case_insensitive,
*escape_char,
*case_insensitive,
);
let rewritten = match kind {
PatternMatchKind::Like => Expr::Like(like),
PatternMatchKind::SimilarTo => Expr::SimilarTo(like),
};
Ok(Transformed::yes(rewritten))
Ok(Some(rewritten))
}
fn rewrite_binary_expr(binary: BinaryExpr, schema: &DFSchemaRef) -> Result<Option<Expr>> {
if let Some(expr) = rewrite_dictionary_string_regex(binary.clone(), schema)? {
fn rewrite_binary_expr(binary: &BinaryExpr, schema: &DFSchemaRef) -> Result<Option<Expr>> {
if let Some(expr) = rewrite_dictionary_string_regex(binary, schema)? {
return Ok(Some(expr));
}
@@ -226,9 +194,7 @@ fn rewrite_binary_expr(binary: BinaryExpr, schema: &DFSchemaRef) -> Result<Optio
}
let BinaryExpr { left, op, right } = binary;
let left = *left;
let right = *right;
if let Some(expr) = rewrite_binary_side(left.clone(), op, right.clone(), schema)? {
if let Some(expr) = rewrite_binary_side(left, *op, right, schema)? {
return Ok(Some(expr));
}
@@ -244,12 +210,12 @@ fn rewrite_binary_expr(binary: BinaryExpr, schema: &DFSchemaRef) -> Result<Optio
/// Keeping the dictionary input lets DataFusion's physical regex kernel evaluate scalar patterns
/// against dictionary values instead of materializing the string column first.
fn rewrite_dictionary_string_regex(
binary: BinaryExpr,
binary: &BinaryExpr,
schema: &DFSchemaRef,
) -> Result<Option<Expr>> {
let BinaryExpr { left, op, right } = binary;
if !matches!(
&op,
op,
Operator::RegexMatch
| Operator::RegexIMatch
| Operator::RegexNotMatch
@@ -259,7 +225,7 @@ fn rewrite_dictionary_string_regex(
return Ok(None);
}
let Some((CastInputKind::Cast, source, DataType::Utf8)) = extract_cast_input(&left) else {
let Some((CastInputKind::Cast, source, DataType::Utf8)) = extract_cast_input(left) else {
return Ok(None);
};
if !matches!(source, Expr::Column(_))
@@ -274,19 +240,18 @@ fn rewrite_dictionary_string_regex(
Ok(Some(Expr::BinaryExpr(BinaryExpr {
left: Box::new(source.clone()),
op,
right,
op: *op,
right: right.clone(),
})))
}
fn rewrite_binary_side(
target_expr: Expr,
target_expr: &Expr,
op: Operator,
constant_expr: Expr,
constant_expr: &Expr,
schema: &DFSchemaRef,
) -> Result<Option<Expr>> {
let Some((target, constants)) =
extract_rewrite_operands(&target_expr, std::slice::from_ref(&constant_expr), schema)?
let Some((target, constants)) = extract_rewrite_operands(target_expr, [constant_expr], schema)?
else {
return Ok(None);
};
@@ -305,11 +270,11 @@ fn rewrite_binary_side(
Ok(target.normalize_timestamp_binary(op, &constants[0]))
}
fn extract_rewrite_operands(
target_expr: &Expr,
constant_exprs: &[Expr],
fn extract_rewrite_operands<'a, 'b>(
target_expr: &'a Expr,
constant_exprs: impl IntoIterator<Item = &'b Expr>,
schema: &DFSchemaRef,
) -> Result<Option<(NormalizationTarget, Vec<ScalarValue>)>> {
) -> Result<Option<(NormalizationTarget<'a>, Vec<ScalarValue>)>> {
let Some(target) = extract_normalization_target(target_expr, schema)? else {
return Ok(None);
};
@@ -319,8 +284,8 @@ fn extract_rewrite_operands(
}
#[derive(Clone)]
struct NormalizationTarget {
expr: Expr,
struct NormalizationTarget<'a> {
expr: &'a Expr,
data_type: DataType,
kind: NormalizationKind,
}
@@ -337,7 +302,7 @@ enum NormalizationKind {
},
}
impl NormalizationTarget {
impl NormalizationTarget<'_> {
/// Normalizes constants for rewrites that can preserve the original predicate with a direct
/// literal cast. Timestamp precision-changing casts are handled by timestamp-specific helpers.
fn normalize_constants(&self, constants: &[ScalarValue]) -> Option<Vec<ScalarValue>> {
@@ -438,17 +403,17 @@ impl NormalizationTarget {
///
/// Plain expressions normalize literals to their own type. Cast expressions only participate when
/// the cast is lossless or when timestamp downcasts can be rewritten as wider source-side bounds.
fn extract_normalization_target(
expr: &Expr,
fn extract_normalization_target<'a>(
expr: &'a Expr,
schema: &DFSchemaRef,
) -> Result<Option<NormalizationTarget>> {
) -> Result<Option<NormalizationTarget<'a>>> {
if extract_constant_scalar(expr)?.is_some() {
return Ok(None);
}
let Some((_, source_expr, target_type)) = extract_cast_input(expr) else {
return Ok(Some(NormalizationTarget {
expr: expr.clone(),
expr,
data_type: expr.get_type(schema)?,
kind: NormalizationKind::Lossless,
}));
@@ -460,7 +425,7 @@ fn extract_normalization_target(
};
Ok(Some(NormalizationTarget {
expr: source_expr.clone(),
expr: source_expr,
data_type,
kind,
}))
@@ -520,8 +485,11 @@ enum PatternMatchKind {
SimilarTo,
}
fn extract_constant_scalars(exprs: &[Expr]) -> Result<Option<Vec<ScalarValue>>> {
let mut values = Vec::with_capacity(exprs.len());
fn extract_constant_scalars<'a>(
exprs: impl IntoIterator<Item = &'a Expr>,
) -> Result<Option<Vec<ScalarValue>>> {
let exprs = exprs.into_iter();
let mut values = Vec::with_capacity(exprs.size_hint().0);
for expr in exprs {
let Some(value) = extract_constant_scalar(expr)? else {
return Ok(None);
@@ -672,6 +640,41 @@ mod tests {
rewrite_dictionary_string_regex, try_cast_literal_to_type,
};
#[test]
fn test_normalize_deep_predicate() {
std::thread::Builder::new()
.stack_size(2 * 1024 * 1024)
.spawn(|| {
let mut predicate = col("t.v").eq(lit(0_i64));
for value in 1..1024_i64 {
predicate = predicate.or(col("t.v").eq(lit(value)));
}
let schema = test_schema(vec![Field::new("v", DataType::Int32, false)]);
// Construct the filter directly to isolate this rule from other planner passes.
let plan = analyze_plan(LogicalPlan::Filter(
datafusion_expr::Filter::try_new(predicate, Arc::new(build_scan_plan(schema)))
.unwrap(),
));
let LogicalPlan::Filter(filter) = plan else {
panic!("expected filter");
};
// Inspect and consume the chain without recursive formatting or equality.
let mut predicate = filter.predicate;
for value in (1..1024_i32).rev() {
let Expr::BinaryExpr(BinaryExpr { left, op, right }) = predicate else {
panic!("expected binary predicate");
};
assert_eq!(op, Operator::Or);
assert_eq!(*right, col("t.v").eq(lit(value)));
predicate = *left;
}
assert_eq!(predicate, col("t.v").eq(lit(0_i32)));
})
.unwrap()
.join()
.unwrap();
}
#[test]
fn test_normalize_direct_integer_cast_comparison() {
assert_filter_plan(
@@ -1411,7 +1414,7 @@ mod tests {
op: Operator,
) -> Option<Expr> {
rewrite_dictionary_string_regex(
BinaryExpr {
&BinaryExpr {
left: Box::new(left),
op,
right: Box::new(right),
+223
View File
@@ -0,0 +1,223 @@
// Copyright 2023 Greptime Team
//
// Licensed 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 datafusion_common::config::ConfigOptions;
use datafusion_common::tree_node::{TreeNode, TreeNodeRecursion};
use datafusion_common::{Result, plan_err};
use datafusion_expr::{Expr, LogicalPlan, Partitioning};
use datafusion_optimizer::analyzer::AnalyzerRule;
use promql::extension_plan::EmptyMetric;
use crate::dist_plan::{MergeScanLogicalPlan, MergeSortLogicalPlan};
use crate::range_select::plan::RangeSelect;
// A conservative optimizer input limit, not a guarantee for every recursive operation
// performed while parsing, planning, or dropping an arbitrarily deep query.
const MAX_EXPR_DEPTH: usize = 128;
/// Checks the analyzer output, including the SessionState EXPLAIN path.
#[derive(Debug)]
pub(crate) struct ExprDepthRule;
impl AnalyzerRule for ExprDepthRule {
fn name(&self) -> &str {
"expr_depth"
}
fn analyze(&self, plan: LogicalPlan, _: &ConfigOptions) -> Result<LogicalPlan> {
validate_plan(&plan)?;
Ok(plan)
}
}
/// Rejects deep expressions before recursive optimizer traversal, cloning, or display.
/// Both plan and expression traversal are iterative. Subquery Arcs are retained so
/// expressions supplied temporarily by extension nodes can also be inspected safely.
pub(crate) fn validate_plan(plan: &LogicalPlan) -> Result<()> {
let mut subqueries = Vec::new();
validate_plan_exprs(plan, &mut subqueries)?;
while let Some(plan) = subqueries.pop() {
validate_plan_exprs(&plan, &mut subqueries)?;
}
Ok(())
}
fn validate_plan_exprs(plan: &LogicalPlan, subqueries: &mut Vec<Arc<LogicalPlan>>) -> Result<()> {
let mut plans = vec![plan];
while let Some(plan) = plans.pop() {
let mut check = |expr: &Expr| {
validate_expr(expr, subqueries)?;
Ok(TreeNodeRecursion::Continue)
};
match plan {
// apply_expressions clones range ordering expressions and extension
// expressions. Borrow the built-in nodes that can contain deep trees.
LogicalPlan::Repartition(repartition)
if let Partitioning::Range(range) = &repartition.partitioning_scheme =>
{
for sort in range.ordering() {
check(&sort.expr)?;
}
}
LogicalPlan::Extension(extension) => {
let node = extension.node.as_any();
if let Some(range) = node.downcast_ref::<RangeSelect>() {
for expr in &range.range_expr {
check(&expr.expr)?;
}
check(&range.time_expr)?;
for expr in &range.by {
check(expr)?;
}
} else if let Some(sort) = node.downcast_ref::<MergeSortLogicalPlan>() {
for expr in &sort.expr {
check(&expr.expr)?;
}
} else if let Some(metric) = node.downcast_ref::<EmptyMetric>() {
if let Some(expr) = metric.expr() {
check(expr)?;
}
} else if let Some(scan) = node.downcast_ref::<MergeScanLogicalPlan>() {
// MergeScan deliberately hides this input from optimizer visitors.
plans.push(scan.input());
} else {
// The remaining built-in nodes only expose column expressions.
// External extensions own the safety of their expressions() implementation.
plan.apply_expressions(check)?;
}
}
LogicalPlan::Subquery(subquery) => {
for expr in &subquery.outer_ref_columns {
check(expr)?;
}
}
_ => {
plan.apply_expressions(check)?;
}
}
plans.extend(plan.inputs());
}
Ok(())
}
fn validate_expr(expr: &Expr, subqueries: &mut Vec<Arc<LogicalPlan>>) -> Result<()> {
let mut expressions = vec![(expr, 1)];
while let Some((expr, depth)) = expressions.pop() {
if depth > MAX_EXPR_DEPTH {
return plan_err!(
"Expression depth exceeds the maximum supported depth of {MAX_EXPR_DEPTH}"
);
}
let subquery = match expr {
Expr::Exists(exists) => Some(&exists.subquery),
Expr::ScalarSubquery(subquery) => Some(subquery),
Expr::InSubquery(in_subquery) => Some(&in_subquery.subquery),
Expr::SetComparison(comparison) => Some(&comparison.subquery),
_ => None,
};
if let Some(subquery) = subquery {
subqueries.push(Arc::clone(&subquery.subquery));
expressions.extend(subquery.outer_ref_columns.iter().map(|expr| (expr, 1)));
}
expr.apply_children(|child| {
expressions.push((child, depth + 1));
Ok(TreeNodeRecursion::Continue)
})?;
}
Ok(())
}
#[cfg(test)]
mod tests {
use arrow_schema::{DataType, Field, Schema};
use datafusion_common::DFSchema;
use datafusion_expr::expr::{Exists, InSubquery, SetComparison, SetQuantifier};
use datafusion_expr::{EmptyRelation, Filter, Operator, Subquery, lit};
use super::*;
fn filter(predicate: Expr) -> LogicalPlan {
LogicalPlan::Filter(
Filter::try_new(
predicate,
Arc::new(LogicalPlan::EmptyRelation(EmptyRelation {
produce_one_row: true,
schema: Arc::new(
DFSchema::try_from(Schema::new(vec![Field::new(
"value",
DataType::Boolean,
false,
)]))
.unwrap(),
),
})),
)
.unwrap(),
)
}
#[test]
fn test_depth_in_subqueries_and_hidden_input() {
std::thread::Builder::new()
.stack_size(2 * 1024 * 1024)
.spawn(|| {
for depth in [MAX_EXPR_DEPTH, MAX_EXPR_DEPTH + 1, 512] {
let mut predicate = lit(true);
for _ in 1..depth {
predicate = predicate.or(lit(false));
}
// MergeScan hides its input, and apply_children on an Expr
// does not visit the plan embedded in its subquery.
let plan = MergeScanLogicalPlan::new(filter(predicate), false)
.into_logical_plan();
let subquery = Subquery {
subquery: Arc::new(plan),
outer_ref_columns: vec![],
spans: Default::default(),
};
for expr in [
Expr::Exists(Exists::new(subquery.clone(), false)),
Expr::ScalarSubquery(subquery.clone()),
Expr::InSubquery(InSubquery::new(
Box::new(lit(true)),
subquery.clone(),
false,
)),
Expr::SetComparison(SetComparison::new(
Box::new(lit(true)),
subquery.clone(),
Operator::Eq,
SetQuantifier::Any,
)),
] {
let result = validate_plan(&filter(expr));
if depth <= MAX_EXPR_DEPTH {
result.unwrap();
} else {
assert!(matches!(
result,
Err(datafusion_common::DataFusionError::Plan(message))
if message == "Expression depth exceeds the maximum supported depth of 128"
));
}
}
}
})
.unwrap()
.join()
.unwrap();
}
}
+2
View File
@@ -55,6 +55,7 @@ use crate::error::{
UnimplementedSnafu,
};
use crate::log_query::planner::LogQueryPlanner;
use crate::optimizer::expr_depth;
use crate::parser::{DEFAULT_LOOKBACK_STRING, PromQuery, QueryLanguageParser, QueryStatement};
use crate::promql::planner::PromPlanner;
use crate::query_engine::{DefaultPlanDecoder, QueryEngineState};
@@ -267,6 +268,7 @@ impl DfLogicalPlanner {
.context(PlanSqlSnafu)?
};
expr_depth::validate_plan(&result)?;
common_telemetry::debug!("Logical planner, statement to plan result: {result}");
let plan = RangePlanRewriter::new(table_provider, query_ctx.clone())
.rewrite(result)
+8 -1
View File
@@ -66,6 +66,7 @@ use crate::optimizer::constant_term::MatchesConstantTermOptimizer;
use crate::optimizer::count_nest_aggr::CountNestAggrRule;
use crate::optimizer::count_wildcard::CountWildcardToTimeIndexRule;
use crate::optimizer::enforce_sorting::EnforceSorting;
use crate::optimizer::expr_depth;
use crate::optimizer::global_limit::EnsureGlobalLimitForFetch;
use crate::optimizer::json_schema_concretize::JsonSchemaConcretizeRule;
use crate::optimizer::json_type_concretize::JsonTypeConcretizeRule;
@@ -212,6 +213,7 @@ impl QueryEngineState {
analyzer.rules.push(Arc::new(JsonSchemaConcretizeRule));
}
analyzer.rules.push(Arc::new(FixStateUdafOrderingAnalyzer));
analyzer.rules.push(Arc::new(expr_depth::ExprDepthRule));
// Note: Postgres oid-alias string coercion (`'x'::regclass`,
// `'public'::regnamespace`, ...) is handled by the
@@ -311,15 +313,20 @@ impl QueryEngineState {
plan: DfLogicalPlan,
context: &QueryEngineContext,
) -> DfResult<DfLogicalPlan> {
expr_depth::validate_plan(&plan)?;
self.extension_rules
.iter()
.try_fold(plan, |acc_plan, rule| {
rule.analyze(acc_plan, context, self.session_state().config_options())
let plan =
rule.analyze(acc_plan, context, self.session_state().config_options())?;
expr_depth::validate_plan(&plan)?;
Ok(plan)
})
}
/// Run the full logical plan optimize phase for the given plan.
pub fn optimize_logical_plan(&self, plan: DfLogicalPlan) -> DfResult<DfLogicalPlan> {
expr_depth::validate_plan(&plan)?;
self.session_state().optimize(&plan)
}
+51
View File
@@ -86,6 +86,7 @@ macro_rules! sql_tests {
test_postgres_datestyle,
test_postgres_intervalstyle,
test_postgres_parameter_inference,
test_postgres_expression_depth,
test_postgres_regclass_bind_parameter,
test_postgres_uint64_parameter,
test_postgres_explain_bind_parameter,
@@ -1602,6 +1603,56 @@ pub async fn test_postgres_parameter_inference(store_type: StorageType) {
guard.remove_all().await;
}
// Regression for #9553: reject the predicate before DataFusion's simplifier
// overflows the worker stack, and keep the extended-query connection usable.
pub async fn test_postgres_expression_depth(store_type: StorageType) {
let (mut guard, server) = setup_pg_server(store_type, "test_postgres_expression_depth").await;
let addr = server.bind_addr().unwrap();
let (client, connection) = tokio_postgres::connect(&format!("postgres://{addr}/public"), NoTls)
.await
.unwrap();
let connection = tokio::spawn(connection);
client
.simple_query(
"CREATE TABLE depth_guard (ts TIMESTAMP TIME INDEX, v INT); \
INSERT INTO depth_guard VALUES (0, 42)",
)
.await
.unwrap();
let query = |terms: usize| {
let predicate = std::iter::once("CAST(v AS BIGINT) = $1".to_string())
.chain((1..terms).map(|i| format!("CAST(v AS BIGINT) = {i}")))
.collect::<Vec<_>>()
.join(" OR ");
format!("SELECT count(*) FROM depth_guard WHERE {predicate}")
};
let row = client.query_one(&query(64), &[&42_i64]).await.unwrap();
assert_eq!(row.get::<_, i64>(0), 1);
for prefix in ["", "EXPLAIN ", "EXPLAIN ANALYZE "] {
let error = client
.query(&format!("{prefix}{}", query(512)), &[&42_i64])
.await
.unwrap_err();
assert!(
error
.as_db_error()
.unwrap()
.message()
.contains("Expression depth exceeds the maximum supported depth of 128"),
"{error}"
);
let row = client.query_one("SELECT 42::BIGINT", &[]).await.unwrap();
assert_eq!(row.get::<_, i64>(0), 42);
}
drop(client);
connection.await.unwrap().unwrap();
server.shutdown().await.unwrap();
guard.remove_all().await;
}
pub async fn test_postgres_regclass_bind_parameter(store_type: StorageType) {
let (mut guard, fe_pg_server) =
setup_pg_server(store_type, "test_postgres_regclass_bind_parameter").await;
@@ -0,0 +1,24 @@
CREATE TABLE depth_guard (ts TIMESTAMP TIME INDEX, v INT);
Affected Rows: 0
INSERT INTO depth_guard VALUES (0, 42);
Affected Rows: 1
SELECT count(*) FROM depth_guard WHERE CAST(v AS BIGINT) = 0 OR CAST(v AS BIGINT) = 1 OR CAST(v AS BIGINT) = 2 OR CAST(v AS BIGINT) = 3 OR CAST(v AS BIGINT) = 4 OR CAST(v AS BIGINT) = 5 OR CAST(v AS BIGINT) = 6 OR CAST(v AS BIGINT) = 7 OR CAST(v AS BIGINT) = 8 OR CAST(v AS BIGINT) = 9 OR CAST(v AS BIGINT) = 10 OR CAST(v AS BIGINT) = 11 OR CAST(v AS BIGINT) = 12 OR CAST(v AS BIGINT) = 13 OR CAST(v AS BIGINT) = 14 OR CAST(v AS BIGINT) = 15 OR CAST(v AS BIGINT) = 16 OR CAST(v AS BIGINT) = 17 OR CAST(v AS BIGINT) = 18 OR CAST(v AS BIGINT) = 19 OR CAST(v AS BIGINT) = 20 OR CAST(v AS BIGINT) = 21 OR CAST(v AS BIGINT) = 22 OR CAST(v AS BIGINT) = 23 OR CAST(v AS BIGINT) = 24 OR CAST(v AS BIGINT) = 25 OR CAST(v AS BIGINT) = 26 OR CAST(v AS BIGINT) = 27 OR CAST(v AS BIGINT) = 28 OR CAST(v AS BIGINT) = 29 OR CAST(v AS BIGINT) = 30 OR CAST(v AS BIGINT) = 31 OR CAST(v AS BIGINT) = 32 OR CAST(v AS BIGINT) = 33 OR CAST(v AS BIGINT) = 34 OR CAST(v AS BIGINT) = 35 OR CAST(v AS BIGINT) = 36 OR CAST(v AS BIGINT) = 37 OR CAST(v AS BIGINT) = 38 OR CAST(v AS BIGINT) = 39 OR CAST(v AS BIGINT) = 40 OR CAST(v AS BIGINT) = 41 OR CAST(v AS BIGINT) = 42 OR CAST(v AS BIGINT) = 43 OR CAST(v AS BIGINT) = 44 OR CAST(v AS BIGINT) = 45 OR CAST(v AS BIGINT) = 46 OR CAST(v AS BIGINT) = 47 OR CAST(v AS BIGINT) = 48 OR CAST(v AS BIGINT) = 49 OR CAST(v AS BIGINT) = 50 OR CAST(v AS BIGINT) = 51 OR CAST(v AS BIGINT) = 52 OR CAST(v AS BIGINT) = 53 OR CAST(v AS BIGINT) = 54 OR CAST(v AS BIGINT) = 55 OR CAST(v AS BIGINT) = 56 OR CAST(v AS BIGINT) = 57 OR CAST(v AS BIGINT) = 58 OR CAST(v AS BIGINT) = 59 OR CAST(v AS BIGINT) = 60 OR CAST(v AS BIGINT) = 61 OR CAST(v AS BIGINT) = 62 OR CAST(v AS BIGINT) = 63 OR CAST(v AS BIGINT) = 64 OR CAST(v AS BIGINT) = 65 OR CAST(v AS BIGINT) = 66 OR CAST(v AS BIGINT) = 67 OR CAST(v AS BIGINT) = 68 OR CAST(v AS BIGINT) = 69 OR CAST(v AS BIGINT) = 70 OR CAST(v AS BIGINT) = 71 OR CAST(v AS BIGINT) = 72 OR CAST(v AS BIGINT) = 73 OR CAST(v AS BIGINT) = 74 OR CAST(v AS BIGINT) = 75 OR CAST(v AS BIGINT) = 76 OR CAST(v AS BIGINT) = 77 OR CAST(v AS BIGINT) = 78 OR CAST(v AS BIGINT) = 79 OR CAST(v AS BIGINT) = 80 OR CAST(v AS BIGINT) = 81 OR CAST(v AS BIGINT) = 82 OR CAST(v AS BIGINT) = 83 OR CAST(v AS BIGINT) = 84 OR CAST(v AS BIGINT) = 85 OR CAST(v AS BIGINT) = 86 OR CAST(v AS BIGINT) = 87 OR CAST(v AS BIGINT) = 88 OR CAST(v AS BIGINT) = 89 OR CAST(v AS BIGINT) = 90 OR CAST(v AS BIGINT) = 91 OR CAST(v AS BIGINT) = 92 OR CAST(v AS BIGINT) = 93 OR CAST(v AS BIGINT) = 94 OR CAST(v AS BIGINT) = 95 OR CAST(v AS BIGINT) = 96 OR CAST(v AS BIGINT) = 97 OR CAST(v AS BIGINT) = 98 OR CAST(v AS BIGINT) = 99 OR CAST(v AS BIGINT) = 100 OR CAST(v AS BIGINT) = 101 OR CAST(v AS BIGINT) = 102 OR CAST(v AS BIGINT) = 103 OR CAST(v AS BIGINT) = 104 OR CAST(v AS BIGINT) = 105 OR CAST(v AS BIGINT) = 106 OR CAST(v AS BIGINT) = 107 OR CAST(v AS BIGINT) = 108 OR CAST(v AS BIGINT) = 109 OR CAST(v AS BIGINT) = 110 OR CAST(v AS BIGINT) = 111 OR CAST(v AS BIGINT) = 112 OR CAST(v AS BIGINT) = 113 OR CAST(v AS BIGINT) = 114 OR CAST(v AS BIGINT) = 115 OR CAST(v AS BIGINT) = 116 OR CAST(v AS BIGINT) = 117 OR CAST(v AS BIGINT) = 118 OR CAST(v AS BIGINT) = 119 OR CAST(v AS BIGINT) = 120 OR CAST(v AS BIGINT) = 121 OR CAST(v AS BIGINT) = 122 OR CAST(v AS BIGINT) = 123 OR CAST(v AS BIGINT) = 124 OR CAST(v AS BIGINT) = 125 OR CAST(v AS BIGINT) = 126 OR CAST(v AS BIGINT) = 127 OR CAST(v AS BIGINT) = 128 OR CAST(v AS BIGINT) = 129;
Error: 3000(PlanQuery), Error during planning: Expression depth exceeds the maximum supported depth of 128
SELECT count(*) FROM depth_guard WHERE v = 42;
+----------+
| count(*) |
+----------+
| 1 |
+----------+
DROP TABLE depth_guard;
Affected Rows: 0
@@ -0,0 +1,9 @@
CREATE TABLE depth_guard (ts TIMESTAMP TIME INDEX, v INT);
INSERT INTO depth_guard VALUES (0, 42);
SELECT count(*) FROM depth_guard WHERE CAST(v AS BIGINT) = 0 OR CAST(v AS BIGINT) = 1 OR CAST(v AS BIGINT) = 2 OR CAST(v AS BIGINT) = 3 OR CAST(v AS BIGINT) = 4 OR CAST(v AS BIGINT) = 5 OR CAST(v AS BIGINT) = 6 OR CAST(v AS BIGINT) = 7 OR CAST(v AS BIGINT) = 8 OR CAST(v AS BIGINT) = 9 OR CAST(v AS BIGINT) = 10 OR CAST(v AS BIGINT) = 11 OR CAST(v AS BIGINT) = 12 OR CAST(v AS BIGINT) = 13 OR CAST(v AS BIGINT) = 14 OR CAST(v AS BIGINT) = 15 OR CAST(v AS BIGINT) = 16 OR CAST(v AS BIGINT) = 17 OR CAST(v AS BIGINT) = 18 OR CAST(v AS BIGINT) = 19 OR CAST(v AS BIGINT) = 20 OR CAST(v AS BIGINT) = 21 OR CAST(v AS BIGINT) = 22 OR CAST(v AS BIGINT) = 23 OR CAST(v AS BIGINT) = 24 OR CAST(v AS BIGINT) = 25 OR CAST(v AS BIGINT) = 26 OR CAST(v AS BIGINT) = 27 OR CAST(v AS BIGINT) = 28 OR CAST(v AS BIGINT) = 29 OR CAST(v AS BIGINT) = 30 OR CAST(v AS BIGINT) = 31 OR CAST(v AS BIGINT) = 32 OR CAST(v AS BIGINT) = 33 OR CAST(v AS BIGINT) = 34 OR CAST(v AS BIGINT) = 35 OR CAST(v AS BIGINT) = 36 OR CAST(v AS BIGINT) = 37 OR CAST(v AS BIGINT) = 38 OR CAST(v AS BIGINT) = 39 OR CAST(v AS BIGINT) = 40 OR CAST(v AS BIGINT) = 41 OR CAST(v AS BIGINT) = 42 OR CAST(v AS BIGINT) = 43 OR CAST(v AS BIGINT) = 44 OR CAST(v AS BIGINT) = 45 OR CAST(v AS BIGINT) = 46 OR CAST(v AS BIGINT) = 47 OR CAST(v AS BIGINT) = 48 OR CAST(v AS BIGINT) = 49 OR CAST(v AS BIGINT) = 50 OR CAST(v AS BIGINT) = 51 OR CAST(v AS BIGINT) = 52 OR CAST(v AS BIGINT) = 53 OR CAST(v AS BIGINT) = 54 OR CAST(v AS BIGINT) = 55 OR CAST(v AS BIGINT) = 56 OR CAST(v AS BIGINT) = 57 OR CAST(v AS BIGINT) = 58 OR CAST(v AS BIGINT) = 59 OR CAST(v AS BIGINT) = 60 OR CAST(v AS BIGINT) = 61 OR CAST(v AS BIGINT) = 62 OR CAST(v AS BIGINT) = 63 OR CAST(v AS BIGINT) = 64 OR CAST(v AS BIGINT) = 65 OR CAST(v AS BIGINT) = 66 OR CAST(v AS BIGINT) = 67 OR CAST(v AS BIGINT) = 68 OR CAST(v AS BIGINT) = 69 OR CAST(v AS BIGINT) = 70 OR CAST(v AS BIGINT) = 71 OR CAST(v AS BIGINT) = 72 OR CAST(v AS BIGINT) = 73 OR CAST(v AS BIGINT) = 74 OR CAST(v AS BIGINT) = 75 OR CAST(v AS BIGINT) = 76 OR CAST(v AS BIGINT) = 77 OR CAST(v AS BIGINT) = 78 OR CAST(v AS BIGINT) = 79 OR CAST(v AS BIGINT) = 80 OR CAST(v AS BIGINT) = 81 OR CAST(v AS BIGINT) = 82 OR CAST(v AS BIGINT) = 83 OR CAST(v AS BIGINT) = 84 OR CAST(v AS BIGINT) = 85 OR CAST(v AS BIGINT) = 86 OR CAST(v AS BIGINT) = 87 OR CAST(v AS BIGINT) = 88 OR CAST(v AS BIGINT) = 89 OR CAST(v AS BIGINT) = 90 OR CAST(v AS BIGINT) = 91 OR CAST(v AS BIGINT) = 92 OR CAST(v AS BIGINT) = 93 OR CAST(v AS BIGINT) = 94 OR CAST(v AS BIGINT) = 95 OR CAST(v AS BIGINT) = 96 OR CAST(v AS BIGINT) = 97 OR CAST(v AS BIGINT) = 98 OR CAST(v AS BIGINT) = 99 OR CAST(v AS BIGINT) = 100 OR CAST(v AS BIGINT) = 101 OR CAST(v AS BIGINT) = 102 OR CAST(v AS BIGINT) = 103 OR CAST(v AS BIGINT) = 104 OR CAST(v AS BIGINT) = 105 OR CAST(v AS BIGINT) = 106 OR CAST(v AS BIGINT) = 107 OR CAST(v AS BIGINT) = 108 OR CAST(v AS BIGINT) = 109 OR CAST(v AS BIGINT) = 110 OR CAST(v AS BIGINT) = 111 OR CAST(v AS BIGINT) = 112 OR CAST(v AS BIGINT) = 113 OR CAST(v AS BIGINT) = 114 OR CAST(v AS BIGINT) = 115 OR CAST(v AS BIGINT) = 116 OR CAST(v AS BIGINT) = 117 OR CAST(v AS BIGINT) = 118 OR CAST(v AS BIGINT) = 119 OR CAST(v AS BIGINT) = 120 OR CAST(v AS BIGINT) = 121 OR CAST(v AS BIGINT) = 122 OR CAST(v AS BIGINT) = 123 OR CAST(v AS BIGINT) = 124 OR CAST(v AS BIGINT) = 125 OR CAST(v AS BIGINT) = 126 OR CAST(v AS BIGINT) = 127 OR CAST(v AS BIGINT) = 128 OR CAST(v AS BIGINT) = 129;
SELECT count(*) FROM depth_guard WHERE v = 42;
DROP TABLE depth_guard;
@@ -102,6 +102,7 @@ TQL EXPLAIN VERBOSE (0, 10, '5s') test;
|_| ]]_|
| logical_plan after JsonSchemaConcretizeRule_| SAME TEXT AS ABOVE_|
| logical_plan after FixStateUdafOrderingAnalyzer_| SAME TEXT AS ABOVE_|
| logical_plan after expr_depth_| SAME TEXT AS ABOVE_|
| analyzed_logical_plan_| SAME TEXT AS ABOVE_|
| logical_plan after rewrite_set_comparison_| SAME TEXT AS ABOVE_|
| logical_plan after optimize_unions_| SAME TEXT AS ABOVE_|
@@ -256,6 +257,7 @@ TQL EXPLAIN VERBOSE (0, 10, '5s') test AS series;
|_| ]]_|
| logical_plan after JsonSchemaConcretizeRule_| SAME TEXT AS ABOVE_|
| logical_plan after FixStateUdafOrderingAnalyzer_| SAME TEXT AS ABOVE_|
| logical_plan after expr_depth_| SAME TEXT AS ABOVE_|
| analyzed_logical_plan_| SAME TEXT AS ABOVE_|
| logical_plan after rewrite_set_comparison_| SAME TEXT AS ABOVE_|
| logical_plan after optimize_unions_| SAME TEXT AS ABOVE_|
@@ -442,6 +444,7 @@ TQL EXPLAIN VERBOSE (0, 10, '5s') test_nano;
|_| ]]_|
| logical_plan after JsonSchemaConcretizeRule_| SAME TEXT AS ABOVE_|
| logical_plan after FixStateUdafOrderingAnalyzer_| SAME TEXT AS ABOVE_|
| logical_plan after expr_depth_| SAME TEXT AS ABOVE_|
| analyzed_logical_plan_| SAME TEXT AS ABOVE_|
| logical_plan after rewrite_set_comparison_| SAME TEXT AS ABOVE_|
| logical_plan after optimize_unions_| SAME TEXT AS ABOVE_|