diff --git a/src/query/src/optimizer.rs b/src/query/src/optimizer.rs index bdc12ad8fed..62894a0862b 100644 --- a/src/query/src/optimizer.rs +++ b/src/query/src/optimizer.rs @@ -23,6 +23,7 @@ pub(crate) mod infer_aggregate_input_bounds; pub(crate) mod insert_assignment; pub(crate) mod json_schema_concretize; pub(crate) mod json_type_concretize; +pub(crate) mod merge_scan_projection; pub mod parallelize_scan; pub mod promql_tsid_narrow_join; pub mod remove_duplicate; diff --git a/src/query/src/optimizer/merge_scan_projection.rs b/src/query/src/optimizer/merge_scan_projection.rs new file mode 100644 index 00000000000..12a3a534df6 --- /dev/null +++ b/src/query/src/optimizer/merge_scan_projection.rs @@ -0,0 +1,141 @@ +// 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 datafusion_common::Result; +use datafusion_common::tree_node::Transformed; +use datafusion_expr::{Expr, LogicalPlan, LogicalPlanBuilder}; +use datafusion_optimizer::optimizer::ApplyOrder; +use datafusion_optimizer::{OptimizerConfig, OptimizerRule}; + +use crate::dist_plan::MergeScanLogicalPlan; + +#[derive(Debug)] +pub(crate) struct MergeScanProjection; + +impl OptimizerRule for MergeScanProjection { + fn name(&self) -> &str { + "merge_scan_projection" + } + + fn apply_order(&self) -> Option { + Some(ApplyOrder::BottomUp) + } + + fn rewrite( + &self, + plan: LogicalPlan, + _config: &dyn OptimizerConfig, + ) -> Result> { + let LogicalPlan::Projection(projection) = &plan else { + return Ok(Transformed::no(plan)); + }; + // Remote batches do not yet preserve the row count of empty projections. + if projection.expr.is_empty() + || !projection + .expr + .iter() + .all(|expr| matches!(expr, Expr::Column(_))) + { + return Ok(Transformed::no(plan)); + } + let LogicalPlan::Extension(extension) = projection.input.as_ref() else { + return Ok(Transformed::no(plan)); + }; + let Some(scan) = extension + .node + .as_any() + .downcast_ref::() + else { + return Ok(Transformed::no(plan)); + }; + if scan.is_placeholder() { + return Ok(Transformed::no(plan)); + } + // Project after the remote subtree, preserving its sort, limit, and aggregation scope. + let remote = LogicalPlanBuilder::from(scan.input().clone()) + .project(projection.expr.clone())? + .build()?; + let mut projected = + MergeScanLogicalPlan::new(remote, false).with_output_schema(projection.schema.clone()); + if let Some(id) = scan.remote_dyn_filter_producer_id() { + projected = projected.with_remote_dyn_filter_producer_id(id); + } + Ok(Transformed::yes(projected.into_logical_plan())) + } +} + +#[cfg(test)] +mod tests { + use datafusion_expr::{col, lit}; + use datafusion_optimizer::optimizer::OptimizerContext; + + use super::*; + + #[test] + fn projects_remote_columns_without_changing_limit_scope() { + let input = LogicalPlanBuilder::values(vec![vec![lit(1), lit(2)]]) + .unwrap() + .limit(0, Some(1)) + .unwrap() + .build() + .unwrap(); + let producer = crate::dist_plan::RemoteDynFilterProducerId::new(42); + let scan = MergeScanLogicalPlan::new(input.clone(), false) + .with_remote_dyn_filter_producer_id(producer); + let plan = LogicalPlanBuilder::from(scan.into_logical_plan()) + .project(vec![col("column2")]) + .unwrap() + .build() + .unwrap(); + let schema = plan.schema().clone(); + let result = MergeScanProjection + .rewrite(plan, &OptimizerContext::new()) + .unwrap(); + assert!(result.transformed); + assert_eq!(result.data.schema(), &schema); + let LogicalPlan::Extension(extension) = result.data else { + panic!("expected MergeScan") + }; + let scan = extension + .node + .as_any() + .downcast_ref::() + .unwrap(); + assert_eq!(scan.remote_dyn_filter_producer_id(), Some(producer)); + let LogicalPlan::Projection(projection) = scan.input() else { + panic!("expected remote projection") + }; + assert_eq!(projection.expr, vec![col("column2")]); + assert_eq!(*projection.input, input); + } + + #[test] + fn keeps_computed_expressions_local() { + let input = LogicalPlanBuilder::values(vec![vec![lit(1), lit(2)]]) + .unwrap() + .build() + .unwrap(); + let plan = + LogicalPlanBuilder::from(MergeScanLogicalPlan::new(input, false).into_logical_plan()) + .project(vec![col("column2") + lit(1)]) + .unwrap() + .build() + .unwrap(); + let result = MergeScanProjection + .rewrite(plan.clone(), &OptimizerContext::new()) + .unwrap(); + assert!(!result.transformed); + assert_eq!(result.data, plan); + } +} diff --git a/src/query/src/query_engine/state.rs b/src/query/src/query_engine/state.rs index bae7f7fb8ba..f93ce49f7ee 100644 --- a/src/query/src/query_engine/state.rs +++ b/src/query/src/query_engine/state.rs @@ -73,6 +73,7 @@ use crate::optimizer::global_limit::EnsureGlobalLimitForFetch; use crate::optimizer::infer_aggregate_input_bounds::InferAggregateInputBoundsRule; use crate::optimizer::json_schema_concretize::JsonSchemaConcretizeRule; use crate::optimizer::json_type_concretize::JsonTypeConcretizeRule; +use crate::optimizer::merge_scan_projection::MergeScanProjection; use crate::optimizer::parallelize_scan::ParallelizeScan; use crate::optimizer::promql_tsid_narrow_join::PromqlTsidNarrowJoin; use crate::optimizer::remove_duplicate::RemoveDuplicate; @@ -246,6 +247,9 @@ impl QueryEngineState { filter_pushdown_index, Arc::new(InferAggregateInputBoundsRule), ); + if with_dist_planner { + optimizer.rules.push(Arc::new(MergeScanProjection)); + } optimizer.rules.push(Arc::new(ScanHintRule)); optimizer.rules.push(Arc::new(JsonTypeConcretizeRule)); diff --git a/tests/cases/distributed/explain/step_aggr_advance.result b/tests/cases/distributed/explain/step_aggr_advance.result index 4186cb18c3a..95f71de8ae1 100644 --- a/tests/cases/distributed/explain/step_aggr_advance.result +++ b/tests/cases/distributed/explain/step_aggr_advance.result @@ -1084,9 +1084,9 @@ EXPLAIN SELECT COUNT(DISTINCT val_col_1) FROM step_aggr_extended; | logical_plan_| Projection: count(alias1) AS count(DISTINCT step_aggr_extended.val_col_1)_| |_|_Aggregate: groupBy=[[]], aggr=[[count(alias1)]]_| |_|_Aggregate: groupBy=[[step_aggr_extended.val_col_1 AS alias1]], aggr=[[]]_| -|_|_Projection: step_aggr_extended.val_col_1_| |_|_MergeScan [is_placeholder=false, remote_input=[_| -|_| TableScan: step_aggr_extended_| +|_| Projection: step_aggr_extended.val_col_1_| +|_|_TableScan: step_aggr_extended_| |_| ]]_| | physical_plan | ProjectionExec: expr=[count(alias1)@0 as count(DISTINCT step_aggr_extended.val_col_1)]_| |_|_AggregateExec: mode=Final, gby=[], aggr=[count(alias1)]_| @@ -1096,7 +1096,6 @@ EXPLAIN SELECT COUNT(DISTINCT val_col_1) FROM step_aggr_extended; |_|_RepartitionExec: partitioning=REDACTED |_|_AggregateExec: mode=Partial, gby=[val_col_1@0 as alias1], aggr=[]_| |_|_RepartitionExec: partitioning=REDACTED -|_|_ProjectionExec: expr=[val_col_1@2 as val_col_1]_| |_|_MergeScanExec: REDACTED |_|_| +-+-+ diff --git a/tests/cases/distributed/explain/subqueries.result b/tests/cases/distributed/explain/subqueries.result index 3d3f24777db..657b7cb3e30 100644 --- a/tests/cases/distributed/explain/subqueries.result +++ b/tests/cases/distributed/explain/subqueries.result @@ -89,9 +89,9 @@ order by t.i desc; | logical_plan_| Sort: t.i DESC NULLS FIRST_| |_|_SubqueryAlias: t_| |_|_Cross Join:_| -|_|_Projection: integers.i_| |_|_MergeScan [is_placeholder=false, remote_input=[_| -|_| Filter: integers.i IS NOT NULL_| +|_| Projection: integers.i_| +|_|_Filter: integers.i IS NOT NULL_| |_|_TableScan: integers, partial_filters=[integers.i IS NOT NULL, Boolean(true)] | |_| ]]_| |_|_Projection:_| @@ -101,7 +101,6 @@ order by t.i desc; | physical_plan | SortPreservingMergeExec: [i@0 DESC]_| |_|_SortExec: expr=[i@0 DESC], preserve_partitioning=[true]_| |_|_CrossJoinExec_| -|_|_ProjectionExec: expr=[i@0 as i]_| |_|_MergeScanExec: REDACTED |_|_RepartitionExec: partitioning=REDACTED |_|_ProjectionExec: expr=[]_| @@ -129,9 +128,9 @@ EXPLAIN INSERT INTO other SELECT i, 2 FROM integers WHERE i=(SELECT MAX(i) FROM | | Aggregate: groupBy=[[]], aggr=[[max(integers.i)]] | | | TableScan: integers | | | ]] | -| | Projection: integers.i | -| | MergeScan [is_placeholder=false, remote_input=[ | -| | TableScan: integers | +| | MergeScan [is_placeholder=false, remote_input=[ | +| | Projection: integers.i | +| | TableScan: integers | | | ]] | | physical_plan_error | This feature is not implemented: Insert into not implemented for this table | +---------------------+-----------------------------------------------------------------------------+ @@ -203,9 +202,9 @@ EXPLAIN SELECT * FROM integers i1 WHERE EXISTS(SELECT count(i) FROM integers WHE |_|_Filter: integers.i IS NOT NULL_| |_|_TableScan: integers, partial_filters=[integers.i IS NOT NULL]_| |_| ]]_| -|_|_Projection: __correlated_sq_1.i_| |_|_MergeScan [is_placeholder=false, remote_input=[_| -|_| SubqueryAlias: __correlated_sq_1_| +|_| Projection: __correlated_sq_1.i_| +|_|_SubqueryAlias: __correlated_sq_1_| |_|_Projection: count(integers.i), integers.i_| |_|_Aggregate: groupBy=[[integers.i]], aggr=[[count(integers.i)]]_| |_|_Filter: integers.i IS NOT NULL_| @@ -217,7 +216,6 @@ EXPLAIN SELECT * FROM integers i1 WHERE EXISTS(SELECT count(i) FROM integers WHE |_|_RepartitionExec: partitioning=REDACTED |_|_MergeScanExec: REDACTED |_|_RepartitionExec: partitioning=REDACTED -|_|_ProjectionExec: expr=[i@1 as i]_| |_|_MergeScanExec: REDACTED |_|_| +-+-+ @@ -773,9 +771,9 @@ EXPLAIN SELECT (SELECT SUM(v) FROM nested_scalar_part WHERE v > (SELECT AVG(v) F |_| Aggregate: groupBy=[[]], aggr=[[__avg_state(CAST(nested_scalar_part.v AS Float64))]]_| |_|_TableScan: nested_scalar_part_| |_| ]]_| -|_|_Projection: nested_scalar_part.v_| |_|_MergeScan [is_placeholder=false, remote_input=[_| -|_| TableScan: nested_scalar_part_| +|_| Projection: nested_scalar_part.v_| +|_|_TableScan: nested_scalar_part_| |_| ]]_| |_|_Projection:_| |_|_Limit: skip=0, fetch=1_| @@ -795,7 +793,6 @@ EXPLAIN SELECT (SELECT SUM(v) FROM nested_scalar_part WHERE v > (SELECT AVG(v) F |_|_AggregateExec: mode=Partial, gby=[], aggr=[sum(nested_scalar_part.v)]_| |_|_FilterExec: CAST(v@0 AS Float64) > scalar_subquery()_| |_|_RepartitionExec: partitioning=REDACTED -|_|_ProjectionExec: expr=[v@1 as v]_| |_|_MergeScanExec: REDACTED |_|_AggregateExec: mode=Final, gby=[], aggr=[__avg_merge(__avg_state(nested_scalar_part.v)) as avg(nested_scalar_part.v)]_| |_|_CoalescePartitionsExec_| diff --git a/tests/cases/distributed/optimizer/pass_distribution_partition_subset_join.result b/tests/cases/distributed/optimizer/pass_distribution_partition_subset_join.result index 91c0e3e851d..7b2552b70e7 100644 --- a/tests/cases/distributed/optimizer/pass_distribution_partition_subset_join.result +++ b/tests/cases/distributed/optimizer/pass_distribution_partition_subset_join.result @@ -71,27 +71,25 @@ ORDER BY l.area, r.area; | logical_plan | Sort: l.area ASC NULLS LAST, r.area ASC NULLS LAST | | | Projection: l.area, r.area | | | Inner Join: l.device_id = r.device_id | -| | Projection: l.device_id, l.area | -| | MergeScan [is_placeholder=false, remote_input=[ | -| | SubqueryAlias: l | -| | Filter: pd_sub_l.device_id IS NOT NULL | -| | TableScan: pd_sub_l, partial_filters=[pd_sub_l.device_id IS NOT NULL] | +| | MergeScan [is_placeholder=false, remote_input=[ | +| | Projection: l.device_id, l.area | +| | SubqueryAlias: l | +| | Filter: pd_sub_l.device_id IS NOT NULL | +| | TableScan: pd_sub_l, partial_filters=[pd_sub_l.device_id IS NOT NULL] | | | ]] | -| | Projection: r.device_id, r.area | -| | MergeScan [is_placeholder=false, remote_input=[ | -| | SubqueryAlias: r | -| | Filter: pd_sub_r.device_id IS NOT NULL | -| | TableScan: pd_sub_r, partial_filters=[pd_sub_r.device_id IS NOT NULL] | +| | MergeScan [is_placeholder=false, remote_input=[ | +| | Projection: r.device_id, r.area | +| | SubqueryAlias: r | +| | Filter: pd_sub_r.device_id IS NOT NULL | +| | TableScan: pd_sub_r, partial_filters=[pd_sub_r.device_id IS NOT NULL] | | | ]] | | physical_plan | SortPreservingMergeExec: [area@0 ASC NULLS LAST, area@1 ASC NULLS LAST] | | | SortExec: expr=[area@0 ASC NULLS LAST, area@1 ASC NULLS LAST], preserve_partitioning=[true] | | | HashJoinExec: mode=Partitioned, join_type=Inner, on=[(device_id@0, device_id@0)], projection=[area@1, area@3] | | | RepartitionExec: partitioning=Hash([device_id@0],REDACTED_PARTITIONS), input_partitions=REDACTED| -| | ProjectionExec: expr=[device_id@1 as device_id, area@2 as area] | -| | MergeScanExec: peers=[REDACTED_REGION, REDACTED_REGION, REDACTED_REGION, ] | +| | MergeScanExec: peers=[REDACTED_REGION, REDACTED_REGION, REDACTED_REGION, ] | | | RepartitionExec: partitioning=Hash([device_id@0],REDACTED_PARTITIONS), input_partitions=REDACTED| -| | ProjectionExec: expr=[device_id@1 as device_id, area@2 as area] | -| | MergeScanExec: peers=[REDACTED_REGION, REDACTED_REGION, REDACTED_REGION, ] | +| | MergeScanExec: peers=[REDACTED_REGION, REDACTED_REGION, REDACTED_REGION, ] | | | | +---------------+-------------------------------------------------------------------------------------------------------------------+ diff --git a/tests/cases/distributed/optimizer/time_index_filter_pushdown.result b/tests/cases/distributed/optimizer/time_index_filter_pushdown.result index 5530edbcead..0da80725d67 100644 --- a/tests/cases/distributed/optimizer/time_index_filter_pushdown.result +++ b/tests/cases/distributed/optimizer/time_index_filter_pushdown.result @@ -122,13 +122,12 @@ WHERE |_|_Projection: cpu.greptime_timestamp_| |_|_TableScan: cpu_| |_| ]]_| -|_|_Projection: cpu.rack, cpu.os, cpu.greptime_timestamp_| |_|_MergeScan [is_placeholder=false, remote_input=[_| -|_| TableScan: cpu_| +|_| Projection: cpu.rack, cpu.os, cpu.greptime_timestamp_| +|_|_TableScan: cpu_| |_| ]]_| | physical_plan | ScalarSubqueryExec: subqueries=1_| |_|_FilterExec: greptime_timestamp@2 = scalar_subquery()_| -|_|_ProjectionExec: expr=[rack@0 as rack, os@1 as os, greptime_timestamp@3 as greptime_timestamp]_| |_|_CooperativeExec_| |_|_MergeScanExec: REDACTED |_|_MergeSortExec: [greptime_timestamp@0 DESC], fetch=1_| @@ -198,26 +197,25 @@ WHERE +-+-+-+ | 0_| 0_|_ScalarSubqueryExec REDACTED |_|_|_FilterExec: greptime_timestamp@2 = scalar_subquery(1686535800000000000) REDACTED -|_|_|_ProjectionExec: expr=[rack@0 as rack, os@1 as os, greptime_timestamp@3 as greptime_timestamp] REDACTED |_|_|_CooperativeExec REDACTED |_|_|_MergeScanExec: REDACTED |_|_|_MergeSortExec: [greptime_timestamp@0 DESC], fetch=1 REDACTED |_|_|_MergeScanExec: REDACTED |_|_|_| | 1_| 0_|_CooperativeExec REDACTED -|_|_|_UnorderedScan: REDACTED, "dyn_filters": ["DynamicFilter [ greptime_timestamp@3 = 1686535800000000000 ]"] +|_|_|_UnorderedScan: REDACTED, "dyn_filters": ["DynamicFilter [ greptime_timestamp@2 = 1686535800000000000 ]"] |_|_|_| | 1_| 1_|_CooperativeExec REDACTED -|_|_|_UnorderedScan: REDACTED, "dyn_filters": ["DynamicFilter [ greptime_timestamp@3 = 1686535800000000000 ]"] +|_|_|_UnorderedScan: REDACTED, "dyn_filters": ["DynamicFilter [ greptime_timestamp@2 = 1686535800000000000 ]"] |_|_|_| | 1_| 2_|_CooperativeExec REDACTED -|_|_|_UnorderedScan: REDACTED, "dyn_filters": ["DynamicFilter [ greptime_timestamp@3 = 1686535800000000000 ]"] +|_|_|_UnorderedScan: REDACTED, "dyn_filters": ["DynamicFilter [ greptime_timestamp@2 = 1686535800000000000 ]"] |_|_|_| | 1_| 3_|_CooperativeExec REDACTED -|_|_|_UnorderedScan: REDACTED, "dyn_filters": ["DynamicFilter [ greptime_timestamp@3 = 1686535800000000000 ]"] +|_|_|_UnorderedScan: REDACTED, "dyn_filters": ["DynamicFilter [ greptime_timestamp@2 = 1686535800000000000 ]"] |_|_|_| | 1_| 4_|_CooperativeExec REDACTED -|_|_|_UnorderedScan: REDACTED, "dyn_filters": ["DynamicFilter [ greptime_timestamp@3 = 1686535800000000000 ]"] +|_|_|_UnorderedScan: REDACTED, "dyn_filters": ["DynamicFilter [ greptime_timestamp@2 = 1686535800000000000 ]"] |_|_|_| | 1_| 0_|_FilterExec: DynamicFilter [ REDACTED ] REDACTED |_|_|_SortPreservingMergeExec: [greptime_timestamp@0 DESC], fetch=1 REDACTED diff --git a/tests/cases/standalone/common/cte/cte_join_build_side.result b/tests/cases/standalone/common/cte/cte_join_build_side.result index d549b21c241..f697d85999a 100644 --- a/tests/cases/standalone/common/cte/cte_join_build_side.result +++ b/tests/cases/standalone/common/cte/cte_join_build_side.result @@ -53,44 +53,42 @@ FROM cte_join_logs AS t JOIN top_dbs td ON t."db" = td."db" GROUP BY t."db" ORDER BY c DESC, t."db"; -+---------------+----------------------------------------------------------------------------------------------------------+ -| plan_type | plan | -+---------------+----------------------------------------------------------------------------------------------------------+ -| logical_plan | Sort: c DESC NULLS FIRST, t.db ASC NULLS LAST | -| | Projection: t.db, count(Int64(1)) AS count(*) AS c | -| | Aggregate: groupBy=[[t.db]], aggr=[[count(Int64(1))]] | -| | LeftSemi Join: t.db = td.db | -| | Projection: t.db | -| | MergeScan [is_placeholder=false, remote_input=[ | -| | SubqueryAlias: t | -| | Filter: cte_join_logs.db IS NOT NULL | -| | TableScan: cte_join_logs, partial_filters=[cte_join_logs.db IS NOT NULL] | -| | ]] | -| | Projection: td.db | -| | MergeScan [is_placeholder=false, remote_input=[ | -| | SubqueryAlias: td | -| | SubqueryAlias: top_dbs | -| | Filter: cte_join_logs.db IS NOT NULL | -| | Limit: skip=0, fetch=5 | -| | Sort: cnt DESC NULLS FIRST | -| | Projection: cte_join_logs.db, count(Int64(1)) AS count(*) AS cnt | -| | Aggregate: groupBy=[[cte_join_logs.db]], aggr=[[count(cte_join_logs.ts) AS count(Int64(1))]] | -| | TableScan: cte_join_logs | -| | ]] | -| physical_plan | ProjectionExec: expr=[db@0 as db, count(Int64(1))@1 as c] | -| | SortPreservingMergeExec: [count(Int64(1))@1 DESC, db@0 ASC NULLS LAST] | ++---------------+------------------------------------------------------------------------------------------------------------+ +| plan_type | plan | ++---------------+------------------------------------------------------------------------------------------------------------+ +| logical_plan | Sort: c DESC NULLS FIRST, t.db ASC NULLS LAST | +| | Projection: t.db, count(Int64(1)) AS count(*) AS c | +| | Aggregate: groupBy=[[t.db]], aggr=[[count(Int64(1))]] | +| | LeftSemi Join: t.db = td.db | +| | MergeScan [is_placeholder=false, remote_input=[ | +| | Projection: t.db | +| | SubqueryAlias: t | +| | Filter: cte_join_logs.db IS NOT NULL | +| | TableScan: cte_join_logs, partial_filters=[cte_join_logs.db IS NOT NULL] | +| | ]] | +| | MergeScan [is_placeholder=false, remote_input=[ | +| | Projection: td.db | +| | SubqueryAlias: td | +| | SubqueryAlias: top_dbs | +| | Filter: cte_join_logs.db IS NOT NULL | +| | Limit: skip=0, fetch=5 | +| | Sort: cnt DESC NULLS FIRST | +| | Projection: cte_join_logs.db, count(Int64(1)) AS count(*) AS cnt | +| | Aggregate: groupBy=[[cte_join_logs.db]], aggr=[[count(cte_join_logs.ts) AS count(Int64(1))]] | +| | TableScan: cte_join_logs | +| | ]] | +| physical_plan | ProjectionExec: expr=[db@0 as db, count(Int64(1))@1 as c] | +| | SortPreservingMergeExec: [count(Int64(1))@1 DESC, db@0 ASC NULLS LAST] | | | SortExec: expr=[count(Int64(1))@1 DESC, db@0 ASC NULLS LAST], preserve_REDACTED -| | AggregateExec: mode=FinalPartitioned, gby=[db@0 as db], aggr=[count(Int64(1))] | +| | AggregateExec: mode=FinalPartitioned, gby=[db@0 as db], aggr=[count(Int64(1))] | | | RepartitionExec: REDACTED -| | AggregateExec: mode=Partial, gby=[db@0 as db], aggr=[count(Int64(1))] | -| | HashJoinExec: mode=CollectLeft, join_type=RightSemi, on=[(db@0, db@0)] | -| | ProjectionExec: expr=[db@0 as db] | -| | MergeScanExec: REDACTED +| | AggregateExec: mode=Partial, gby=[db@0 as db], aggr=[count(Int64(1))] | +| | HashJoinExec: mode=CollectLeft, join_type=RightSemi, on=[(db@0, db@0)] | +| | MergeScanExec: REDACTED | | RepartitionExec: REDACTED -| | ProjectionExec: expr=[db@1 as db] | -| | MergeScanExec: REDACTED -| | | -+---------------+----------------------------------------------------------------------------------------------------------+ +| | MergeScanExec: REDACTED +| | | ++---------------+------------------------------------------------------------------------------------------------------------+ WITH top_dbs AS ( SELECT "db", count(*) AS cnt FROM cte_join_logs diff --git a/tests/cases/standalone/common/filter/hash_join_dyn_filter.result b/tests/cases/standalone/common/filter/hash_join_dyn_filter.result index e534f55791c..266a5e11bfe 100644 --- a/tests/cases/standalone/common/filter/hash_join_dyn_filter.result +++ b/tests/cases/standalone/common/filter/hash_join_dyn_filter.result @@ -39,25 +39,23 @@ WHERE c.tier = 'gold'; +---------------+-----------------------------------------------------------------------------------------------------------------------------------+ | logical_plan | Projection: o.id, o.amount, c.name, c.tier | | | Inner Join: o.customer_id = c.customer_id | -| | Projection: o.id, o.customer_id, o.amount | -| | MergeScan [is_placeholder=false, remote_input=[ | -| | SubqueryAlias: o | -| | Filter: orders.customer_id IS NOT NULL | -| | TableScan: orders, partial_filters=[orders.customer_id IS NOT NULL] | +| | MergeScan [is_placeholder=false, remote_input=[ | +| | Projection: o.id, o.customer_id, o.amount | +| | SubqueryAlias: o | +| | Filter: orders.customer_id IS NOT NULL | +| | TableScan: orders, partial_filters=[orders.customer_id IS NOT NULL] | | | ]] | -| | Projection: c.customer_id, c.name, c.tier | -| | MergeScan [is_placeholder=false, remote_input=[ | -| | SubqueryAlias: c | -| | Filter: customers.customer_id IS NOT NULL AND customers.tier = Utf8("gold") | -| | TableScan: customers, partial_filters=[customers.customer_id IS NOT NULL, customers.tier = Utf8("gold")] | +| | MergeScan [is_placeholder=false, remote_input=[ | +| | Projection: c.customer_id, c.name, c.tier | +| | SubqueryAlias: c | +| | Filter: customers.customer_id IS NOT NULL AND customers.tier = Utf8("gold") | +| | TableScan: customers, partial_filters=[customers.customer_id IS NOT NULL, customers.tier = Utf8("gold")] | | | ]] | | physical_plan | HashJoinExec: mode=Partitioned, join_type=Inner, on=[(customer_id@1, customer_id@0)], projection=[id@0, amount@2, name@4, tier@5] | | | RepartitionExec: REDACTED -| | ProjectionExec: expr=[id@0 as id, customer_id@1 as customer_id, amount@2 as amount] | -| | MergeScanExec: REDACTED +| | MergeScanExec: REDACTED | | RepartitionExec: REDACTED -| | ProjectionExec: expr=[customer_id@0 as customer_id, name@1 as name, tier@2 as tier] | -| | MergeScanExec: REDACTED +| | MergeScanExec: REDACTED | | | +---------------+-----------------------------------------------------------------------------------------------------------------------------------+ @@ -89,19 +87,17 @@ WHERE c.tier = 'gold'; +-+-+-+ | 0_| 0_|_HashJoinExec: mode=Partitioned, join_type=Inner, on=[(customer_id@1, customer_id@0)], projection=[id@0, amount@2, name@4, tier@5] metrics=REDACTED_| |_|_|_RepartitionExec: partitioning=REDACTED -|_|_|_ProjectionExec: expr=[id@0 as id, customer_id@1 as customer_id, amount@2 as amount] metrics=REDACTED_| |_|_|_MergeScanExec: REDACTED |_|_|_RepartitionExec: partitioning=REDACTED -|_|_|_ProjectionExec: expr=[customer_id@0 as customer_id, name@1 as name, tier@2 as tier] metrics=REDACTED_| |_|_|_MergeScanExec: REDACTED |_|_|_| | 1_| 0_|_FilterExec: customer_id@1 IS NOT NULL metrics=REDACTED_| |_|_|_CooperativeExec metrics=REDACTED_| -|_|_|_SeqScan: region=REDACTED, {"partition_count":REDACTED, "projection": ["id", "customer_id", "amount", "ts"], "filters": ["customer_id IS NOT NULL"], "time_range":REDACTED +|_|_|_SeqScan: region=REDACTED, {"partition_count":REDACTED, "projection": ["id", "customer_id", "amount"], "filters": ["customer_id IS NOT NULL"], "time_range":REDACTED |_|_|_| | 1_| 0_|_FilterExec: customer_id@0 IS NOT NULL AND tier@2 = gold metrics=REDACTED_| |_|_|_CooperativeExec metrics=REDACTED_| -|_|_|_SeqScan: region=REDACTED, {"partition_count":REDACTED, "projection": ["customer_id", "name", "tier", "ts"], "filters": ["customer_id IS NOT NULL", "tier = Utf8(\"gold\")"], "dyn_filters":REDACTED +|_|_|_SeqScan: region=REDACTED, {"partition_count":REDACTED, "projection": ["customer_id", "name", "tier"], "filters": ["customer_id IS NOT NULL", "tier = Utf8(\"gold\")"], "dyn_filters":REDACTED |_|_|_| |_|_| Total rows: REDACTED_| +-+-+-+ @@ -162,33 +158,31 @@ FROM (SELECT "id", customer_id as cid, amount, ts FROM orders) o JOIN customers c ON o.cid = c.customer_id WHERE c.tier IN ('gold', 'silver'); -+---------------+-------------------------------------------------------------------------------------------------------------------------------------------------+ -| plan_type | plan | -+---------------+-------------------------------------------------------------------------------------------------------------------------------------------------+ -| logical_plan | Projection: o.id, o.amount, c.name, c.tier | -| | Inner Join: o.cid = c.customer_id | -| | Projection: o.id, o.cid, o.amount | -| | MergeScan [is_placeholder=false, remote_input=[ | -| | SubqueryAlias: o | -| | Projection: orders.id, orders.customer_id AS cid, orders.amount, orders.ts | -| | Filter: orders.customer_id IS NOT NULL | -| | TableScan: orders, partial_filters=[orders.customer_id IS NOT NULL] | -| | ]] | -| | Projection: c.customer_id, c.name, c.tier | -| | MergeScan [is_placeholder=false, remote_input=[ | -| | SubqueryAlias: c | -| | Filter: customers.customer_id IS NOT NULL AND (customers.tier = Utf8("gold") OR customers.tier = Utf8("silver")) | -| | TableScan: customers, partial_filters=[customers.customer_id IS NOT NULL, customers.tier = Utf8("gold") OR customers.tier = Utf8("silver")] | -| | ]] | -| physical_plan | HashJoinExec: mode=Partitioned, join_type=Inner, on=[(cid@1, customer_id@0)], projection=[id@0, amount@2, name@4, tier@5] | ++---------------+---------------------------------------------------------------------------------------------------------------------------------------------------+ +| plan_type | plan | ++---------------+---------------------------------------------------------------------------------------------------------------------------------------------------+ +| logical_plan | Projection: o.id, o.amount, c.name, c.tier | +| | Inner Join: o.cid = c.customer_id | +| | MergeScan [is_placeholder=false, remote_input=[ | +| | Projection: o.id, o.cid, o.amount | +| | SubqueryAlias: o | +| | Projection: orders.id, orders.customer_id AS cid, orders.amount, orders.ts | +| | Filter: orders.customer_id IS NOT NULL | +| | TableScan: orders, partial_filters=[orders.customer_id IS NOT NULL] | +| | ]] | +| | MergeScan [is_placeholder=false, remote_input=[ | +| | Projection: c.customer_id, c.name, c.tier | +| | SubqueryAlias: c | +| | Filter: customers.customer_id IS NOT NULL AND (customers.tier = Utf8("gold") OR customers.tier = Utf8("silver")) | +| | TableScan: customers, partial_filters=[customers.customer_id IS NOT NULL, customers.tier = Utf8("gold") OR customers.tier = Utf8("silver")] | +| | ]] | +| physical_plan | HashJoinExec: mode=Partitioned, join_type=Inner, on=[(cid@1, customer_id@0)], projection=[id@0, amount@2, name@4, tier@5] | | | RepartitionExec: REDACTED -| | ProjectionExec: expr=[id@0 as id, cid@1 as cid, amount@2 as amount] | -| | MergeScanExec: REDACTED +| | MergeScanExec: REDACTED | | RepartitionExec: REDACTED -| | ProjectionExec: expr=[customer_id@0 as customer_id, name@1 as name, tier@2 as tier] | -| | MergeScanExec: REDACTED -| | | -+---------------+-------------------------------------------------------------------------------------------------------------------------------------------------+ +| | MergeScanExec: REDACTED +| | | ++---------------+---------------------------------------------------------------------------------------------------------------------------------------------------+ -- SQLNESS REPLACE ("metrics_per_partition":\s*.*metrics=) "metrics_per_partition": REDACTED metrics= -- SQLNESS REPLACE (metrics=\{.*\}) metrics=REDACTED @@ -218,20 +212,18 @@ WHERE c.tier IN ('gold', 'silver'); +-+-+-+ | 0_| 0_|_HashJoinExec: mode=Partitioned, join_type=Inner, on=[(cid@1, customer_id@0)], projection=[id@0, amount@2, name@4, tier@5] metrics=REDACTED_| |_|_|_RepartitionExec: partitioning=REDACTED -|_|_|_ProjectionExec: expr=[id@0 as id, cid@1 as cid, amount@2 as amount] metrics=REDACTED_| |_|_|_MergeScanExec: REDACTED |_|_|_RepartitionExec: partitioning=REDACTED -|_|_|_ProjectionExec: expr=[customer_id@0 as customer_id, name@1 as name, tier@2 as tier] metrics=REDACTED_| |_|_|_MergeScanExec: REDACTED |_|_|_| -| 1_| 0_|_ProjectionExec: expr=[id@0 as id, customer_id@1 as cid, amount@2 as amount, ts@3 as ts] metrics=REDACTED_| +| 1_| 0_|_ProjectionExec: expr=[id@0 as id, customer_id@1 as cid, amount@2 as amount] metrics=REDACTED_| |_|_|_FilterExec: customer_id@1 IS NOT NULL metrics=REDACTED_| |_|_|_CooperativeExec metrics=REDACTED_| -|_|_|_SeqScan: region=REDACTED, {"partition_count":REDACTED, "projection": ["id", "customer_id", "amount", "ts"], "filters": ["customer_id IS NOT NULL"], "time_range":REDACTED +|_|_|_SeqScan: region=REDACTED, {"partition_count":REDACTED, "projection": ["id", "customer_id", "amount"], "filters": ["customer_id IS NOT NULL"], "time_range":REDACTED |_|_|_| | 1_| 0_|_FilterExec: customer_id@0 IS NOT NULL AND (tier@2 = gold OR tier@2 = silver) metrics=REDACTED_| |_|_|_CooperativeExec metrics=REDACTED_| -|_|_|_SeqScan: region=REDACTED, {"partition_count":REDACTED, "projection": ["customer_id", "name", "tier", "ts"], "filters": ["customer_id IS NOT NULL", "tier = Utf8(\"gold\") OR tier = Utf8(\"silver\")"], "dyn_filters":REDACTED +|_|_|_SeqScan: region=REDACTED, {"partition_count":REDACTED, "projection": ["customer_id", "name", "tier"], "filters": ["customer_id IS NOT NULL", "tier = Utf8(\"gold\") OR tier = Utf8(\"silver\")"], "dyn_filters":REDACTED |_|_|_| |_|_| Total rows: REDACTED_| +-+-+-+ diff --git a/tests/cases/standalone/common/filter/hash_join_topk_dyn_filter.result b/tests/cases/standalone/common/filter/hash_join_topk_dyn_filter.result index 1c2d05bbcbb..a218c5b2c31 100644 --- a/tests/cases/standalone/common/filter/hash_join_topk_dyn_filter.result +++ b/tests/cases/standalone/common/filter/hash_join_topk_dyn_filter.result @@ -47,35 +47,33 @@ FROM ( JOIN customers c ON top_orders.customer_id = c.customer_id WHERE c.tier IN ('gold', 'bronze'); -+---------------+-------------------------------------------------------------------------------------------------------------------------------------------------+ -| plan_type | plan | -+---------------+-------------------------------------------------------------------------------------------------------------------------------------------------+ -| logical_plan | Projection: top_orders.id, top_orders.amount, c.name, c.tier | -| | Inner Join: top_orders.customer_id = c.customer_id | -| | Projection: top_orders.id, top_orders.customer_id, top_orders.amount | -| | MergeScan [is_placeholder=false, remote_input=[ | -| | SubqueryAlias: top_orders | -| | Filter: orders.customer_id IS NOT NULL | -| | Limit: skip=0, fetch=5 | -| | Sort: orders.amount DESC NULLS FIRST | -| | Projection: orders.id, orders.customer_id, orders.amount, orders.ts | -| | TableScan: orders | -| | ]] | -| | Projection: c.customer_id, c.name, c.tier | -| | MergeScan [is_placeholder=false, remote_input=[ | -| | SubqueryAlias: c | -| | Filter: customers.customer_id IS NOT NULL AND (customers.tier = Utf8("gold") OR customers.tier = Utf8("bronze")) | -| | TableScan: customers, partial_filters=[customers.customer_id IS NOT NULL, customers.tier = Utf8("gold") OR customers.tier = Utf8("bronze")] | -| | ]] | -| physical_plan | HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(customer_id@1, customer_id@0)], projection=[id@0, amount@2, name@4, tier@5] | -| | ProjectionExec: expr=[id@0 as id, customer_id@1 as customer_id, amount@2 as amount] | -| | CooperativeExec | -| | MergeScanExec: REDACTED -| | RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1 | -| | ProjectionExec: expr=[customer_id@0 as customer_id, name@1 as name, tier@2 as tier] | -| | MergeScanExec: REDACTED -| | | -+---------------+-------------------------------------------------------------------------------------------------------------------------------------------------+ ++---------------+---------------------------------------------------------------------------------------------------------------------------------------------------+ +| plan_type | plan | ++---------------+---------------------------------------------------------------------------------------------------------------------------------------------------+ +| logical_plan | Projection: top_orders.id, top_orders.amount, c.name, c.tier | +| | Inner Join: top_orders.customer_id = c.customer_id | +| | MergeScan [is_placeholder=false, remote_input=[ | +| | Projection: top_orders.id, top_orders.customer_id, top_orders.amount | +| | SubqueryAlias: top_orders | +| | Filter: orders.customer_id IS NOT NULL | +| | Limit: skip=0, fetch=5 | +| | Sort: orders.amount DESC NULLS FIRST | +| | Projection: orders.id, orders.customer_id, orders.amount, orders.ts | +| | TableScan: orders | +| | ]] | +| | MergeScan [is_placeholder=false, remote_input=[ | +| | Projection: c.customer_id, c.name, c.tier | +| | SubqueryAlias: c | +| | Filter: customers.customer_id IS NOT NULL AND (customers.tier = Utf8("gold") OR customers.tier = Utf8("bronze")) | +| | TableScan: customers, partial_filters=[customers.customer_id IS NOT NULL, customers.tier = Utf8("gold") OR customers.tier = Utf8("bronze")] | +| | ]] | +| physical_plan | HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(customer_id@1, customer_id@0)], projection=[id@0, amount@2, name@4, tier@5] | +| | CooperativeExec | +| | MergeScanExec: REDACTED +| | RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1 | +| | MergeScanExec: REDACTED +| | | ++---------------+---------------------------------------------------------------------------------------------------------------------------------------------------+ -- SQLNESS REPLACE ("metrics_per_partition":\s*.*metrics=) "metrics_per_partition": REDACTED metrics= -- SQLNESS REPLACE (metrics=\{.*\}) metrics=REDACTED @@ -109,22 +107,20 @@ WHERE c.tier IN ('gold', 'bronze'); | stage | node | plan_| +-+-+-+ | 0_| 0_|_HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(customer_id@1, customer_id@0)], projection=[id@0, amount@2, name@4, tier@5] metrics=REDACTED_| -|_|_|_ProjectionExec: expr=[id@0 as id, customer_id@1 as customer_id, amount@2 as amount] metrics=REDACTED_| |_|_|_CooperativeExec metrics=REDACTED_| |_|_|_MergeScanExec: REDACTED |_|_|_RepartitionExec: partitioning=REDACTED -|_|_|_ProjectionExec: expr=[customer_id@0 as customer_id, name@1 as name, tier@2 as tier] metrics=REDACTED_| |_|_|_MergeScanExec: REDACTED |_|_|_| | 1_| 0_|_FilterExec: customer_id@1 IS NOT NULL metrics=REDACTED_| |_|_|_RepartitionExec: partitioning=REDACTED |_|_|_SortPreservingMergeExec: [amount@2 DESC], fetch=5 metrics=REDACTED_| |_|_|_SortExec: TopK(fetch=5), expr=[amount@2 DESC], preserve_partitioning=[true] metrics=REDACTED_| -|_|_|_SeqScan: region=REDACTED, {"partition_count":REDACTED, "projection": ["id", "customer_id", "amount", "ts"], "dyn_filters":REDACTED +|_|_|_SeqScan: region=REDACTED, {"partition_count":REDACTED, "projection": ["id", "customer_id", "amount"], "dyn_filters":REDACTED |_|_|_| | 1_| 0_|_FilterExec: customer_id@0 IS NOT NULL AND (tier@2 = gold OR tier@2 = bronze) metrics=REDACTED_| |_|_|_CooperativeExec metrics=REDACTED_| -|_|_|_SeqScan: region=REDACTED, {"partition_count":REDACTED, "projection": ["customer_id", "name", "tier", "ts"], "filters": ["customer_id IS NOT NULL", "tier = Utf8(\"gold\") OR tier = Utf8(\"bronze\")"], "dyn_filters":REDACTED +|_|_|_SeqScan: region=REDACTED, {"partition_count":REDACTED, "projection": ["customer_id", "name", "tier"], "filters": ["customer_id IS NOT NULL", "tier = Utf8(\"gold\") OR tier = Utf8(\"bronze\")"], "dyn_filters":REDACTED |_|_|_| |_|_| Total rows: REDACTED_| +-+-+-+ @@ -211,27 +207,25 @@ LIMIT 4; | logical_plan | Sort: o.amount DESC NULLS FIRST, fetch=4 | | | Projection: o.id, o.customer_id, c.name, c.tier, o.amount | | | Inner Join: o.customer_id = c.customer_id | -| | Projection: o.id, o.customer_id, o.amount | -| | MergeScan [is_placeholder=false, remote_input=[ | -| | SubqueryAlias: o | -| | Filter: orders.customer_id IS NOT NULL | -| | TableScan: orders, partial_filters=[orders.customer_id IS NOT NULL] | +| | MergeScan [is_placeholder=false, remote_input=[ | +| | Projection: o.id, o.customer_id, o.amount | +| | SubqueryAlias: o | +| | Filter: orders.customer_id IS NOT NULL | +| | TableScan: orders, partial_filters=[orders.customer_id IS NOT NULL] | | | ]] | -| | Projection: c.customer_id, c.name, c.tier | -| | MergeScan [is_placeholder=false, remote_input=[ | -| | SubqueryAlias: c | -| | Filter: customers.customer_id IS NOT NULL AND (customers.tier = Utf8("gold") OR customers.tier = Utf8("silver")) | -| | TableScan: customers, partial_filters=[customers.customer_id IS NOT NULL, customers.tier = Utf8("gold") OR customers.tier = Utf8("silver")] | +| | MergeScan [is_placeholder=false, remote_input=[ | +| | Projection: c.customer_id, c.name, c.tier | +| | SubqueryAlias: c | +| | Filter: customers.customer_id IS NOT NULL AND (customers.tier = Utf8("gold") OR customers.tier = Utf8("silver")) | +| | TableScan: customers, partial_filters=[customers.customer_id IS NOT NULL, customers.tier = Utf8("gold") OR customers.tier = Utf8("silver")] | | | ]] | | physical_plan | SortPreservingMergeExec: [amount@4 DESC], fetch=4 | | | SortExec: TopK(fetch=4), expr=[amount@4 DESC], preserve_partitioning=[true] | | | HashJoinExec: mode=Partitioned, join_type=Inner, on=[(customer_id@1, customer_id@0)], projection=[id@0, customer_id@1, name@4, tier@5, amount@2] | | | RepartitionExec: partitioning=REDACTED -| | ProjectionExec: expr=[id@0 as id, customer_id@1 as customer_id, amount@2 as amount] | -| | MergeScanExec: REDACTED +| | MergeScanExec: REDACTED | | RepartitionExec: partitioning=REDACTED -| | ProjectionExec: expr=[customer_id@0 as customer_id, name@1 as name, tier@2 as tier] | -| | MergeScanExec: REDACTED +| | MergeScanExec: REDACTED | | | +---------------+------------------------------------------------------------------------------------------------------------------------------------------------------+ @@ -270,19 +264,17 @@ LIMIT 4; |_|_|_SortExec: TopK(fetch=4), expr=[amount@4 DESC], preserve_partitioning=[true] metrics=REDACTED_| |_|_|_HashJoinExec: mode=Partitioned, join_type=Inner, on=[(customer_id@1, customer_id@0)], projection=[id@0, customer_id@1, name@4, tier@5, amount@2] metrics=REDACTED_| |_|_|_RepartitionExec: partitioning=REDACTED -|_|_|_ProjectionExec: expr=[id@0 as id, customer_id@1 as customer_id, amount@2 as amount] metrics=REDACTED_| |_|_|_MergeScanExec: REDACTED |_|_|_RepartitionExec: partitioning=REDACTED -|_|_|_ProjectionExec: expr=[customer_id@0 as customer_id, name@1 as name, tier@2 as tier] metrics=REDACTED_| |_|_|_MergeScanExec: REDACTED |_|_|_| | 1_| 0_|_FilterExec: customer_id@1 IS NOT NULL metrics=REDACTED_| |_|_|_CooperativeExec metrics=REDACTED_| -|_|_|_SeqScan: region=REDACTED, {"partition_count":REDACTED, "projection": ["id", "customer_id", "amount", "ts"], "filters": ["customer_id IS NOT NULL"], "dyn_filters":REDACTED +|_|_|_SeqScan: region=REDACTED, {"partition_count":REDACTED, "projection": ["id", "customer_id", "amount"], "filters": ["customer_id IS NOT NULL"], "dyn_filters":REDACTED |_|_|_| | 1_| 0_|_FilterExec: customer_id@0 IS NOT NULL AND (tier@2 = gold OR tier@2 = silver) metrics=REDACTED_| |_|_|_CooperativeExec metrics=REDACTED_| -|_|_|_SeqScan: region=REDACTED, {"partition_count":REDACTED, "projection": ["customer_id", "name", "tier", "ts"], "filters": ["customer_id IS NOT NULL", "tier = Utf8(\"gold\") OR tier = Utf8(\"silver\")"], "dyn_filters":REDACTED +|_|_|_SeqScan: region=REDACTED, {"partition_count":REDACTED, "projection": ["customer_id", "name", "tier"], "filters": ["customer_id IS NOT NULL", "tier = Utf8(\"gold\") OR tier = Utf8(\"silver\")"], "dyn_filters":REDACTED |_|_|_| |_|_| Total rows: REDACTED_| +-+-+-+ @@ -377,37 +369,34 @@ WHERE c.tier IN ('gold', 'silver') | | Inner Join: o.product_id = p.product_id | | | Projection: o.id, o.product_id, o.amount, c.name, c.tier | | | Inner Join: o.customer_id = c.customer_id | -| | Projection: o.id, o.customer_id, o.product_id, o.amount | -| | MergeScan [is_placeholder=false, remote_input=[ | -| | SubqueryAlias: o | -| | Filter: orders.customer_id IS NOT NULL AND orders.product_id IS NOT NULL | -| | TableScan: orders, partial_filters=[orders.customer_id IS NOT NULL, orders.product_id IS NOT NULL] | +| | MergeScan [is_placeholder=false, remote_input=[ | +| | Projection: o.id, o.customer_id, o.product_id, o.amount | +| | SubqueryAlias: o | +| | Filter: orders.customer_id IS NOT NULL AND orders.product_id IS NOT NULL | +| | TableScan: orders, partial_filters=[orders.customer_id IS NOT NULL, orders.product_id IS NOT NULL] | | | ]] | -| | Projection: c.customer_id, c.name, c.tier | -| | MergeScan [is_placeholder=false, remote_input=[ | -| | SubqueryAlias: c | -| | Filter: customers.customer_id IS NOT NULL AND (customers.tier = Utf8("gold") OR customers.tier = Utf8("silver")) | -| | TableScan: customers, partial_filters=[customers.customer_id IS NOT NULL, customers.tier = Utf8("gold") OR customers.tier = Utf8("silver")] | +| | MergeScan [is_placeholder=false, remote_input=[ | +| | Projection: c.customer_id, c.name, c.tier | +| | SubqueryAlias: c | +| | Filter: customers.customer_id IS NOT NULL AND (customers.tier = Utf8("gold") OR customers.tier = Utf8("silver")) | +| | TableScan: customers, partial_filters=[customers.customer_id IS NOT NULL, customers.tier = Utf8("gold") OR customers.tier = Utf8("silver")] | | | ]] | -| | Projection: p.product_id, p.name, p.category | -| | MergeScan [is_placeholder=false, remote_input=[ | -| | SubqueryAlias: p | -| | Filter: products.product_id IS NOT NULL AND products.category = Utf8("electronics") | -| | TableScan: products, partial_filters=[products.product_id IS NOT NULL, products.category = Utf8("electronics")] | +| | MergeScan [is_placeholder=false, remote_input=[ | +| | Projection: p.product_id, p.name, p.category | +| | SubqueryAlias: p | +| | Filter: products.product_id IS NOT NULL AND products.category = Utf8("electronics") | +| | TableScan: products, partial_filters=[products.product_id IS NOT NULL, products.category = Utf8("electronics")] | | | ]] | | physical_plan | ProjectionExec: expr=[id@0 as id, amount@1 as amount, name@2 as name, tier@3 as tier, name@4 as product_name, category@5 as category] | | | HashJoinExec: mode=Partitioned, join_type=Inner, on=[(product_id@1, product_id@0)], projection=[id@0, amount@2, name@3, tier@4, name@6, category@7] | | | RepartitionExec: partitioning=REDACTED | | HashJoinExec: mode=Partitioned, join_type=Inner, on=[(customer_id@1, customer_id@0)], projection=[id@0, product_id@2, amount@3, name@5, tier@6] | | | RepartitionExec: partitioning=REDACTED -| | ProjectionExec: expr=[id@0 as id, customer_id@1 as customer_id, product_id@2 as product_id, amount@3 as amount] | -| | MergeScanExec: REDACTED +| | MergeScanExec: REDACTED | | RepartitionExec: partitioning=REDACTED -| | ProjectionExec: expr=[customer_id@0 as customer_id, name@1 as name, tier@2 as tier] | -| | MergeScanExec: REDACTED +| | MergeScanExec: REDACTED | | RepartitionExec: partitioning=REDACTED -| | ProjectionExec: expr=[product_id@0 as product_id, name@1 as name, category@2 as category] | -| | MergeScanExec: REDACTED +| | MergeScanExec: REDACTED | | | +---------------+-------------------------------------------------------------------------------------------------------------------------------------------------------+ @@ -444,26 +433,23 @@ WHERE c.tier IN ('gold', 'silver') |_|_|_RepartitionExec: partitioning=REDACTED |_|_|_HashJoinExec: mode=Partitioned, join_type=Inner, on=[(customer_id@1, customer_id@0)], projection=[id@0, product_id@2, amount@3, name@5, tier@6] metrics=REDACTED_| |_|_|_RepartitionExec: partitioning=REDACTED -|_|_|_ProjectionExec: expr=[id@0 as id, customer_id@1 as customer_id, product_id@2 as product_id, amount@3 as amount] metrics=REDACTED_| |_|_|_MergeScanExec: REDACTED |_|_|_RepartitionExec: partitioning=REDACTED -|_|_|_ProjectionExec: expr=[customer_id@0 as customer_id, name@1 as name, tier@2 as tier] metrics=REDACTED_| |_|_|_MergeScanExec: REDACTED |_|_|_RepartitionExec: partitioning=REDACTED -|_|_|_ProjectionExec: expr=[product_id@0 as product_id, name@1 as name, category@2 as category] metrics=REDACTED_| |_|_|_MergeScanExec: REDACTED |_|_|_| | 1_| 0_|_FilterExec: customer_id@1 IS NOT NULL AND product_id@2 IS NOT NULL metrics=REDACTED_| |_|_|_CooperativeExec metrics=REDACTED_| -|_|_|_SeqScan: region=REDACTED, {"partition_count":REDACTED, "projection": ["id", "customer_id", "product_id", "amount", "ts"], "filters": ["customer_id IS NOT NULL", "product_id IS NOT NULL"], "time_range":REDACTED +|_|_|_SeqScan: region=REDACTED, {"partition_count":REDACTED, "projection": ["id", "customer_id", "product_id", "amount"], "filters": ["customer_id IS NOT NULL", "product_id IS NOT NULL"], "time_range":REDACTED |_|_|_| | 1_| 0_|_FilterExec: customer_id@0 IS NOT NULL AND (tier@2 = gold OR tier@2 = silver) metrics=REDACTED_| |_|_|_CooperativeExec metrics=REDACTED_| -|_|_|_SeqScan: region=REDACTED, {"partition_count":REDACTED, "projection": ["customer_id", "name", "tier", "ts"], "filters": ["customer_id IS NOT NULL", "tier = Utf8(\"gold\") OR tier = Utf8(\"silver\")"], "dyn_filters":REDACTED +|_|_|_SeqScan: region=REDACTED, {"partition_count":REDACTED, "projection": ["customer_id", "name", "tier"], "filters": ["customer_id IS NOT NULL", "tier = Utf8(\"gold\") OR tier = Utf8(\"silver\")"], "dyn_filters":REDACTED |_|_|_| | 1_| 0_|_FilterExec: product_id@0 IS NOT NULL AND category@2 = electronics metrics=REDACTED_| |_|_|_CooperativeExec metrics=REDACTED_| -|_|_|_SeqScan: region=REDACTED, {"partition_count":REDACTED, "projection": ["product_id", "name", "category", "ts"], "filters": ["product_id IS NOT NULL", "category = Utf8(\"electronics\")"], "dyn_filters":REDACTED +|_|_|_SeqScan: region=REDACTED, {"partition_count":REDACTED, "projection": ["product_id", "name", "category"], "filters": ["product_id IS NOT NULL", "category = Utf8(\"electronics\")"], "dyn_filters":REDACTED |_|_|_| |_|_| Total rows: REDACTED_| +-+-+-+ diff --git a/tests/cases/standalone/common/join/merge_scan_projection.result b/tests/cases/standalone/common/join/merge_scan_projection.result new file mode 100644 index 00000000000..2607a776d93 --- /dev/null +++ b/tests/cases/standalone/common/join/merge_scan_projection.result @@ -0,0 +1,103 @@ +CREATE TABLE projection_spans ( + trace_id STRING, + span_id STRING, + parent_span_id STRING, + service_name STRING, + payload STRING, + ts TIMESTAMP TIME INDEX, + PRIMARY KEY (service_name) +) +PARTITION ON COLUMNS (service_name) ( + service_name < 'storage', service_name >= 'storage' +) +WITH (append_mode = 'true'); + +Affected Rows: 0 + +INSERT INTO projection_spans VALUES + ('trace-a', 'root', NULL, 'front', 'root payload', 0), + ('trace-a', 'child', 'root', 'storage', 'child payload', 1), + ('trace-a', 'same', 'root', 'front', 'same service', 2), + ('trace-b', 'child', 'root', 'storage', 'unmatched trace', 3); + +Affected Rows: 4 + +ADMIN FLUSH_TABLE('projection_spans'); + ++---------------------------------------+ +| ADMIN FLUSH_TABLE('projection_spans') | ++---------------------------------------+ +| 0 | ++---------------------------------------+ + +-- Each remote scan supplies the three join columns without a local projection. +-- SQLNESS REPLACE (peers.*) peers=REDACTED +EXPLAIN SELECT p.service_name AS parent_service, + c.service_name AS child_service, COUNT(*) AS call_count +FROM projection_spans c JOIN projection_spans p + ON c.trace_id = p.trace_id AND c.parent_span_id = p.span_id +WHERE c.service_name <> p.service_name +GROUP BY p.service_name, c.service_name +ORDER BY p.service_name, c.service_name; + ++---------------+-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ +| plan_type | plan | ++---------------+-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ +| logical_plan | Sort: parent_service ASC NULLS LAST, child_service ASC NULLS LAST | +| | Projection: p.service_name AS parent_service, c.service_name AS child_service, count(Int64(1)) AS count(*) AS call_count | +| | Aggregate: groupBy=[[p.service_name, c.service_name]], aggr=[[count(Int64(1))]] | +| | Projection: c.service_name, p.service_name | +| | Inner Join: c.trace_id = p.trace_id, c.parent_span_id = p.span_id Filter: p.service_name != c.service_name | +| | MergeScan [is_placeholder=false, remote_input=[ | +| | Projection: c.trace_id, c.parent_span_id, c.service_name | +| | SubqueryAlias: c | +| | Filter: projection_spans.trace_id IS NOT NULL AND projection_spans.parent_span_id IS NOT NULL | +| | TableScan: projection_spans, partial_filters=[projection_spans.trace_id IS NOT NULL, projection_spans.parent_span_id IS NOT NULL] | +| | ]] | +| | MergeScan [is_placeholder=false, remote_input=[ | +| | Projection: p.trace_id, p.span_id, p.service_name | +| | SubqueryAlias: p | +| | Filter: projection_spans.trace_id IS NOT NULL AND projection_spans.span_id IS NOT NULL | +| | TableScan: projection_spans, partial_filters=[projection_spans.trace_id IS NOT NULL, projection_spans.span_id IS NOT NULL] | +| | ]] | +| physical_plan | SortPreservingMergeExec: [parent_service@0 ASC NULLS LAST, child_service@1 ASC NULLS LAST] | +| | ProjectionExec: expr=[service_name@0 as parent_service, service_name@1 as child_service, count(Int64(1))@2 as call_count] | +| | SortExec: expr=[service_name@0 ASC NULLS LAST, service_name@1 ASC NULLS LAST], preserve_partitioning=[true] | +| | AggregateExec: mode=FinalPartitioned, gby=[service_name@0 as service_name, service_name@1 as service_name], aggr=[count(Int64(1))] | +| | RepartitionExec: partitioning=Hash([service_name@0, service_name@1], 4), input_partitions=4 | +| | AggregateExec: mode=Partial, gby=[service_name@1 as service_name, service_name@0 as service_name], aggr=[count(Int64(1))] | +| | HashJoinExec: mode=Partitioned, join_type=Inner, on=[(trace_id@0, trace_id@0), (parent_span_id@1, span_id@1)], filter=service_name@1 != service_name@0, projection=[service_name@2, service_name@5] | +| | RepartitionExec: partitioning=Hash([trace_id@0, parent_span_id@1], 4), input_partitions=2 | +| | MergeScanExec: peers=REDACTED +| | RepartitionExec: partitioning=Hash([trace_id@0, span_id@1], 4), input_partitions=2 | +| | MergeScanExec: peers=REDACTED +| | | ++---------------+-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ + +SELECT p.service_name AS parent_service, + c.service_name AS child_service, COUNT(*) AS call_count +FROM projection_spans c JOIN projection_spans p + ON c.trace_id = p.trace_id AND c.parent_span_id = p.span_id +WHERE c.service_name <> p.service_name +GROUP BY p.service_name, c.service_name +ORDER BY p.service_name, c.service_name; + ++----------------+---------------+------------+ +| parent_service | child_service | call_count | ++----------------+---------------+------------+ +| front | storage | 1 | ++----------------+---------------+------------+ + +SELECT COUNT(*) AS cross_rows +FROM projection_spans a CROSS JOIN projection_spans b; + ++------------+ +| cross_rows | ++------------+ +| 16 | ++------------+ + +DROP TABLE projection_spans; + +Affected Rows: 0 + diff --git a/tests/cases/standalone/common/join/merge_scan_projection.sql b/tests/cases/standalone/common/join/merge_scan_projection.sql new file mode 100644 index 00000000000..a9297d028b2 --- /dev/null +++ b/tests/cases/standalone/common/join/merge_scan_projection.sql @@ -0,0 +1,44 @@ +CREATE TABLE projection_spans ( + trace_id STRING, + span_id STRING, + parent_span_id STRING, + service_name STRING, + payload STRING, + ts TIMESTAMP TIME INDEX, + PRIMARY KEY (service_name) +) +PARTITION ON COLUMNS (service_name) ( + service_name < 'storage', service_name >= 'storage' +) +WITH (append_mode = 'true'); + +INSERT INTO projection_spans VALUES + ('trace-a', 'root', NULL, 'front', 'root payload', 0), + ('trace-a', 'child', 'root', 'storage', 'child payload', 1), + ('trace-a', 'same', 'root', 'front', 'same service', 2), + ('trace-b', 'child', 'root', 'storage', 'unmatched trace', 3); + +ADMIN FLUSH_TABLE('projection_spans'); + +-- Each remote scan supplies the three join columns without a local projection. +-- SQLNESS REPLACE (peers.*) peers=REDACTED +EXPLAIN SELECT p.service_name AS parent_service, + c.service_name AS child_service, COUNT(*) AS call_count +FROM projection_spans c JOIN projection_spans p + ON c.trace_id = p.trace_id AND c.parent_span_id = p.span_id +WHERE c.service_name <> p.service_name +GROUP BY p.service_name, c.service_name +ORDER BY p.service_name, c.service_name; + +SELECT p.service_name AS parent_service, + c.service_name AS child_service, COUNT(*) AS call_count +FROM projection_spans c JOIN projection_spans p + ON c.trace_id = p.trace_id AND c.parent_span_id = p.span_id +WHERE c.service_name <> p.service_name +GROUP BY p.service_name, c.service_name +ORDER BY p.service_name, c.service_name; + +SELECT COUNT(*) AS cross_rows +FROM projection_spans a CROSS JOIN projection_spans b; + +DROP TABLE projection_spans; diff --git a/tests/cases/standalone/common/join/merge_scan_semi_join.result b/tests/cases/standalone/common/join/merge_scan_semi_join.result index 598e2099662..d56fd6a6877 100644 --- a/tests/cases/standalone/common/join/merge_scan_semi_join.result +++ b/tests/cases/standalone/common/join/merge_scan_semi_join.result @@ -42,36 +42,35 @@ WHERE key_id IN ( ) GROUP BY key_id ORDER BY key_id; -+---------------+---------------------------------------------------------------------------------------------------------------------------------------+ -| plan_type | plan | -+---------------+---------------------------------------------------------------------------------------------------------------------------------------+ -| logical_plan | Sort: merge_semi_data.key_id ASC NULLS LAST | -| | Projection: merge_semi_data.key_id, min(merge_semi_data.ts) AS first_ts | -| | Aggregate: groupBy=[[merge_semi_data.key_id]], aggr=[[min(merge_semi_data.ts)]] | -| | LeftSemi Join: merge_semi_data.key_id = __correlated_sq_1.key_id | -| | Projection: merge_semi_data.ts, merge_semi_data.key_id | -| | MergeScan [is_placeholder=false, remote_input=[ | -| | Filter: merge_semi_data.key_id IS NOT NULL | -| | TableScan: merge_semi_data, partial_filters=[merge_semi_data.key_id IS NOT NULL] | -| | ]] | -| | MergeScan [is_placeholder=false, remote_input=[ | -| | SubqueryAlias: __correlated_sq_1 | -| | Projection: merge_semi_data.key_id | -| | Filter: merge_semi_data.is_match AND merge_semi_data.key_id IS NOT NULL | -| | TableScan: merge_semi_data, partial_filters=[merge_semi_data.is_match, merge_semi_data.key_id IS NOT NULL] | -| | ]] | -| physical_plan | SortPreservingMergeExec: [key_id@0 ASC NULLS LAST] | -| | ProjectionExec: expr=[key_id@0 as key_id, min(merge_semi_data.ts)@1 as first_ts] | -| | SortExec: expr=[key_id@0 ASC NULLS LAST], preserve_partitioning=[true] | -| | AggregateExec: mode=SinglePartitioned, gby=[key_id@1 as key_id], aggr=[min(merge_semi_data.ts)] | -| | HashJoinExec: mode=Partitioned, join_type=LeftSemi, on=[(key_id@1, key_id@0)] | -| | RepartitionExec: partitioning=Hash([key_id@1], 4), input_partitions=4 | -| | ProjectionExec: expr=[ts@0 as ts, key_id@1 as key_id] | -| | MergeScanExec: peers=REDACTED -| | RepartitionExec: partitioning=Hash([key_id@0], 4), input_partitions=4 | ++---------------+-------------------------------------------------------------------------------------------------------------------------------------+ +| plan_type | plan | ++---------------+-------------------------------------------------------------------------------------------------------------------------------------+ +| logical_plan | Sort: merge_semi_data.key_id ASC NULLS LAST | +| | Projection: merge_semi_data.key_id, min(merge_semi_data.ts) AS first_ts | +| | Aggregate: groupBy=[[merge_semi_data.key_id]], aggr=[[min(merge_semi_data.ts)]] | +| | LeftSemi Join: merge_semi_data.key_id = __correlated_sq_1.key_id | +| | MergeScan [is_placeholder=false, remote_input=[ | +| | Projection: merge_semi_data.ts, merge_semi_data.key_id | +| | Filter: merge_semi_data.key_id IS NOT NULL | +| | TableScan: merge_semi_data, partial_filters=[merge_semi_data.key_id IS NOT NULL] | +| | ]] | +| | MergeScan [is_placeholder=false, remote_input=[ | +| | SubqueryAlias: __correlated_sq_1 | +| | Projection: merge_semi_data.key_id | +| | Filter: merge_semi_data.is_match AND merge_semi_data.key_id IS NOT NULL | +| | TableScan: merge_semi_data, partial_filters=[merge_semi_data.is_match, merge_semi_data.key_id IS NOT NULL] | +| | ]] | +| physical_plan | SortPreservingMergeExec: [key_id@0 ASC NULLS LAST] | +| | ProjectionExec: expr=[key_id@0 as key_id, min(merge_semi_data.ts)@1 as first_ts] | +| | SortExec: expr=[key_id@0 ASC NULLS LAST], preserve_partitioning=[true] | +| | AggregateExec: mode=SinglePartitioned, gby=[key_id@1 as key_id], aggr=[min(merge_semi_data.ts)] | +| | HashJoinExec: mode=Partitioned, join_type=LeftSemi, on=[(key_id@1, key_id@0)] | +| | RepartitionExec: partitioning=Hash([key_id@1], 4), input_partitions=4 | | | MergeScanExec: peers=REDACTED -| | | -+---------------+---------------------------------------------------------------------------------------------------------------------------------------+ +| | RepartitionExec: partitioning=Hash([key_id@0], 4), input_partitions=4 | +| | MergeScanExec: peers=REDACTED +| | | ++---------------+-------------------------------------------------------------------------------------------------------------------------------------+ -- Every key matches and reports its earliest (false-row) timestamp. SELECT key_id, MIN(ts) AS first_ts @@ -131,36 +130,35 @@ WHERE key_id IN ( ) GROUP BY key_id ORDER BY key_id; -+---------------+---------------------------------------------------------------------------------------------------------------------------------------+ -| plan_type | plan | -+---------------+---------------------------------------------------------------------------------------------------------------------------------------+ -| logical_plan | Sort: merge_semi_data.key_id ASC NULLS LAST | -| | Projection: merge_semi_data.key_id, min(merge_semi_data.ts) AS first_ts | -| | Aggregate: groupBy=[[merge_semi_data.key_id]], aggr=[[min(merge_semi_data.ts)]] | -| | LeftSemi Join: merge_semi_data.key_id = __correlated_sq_1.key_id | -| | Projection: merge_semi_data.ts, merge_semi_data.key_id | -| | MergeScan [is_placeholder=false, remote_input=[ | -| | Filter: merge_semi_data.key_id IS NOT NULL | -| | TableScan: merge_semi_data, partial_filters=[merge_semi_data.key_id IS NOT NULL] | -| | ]] | -| | MergeScan [is_placeholder=false, remote_input=[ | -| | SubqueryAlias: __correlated_sq_1 | -| | Projection: merge_semi_data.key_id | -| | Filter: merge_semi_data.is_match AND merge_semi_data.key_id IS NOT NULL | -| | TableScan: merge_semi_data, partial_filters=[merge_semi_data.is_match, merge_semi_data.key_id IS NOT NULL] | -| | ]] | -| physical_plan | SortPreservingMergeExec: [key_id@0 ASC NULLS LAST] | -| | ProjectionExec: expr=[key_id@0 as key_id, min(merge_semi_data.ts)@1 as first_ts] | -| | SortExec: expr=[key_id@0 ASC NULLS LAST], preserve_partitioning=[true] | -| | AggregateExec: mode=SinglePartitioned, gby=[key_id@1 as key_id], aggr=[min(merge_semi_data.ts)] | -| | HashJoinExec: mode=Partitioned, join_type=LeftSemi, on=[(key_id@1, key_id@0)] | -| | RepartitionExec: partitioning=Hash([key_id@1], 2), input_partitions=2 | -| | ProjectionExec: expr=[ts@0 as ts, key_id@1 as key_id] | -| | MergeScanExec: peers=REDACTED -| | RepartitionExec: partitioning=Hash([key_id@0], 2), input_partitions=2 | ++---------------+-------------------------------------------------------------------------------------------------------------------------------------+ +| plan_type | plan | ++---------------+-------------------------------------------------------------------------------------------------------------------------------------+ +| logical_plan | Sort: merge_semi_data.key_id ASC NULLS LAST | +| | Projection: merge_semi_data.key_id, min(merge_semi_data.ts) AS first_ts | +| | Aggregate: groupBy=[[merge_semi_data.key_id]], aggr=[[min(merge_semi_data.ts)]] | +| | LeftSemi Join: merge_semi_data.key_id = __correlated_sq_1.key_id | +| | MergeScan [is_placeholder=false, remote_input=[ | +| | Projection: merge_semi_data.ts, merge_semi_data.key_id | +| | Filter: merge_semi_data.key_id IS NOT NULL | +| | TableScan: merge_semi_data, partial_filters=[merge_semi_data.key_id IS NOT NULL] | +| | ]] | +| | MergeScan [is_placeholder=false, remote_input=[ | +| | SubqueryAlias: __correlated_sq_1 | +| | Projection: merge_semi_data.key_id | +| | Filter: merge_semi_data.is_match AND merge_semi_data.key_id IS NOT NULL | +| | TableScan: merge_semi_data, partial_filters=[merge_semi_data.is_match, merge_semi_data.key_id IS NOT NULL] | +| | ]] | +| physical_plan | SortPreservingMergeExec: [key_id@0 ASC NULLS LAST] | +| | ProjectionExec: expr=[key_id@0 as key_id, min(merge_semi_data.ts)@1 as first_ts] | +| | SortExec: expr=[key_id@0 ASC NULLS LAST], preserve_partitioning=[true] | +| | AggregateExec: mode=SinglePartitioned, gby=[key_id@1 as key_id], aggr=[min(merge_semi_data.ts)] | +| | HashJoinExec: mode=Partitioned, join_type=LeftSemi, on=[(key_id@1, key_id@0)] | +| | RepartitionExec: partitioning=Hash([key_id@1], 2), input_partitions=2 | | | MergeScanExec: peers=REDACTED -| | | -+---------------+---------------------------------------------------------------------------------------------------------------------------------------+ +| | RepartitionExec: partitioning=Hash([key_id@0], 2), input_partitions=2 | +| | MergeScanExec: peers=REDACTED +| | | ++---------------+-------------------------------------------------------------------------------------------------------------------------------------+ -- SQLNESS PROTOCOL MYSQL SELECT key_id, MIN(ts) AS first_ts @@ -220,35 +218,34 @@ WHERE key_id IN ( ) GROUP BY key_id ORDER BY key_id; -+---------------+-------------------------------------------------------------------------------------------------------------------------------------+ -| plan_type | plan | -+---------------+-------------------------------------------------------------------------------------------------------------------------------------+ -| logical_plan | Sort: merge_semi_data.key_id ASC NULLS LAST | -| | Projection: merge_semi_data.key_id, min(merge_semi_data.ts) AS first_ts | -| | Aggregate: groupBy=[[merge_semi_data.key_id]], aggr=[[min(merge_semi_data.ts)]] | -| | LeftSemi Join: merge_semi_data.key_id = __correlated_sq_1.key_id | -| | Projection: merge_semi_data.ts, merge_semi_data.key_id | -| | MergeScan [is_placeholder=false, remote_input=[ | -| | Filter: merge_semi_data.key_id IS NOT NULL | -| | TableScan: merge_semi_data, partial_filters=[merge_semi_data.key_id IS NOT NULL] | -| | ]] | -| | MergeScan [is_placeholder=false, remote_input=[ | -| | SubqueryAlias: __correlated_sq_1 | -| | Projection: merge_semi_data.key_id | -| | Filter: merge_semi_data.is_match AND merge_semi_data.key_id IS NOT NULL | -| | TableScan: merge_semi_data, partial_filters=[merge_semi_data.is_match, merge_semi_data.key_id IS NOT NULL] | -| | ]] | -| physical_plan | ProjectionExec: expr=[key_id@0 as key_id, min(merge_semi_data.ts)@1 as first_ts] | -| | SortExec: expr=[key_id@0 ASC NULLS LAST], preserve_partitioning=[false] | -| | AggregateExec: mode=Single, gby=[key_id@1 as key_id], aggr=[min(merge_semi_data.ts)] | -| | HashJoinExec: mode=CollectLeft, join_type=LeftSemi, on=[(key_id@1, key_id@0)] | -| | ProjectionExec: expr=[ts@0 as ts, key_id@1 as key_id] | -| | CooperativeExec | -| | MergeScanExec: peers=REDACTED -| | CooperativeExec | ++---------------+-----------------------------------------------------------------------------------------------------------------------------------+ +| plan_type | plan | ++---------------+-----------------------------------------------------------------------------------------------------------------------------------+ +| logical_plan | Sort: merge_semi_data.key_id ASC NULLS LAST | +| | Projection: merge_semi_data.key_id, min(merge_semi_data.ts) AS first_ts | +| | Aggregate: groupBy=[[merge_semi_data.key_id]], aggr=[[min(merge_semi_data.ts)]] | +| | LeftSemi Join: merge_semi_data.key_id = __correlated_sq_1.key_id | +| | MergeScan [is_placeholder=false, remote_input=[ | +| | Projection: merge_semi_data.ts, merge_semi_data.key_id | +| | Filter: merge_semi_data.key_id IS NOT NULL | +| | TableScan: merge_semi_data, partial_filters=[merge_semi_data.key_id IS NOT NULL] | +| | ]] | +| | MergeScan [is_placeholder=false, remote_input=[ | +| | SubqueryAlias: __correlated_sq_1 | +| | Projection: merge_semi_data.key_id | +| | Filter: merge_semi_data.is_match AND merge_semi_data.key_id IS NOT NULL | +| | TableScan: merge_semi_data, partial_filters=[merge_semi_data.is_match, merge_semi_data.key_id IS NOT NULL] | +| | ]] | +| physical_plan | ProjectionExec: expr=[key_id@0 as key_id, min(merge_semi_data.ts)@1 as first_ts] | +| | SortExec: expr=[key_id@0 ASC NULLS LAST], preserve_partitioning=[false] | +| | AggregateExec: mode=Single, gby=[key_id@1 as key_id], aggr=[min(merge_semi_data.ts)] | +| | HashJoinExec: mode=CollectLeft, join_type=LeftSemi, on=[(key_id@1, key_id@0)] | +| | CooperativeExec | | | MergeScanExec: peers=REDACTED -| | | -+---------------+-------------------------------------------------------------------------------------------------------------------------------------+ +| | CooperativeExec | +| | MergeScanExec: peers=REDACTED +| | | ++---------------+-----------------------------------------------------------------------------------------------------------------------------------+ -- SQLNESS PROTOCOL MYSQL SELECT key_id, MIN(ts) AS first_ts @@ -313,36 +310,35 @@ WHERE key_id IN ( ) GROUP BY key_id ORDER BY key_id; -+---------------+---------------------------------------------------------------------------------------------------------------------------------------+ -| plan_type | plan | -+---------------+---------------------------------------------------------------------------------------------------------------------------------------+ -| logical_plan | Sort: merge_semi_data.key_id ASC NULLS LAST | -| | Projection: merge_semi_data.key_id, min(merge_semi_data.ts) AS first_ts | -| | Aggregate: groupBy=[[merge_semi_data.key_id]], aggr=[[min(merge_semi_data.ts)]] | -| | LeftSemi Join: merge_semi_data.key_id = __correlated_sq_1.key_id | -| | Projection: merge_semi_data.ts, merge_semi_data.key_id | -| | MergeScan [is_placeholder=false, remote_input=[ | -| | Filter: merge_semi_data.key_id IS NOT NULL | -| | TableScan: merge_semi_data, partial_filters=[merge_semi_data.key_id IS NOT NULL] | -| | ]] | -| | MergeScan [is_placeholder=false, remote_input=[ | -| | SubqueryAlias: __correlated_sq_1 | -| | Projection: merge_semi_data.key_id | -| | Filter: merge_semi_data.is_match AND merge_semi_data.key_id IS NOT NULL | -| | TableScan: merge_semi_data, partial_filters=[merge_semi_data.is_match, merge_semi_data.key_id IS NOT NULL] | -| | ]] | -| physical_plan | SortPreservingMergeExec: [key_id@0 ASC NULLS LAST] | -| | ProjectionExec: expr=[key_id@0 as key_id, min(merge_semi_data.ts)@1 as first_ts] | -| | SortExec: expr=[key_id@0 ASC NULLS LAST], preserve_partitioning=[true] | -| | AggregateExec: mode=SinglePartitioned, gby=[key_id@1 as key_id], aggr=[min(merge_semi_data.ts)] | -| | HashJoinExec: mode=Partitioned, join_type=LeftSemi, on=[(key_id@1, key_id@0)] | -| | RepartitionExec: partitioning=Hash([key_id@1], 2), input_partitions=2 | -| | ProjectionExec: expr=[ts@0 as ts, key_id@1 as key_id] | -| | MergeScanExec: peers=REDACTED -| | RepartitionExec: partitioning=Hash([key_id@0], 2), input_partitions=2 | ++---------------+-------------------------------------------------------------------------------------------------------------------------------------+ +| plan_type | plan | ++---------------+-------------------------------------------------------------------------------------------------------------------------------------+ +| logical_plan | Sort: merge_semi_data.key_id ASC NULLS LAST | +| | Projection: merge_semi_data.key_id, min(merge_semi_data.ts) AS first_ts | +| | Aggregate: groupBy=[[merge_semi_data.key_id]], aggr=[[min(merge_semi_data.ts)]] | +| | LeftSemi Join: merge_semi_data.key_id = __correlated_sq_1.key_id | +| | MergeScan [is_placeholder=false, remote_input=[ | +| | Projection: merge_semi_data.ts, merge_semi_data.key_id | +| | Filter: merge_semi_data.key_id IS NOT NULL | +| | TableScan: merge_semi_data, partial_filters=[merge_semi_data.key_id IS NOT NULL] | +| | ]] | +| | MergeScan [is_placeholder=false, remote_input=[ | +| | SubqueryAlias: __correlated_sq_1 | +| | Projection: merge_semi_data.key_id | +| | Filter: merge_semi_data.is_match AND merge_semi_data.key_id IS NOT NULL | +| | TableScan: merge_semi_data, partial_filters=[merge_semi_data.is_match, merge_semi_data.key_id IS NOT NULL] | +| | ]] | +| physical_plan | SortPreservingMergeExec: [key_id@0 ASC NULLS LAST] | +| | ProjectionExec: expr=[key_id@0 as key_id, min(merge_semi_data.ts)@1 as first_ts] | +| | SortExec: expr=[key_id@0 ASC NULLS LAST], preserve_partitioning=[true] | +| | AggregateExec: mode=SinglePartitioned, gby=[key_id@1 as key_id], aggr=[min(merge_semi_data.ts)] | +| | HashJoinExec: mode=Partitioned, join_type=LeftSemi, on=[(key_id@1, key_id@0)] | +| | RepartitionExec: partitioning=Hash([key_id@1], 2), input_partitions=2 | | | MergeScanExec: peers=REDACTED -| | | -+---------------+---------------------------------------------------------------------------------------------------------------------------------------+ +| | RepartitionExec: partitioning=Hash([key_id@0], 2), input_partitions=2 | +| | MergeScanExec: peers=REDACTED +| | | ++---------------+-------------------------------------------------------------------------------------------------------------------------------------+ -- SQLNESS PROTOCOL MYSQL SELECT key_id, MIN(ts) AS first_ts @@ -398,36 +394,35 @@ WHERE key_id IN ( ) GROUP BY key_id ORDER BY key_id; -+---------------+---------------------------------------------------------------------------------------------------------------------------------------+ -| plan_type | plan | -+---------------+---------------------------------------------------------------------------------------------------------------------------------------+ -| logical_plan | Sort: merge_semi_data.key_id ASC NULLS LAST | -| | Projection: merge_semi_data.key_id, min(merge_semi_data.ts) AS first_ts | -| | Aggregate: groupBy=[[merge_semi_data.key_id]], aggr=[[min(merge_semi_data.ts)]] | -| | LeftSemi Join: merge_semi_data.key_id = __correlated_sq_1.key_id | -| | Projection: merge_semi_data.ts, merge_semi_data.key_id | -| | MergeScan [is_placeholder=false, remote_input=[ | -| | Filter: merge_semi_data.key_id IS NOT NULL | -| | TableScan: merge_semi_data, partial_filters=[merge_semi_data.key_id IS NOT NULL] | -| | ]] | -| | MergeScan [is_placeholder=false, remote_input=[ | -| | SubqueryAlias: __correlated_sq_1 | -| | Projection: merge_semi_data.key_id | -| | Filter: merge_semi_data.is_match AND merge_semi_data.key_id IS NOT NULL | -| | TableScan: merge_semi_data, partial_filters=[merge_semi_data.is_match, merge_semi_data.key_id IS NOT NULL] | -| | ]] | -| physical_plan | SortPreservingMergeExec: [key_id@0 ASC NULLS LAST] | -| | ProjectionExec: expr=[key_id@0 as key_id, min(merge_semi_data.ts)@1 as first_ts] | -| | SortExec: expr=[key_id@0 ASC NULLS LAST], preserve_partitioning=[true] | -| | AggregateExec: mode=SinglePartitioned, gby=[key_id@1 as key_id], aggr=[min(merge_semi_data.ts)] | -| | HashJoinExec: mode=Partitioned, join_type=LeftSemi, on=[(key_id@1, key_id@0)] | -| | RepartitionExec: partitioning=Hash([key_id@1], 4), input_partitions=4 | -| | ProjectionExec: expr=[ts@0 as ts, key_id@1 as key_id] | -| | MergeScanExec: peers=REDACTED -| | RepartitionExec: partitioning=Hash([key_id@0], 4), input_partitions=4 | ++---------------+-------------------------------------------------------------------------------------------------------------------------------------+ +| plan_type | plan | ++---------------+-------------------------------------------------------------------------------------------------------------------------------------+ +| logical_plan | Sort: merge_semi_data.key_id ASC NULLS LAST | +| | Projection: merge_semi_data.key_id, min(merge_semi_data.ts) AS first_ts | +| | Aggregate: groupBy=[[merge_semi_data.key_id]], aggr=[[min(merge_semi_data.ts)]] | +| | LeftSemi Join: merge_semi_data.key_id = __correlated_sq_1.key_id | +| | MergeScan [is_placeholder=false, remote_input=[ | +| | Projection: merge_semi_data.ts, merge_semi_data.key_id | +| | Filter: merge_semi_data.key_id IS NOT NULL | +| | TableScan: merge_semi_data, partial_filters=[merge_semi_data.key_id IS NOT NULL] | +| | ]] | +| | MergeScan [is_placeholder=false, remote_input=[ | +| | SubqueryAlias: __correlated_sq_1 | +| | Projection: merge_semi_data.key_id | +| | Filter: merge_semi_data.is_match AND merge_semi_data.key_id IS NOT NULL | +| | TableScan: merge_semi_data, partial_filters=[merge_semi_data.is_match, merge_semi_data.key_id IS NOT NULL] | +| | ]] | +| physical_plan | SortPreservingMergeExec: [key_id@0 ASC NULLS LAST] | +| | ProjectionExec: expr=[key_id@0 as key_id, min(merge_semi_data.ts)@1 as first_ts] | +| | SortExec: expr=[key_id@0 ASC NULLS LAST], preserve_partitioning=[true] | +| | AggregateExec: mode=SinglePartitioned, gby=[key_id@1 as key_id], aggr=[min(merge_semi_data.ts)] | +| | HashJoinExec: mode=Partitioned, join_type=LeftSemi, on=[(key_id@1, key_id@0)] | +| | RepartitionExec: partitioning=Hash([key_id@1], 4), input_partitions=4 | | | MergeScanExec: peers=REDACTED -| | | -+---------------+---------------------------------------------------------------------------------------------------------------------------------------+ +| | RepartitionExec: partitioning=Hash([key_id@0], 4), input_partitions=4 | +| | MergeScanExec: peers=REDACTED +| | | ++---------------+-------------------------------------------------------------------------------------------------------------------------------------+ DROP TABLE merge_semi_data; diff --git a/tests/cases/standalone/common/promql/tsid_binary_join_regression.result b/tests/cases/standalone/common/promql/tsid_binary_join_regression.result index 144f9defc35..4219b3a4787 100644 --- a/tests/cases/standalone/common/promql/tsid_binary_join_regression.result +++ b/tests/cases/standalone/common/promql/tsid_binary_join_regression.result @@ -119,7 +119,6 @@ TQL ANALYZE (0, 5, '5s') tsid_binary_join_left / tsid_binary_join_right; +-+-+-+ | 0_| 0_|_ProjectionExec: expr=[host@0 as host, job@1 as job, ts@2 as ts, __tsid@3 as __tsid, greptime_value@4 / greptime_value@5 as tsid_binary_join_left.greptime_value / tsid_binary_join_right.greptime_value] REDACTED |_|_|_HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(__tsid@1, __tsid@3), (ts@2, ts@4)], projection=[host@4, job@5, ts@7, __tsid@6, greptime_value@0, greptime_value@3], NullsEqual: true REDACTED -|_|_|_ProjectionExec: expr=[greptime_value@0 as greptime_value, __tsid@3 as __tsid, ts@4 as ts] REDACTED |_|_|_CooperativeExec REDACTED |_|_|_MergeScanExec: REDACTED |_|_|_RepartitionExec: partitioning=REDACTED @@ -127,9 +126,9 @@ TQL ANALYZE (0, 5, '5s') tsid_binary_join_left / tsid_binary_join_right; |_|_|_| | 1_| 0_|_PromInstantManipulateExec: range=[0..5000], lookback=[300000], interval=[5000], time index=[ts] REDACTED |_|_|_PromSeriesDivideExec: tags=["__tsid"] REDACTED -|_|_|_ProjectionExec: expr=[greptime_value@1 as greptime_value, host@3 as host, job@4 as job, __tsid@2 as __tsid, ts@0 as ts] REDACTED +|_|_|_ProjectionExec: expr=[greptime_value@1 as greptime_value, __tsid@2 as __tsid, ts@0 as ts] REDACTED |_|_|_CooperativeExec REDACTED -|_|_|_SeriesScan: region=REDACTED, "partition_count":REDACTED, "distribution":"PerSeries", "projection": ["ts", "greptime_value", "__tsid", "host", "job"], "filters": ["__table_id=UInt32(REDACTED)", "ts >= TimestampMillisecond(-299999, None)", "ts <= TimestampMillisecond(5000, None)"], "time_range":REDACTED +|_|_|_SeriesScan: region=REDACTED, "partition_count":REDACTED, "distribution":"PerSeries", "projection": ["ts", "greptime_value", "__tsid"], "filters": ["__table_id=UInt32(REDACTED)", "ts >= TimestampMillisecond(-299999, None)", "ts <= TimestampMillisecond(5000, None)"], "time_range":REDACTED |_|_|_| | 1_| 0_|_PromInstantManipulateExec: range=[0..5000], lookback=[300000], interval=[5000], time index=[ts] REDACTED |_|_|_PromSeriesDivideExec: tags=["__tsid"] REDACTED @@ -164,7 +163,6 @@ TQL ANALYZE (0, 5, '5s') (tsid_binary_join_left + tsid_binary_join_right) / tsid |_|_|_RepartitionExec: partitioning=Hash([__tsid@3, ts@4],REDACTED |_|_|_MergeScanExec: REDACTED |_|_|_RepartitionExec: partitioning=Hash([__tsid@1, ts@2],REDACTED -|_|_|_ProjectionExec: expr=[greptime_value@0 as greptime_value, __tsid@3 as __tsid, ts@4 as ts] REDACTED |_|_|_MergeScanExec: REDACTED |_|_|_| | 1_| 0_|_PromInstantManipulateExec: range=[0..5000], lookback=[300000], interval=[5000], time index=[ts] REDACTED @@ -175,9 +173,9 @@ TQL ANALYZE (0, 5, '5s') (tsid_binary_join_left + tsid_binary_join_right) / tsid |_|_|_| | 1_| 0_|_PromInstantManipulateExec: range=[0..5000], lookback=[300000], interval=[5000], time index=[ts] REDACTED |_|_|_PromSeriesDivideExec: tags=["__tsid"] REDACTED -|_|_|_ProjectionExec: expr=[greptime_value@1 as greptime_value, host@3 as host, job@4 as job, __tsid@2 as __tsid, ts@0 as ts] REDACTED +|_|_|_ProjectionExec: expr=[greptime_value@1 as greptime_value, __tsid@2 as __tsid, ts@0 as ts] REDACTED |_|_|_CooperativeExec REDACTED -|_|_|_SeriesScan: region=REDACTED, "partition_count":REDACTED, "distribution":"PerSeries", "projection": ["ts", "greptime_value", "__tsid", "host", "job"], "filters": ["__table_id=UInt32(REDACTED)", "ts >= TimestampMillisecond(-299999, None)", "ts <= TimestampMillisecond(5000, None)"], "time_range":REDACTED +|_|_|_SeriesScan: region=REDACTED, "partition_count":REDACTED, "distribution":"PerSeries", "projection": ["ts", "greptime_value", "__tsid"], "filters": ["__table_id=UInt32(REDACTED)", "ts >= TimestampMillisecond(-299999, None)", "ts <= TimestampMillisecond(5000, None)"], "time_range":REDACTED |_|_|_| |_|_| Total rows: 4_| +-+-+-+ @@ -206,10 +204,8 @@ TQL ANALYZE (0, 5, '5s') ((tsid_binary_join_left + tsid_binary_join_right) * (ts |_|_|_RepartitionExec: partitioning=Hash([REDACTED |_|_|_MergeScanExec: REDACTED |_|_|_RepartitionExec: partitioning=Hash([REDACTED -|_|_|_ProjectionExec: expr=[greptime_value@0 as greptime_value, __tsid@3 as __tsid, ts@4 as ts] REDACTED |_|_|_MergeScanExec: REDACTED |_|_|_RepartitionExec: partitioning=Hash([REDACTED -|_|_|_ProjectionExec: expr=[greptime_value@0 as greptime_value, __tsid@3 as __tsid, ts@4 as ts] REDACTED |_|_|_MergeScanExec: REDACTED |_|_|_| | 1_| 0_|_PromInstantManipulateExec: range=[0..5000], lookback=[300000], interval=[5000], time index=[ts] REDACTED @@ -220,15 +216,15 @@ TQL ANALYZE (0, 5, '5s') ((tsid_binary_join_left + tsid_binary_join_right) * (ts |_|_|_| | 1_| 0_|_PromInstantManipulateExec: range=[0..5000], lookback=[300000], interval=[5000], time index=[ts] REDACTED |_|_|_PromSeriesDivideExec: tags=["__tsid"] REDACTED -|_|_|_ProjectionExec: expr=[greptime_value@1 as greptime_value, host@3 as host, job@4 as job, __tsid@2 as __tsid, ts@0 as ts] REDACTED +|_|_|_ProjectionExec: expr=[greptime_value@1 as greptime_value, __tsid@2 as __tsid, ts@0 as ts] REDACTED |_|_|_CooperativeExec REDACTED -|_|_|_SeriesScan: region=REDACTED, "partition_count":REDACTED, "distribution":"PerSeries", "projection": ["ts", "greptime_value", "__tsid", "host", "job"], "filters": ["__table_id=UInt32(REDACTED)", "ts >= TimestampMillisecond(-299999, None)", "ts <= TimestampMillisecond(5000, None)"], "time_range":REDACTED +|_|_|_SeriesScan: region=REDACTED, "partition_count":REDACTED, "distribution":"PerSeries", "projection": ["ts", "greptime_value", "__tsid"], "filters": ["__table_id=UInt32(REDACTED)", "ts >= TimestampMillisecond(-299999, None)", "ts <= TimestampMillisecond(5000, None)"], "time_range":REDACTED |_|_|_| | 1_| 0_|_PromInstantManipulateExec: range=[0..5000], lookback=[300000], interval=[5000], time index=[ts] REDACTED |_|_|_PromSeriesDivideExec: tags=["__tsid"] REDACTED -|_|_|_ProjectionExec: expr=[greptime_value@1 as greptime_value, host@3 as host, job@4 as job, __tsid@2 as __tsid, ts@0 as ts] REDACTED +|_|_|_ProjectionExec: expr=[greptime_value@1 as greptime_value, __tsid@2 as __tsid, ts@0 as ts] REDACTED |_|_|_CooperativeExec REDACTED -|_|_|_SeriesScan: region=REDACTED, "partition_count":REDACTED, "distribution":"PerSeries", "projection": ["ts", "greptime_value", "__tsid", "host", "job"], "filters": ["__table_id=UInt32(REDACTED)", "ts >= TimestampMillisecond(-299999, None)", "ts <= TimestampMillisecond(5000, None)"], "time_range":REDACTED +|_|_|_SeriesScan: region=REDACTED, "partition_count":REDACTED, "distribution":"PerSeries", "projection": ["ts", "greptime_value", "__tsid"], "filters": ["__table_id=UInt32(REDACTED)", "ts >= TimestampMillisecond(-299999, None)", "ts <= TimestampMillisecond(5000, None)"], "time_range":REDACTED |_|_|_| |_|_| Total rows: 4_| +-+-+-+ @@ -257,26 +253,26 @@ TQL ANALYZE (0, 5, '5s') tsid_binary_join_left / ignoring(host) tsid_binary_join |_|_|_WindowAggExec: wdw=[__promql_match_group_count: Ok(Field { name: "__promql_match_group_count", data_type: Int64 }), frame: WindowFrame { units: Rows, start_bound: Preceding(UInt64(NULL)), end_bound: Following(UInt64(NULL)), is_causal: false }] REDACTED |_|_|_HashJoinExec: mode=Partitioned, join_type=Inner, on=[(job@1, job@1), (ts@2, ts@2)], projection=[greptime_value@0, job@1, ts@2, greptime_value@3, ts@5], NullsEqual: true REDACTED |_|_|_RepartitionExec: partitioning=Hash([job@1, ts@2],REDACTED -|_|_|_ProjectionExec: expr=[greptime_value@0 as greptime_value, job@2 as job, ts@4 as ts] REDACTED |_|_|_MergeScanExec: REDACTED |_|_|_FilterExec: prom_assert_unique_match_group(__promql_match_group_count@3, job@1), projection=[greptime_value@0, job@1, ts@2] REDACTED |_|_|_WindowAggExec: wdw=[__promql_match_group_count: Ok(Field { name: "__promql_match_group_count", data_type: Int64 }), frame: WindowFrame { units: Rows, start_bound: Preceding(UInt64(NULL)), end_bound: Following(UInt64(NULL)), is_causal: false }] REDACTED |_|_|_SortExec: expr=[job@1 ASC NULLS LAST, ts@2 ASC NULLS LAST], preserve_partitioning=[true] REDACTED |_|_|_RepartitionExec: partitioning=Hash([job@1, ts@2],REDACTED -|_|_|_ProjectionExec: expr=[greptime_value@0 as greptime_value, job@2 as job, ts@4 as ts] REDACTED |_|_|_MergeScanExec: REDACTED |_|_|_| -| 1_| 0_|_PromInstantManipulateExec: range=[0..5000], lookback=[300000], interval=[5000], time index=[ts] REDACTED +| 1_| 0_|_ProjectionExec: expr=[greptime_value@0 as greptime_value, job@1 as job, ts@3 as ts] REDACTED +|_|_|_PromInstantManipulateExec: range=[0..5000], lookback=[300000], interval=[5000], time index=[ts] REDACTED |_|_|_PromSeriesDivideExec: tags=["__tsid"] REDACTED -|_|_|_ProjectionExec: expr=[greptime_value@1 as greptime_value, host@3 as host, job@4 as job, __tsid@2 as __tsid, ts@0 as ts] REDACTED +|_|_|_ProjectionExec: expr=[greptime_value@1 as greptime_value, job@3 as job, __tsid@2 as __tsid, ts@0 as ts] REDACTED |_|_|_CooperativeExec REDACTED -|_|_|_SeriesScan: region=REDACTED, "partition_count":REDACTED, "distribution":"PerSeries", "projection": ["ts", "greptime_value", "__tsid", "host", "job"], "filters": ["__table_id=UInt32(REDACTED)", "ts >= TimestampMillisecond(-299999, None)", "ts <= TimestampMillisecond(5000, None)"], "time_range":REDACTED +|_|_|_SeriesScan: region=REDACTED, "partition_count":REDACTED, "distribution":"PerSeries", "projection": ["ts", "greptime_value", "__tsid", "job"], "filters": ["__table_id=UInt32(REDACTED)", "ts >= TimestampMillisecond(-299999, None)", "ts <= TimestampMillisecond(5000, None)"], "time_range":REDACTED |_|_|_| -| 1_| 0_|_PromInstantManipulateExec: range=[0..5000], lookback=[300000], interval=[5000], time index=[ts] REDACTED +| 1_| 0_|_ProjectionExec: expr=[greptime_value@0 as greptime_value, job@1 as job, ts@3 as ts] REDACTED +|_|_|_PromInstantManipulateExec: range=[0..5000], lookback=[300000], interval=[5000], time index=[ts] REDACTED |_|_|_PromSeriesDivideExec: tags=["__tsid"] REDACTED -|_|_|_ProjectionExec: expr=[greptime_value@1 as greptime_value, host@3 as host, job@4 as job, __tsid@2 as __tsid, ts@0 as ts] REDACTED +|_|_|_ProjectionExec: expr=[greptime_value@1 as greptime_value, job@3 as job, __tsid@2 as __tsid, ts@0 as ts] REDACTED |_|_|_CooperativeExec REDACTED -|_|_|_SeriesScan: region=REDACTED, "partition_count":REDACTED, "distribution":"PerSeries", "projection": ["ts", "greptime_value", "__tsid", "host", "job"], "filters": ["__table_id=UInt32(REDACTED)", "ts >= TimestampMillisecond(-299999, None)", "ts <= TimestampMillisecond(5000, None)"], "time_range":REDACTED +|_|_|_SeriesScan: region=REDACTED, "partition_count":REDACTED, "distribution":"PerSeries", "projection": ["ts", "greptime_value", "__tsid", "job"], "filters": ["__table_id=UInt32(REDACTED)", "ts >= TimestampMillisecond(-299999, None)", "ts <= TimestampMillisecond(5000, None)"], "time_range":REDACTED |_|_|_| |_|_| Total rows: 4_| +-+-+-+ @@ -308,19 +304,19 @@ TQL ANALYZE (0, 5, '5s') tsid_binary_join_left / on(job) tsid_binary_join_right_ |_|_|_CoalescePartitionsExec REDACTED |_|_|_HashJoinExec: mode=Partitioned, join_type=Inner, on=[(job@1, job@1), (ts@2, ts@2)], projection=[greptime_value@0, job@1, ts@2, greptime_value@3, ts@5], NullsEqual: true REDACTED |_|_|_RepartitionExec: partitioning=Hash([job@1, ts@2],REDACTED -|_|_|_ProjectionExec: expr=[greptime_value@0 as greptime_value, job@2 as job, ts@4 as ts] REDACTED |_|_|_MergeScanExec: REDACTED |_|_|_RepartitionExec: partitioning=Hash([job@1, ts@2],REDACTED -|_|_|_ProjectionExec: expr=[greptime_value@0 as greptime_value, job@1 as job, ts@3 as ts] REDACTED |_|_|_MergeScanExec: REDACTED |_|_|_| -| 1_| 0_|_PromInstantManipulateExec: range=[0..5000], lookback=[300000], interval=[5000], time index=[ts] REDACTED +| 1_| 0_|_ProjectionExec: expr=[greptime_value@0 as greptime_value, job@1 as job, ts@3 as ts] REDACTED +|_|_|_PromInstantManipulateExec: range=[0..5000], lookback=[300000], interval=[5000], time index=[ts] REDACTED |_|_|_PromSeriesDivideExec: tags=["__tsid"] REDACTED -|_|_|_ProjectionExec: expr=[greptime_value@1 as greptime_value, host@3 as host, job@4 as job, __tsid@2 as __tsid, ts@0 as ts] REDACTED +|_|_|_ProjectionExec: expr=[greptime_value@1 as greptime_value, job@3 as job, __tsid@2 as __tsid, ts@0 as ts] REDACTED |_|_|_CooperativeExec REDACTED -|_|_|_SeriesScan: region=REDACTED, "partition_count":REDACTED, "distribution":"PerSeries", "projection": ["ts", "greptime_value", "__tsid", "host", "job"], "filters": ["__table_id=UInt32(REDACTED)", "ts >= TimestampMillisecond(-299999, None)", "ts <= TimestampMillisecond(5000, None)"], "time_range":REDACTED +|_|_|_SeriesScan: region=REDACTED, "partition_count":REDACTED, "distribution":"PerSeries", "projection": ["ts", "greptime_value", "__tsid", "job"], "filters": ["__table_id=UInt32(REDACTED)", "ts >= TimestampMillisecond(-299999, None)", "ts <= TimestampMillisecond(5000, None)"], "time_range":REDACTED |_|_|_| -| 1_| 0_|_PromInstantManipulateExec: range=[0..5000], lookback=[300000], interval=[5000], time index=[ts] REDACTED +| 1_| 0_|_ProjectionExec: expr=[greptime_value@0 as greptime_value, job@1 as job, ts@3 as ts] REDACTED +|_|_|_PromInstantManipulateExec: range=[0..5000], lookback=[300000], interval=[5000], time index=[ts] REDACTED |_|_|_PromSeriesDivideExec: tags=["__tsid"] REDACTED |_|_|_ProjectionExec: expr=[greptime_value@1 as greptime_value, job@3 as job, __tsid@2 as __tsid, ts@0 as ts] REDACTED |_|_|_CooperativeExec REDACTED @@ -352,7 +348,6 @@ TQL ANALYZE (0, 5, '5s') tsid_binary_join_left > tsid_binary_join_right; |_|_|_RepartitionExec: partitioning=Hash([__tsid@3, ts@4],REDACTED |_|_|_MergeScanExec: REDACTED |_|_|_RepartitionExec: partitioning=Hash([__tsid@1, ts@2],REDACTED -|_|_|_ProjectionExec: expr=[greptime_value@0 as greptime_value, __tsid@3 as __tsid, ts@4 as ts] REDACTED |_|_|_MergeScanExec: REDACTED |_|_|_| | 1_| 0_|_PromInstantManipulateExec: range=[0..5000], lookback=[300000], interval=[5000], time index=[ts] REDACTED @@ -363,9 +358,9 @@ TQL ANALYZE (0, 5, '5s') tsid_binary_join_left > tsid_binary_join_right; |_|_|_| | 1_| 0_|_PromInstantManipulateExec: range=[0..5000], lookback=[300000], interval=[5000], time index=[ts] REDACTED |_|_|_PromSeriesDivideExec: tags=["__tsid"] REDACTED -|_|_|_ProjectionExec: expr=[greptime_value@1 as greptime_value, host@3 as host, job@4 as job, __tsid@2 as __tsid, ts@0 as ts] REDACTED +|_|_|_ProjectionExec: expr=[greptime_value@1 as greptime_value, __tsid@2 as __tsid, ts@0 as ts] REDACTED |_|_|_CooperativeExec REDACTED -|_|_|_SeriesScan: region=REDACTED, "partition_count":REDACTED, "distribution":"PerSeries", "projection": ["ts", "greptime_value", "__tsid", "host", "job"], "filters": ["__table_id=UInt32(REDACTED)", "ts >= TimestampMillisecond(-299999, None)", "ts <= TimestampMillisecond(5000, None)"], "time_range":REDACTED +|_|_|_SeriesScan: region=REDACTED, "partition_count":REDACTED, "distribution":"PerSeries", "projection": ["ts", "greptime_value", "__tsid"], "filters": ["__table_id=UInt32(REDACTED)", "ts >= TimestampMillisecond(-299999, None)", "ts <= TimestampMillisecond(5000, None)"], "time_range":REDACTED |_|_|_| |_|_| Total rows: 4_| +-+-+-+ @@ -390,7 +385,6 @@ TQL ANALYZE (0, 5, '5s') tsid_binary_join_left > bool tsid_binary_join_right; +-+-+-+ | 0_| 0_|_ProjectionExec: expr=[host@0 as host, job@1 as job, ts@2 as ts, __tsid@3 as __tsid, CAST(prom_gt(greptime_value@4, greptime_value@5) AS Float64) as tsid_binary_join_left.greptime_value > tsid_binary_join_right.greptime_value] REDACTED |_|_|_HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(__tsid@1, __tsid@3), (ts@2, ts@4)], projection=[host@4, job@5, ts@7, __tsid@6, greptime_value@0, greptime_value@3], NullsEqual: true REDACTED -|_|_|_ProjectionExec: expr=[greptime_value@0 as greptime_value, __tsid@3 as __tsid, ts@4 as ts] REDACTED |_|_|_CooperativeExec REDACTED |_|_|_MergeScanExec: REDACTED |_|_|_RepartitionExec: partitioning=REDACTED @@ -398,9 +392,9 @@ TQL ANALYZE (0, 5, '5s') tsid_binary_join_left > bool tsid_binary_join_right; |_|_|_| | 1_| 0_|_PromInstantManipulateExec: range=[0..5000], lookback=[300000], interval=[5000], time index=[ts] REDACTED |_|_|_PromSeriesDivideExec: tags=["__tsid"] REDACTED -|_|_|_ProjectionExec: expr=[greptime_value@1 as greptime_value, host@3 as host, job@4 as job, __tsid@2 as __tsid, ts@0 as ts] REDACTED +|_|_|_ProjectionExec: expr=[greptime_value@1 as greptime_value, __tsid@2 as __tsid, ts@0 as ts] REDACTED |_|_|_CooperativeExec REDACTED -|_|_|_SeriesScan: region=REDACTED, "partition_count":REDACTED, "distribution":"PerSeries", "projection": ["ts", "greptime_value", "__tsid", "host", "job"], "filters": ["__table_id=UInt32(REDACTED)", "ts >= TimestampMillisecond(-299999, None)", "ts <= TimestampMillisecond(5000, None)"], "time_range":REDACTED +|_|_|_SeriesScan: region=REDACTED, "partition_count":REDACTED, "distribution":"PerSeries", "projection": ["ts", "greptime_value", "__tsid"], "filters": ["__table_id=UInt32(REDACTED)", "ts >= TimestampMillisecond(-299999, None)", "ts <= TimestampMillisecond(5000, None)"], "time_range":REDACTED |_|_|_| | 1_| 0_|_PromInstantManipulateExec: range=[0..5000], lookback=[300000], interval=[5000], time index=[ts] REDACTED |_|_|_PromSeriesDivideExec: tags=["__tsid"] REDACTED @@ -434,25 +428,23 @@ TQL ANALYZE (0, 5, '5s') ((tsid_binary_join_left > tsid_binary_join_right) / tsi |_|_|_CoalescePartitionsExec REDACTED |_|_|_HashJoinExec: mode=Partitioned, join_type=Inner, on=[(__tsid@1, __tsid@1), (ts@2, ts@2)], filter=prom_gt(greptime_value@0, greptime_value@1), projection=[ts@2, greptime_value@0, __tsid@1], NullsEqual: true REDACTED |_|_|_RepartitionExec: partitioning=Hash([REDACTED -|_|_|_ProjectionExec: expr=[greptime_value@0 as greptime_value, __tsid@3 as __tsid, ts@4 as ts] REDACTED |_|_|_MergeScanExec: REDACTED |_|_|_RepartitionExec: partitioning=Hash([REDACTED -|_|_|_ProjectionExec: expr=[greptime_value@0 as greptime_value, __tsid@3 as __tsid, ts@4 as ts] REDACTED |_|_|_MergeScanExec: REDACTED |_|_|_RepartitionExec: partitioning=REDACTED |_|_|_MergeScanExec: REDACTED |_|_|_| | 1_| 0_|_PromInstantManipulateExec: range=[0..5000], lookback=[300000], interval=[5000], time index=[ts] REDACTED |_|_|_PromSeriesDivideExec: tags=["__tsid"] REDACTED -|_|_|_ProjectionExec: expr=[greptime_value@1 as greptime_value, host@3 as host, job@4 as job, __tsid@2 as __tsid, ts@0 as ts] REDACTED +|_|_|_ProjectionExec: expr=[greptime_value@1 as greptime_value, __tsid@2 as __tsid, ts@0 as ts] REDACTED |_|_|_CooperativeExec REDACTED -|_|_|_SeriesScan: region=REDACTED, "partition_count":REDACTED, "distribution":"PerSeries", "projection": ["ts", "greptime_value", "__tsid", "host", "job"], "filters": ["__table_id=UInt32(REDACTED)", "ts >= TimestampMillisecond(-299999, None)", "ts <= TimestampMillisecond(5000, None)"], "time_range":REDACTED +|_|_|_SeriesScan: region=REDACTED, "partition_count":REDACTED, "distribution":"PerSeries", "projection": ["ts", "greptime_value", "__tsid"], "filters": ["__table_id=UInt32(REDACTED)", "ts >= TimestampMillisecond(-299999, None)", "ts <= TimestampMillisecond(5000, None)"], "time_range":REDACTED |_|_|_| | 1_| 0_|_PromInstantManipulateExec: range=[0..5000], lookback=[300000], interval=[5000], time index=[ts] REDACTED |_|_|_PromSeriesDivideExec: tags=["__tsid"] REDACTED -|_|_|_ProjectionExec: expr=[greptime_value@1 as greptime_value, host@3 as host, job@4 as job, __tsid@2 as __tsid, ts@0 as ts] REDACTED +|_|_|_ProjectionExec: expr=[greptime_value@1 as greptime_value, __tsid@2 as __tsid, ts@0 as ts] REDACTED |_|_|_CooperativeExec REDACTED -|_|_|_SeriesScan: region=REDACTED, "partition_count":REDACTED, "distribution":"PerSeries", "projection": ["ts", "greptime_value", "__tsid", "host", "job"], "filters": ["__table_id=UInt32(REDACTED)", "ts >= TimestampMillisecond(-299999, None)", "ts <= TimestampMillisecond(5000, None)"], "time_range":REDACTED +|_|_|_SeriesScan: region=REDACTED, "partition_count":REDACTED, "distribution":"PerSeries", "projection": ["ts", "greptime_value", "__tsid"], "filters": ["__table_id=UInt32(REDACTED)", "ts >= TimestampMillisecond(-299999, None)", "ts <= TimestampMillisecond(5000, None)"], "time_range":REDACTED |_|_|_| | 1_| 0_|_PromInstantManipulateExec: range=[0..5000], lookback=[300000], interval=[5000], time index=[ts] REDACTED |_|_|_PromSeriesDivideExec: tags=["__tsid"] REDACTED @@ -488,34 +480,31 @@ TQL ANALYZE (0, 5, '5s') ((tsid_binary_join_left > bool tsid_binary_join_right) |_|_|_ProjectionExec: expr=[ts@0 as ts, __tsid@1 as __tsid, CAST(prom_gt(greptime_value@2, greptime_value@3) AS Float64) as tsid_binary_join_left.greptime_value > tsid_binary_join_right.greptime_value] REDACTED |_|_|_HashJoinExec: mode=Partitioned, join_type=Inner, on=[(__tsid@1, __tsid@1), (ts@2, ts@2)], projection=[ts@5, __tsid@4, greptime_value@0, greptime_value@3], NullsEqual: true REDACTED |_|_|_RepartitionExec: partitioning=Hash([REDACTED -|_|_|_ProjectionExec: expr=[greptime_value@0 as greptime_value, __tsid@3 as __tsid, ts@4 as ts] REDACTED |_|_|_MergeScanExec: REDACTED |_|_|_RepartitionExec: partitioning=Hash([REDACTED -|_|_|_ProjectionExec: expr=[greptime_value@0 as greptime_value, __tsid@3 as __tsid, ts@4 as ts] REDACTED |_|_|_MergeScanExec: REDACTED |_|_|_RepartitionExec: partitioning=Hash([REDACTED -|_|_|_ProjectionExec: expr=[greptime_value@0 as greptime_value, __tsid@3 as __tsid, ts@4 as ts] REDACTED |_|_|_MergeScanExec: REDACTED |_|_|_RepartitionExec: partitioning=REDACTED |_|_|_MergeScanExec: REDACTED |_|_|_| | 1_| 0_|_PromInstantManipulateExec: range=[0..5000], lookback=[300000], interval=[5000], time index=[ts] REDACTED |_|_|_PromSeriesDivideExec: tags=["__tsid"] REDACTED -|_|_|_ProjectionExec: expr=[greptime_value@1 as greptime_value, host@3 as host, job@4 as job, __tsid@2 as __tsid, ts@0 as ts] REDACTED +|_|_|_ProjectionExec: expr=[greptime_value@1 as greptime_value, __tsid@2 as __tsid, ts@0 as ts] REDACTED |_|_|_CooperativeExec REDACTED -|_|_|_SeriesScan: region=REDACTED, "partition_count":REDACTED, "distribution":"PerSeries", "projection": ["ts", "greptime_value", "__tsid", "host", "job"], "filters": ["__table_id=UInt32(REDACTED)", "ts >= TimestampMillisecond(-299999, None)", "ts <= TimestampMillisecond(5000, None)"], "time_range":REDACTED +|_|_|_SeriesScan: region=REDACTED, "partition_count":REDACTED, "distribution":"PerSeries", "projection": ["ts", "greptime_value", "__tsid"], "filters": ["__table_id=UInt32(REDACTED)", "ts >= TimestampMillisecond(-299999, None)", "ts <= TimestampMillisecond(5000, None)"], "time_range":REDACTED |_|_|_| | 1_| 0_|_PromInstantManipulateExec: range=[0..5000], lookback=[300000], interval=[5000], time index=[ts] REDACTED |_|_|_PromSeriesDivideExec: tags=["__tsid"] REDACTED -|_|_|_ProjectionExec: expr=[greptime_value@1 as greptime_value, host@3 as host, job@4 as job, __tsid@2 as __tsid, ts@0 as ts] REDACTED +|_|_|_ProjectionExec: expr=[greptime_value@1 as greptime_value, __tsid@2 as __tsid, ts@0 as ts] REDACTED |_|_|_CooperativeExec REDACTED -|_|_|_SeriesScan: region=REDACTED, "partition_count":REDACTED, "distribution":"PerSeries", "projection": ["ts", "greptime_value", "__tsid", "host", "job"], "filters": ["__table_id=UInt32(REDACTED)", "ts >= TimestampMillisecond(-299999, None)", "ts <= TimestampMillisecond(5000, None)"], "time_range":REDACTED +|_|_|_SeriesScan: region=REDACTED, "partition_count":REDACTED, "distribution":"PerSeries", "projection": ["ts", "greptime_value", "__tsid"], "filters": ["__table_id=UInt32(REDACTED)", "ts >= TimestampMillisecond(-299999, None)", "ts <= TimestampMillisecond(5000, None)"], "time_range":REDACTED |_|_|_| | 1_| 0_|_PromInstantManipulateExec: range=[0..5000], lookback=[300000], interval=[5000], time index=[ts] REDACTED |_|_|_PromSeriesDivideExec: tags=["__tsid"] REDACTED -|_|_|_ProjectionExec: expr=[greptime_value@1 as greptime_value, host@3 as host, job@4 as job, __tsid@2 as __tsid, ts@0 as ts] REDACTED +|_|_|_ProjectionExec: expr=[greptime_value@1 as greptime_value, __tsid@2 as __tsid, ts@0 as ts] REDACTED |_|_|_CooperativeExec REDACTED -|_|_|_SeriesScan: region=REDACTED, "partition_count":REDACTED, "distribution":"PerSeries", "projection": ["ts", "greptime_value", "__tsid", "host", "job"], "filters": ["__table_id=UInt32(REDACTED)", "ts >= TimestampMillisecond(-299999, None)", "ts <= TimestampMillisecond(5000, None)"], "time_range":REDACTED +|_|_|_SeriesScan: region=REDACTED, "partition_count":REDACTED, "distribution":"PerSeries", "projection": ["ts", "greptime_value", "__tsid"], "filters": ["__table_id=UInt32(REDACTED)", "ts >= TimestampMillisecond(-299999, None)", "ts <= TimestampMillisecond(5000, None)"], "time_range":REDACTED |_|_|_| | 1_| 0_|_PromInstantManipulateExec: range=[0..5000], lookback=[300000], interval=[5000], time index=[ts] REDACTED |_|_|_PromSeriesDivideExec: tags=["__tsid"] REDACTED @@ -604,28 +593,27 @@ TQL ANALYZE (0, 5, '5s') (tsid_binary_join_left / ignoring(host) group_left tsid |_|_|_ProjectionExec: expr=[ts@0 as ts, __tsid@1 as __tsid, greptime_value@2 / greptime_value@3 as tsid_binary_join_left.greptime_value / tsid_binary_join_right.greptime_value] REDACTED |_|_|_HashJoinExec: mode=Partitioned, join_type=Inner, on=[(job@1, job@1), (ts@3, ts@2)], projection=[ts@6, __tsid@2, greptime_value@0, greptime_value@4], NullsEqual: true REDACTED |_|_|_RepartitionExec: partitioning=Hash([REDACTED -|_|_|_ProjectionExec: expr=[greptime_value@0 as greptime_value, job@2 as job, __tsid@3 as __tsid, ts@4 as ts] REDACTED |_|_|_MergeScanExec: REDACTED |_|_|_FilterExec: prom_assert_unique_match_group(__promql_match_group_count@3, job@1), projection=[greptime_value@0, job@1, ts@2] REDACTED |_|_|_WindowAggExec: wdw=[__promql_match_group_count: Ok(Field { name: "__promql_match_group_count", data_type: Int64 }), frame: WindowFrame { units: Rows, start_bound: Preceding(UInt64(NULL)), end_bound: Following(UInt64(NULL)), is_causal: false }] REDACTED |_|_|_SortExec: expr=[job@1 ASC NULLS LAST, ts@2 ASC NULLS LAST], preserve_partitioning=[true] REDACTED |_|_|_RepartitionExec: partitioning=Hash([REDACTED -|_|_|_ProjectionExec: expr=[greptime_value@0 as greptime_value, job@2 as job, ts@4 as ts] REDACTED |_|_|_MergeScanExec: REDACTED |_|_|_RepartitionExec: partitioning=REDACTED |_|_|_MergeScanExec: REDACTED |_|_|_| | 1_| 0_|_PromInstantManipulateExec: range=[0..5000], lookback=[300000], interval=[5000], time index=[ts] REDACTED |_|_|_PromSeriesDivideExec: tags=["__tsid"] REDACTED -|_|_|_ProjectionExec: expr=[greptime_value@1 as greptime_value, host@3 as host, job@4 as job, __tsid@2 as __tsid, ts@0 as ts] REDACTED +|_|_|_ProjectionExec: expr=[greptime_value@1 as greptime_value, job@3 as job, __tsid@2 as __tsid, ts@0 as ts] REDACTED |_|_|_CooperativeExec REDACTED -|_|_|_SeriesScan: region=REDACTED, "partition_count":REDACTED, "distribution":"PerSeries", "projection": ["ts", "greptime_value", "__tsid", "host", "job"], "filters": ["__table_id=UInt32(REDACTED)", "ts >= TimestampMillisecond(-299999, None)", "ts <= TimestampMillisecond(5000, None)"], "time_range":REDACTED +|_|_|_SeriesScan: region=REDACTED, "partition_count":REDACTED, "distribution":"PerSeries", "projection": ["ts", "greptime_value", "__tsid", "job"], "filters": ["__table_id=UInt32(REDACTED)", "ts >= TimestampMillisecond(-299999, None)", "ts <= TimestampMillisecond(5000, None)"], "time_range":REDACTED |_|_|_| -| 1_| 0_|_PromInstantManipulateExec: range=[0..5000], lookback=[300000], interval=[5000], time index=[ts] REDACTED +| 1_| 0_|_ProjectionExec: expr=[greptime_value@0 as greptime_value, job@1 as job, ts@3 as ts] REDACTED +|_|_|_PromInstantManipulateExec: range=[0..5000], lookback=[300000], interval=[5000], time index=[ts] REDACTED |_|_|_PromSeriesDivideExec: tags=["__tsid"] REDACTED -|_|_|_ProjectionExec: expr=[greptime_value@1 as greptime_value, host@3 as host, job@4 as job, __tsid@2 as __tsid, ts@0 as ts] REDACTED +|_|_|_ProjectionExec: expr=[greptime_value@1 as greptime_value, job@3 as job, __tsid@2 as __tsid, ts@0 as ts] REDACTED |_|_|_CooperativeExec REDACTED -|_|_|_SeriesScan: region=REDACTED, "partition_count":REDACTED, "distribution":"PerSeries", "projection": ["ts", "greptime_value", "__tsid", "host", "job"], "filters": ["__table_id=UInt32(REDACTED)", "ts >= TimestampMillisecond(-299999, None)", "ts <= TimestampMillisecond(5000, None)"], "time_range":REDACTED +|_|_|_SeriesScan: region=REDACTED, "partition_count":REDACTED, "distribution":"PerSeries", "projection": ["ts", "greptime_value", "__tsid", "job"], "filters": ["__table_id=UInt32(REDACTED)", "ts >= TimestampMillisecond(-299999, None)", "ts <= TimestampMillisecond(5000, None)"], "time_range":REDACTED |_|_|_| | 1_| 0_|_PromInstantManipulateExec: range=[0..5000], lookback=[300000], interval=[5000], time index=[ts] REDACTED |_|_|_PromSeriesDivideExec: tags=["__tsid"] REDACTED diff --git a/tests/cases/standalone/common/tql-explain-analyze/explain.result b/tests/cases/standalone/common/tql-explain-analyze/explain.result index 355619b3766..77870e8f062 100644 --- a/tests/cases/standalone/common/tql-explain-analyze/explain.result +++ b/tests/cases/standalone/common/tql-explain-analyze/explain.result @@ -138,6 +138,7 @@ TQL EXPLAIN VERBOSE (0, 10, '5s') test; |_|_Filter: test.j >= TimestampMillisecond(-299999, None) AND test.j <= TimestampMillisecond(10000, None)_| |_|_TableScan: test, partial_filters=[test.j >= TimestampMillisecond(-299999, None), test.j <= TimestampMillisecond(10000, None)] | |_| ]]_| +| logical_plan after merge_scan_projection_| SAME TEXT AS ABOVE_| | logical_plan after ScanHintRule_| SAME TEXT AS ABOVE_| | logical_plan after JsonTypeConcretizeRule_| SAME TEXT AS ABOVE_| | logical_plan after rewrite_set_comparison_| SAME TEXT AS ABOVE_| @@ -166,6 +167,7 @@ TQL EXPLAIN VERBOSE (0, 10, '5s') test; | logical_plan after extract_leaf_expressions_| SAME TEXT AS ABOVE_| | logical_plan after push_down_leaf_projections_| SAME TEXT AS ABOVE_| | logical_plan after optimize_projections_| SAME TEXT AS ABOVE_| +| logical_plan after merge_scan_projection_| SAME TEXT AS ABOVE_| | logical_plan after ScanHintRule_| SAME TEXT AS ABOVE_| | logical_plan after JsonTypeConcretizeRule_| SAME TEXT AS ABOVE_| | logical_plan_| MergeScan [is_placeholder=false, remote_input=[_| @@ -297,6 +299,7 @@ TQL EXPLAIN VERBOSE (0, 10, '5s') test AS series; |_|_Filter: test.j >= TimestampMillisecond(-299999, None) AND test.j <= TimestampMillisecond(10000, None)_| |_|_TableScan: test, partial_filters=[test.j >= TimestampMillisecond(-299999, None), test.j <= TimestampMillisecond(10000, None)] | |_| ]]_| +| logical_plan after merge_scan_projection_| SAME TEXT AS ABOVE_| | logical_plan after ScanHintRule_| SAME TEXT AS ABOVE_| | logical_plan after JsonTypeConcretizeRule_| SAME TEXT AS ABOVE_| | logical_plan after rewrite_set_comparison_| SAME TEXT AS ABOVE_| @@ -325,6 +328,7 @@ TQL EXPLAIN VERBOSE (0, 10, '5s') test AS series; | logical_plan after extract_leaf_expressions_| SAME TEXT AS ABOVE_| | logical_plan after push_down_leaf_projections_| SAME TEXT AS ABOVE_| | logical_plan after optimize_projections_| SAME TEXT AS ABOVE_| +| logical_plan after merge_scan_projection_| SAME TEXT AS ABOVE_| | logical_plan after ScanHintRule_| SAME TEXT AS ABOVE_| | logical_plan after JsonTypeConcretizeRule_| SAME TEXT AS ABOVE_| | logical_plan_| MergeScan [is_placeholder=false, remote_input=[_| @@ -487,6 +491,7 @@ TQL EXPLAIN VERBOSE (0, 10, '5s') test_nano; |_|_Filter: test_nano.j > TimestampNanosecond(-300000000000, None) AND test_nano.j <= TimestampNanosecond(10000000000, None)_| |_|_TableScan: test_nano, partial_filters=[test_nano.j > TimestampNanosecond(-300000000000, None), test_nano.j <= TimestampNanosecond(10000000000, None)] | |_| ]]_| +| logical_plan after merge_scan_projection_| SAME TEXT AS ABOVE_| | logical_plan after ScanHintRule_| SAME TEXT AS ABOVE_| | logical_plan after JsonTypeConcretizeRule_| SAME TEXT AS ABOVE_| | logical_plan after rewrite_set_comparison_| SAME TEXT AS ABOVE_| @@ -515,6 +520,7 @@ TQL EXPLAIN VERBOSE (0, 10, '5s') test_nano; | logical_plan after extract_leaf_expressions_| SAME TEXT AS ABOVE_| | logical_plan after push_down_leaf_projections_| SAME TEXT AS ABOVE_| | logical_plan after optimize_projections_| SAME TEXT AS ABOVE_| +| logical_plan after merge_scan_projection_| SAME TEXT AS ABOVE_| | logical_plan after ScanHintRule_| SAME TEXT AS ABOVE_| | logical_plan after JsonTypeConcretizeRule_| SAME TEXT AS ABOVE_| | logical_plan_| MergeScan [is_placeholder=false, remote_input=[_| diff --git a/tests/cases/standalone/common/tql/tql-cte.result b/tests/cases/standalone/common/tql/tql-cte.result index c2bc149e66c..cb8ba2c1eff 100644 --- a/tests/cases/standalone/common/tql/tql-cte.result +++ b/tests/cases/standalone/common/tql/tql-cte.result @@ -360,44 +360,43 @@ WHERE m.ts = l.ts ORDER BY m.ts LIMIT 3; -+---------------+--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ -| plan_type | plan | -+---------------+--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ -| logical_plan | Projection: metric_val, label_val | -| | Sort: m.ts ASC NULLS LAST, fetch=3 | -| | Projection: m.val AS metric_val, l.cpu AS label_val, m.ts | -| | Inner Join: m.ts = l.ts | -| | MergeScan [is_placeholder=false, remote_input=[ | -| | SubqueryAlias: m | -| | SubqueryAlias: metric_data | -| | Projection: metric.ts AS ts, metric.val AS val | -| | PromInstantManipulate: range=[0..40000], lookback=[300000], interval=[10000], time index=[ts] | -| | PromSeriesDivide: tags=[] | -| | Filter: metric.ts >= TimestampMillisecond(-299999, None) AND metric.ts <= TimestampMillisecond(40000, None) | -| | TableScan: metric, partial_filters=[metric.ts >= TimestampMillisecond(-299999, None), metric.ts <= TimestampMillisecond(40000, None)] | -| | ]] | -| | Projection: l.ts, l.cpu | -| | MergeScan [is_placeholder=false, remote_input=[ | -| | SubqueryAlias: l | -| | SubqueryAlias: label_data | -| | Projection: labels.ts AS ts, labels.host AS host, labels.cpu AS cpu | -| | PromInstantManipulate: range=[0..40000], lookback=[300000], interval=[10000], time index=[ts] | -| | PromSeriesDivide: tags=["host"] | -| | Filter: labels.host = Utf8("host2") AND labels.ts >= TimestampMillisecond(-299999, None) AND labels.ts <= TimestampMillisecond(40000, None) | -| | TableScan: labels, partial_filters=[labels.host = Utf8("host2"), labels.ts >= TimestampMillisecond(-299999, None), labels.ts <= TimestampMillisecond(40000, None)] | -| | ]] | -| physical_plan | ProjectionExec: expr=[metric_val@0 as metric_val, label_val@1 as label_val] | -| | SortPreservingMergeExec: [ts@2 ASC NULLS LAST], fetch=3 | ++---------------+----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ +| plan_type | plan | ++---------------+----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ +| logical_plan | Projection: metric_val, label_val | +| | Sort: m.ts ASC NULLS LAST, fetch=3 | +| | Projection: m.val AS metric_val, l.cpu AS label_val, m.ts | +| | Inner Join: m.ts = l.ts | +| | MergeScan [is_placeholder=false, remote_input=[ | +| | SubqueryAlias: m | +| | SubqueryAlias: metric_data | +| | Projection: metric.ts AS ts, metric.val AS val | +| | PromInstantManipulate: range=[0..40000], lookback=[300000], interval=[10000], time index=[ts] | +| | PromSeriesDivide: tags=[] | +| | Filter: metric.ts >= TimestampMillisecond(-299999, None) AND metric.ts <= TimestampMillisecond(40000, None) | +| | TableScan: metric, partial_filters=[metric.ts >= TimestampMillisecond(-299999, None), metric.ts <= TimestampMillisecond(40000, None)] | +| | ]] | +| | MergeScan [is_placeholder=false, remote_input=[ | +| | Projection: l.ts, l.cpu | +| | SubqueryAlias: l | +| | SubqueryAlias: label_data | +| | Projection: labels.ts AS ts, labels.host AS host, labels.cpu AS cpu | +| | PromInstantManipulate: range=[0..40000], lookback=[300000], interval=[10000], time index=[ts] | +| | PromSeriesDivide: tags=["host"] | +| | Filter: labels.host = Utf8("host2") AND labels.ts >= TimestampMillisecond(-299999, None) AND labels.ts <= TimestampMillisecond(40000, None) | +| | TableScan: labels, partial_filters=[labels.host = Utf8("host2"), labels.ts >= TimestampMillisecond(-299999, None), labels.ts <= TimestampMillisecond(40000, None)] | +| | ]] | +| physical_plan | ProjectionExec: expr=[metric_val@0 as metric_val, label_val@1 as label_val] | +| | SortPreservingMergeExec: [ts@2 ASC NULLS LAST], fetch=3 | | | SortExec: TopK(fetch=3), expr=[ts@2 ASC NULLS LAST], preserve_REDACTED -| | ProjectionExec: expr=[val@0 as metric_val, cpu@1 as label_val, ts@2 as ts] | -| | HashJoinExec: mode=Partitioned, join_type=Inner, on=[(ts@0, ts@0)], projection=[val@1, cpu@3, ts@0] | +| | ProjectionExec: expr=[val@0 as metric_val, cpu@1 as label_val, ts@2 as ts] | +| | HashJoinExec: mode=Partitioned, join_type=Inner, on=[(ts@0, ts@0)], projection=[val@1, cpu@3, ts@0] | | | RepartitionExec: REDACTED | | MergeScanExec: REDACTED | | RepartitionExec: REDACTED -| | ProjectionExec: expr=[ts@0 as ts, cpu@2 as cpu] | -| | MergeScanExec: REDACTED -| | | -+---------------+--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ +| | MergeScanExec: REDACTED +| | | ++---------------+----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ -- TQL CTE with mathematical operations WITH computed(ts, val) AS ( @@ -743,24 +742,24 @@ LIMIT 5; | logical_plan | Sort: t.ts ASC NULLS LAST, l.host ASC NULLS LAST, avg_value ASC NULLS LAST, fetch=5 | | | Projection: t.ts, t.cpu AS avg_value, l.host | | | Inner Join: date_trunc(Utf8("second"), t.ts) = date_trunc(Utf8("second"), l.ts) | -| | Projection: t.ts, t.cpu | -| | MergeScan [is_placeholder=false, remote_input=[ | -| | SubqueryAlias: t | -| | SubqueryAlias: tql_summary | -| | Projection: labels.ts AS ts, prom_avg_over_time(ts_range,cpu) AS host, labels.host AS cpu | -| | Filter: prom_avg_over_time(ts_range,cpu) IS NOT NULL | -| | Projection: labels.ts, prom_avg_over_time(ts_range, cpu) AS prom_avg_over_time(ts_range,cpu), labels.host | -| | PromRangeManipulate: req range=[0..40000], interval=[10000], eval range=[30000], time index=[ts], values=["cpu"] | -| | PromSeriesNormalize: offset=[0], time index=[ts], filter NaN: [true] | -| | PromSeriesDivide: tags=["host"] | -| | Filter: labels.ts >= TimestampMillisecond(-29999, None) AND labels.ts <= TimestampMillisecond(40000, None) | -| | TableScan: labels, partial_filters=[labels.ts >= TimestampMillisecond(-29999, None), labels.ts <= TimestampMillisecond(40000, None)] | +| | MergeScan [is_placeholder=false, remote_input=[ | +| | Projection: t.ts, t.cpu | +| | SubqueryAlias: t | +| | SubqueryAlias: tql_summary | +| | Projection: labels.ts AS ts, prom_avg_over_time(ts_range,cpu) AS host, labels.host AS cpu | +| | Filter: prom_avg_over_time(ts_range,cpu) IS NOT NULL | +| | Projection: labels.ts, prom_avg_over_time(ts_range, cpu) AS prom_avg_over_time(ts_range,cpu), labels.host | +| | PromRangeManipulate: req range=[0..40000], interval=[10000], eval range=[30000], time index=[ts], values=["cpu"] | +| | PromSeriesNormalize: offset=[0], time index=[ts], filter NaN: [true] | +| | PromSeriesDivide: tags=["host"] | +| | Filter: labels.ts >= TimestampMillisecond(-29999, None) AND labels.ts <= TimestampMillisecond(40000, None) | +| | TableScan: labels, partial_filters=[labels.ts >= TimestampMillisecond(-29999, None), labels.ts <= TimestampMillisecond(40000, None)] | | | ]] | -| | Projection: l.ts, l.host | -| | MergeScan [is_placeholder=false, remote_input=[ | -| | SubqueryAlias: l | -| | Filter: labels.host = Utf8("host1") | -| | TableScan: labels, partial_filters=[labels.host = Utf8("host1")] | +| | MergeScan [is_placeholder=false, remote_input=[ | +| | Projection: l.ts, l.host | +| | SubqueryAlias: l | +| | Filter: labels.host = Utf8("host1") | +| | TableScan: labels, partial_filters=[labels.host = Utf8("host1")] | | | ]] | | physical_plan | SortPreservingMergeExec: [ts@0 ASC NULLS LAST, host@2 ASC NULLS LAST, avg_value@1 ASC NULLS LAST], fetch=5 | | | SortExec: TopK(fetch=5), expr=[ts@0 ASC NULLS LAST, host@2 ASC NULLS LAST, avg_value@1 ASC NULLS LAST], preserve_REDACTED @@ -769,13 +768,11 @@ LIMIT 5; | | RepartitionExec: REDACTED | | ProjectionExec: expr=[ts@0 as ts, cpu@1 as cpu, date_trunc(second, ts@0) as date_trunc(Utf8("second"),t.ts)] | | | RepartitionExec: REDACTED -| | ProjectionExec: expr=[ts@0 as ts, cpu@2 as cpu] | -| | MergeScanExec: REDACTED +| | MergeScanExec: REDACTED | | RepartitionExec: REDACTED | | ProjectionExec: expr=[ts@0 as ts, host@1 as host, date_trunc(second, ts@0) as date_trunc(Utf8("second"),l.ts)] | | | RepartitionExec: REDACTED -| | ProjectionExec: expr=[ts@0 as ts, host@1 as host] | -| | MergeScanExec: REDACTED +| | MergeScanExec: REDACTED | | | +---------------+----------------------------------------------------------------------------------------------------------------------------------------------------------------------+ @@ -825,24 +822,24 @@ LIMIT 5; | logical_plan | Sort: t.ts ASC NULLS LAST, l.host ASC NULLS LAST, avg_value ASC NULLS LAST, fetch=5 | | | Projection: t.ts, t.cpu AS avg_value, l.host | | | Inner Join: date_trunc(Utf8("second"), t.ts) = date_trunc(Utf8("second"), l.ts) | -| | Projection: t.cpu, t.ts | -| | MergeScan [is_placeholder=false, remote_input=[ | -| | SubqueryAlias: t | -| | SubqueryAlias: tql_summary | -| | Projection: prom_avg_over_time(ts_range,cpu) AS cpu, labels.host, labels.ts | -| | Filter: prom_avg_over_time(ts_range,cpu) IS NOT NULL | -| | Projection: labels.ts, prom_avg_over_time(ts_range, cpu) AS prom_avg_over_time(ts_range,cpu), labels.host | -| | PromRangeManipulate: req range=[0..40000], interval=[10000], eval range=[30000], time index=[ts], values=["cpu"] | -| | PromSeriesNormalize: offset=[0], time index=[ts], filter NaN: [true] | -| | PromSeriesDivide: tags=["host"] | -| | Filter: labels.ts >= TimestampMillisecond(-29999, None) AND labels.ts <= TimestampMillisecond(40000, None) | -| | TableScan: labels, partial_filters=[labels.ts >= TimestampMillisecond(-29999, None), labels.ts <= TimestampMillisecond(40000, None)] | +| | MergeScan [is_placeholder=false, remote_input=[ | +| | Projection: t.cpu, t.ts | +| | SubqueryAlias: t | +| | SubqueryAlias: tql_summary | +| | Projection: prom_avg_over_time(ts_range,cpu) AS cpu, labels.host, labels.ts | +| | Filter: prom_avg_over_time(ts_range,cpu) IS NOT NULL | +| | Projection: labels.ts, prom_avg_over_time(ts_range, cpu) AS prom_avg_over_time(ts_range,cpu), labels.host | +| | PromRangeManipulate: req range=[0..40000], interval=[10000], eval range=[30000], time index=[ts], values=["cpu"] | +| | PromSeriesNormalize: offset=[0], time index=[ts], filter NaN: [true] | +| | PromSeriesDivide: tags=["host"] | +| | Filter: labels.ts >= TimestampMillisecond(-29999, None) AND labels.ts <= TimestampMillisecond(40000, None) | +| | TableScan: labels, partial_filters=[labels.ts >= TimestampMillisecond(-29999, None), labels.ts <= TimestampMillisecond(40000, None)] | | | ]] | -| | Projection: l.ts, l.host | -| | MergeScan [is_placeholder=false, remote_input=[ | -| | SubqueryAlias: l | -| | Filter: labels.host = Utf8("host1") | -| | TableScan: labels, partial_filters=[labels.host = Utf8("host1")] | +| | MergeScan [is_placeholder=false, remote_input=[ | +| | Projection: l.ts, l.host | +| | SubqueryAlias: l | +| | Filter: labels.host = Utf8("host1") | +| | TableScan: labels, partial_filters=[labels.host = Utf8("host1")] | | | ]] | | physical_plan | SortPreservingMergeExec: [ts@0 ASC NULLS LAST, host@2 ASC NULLS LAST, avg_value@1 ASC NULLS LAST], fetch=5 | | | SortExec: TopK(fetch=5), expr=[ts@0 ASC NULLS LAST, host@2 ASC NULLS LAST, avg_value@1 ASC NULLS LAST], preserve_REDACTED @@ -851,13 +848,11 @@ LIMIT 5; | | RepartitionExec: REDACTED | | ProjectionExec: expr=[cpu@0 as cpu, ts@1 as ts, date_trunc(second, ts@1) as date_trunc(Utf8("second"),t.ts)] | | | RepartitionExec: REDACTED -| | ProjectionExec: expr=[cpu@0 as cpu, ts@2 as ts] | -| | MergeScanExec: REDACTED +| | MergeScanExec: REDACTED | | RepartitionExec: REDACTED | | ProjectionExec: expr=[ts@0 as ts, host@1 as host, date_trunc(second, ts@0) as date_trunc(Utf8("second"),l.ts)] | | | RepartitionExec: REDACTED -| | ProjectionExec: expr=[ts@0 as ts, host@1 as host] | -| | MergeScanExec: REDACTED +| | MergeScanExec: REDACTED | | | +---------------+----------------------------------------------------------------------------------------------------------------------------------------------------------------------+ diff --git a/tests/cases/standalone/optimizer/join_filter_pushdown_edge.result b/tests/cases/standalone/optimizer/join_filter_pushdown_edge.result index c9b1f37c9ad..b67375a033e 100644 --- a/tests/cases/standalone/optimizer/join_filter_pushdown_edge.result +++ b/tests/cases/standalone/optimizer/join_filter_pushdown_edge.result @@ -90,47 +90,44 @@ EXPLAIN SELECT f.k, f.val FROM fact f AND f.val > 1.0 AND EXISTS (SELECT 1 FROM vals v WHERE v.x = CAST(f.val AS INT)); -+---------------+---------------------------------------------------------------------------------------------------------------------------------------------------------+ -| plan_type | plan | -+---------------+---------------------------------------------------------------------------------------------------------------------------------------------------------+ -| logical_plan | LeftSemi Join: CAST(f.val AS Int32) = __correlated_sq_1.x | -| | Projection: f.k, f.val | -| | Left Join: f.k = d.k | -| | Projection: f.k, f.val | -| | MergeScan [is_placeholder=false, remote_input=[ | -| | SubqueryAlias: f | -| | Filter: CAST(fact.val AS Int32) IS NOT NULL AND fact.ts >= TimestampMillisecond(1706572800000, None) AND fact.val > Float64(1) | -| | TableScan: fact, partial_filters=[CAST(fact.val AS Int32) IS NOT NULL, fact.ts >= TimestampMillisecond(1706572800000, None), fact.val > Float64(1)] | -| | ]] | -| | Projection: d.k | -| | MergeScan [is_placeholder=false, remote_input=[ | -| | SubqueryAlias: d | -| | Filter: dim.k IS NOT NULL | -| | TableScan: dim, partial_filters=[dim.k IS NOT NULL] | -| | ]] | -| | Projection: __correlated_sq_1.x | -| | MergeScan [is_placeholder=false, remote_input=[ | -| | SubqueryAlias: __correlated_sq_1 | -| | Projection: Int64(1), v.x | -| | SubqueryAlias: v | -| | Filter: vals.x IS NOT NULL | -| | TableScan: vals, partial_filters=[vals.x IS NOT NULL] | -| | ]] | -| physical_plan | HashJoinExec: mode=Partitioned, join_type=LeftSemi, on=[(CAST(f.val AS Int32)@2, x@0)], projection=[k@0, val@1] | ++---------------+-----------------------------------------------------------------------------------------------------------------------------------------------------------+ +| plan_type | plan | ++---------------+-----------------------------------------------------------------------------------------------------------------------------------------------------------+ +| logical_plan | LeftSemi Join: CAST(f.val AS Int32) = __correlated_sq_1.x | +| | Projection: f.k, f.val | +| | Left Join: f.k = d.k | +| | MergeScan [is_placeholder=false, remote_input=[ | +| | Projection: f.k, f.val | +| | SubqueryAlias: f | +| | Filter: CAST(fact.val AS Int32) IS NOT NULL AND fact.ts >= TimestampMillisecond(1706572800000, None) AND fact.val > Float64(1) | +| | TableScan: fact, partial_filters=[CAST(fact.val AS Int32) IS NOT NULL, fact.ts >= TimestampMillisecond(1706572800000, None), fact.val > Float64(1)] | +| | ]] | +| | MergeScan [is_placeholder=false, remote_input=[ | +| | Projection: d.k | +| | SubqueryAlias: d | +| | Filter: dim.k IS NOT NULL | +| | TableScan: dim, partial_filters=[dim.k IS NOT NULL] | +| | ]] | +| | MergeScan [is_placeholder=false, remote_input=[ | +| | Projection: __correlated_sq_1.x | +| | SubqueryAlias: __correlated_sq_1 | +| | Projection: Int64(1), v.x | +| | SubqueryAlias: v | +| | Filter: vals.x IS NOT NULL | +| | TableScan: vals, partial_filters=[vals.x IS NOT NULL] | +| | ]] | +| physical_plan | HashJoinExec: mode=Partitioned, join_type=LeftSemi, on=[(CAST(f.val AS Int32)@2, x@0)], projection=[k@0, val@1] | | | RepartitionExec: partitioning=Hash([REDACTED -| | ProjectionExec: expr=[k@0 as k, val@1 as val, CAST(val@1 AS Int32) as CAST(f.val AS Int32)] | -| | HashJoinExec: mode=Partitioned, join_type=Left, on=[(k@0, k@0)], projection=[k@0, val@1] | +| | ProjectionExec: expr=[k@0 as k, val@1 as val, CAST(val@1 AS Int32) as CAST(f.val AS Int32)] | +| | HashJoinExec: mode=Partitioned, join_type=Left, on=[(k@0, k@0)], projection=[k@0, val@1] | | | RepartitionExec: partitioning=Hash([REDACTED -| | ProjectionExec: expr=[k@1 as k, val@2 as val] | -| | MergeScanExec: REDACTED +| | MergeScanExec: REDACTED | | RepartitionExec: partitioning=Hash([REDACTED -| | ProjectionExec: expr=[k@1 as k] | -| | MergeScanExec: REDACTED +| | MergeScanExec: REDACTED | | RepartitionExec: partitioning=Hash([REDACTED -| | ProjectionExec: expr=[x@1 as x] | -| | MergeScanExec: REDACTED -| | | -+---------------+---------------------------------------------------------------------------------------------------------------------------------------------------------+ +| | MergeScanExec: REDACTED +| | | ++---------------+-----------------------------------------------------------------------------------------------------------------------------------------------------------+ SELECT f.k, f.val, d.label FROM fact f LEFT JOIN dim d ON f.k = d.k diff --git a/tests/cases/standalone/optimizer/lateral_join_guard.result b/tests/cases/standalone/optimizer/lateral_join_guard.result index d2a565089e3..9f78657e6a6 100644 --- a/tests/cases/standalone/optimizer/lateral_join_guard.result +++ b/tests/cases/standalone/optimizer/lateral_join_guard.result @@ -67,11 +67,11 @@ ORDER BY f.k; | logical_plan | Sort: f.k ASC NULLS LAST | | | Projection: f.k, d.threshold | | | Inner Join: f.k = d.k Filter: f.val > d.threshold | -| | Projection: f.k, f.val | -| | MergeScan [is_placeholder=false, remote_input=[ | -| | SubqueryAlias: f | -| | Filter: lateral_fact.k IS NOT NULL | -| | TableScan: lateral_fact, partial_filters=[lateral_fact.k IS NOT NULL] | +| | MergeScan [is_placeholder=false, remote_input=[ | +| | Projection: f.k, f.val | +| | SubqueryAlias: f | +| | Filter: lateral_fact.k IS NOT NULL | +| | TableScan: lateral_fact, partial_filters=[lateral_fact.k IS NOT NULL] | | | ]] | | | MergeScan [is_placeholder=false, remote_input=[ | | | SubqueryAlias: d | @@ -84,8 +84,7 @@ ORDER BY f.k; | | SortExec: expr=[k@0 ASC NULLS LAST], preserve_partitioning=[true] | | | HashJoinExec: mode=Partitioned, join_type=Inner, on=[(k@0, k@1)], filter=val@0 > threshold@1, projection=[k@0, threshold@2] | | | RepartitionExec: partitioning=Hash([k@0], REDACTED), input_partitions=REDACTED| -| | ProjectionExec: expr=[k@1 as k, val@2 as val] | -| | MergeScanExec: REDACTED +| | MergeScanExec: REDACTED | | RepartitionExec: partitioning=Hash([k@1], REDACTED), input_partitions=REDACTED| | | MergeScanExec: REDACTED | | | diff --git a/tests/cases/standalone/optimizer/rewrite_set_comparison.result b/tests/cases/standalone/optimizer/rewrite_set_comparison.result index f7463099de2..af54de5a378 100644 --- a/tests/cases/standalone/optimizer/rewrite_set_comparison.result +++ b/tests/cases/standalone/optimizer/rewrite_set_comparison.result @@ -58,9 +58,9 @@ EXPLAIN SELECT v FROM sc_t WHERE v > ANY(SELECT v FROM sc_s) ORDER BY v; | | LeftMark Join: Filter: sc_t.v > __correlated_sq_2.v IS NULL | | | Filter: __correlated_sq_1.mark OR NOT __correlated_sq_1.mark AND Boolean(NULL) | | | LeftMark Join: Filter: sc_t.v > __correlated_sq_1.v IS TRUE | -| | Projection: sc_t.v | -| | MergeScan [is_placeholder=false, remote_input=[ | -| | TableScan: sc_t | +| | MergeScan [is_placeholder=false, remote_input=[ | +| | Projection: sc_t.v | +| | TableScan: sc_t | | | ]] | | | MergeScan [is_placeholder=false, remote_input=[ | | | Projection: __correlated_sq_1.v | @@ -81,8 +81,7 @@ EXPLAIN SELECT v FROM sc_t WHERE v > ANY(SELECT v FROM sc_s) ORDER BY v; | | CoalescePartitionsExec | | | FilterExec: mark@1 OR NOT mark@1 AND NULL | | | NestedLoopJoinExec: join_type=LeftMark, filter=(v@0 > v@1) IS NOT DISTINCT FROM true | -| | ProjectionExec: expr=[v@1 as v] | -| | MergeScanExec: REDACTED +| | MergeScanExec: REDACTED | | RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1 | | | MergeScanExec: REDACTED | | RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1 | @@ -113,9 +112,9 @@ EXPLAIN SELECT v FROM sc_t WHERE v != ALL(SELECT v FROM sc_s) ORDER BY v; | | Projection: sc_t.v | | | Filter: NOT __correlated_sq_1.mark | | | LeftMark Join: Filter: sc_t.v != __correlated_sq_1.v IS FALSE | -| | Projection: sc_t.v | -| | MergeScan [is_placeholder=false, remote_input=[ | -| | TableScan: sc_t | +| | MergeScan [is_placeholder=false, remote_input=[ | +| | Projection: sc_t.v | +| | TableScan: sc_t | | | ]] | | | MergeScan [is_placeholder=false, remote_input=[ | | | Projection: __correlated_sq_1.v | @@ -136,8 +135,7 @@ EXPLAIN SELECT v FROM sc_t WHERE v != ALL(SELECT v FROM sc_s) ORDER BY v; | | CoalescePartitionsExec | | | FilterExec: NOT mark@1, projection=[v@0] | | | NestedLoopJoinExec: join_type=LeftMark, filter=(v@0 != v@1) IS NOT DISTINCT FROM false | -| | ProjectionExec: expr=[v@1 as v] | -| | MergeScanExec: REDACTED +| | MergeScanExec: REDACTED | | RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1 | | | MergeScanExec: REDACTED | | RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1 | diff --git a/tests/cases/standalone/optimizer/scalar_subquery_prepass.result b/tests/cases/standalone/optimizer/scalar_subquery_prepass.result index 2ed894e016a..f1885e5491d 100644 --- a/tests/cases/standalone/optimizer/scalar_subquery_prepass.result +++ b/tests/cases/standalone/optimizer/scalar_subquery_prepass.result @@ -66,29 +66,27 @@ ORDER BY f.k; | logical_plan | Sort: f.k ASC NULLS LAST | | | Projection: f.k, f.val | | | Inner Join: f.k = __scalar_sq_1.k Filter: f.val > __scalar_sq_1.max(d.threshold) | -| | Projection: f.k, f.val | -| | MergeScan [is_placeholder=false, remote_input=[ | -| | SubqueryAlias: f | -| | TableScan: scalar_fact | +| | MergeScan [is_placeholder=false, remote_input=[ | +| | Projection: f.k, f.val | +| | SubqueryAlias: f | +| | TableScan: scalar_fact | | | ]] | -| | Projection: __scalar_sq_1.max(d.threshold), __scalar_sq_1.k | -| | MergeScan [is_placeholder=false, remote_input=[ | -| | SubqueryAlias: __scalar_sq_1 | -| | Projection: max(d.threshold), d.k, __always_true | -| | Aggregate: groupBy=[[d.k, Boolean(true) AS __always_true]], aggr=[[max(d.threshold)]] | -| | SubqueryAlias: d | -| | Filter: scalar_dim.k IS NOT NULL | -| | TableScan: scalar_dim, partial_filters=[scalar_dim.k IS NOT NULL] | +| | MergeScan [is_placeholder=false, remote_input=[ | +| | Projection: __scalar_sq_1.max(d.threshold), __scalar_sq_1.k | +| | SubqueryAlias: __scalar_sq_1 | +| | Projection: max(d.threshold), d.k, __always_true | +| | Aggregate: groupBy=[[d.k, Boolean(true) AS __always_true]], aggr=[[max(d.threshold)]] | +| | SubqueryAlias: d | +| | Filter: scalar_dim.k IS NOT NULL | +| | TableScan: scalar_dim, partial_filters=[scalar_dim.k IS NOT NULL] | | | ]] | | physical_plan | SortPreservingMergeExec: [k@0 ASC NULLS LAST] | | | SortExec: expr=[k@0 ASC NULLS LAST], preserve_partitioning=[true] | | | HashJoinExec: mode=Partitioned, join_type=Inner, on=[(k@0, k@1)], filter=val@0 > max(d.threshold)@1, projection=[k@0, val@1] | | | RepartitionExec: partitioning=Hash([REDACTED -| | ProjectionExec: expr=[k@1 as k, val@2 as val] | -| | MergeScanExec: REDACTED +| | MergeScanExec: REDACTED | | RepartitionExec: partitioning=Hash([REDACTED -| | ProjectionExec: expr=[max(d.threshold)@0 as max(d.threshold), k@1 as k] | -| | MergeScanExec: REDACTED +| | MergeScanExec: REDACTED | | | +---------------+----------------------------------------------------------------------------------------------------------------------------------+ diff --git a/tests/cases/standalone/optimizer/time_index_filter_pushdown.result b/tests/cases/standalone/optimizer/time_index_filter_pushdown.result index 106a78c49d2..787119875bd 100644 --- a/tests/cases/standalone/optimizer/time_index_filter_pushdown.result +++ b/tests/cases/standalone/optimizer/time_index_filter_pushdown.result @@ -261,26 +261,25 @@ WHERE +-+-+-+ | 0_| 0_|_ScalarSubqueryExec REDACTED |_|_|_FilterExec: greptime_timestamp@2 = scalar_subquery(1686535800000000000) REDACTED -|_|_|_ProjectionExec: expr=[rack@0 as rack, os@1 as os, greptime_timestamp@4 as greptime_timestamp] REDACTED |_|_|_CooperativeExec REDACTED |_|_|_MergeScanExec: REDACTED |_|_|_MergeSortExec: [greptime_timestamp@0 DESC], fetch=1 REDACTED |_|_|_MergeScanExec: REDACTED |_|_|_| | 1_| 0_|_CooperativeExec REDACTED -|_|_|_UnorderedScan: REDACTED, "dyn_filters": ["DynamicFilter [ greptime_timestamp@4 = 1686535800000000000 ]"] +|_|_|_UnorderedScan: REDACTED, "dyn_filters": ["DynamicFilter [ greptime_timestamp@2 = 1686535800000000000 ]"] |_|_|_| | 1_| 1_|_CooperativeExec REDACTED -|_|_|_UnorderedScan: REDACTED, "dyn_filters": ["DynamicFilter [ greptime_timestamp@4 = 1686535800000000000 ]"] +|_|_|_UnorderedScan: REDACTED, "dyn_filters": ["DynamicFilter [ greptime_timestamp@2 = 1686535800000000000 ]"] |_|_|_| | 1_| 2_|_CooperativeExec REDACTED -|_|_|_UnorderedScan: REDACTED, "dyn_filters": ["DynamicFilter [ greptime_timestamp@4 = 1686535800000000000 ]"] +|_|_|_UnorderedScan: REDACTED, "dyn_filters": ["DynamicFilter [ greptime_timestamp@2 = 1686535800000000000 ]"] |_|_|_| | 1_| 3_|_CooperativeExec REDACTED -|_|_|_UnorderedScan: REDACTED, "dyn_filters": ["DynamicFilter [ greptime_timestamp@4 = 1686535800000000000 ]"] +|_|_|_UnorderedScan: REDACTED, "dyn_filters": ["DynamicFilter [ greptime_timestamp@2 = 1686535800000000000 ]"] |_|_|_| | 1_| 4_|_CooperativeExec REDACTED -|_|_|_UnorderedScan: REDACTED, "dyn_filters": ["DynamicFilter [ greptime_timestamp@4 = 1686535800000000000 ]"] +|_|_|_UnorderedScan: REDACTED, "dyn_filters": ["DynamicFilter [ greptime_timestamp@2 = 1686535800000000000 ]"] |_|_|_| | 1_| 0_|_FilterExec: DynamicFilter [ REDACTED ] REDACTED |_|_|_SortPreservingMergeExec: [greptime_timestamp@0 DESC], fetch=1 REDACTED