From f36c5ad744faf9ae160ad4d5616bc466aec680ca Mon Sep 17 00:00:00 2001 From: Ruihang Xia Date: Fri, 5 Jun 2026 11:20:31 +0800 Subject: [PATCH] correct behavior Signed-off-by: Ruihang Xia --- src/query/src/promql/planner.rs | 165 +++++++++++++++--- .../standalone/common/promql/label.result | 35 +--- 2 files changed, 142 insertions(+), 58 deletions(-) diff --git a/src/query/src/promql/planner.rs b/src/query/src/promql/planner.rs index 6b520a280c..f361d4d51e 100644 --- a/src/query/src/promql/planner.rs +++ b/src/query/src/promql/planner.rs @@ -876,14 +876,14 @@ impl PromPlanner { ) -> Result { let only_join_time_index = first_leaf.ctx.tag_columns.is_empty() || right_leaf.ctx.tag_columns.is_empty(); - let (mut left_keys, mut right_keys) = self.binary_join_key_columns( + let (mut left_keys, mut right_keys, force_empty_join) = self.binary_join_key_columns( left.schema(), right_leaf.plan.schema(), &first_leaf.ctx, &right_leaf.ctx, only_join_time_index, &None, - ); + )?; if let (Some(left_time_index_column), Some(right_time_index_column)) = ( first_leaf.ctx.time_index_column.clone(), @@ -907,7 +907,7 @@ impl PromPlanner { .map(|name| Column::new(Some(right_leaf.alias.clone()), name)) .collect::>(), ), - None, + force_empty_join.then_some(lit(false)), NullEquality::NullEqualsNull, ) .context(DataFusionPlanningSnafu)? @@ -1250,6 +1250,14 @@ impl PromPlanner { } let (output_field_columns, field_columns) = Self::align_binary_field_columns(&left_field_columns, &right_field_columns); + let left_aligned_field_columns = field_columns + .iter() + .map(|(left_col_name, _)| (*left_col_name).clone()) + .collect::>(); + let right_aligned_field_columns = field_columns + .iter() + .map(|(_, right_col_name)| (*right_col_name).clone()) + .collect::>(); // PromQL binary arithmetic only combines the shared prefix of value columns. // Keep the output field count aligned with that zipped prefix so planning // remains stable even when the two sides have uneven multi-field schemas. @@ -1302,14 +1310,21 @@ impl PromPlanner { // So we filter on the join result and then project only the side that should // be preserved according to PromQL semantics. let filtered = self.filter_on_field_column(join_plan, bin_expr_builder)?; - let (project_table_ref, project_context) = + let (project_table_ref, mut project_context, project_field_columns) = match (lhs.value_type(), rhs.value_type()) { - (ValueType::Scalar, ValueType::Vector) => { - (&right_table_ref, &right_context) - } - _ => (&left_table_ref, &left_context), + (ValueType::Scalar, ValueType::Vector) => ( + &right_table_ref, + right_context.clone(), + right_aligned_field_columns, + ), + _ => ( + &left_table_ref, + left_context.clone(), + left_aligned_field_columns, + ), }; - self.project_binary_join_side(filtered, project_table_ref, project_context) + project_context.field_columns = project_field_columns; + self.project_binary_join_side(filtered, project_table_ref, &project_context) } else { self.projection_for_each_field_column(join_plan, bin_expr_builder) } @@ -3994,7 +4009,7 @@ impl PromPlanner { right_context: &PromPlannerContext, only_join_time_index: bool, modifier: &Option, - ) -> (BTreeSet, BTreeSet) { + ) -> Result<(BTreeSet, BTreeSet, bool)> { let has_tsid = |schema: &DFSchemaRef| { schema .fields() @@ -4014,16 +4029,22 @@ impl PromPlanner { BTreeSet::from([DATA_SCHEMA_TSID_COLUMN_NAME.to_string()]), ) } else { - let tag_columns = if only_join_time_index { - BTreeSet::new() + if only_join_time_index { + (BTreeSet::new(), BTreeSet::new()) } else { - self.ctx - .tag_columns - .iter() - .cloned() - .collect::>() - }; - (tag_columns.clone(), tag_columns) + ( + left_context + .tag_columns + .iter() + .cloned() + .collect::>(), + right_context + .tag_columns + .iter() + .cloned() + .collect::>(), + ) + } }; if !use_tsid_join @@ -4045,7 +4066,18 @@ impl PromPlanner { } } - (left_tag_columns, right_tag_columns) + let force_empty_join = + !use_tsid_join && !only_join_time_index && left_tag_columns != right_tag_columns; + if force_empty_join { + let common_tag_columns = left_tag_columns + .intersection(&right_tag_columns) + .cloned() + .collect::>(); + left_tag_columns = common_tag_columns.clone(); + right_tag_columns = common_tag_columns; + } + + Ok((left_tag_columns, right_tag_columns, force_empty_join)) } fn binary_modifier_preserves_tsid_join_key( @@ -4102,14 +4134,15 @@ impl PromPlanner { left_context: &PromPlannerContext, right_context: &PromPlannerContext, ) -> Result { - let (mut left_tag_columns, mut right_tag_columns) = self.binary_join_key_columns( - left.schema(), - right.schema(), - left_context, - right_context, - only_join_time_index, - modifier, - ); + let (mut left_tag_columns, mut right_tag_columns, force_empty_join) = self + .binary_join_key_columns( + left.schema(), + right.schema(), + left_context, + right_context, + only_join_time_index, + modifier, + )?; // push time index column if it exists if let (Some(left_time_index_column), Some(right_time_index_column)) = @@ -4142,7 +4175,7 @@ impl PromPlanner { .map(Column::from_name) .collect::>(), ), - None, + force_empty_join.then_some(lit(false)), NullEquality::NullEqualsNull, ) .context(DataFusionPlanningSnafu)? @@ -5543,6 +5576,35 @@ mod test { } } + #[tokio::test] + async fn timestamp_binary_join_rejects_default_matching_on_mismatched_labels() { + let eval_stmt = build_eval_stmt("timestamp(left_host_job) / right_by_job"); + + let table_provider = build_test_table_provider_with_tsid_tag_fields(&[ + ( + (DEFAULT_SCHEMA_NAME.to_string(), "left_host_job".to_string()), + 2, + 1, + ), + ( + (DEFAULT_SCHEMA_NAME.to_string(), "right_by_job".to_string()), + 1, + 1, + ), + ]) + .await; + let plan = + PromPlanner::stmt_to_plan(table_provider, &eval_stmt, &build_query_engine_state()) + .await + .unwrap(); + let plan_str = plan.display_indent_schema().to_string(); + + assert!( + plan_str.contains("Boolean(false)") || plan_str.contains("false"), + "{plan_str}" + ); + } + #[tokio::test] async fn tsid_is_preserved_for_nested_default_binary_joins() { let eval_stmt = build_eval_stmt("(some_metric - some_alt_metric) / some_third_metric"); @@ -5966,6 +6028,51 @@ mod test { assert_eq!(value_columns, 1, "{field_names:?}"); } + #[tokio::test] + async fn comparison_binary_join_uses_shorter_field_side() { + let eval_stmt = build_eval_stmt("two_field_metric > one_field_metric"); + + let table_provider = build_test_table_provider_with_tsid_fields( + &[ + ( + ( + DEFAULT_SCHEMA_NAME.to_string(), + "two_field_metric".to_string(), + ), + 2, + ), + ( + ( + DEFAULT_SCHEMA_NAME.to_string(), + "one_field_metric".to_string(), + ), + 1, + ), + ], + 1, + ) + .await; + let plan = + PromPlanner::stmt_to_plan(table_provider, &eval_stmt, &build_query_engine_state()) + .await + .unwrap(); + + let field_names = plan + .schema() + .fields() + .iter() + .map(|field| field.name().clone()) + .collect::>(); + assert!( + field_names.iter().any(|name| name == "field_0"), + "{field_names:?}" + ); + assert!( + !field_names.iter().any(|name| name == "field_1"), + "{field_names:?}" + ); + } + #[tokio::test] async fn label_matching_modifier_disables_tsid_binary_join() { let eval_stmt = build_eval_stmt("some_metric / ignoring(tag_0) some_alt_metric"); diff --git a/tests/cases/standalone/common/promql/label.result b/tests/cases/standalone/common/promql/label.result index 1bdc37b5aa..f83bbd38e6 100644 --- a/tests/cases/standalone/common/promql/label.result +++ b/tests/cases/standalone/common/promql/label.result @@ -202,38 +202,14 @@ TQL EVAL(0, 15, '5s') label_replace(vector(1), "host", "host1", "", ""); -- SQLNESS SORT_RESULT 3 1 TQL EVAL(0, 15, '5s') {__name__="test",host="host1"} * label_replace(vector(1), "host", "host1", "", ""); -+-------+---------------------+----------------------------+ -| host | time | test.val * .greptime_value | -+-------+---------------------+----------------------------+ -| host1 | 1970-01-01T00:00:00 | 1.0 | -| host1 | 1970-01-01T00:00:05 | 1.0 | -| host1 | 1970-01-01T00:00:05 | 3.0 | -| host1 | 1970-01-01T00:00:10 | 1.0 | -| host1 | 1970-01-01T00:00:10 | 3.0 | -| host1 | 1970-01-01T00:00:10 | 5.0 | -| host1 | 1970-01-01T00:00:15 | 1.0 | -| host1 | 1970-01-01T00:00:15 | 3.0 | -| host1 | 1970-01-01T00:00:15 | 5.0 | -| host1 | 1970-01-01T00:00:15 | 7.0 | -+-------+---------------------+----------------------------+ +++ +++ -- SQLNESS SORT_RESULT 3 1 TQL EVAL(0, 15, '5s') {__name__="test",host="host1"} + label_replace(vector(1), "host", "host1", "", ""); -+-------+---------------------+----------------------------+ -| host | time | test.val + .greptime_value | -+-------+---------------------+----------------------------+ -| host1 | 1970-01-01T00:00:00 | 2.0 | -| host1 | 1970-01-01T00:00:05 | 2.0 | -| host1 | 1970-01-01T00:00:05 | 4.0 | -| host1 | 1970-01-01T00:00:10 | 2.0 | -| host1 | 1970-01-01T00:00:10 | 4.0 | -| host1 | 1970-01-01T00:00:10 | 6.0 | -| host1 | 1970-01-01T00:00:15 | 2.0 | -| host1 | 1970-01-01T00:00:15 | 4.0 | -| host1 | 1970-01-01T00:00:15 | 6.0 | -| host1 | 1970-01-01T00:00:15 | 8.0 | -+-------+---------------------+----------------------------+ +++ +++ -- Empty regex with existing source label -- SQLNESS SORT_RESULT 3 1 @@ -303,7 +279,8 @@ TQL EVAL(0, 15, '5s') {__name__="test",host="host1"} * label_replace(vector(1), -- SQLNESS SORT_RESULT 3 1 TQL EVAL(0, 15, '5s') {__name__="test",host="host1"} * label_replace(vector(1), "addr", "host1", "instance", ""); -Error: 3000(PlanQuery), Internal error during building DataFusion plan: No field named addr. Valid fields are test.ts, test.host, test.idc, test.val. +++ +++ TQL EVAL label_replace(demo_num_cpus, "~invalid", "", "src", "(.*)");