rust/greptimedb
1
0
Fork 0
mirror of https://github.com/GreptimeTeam/greptimedb.git synced 2026-11-13 14:27:14 +00:00
Code Issues Packages Projects Releases Wiki Activity

perf(query): push column projections through MergeScan (#10221)

Browse Source 
* perf(query): push column projections through MergeScan

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

* fix(query): retain empty projections and refresh sqlness plans

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

* test(query): update TSID join plans after projection pushdown

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

---------

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>
This commit is contained in:
dennis zhuang
2026-11-09 22:14:09 +00:00
committed by GitHub
parent 8da533c35f
commit 16123d806b
21 changed files with 785 additions and 549 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/query/src/optimizer.rs
+1
View File
@@ -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;
src/query/src/optimizer/merge_scan_projection.rs
+141
View File
@@ -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<ApplyOrder> {
Some(ApplyOrder::BottomUp)
}
fn rewrite(
&self,
plan: LogicalPlan,
_config: &dyn OptimizerConfig,
) -> Result<Transformed<LogicalPlan>> {
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::<MergeScanLogicalPlan>()
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::<MergeScanLogicalPlan>()
.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);
}
}
src/query/src/query_engine/state.rs
+4
View File
@@ -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));
tests/cases/distributed/explain/step_aggr_advance.result
+2 -3
View File
@@ -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
|_|_|
+-+-+
tests/cases/distributed/explain/subqueries.result
+9 -12
View File
@@ -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(<pending>)_|
|_|_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_|
tests/cases/distributed/optimizer/pass_distribution_partition_subset_join.result
+12 -14
View File
@@ -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, ] |
| | |
+---------------+-------------------------------------------------------------------------------------------------------------------+
tests/cases/distributed/optimizer/time_index_filter_pushdown.result
+7 -9
View File
@@ -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(<pending>)_|
|_|_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
tests/cases/standalone/common/cte/cte_join_build_side.result
+33 -35
View File
@@ -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
tests/cases/standalone/common/filter/hash_join_dyn_filter.result
+40 -48
View File
@@ -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_|
+-+-+-+
tests/cases/standalone/common/filter/hash_join_topk_dyn_filter.result
+64 -78
View File
@@ -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_|
+-+-+-+
tests/cases/standalone/common/join/merge_scan_projection.result
+103
View File
@@ -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
tests/cases/standalone/common/join/merge_scan_projection.sql
+44
View File
@@ -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;
tests/cases/standalone/common/join/merge_scan_semi_join.result
+139 -144
View File
@@ -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;
tests/cases/standalone/common/promql/tsid_binary_join_regression.result
+42 -54
View File
@@ -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
tests/cases/standalone/common/tql-explain-analyze/explain.result
+6
View File
@@ -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=[_|
tests/cases/standalone/common/tql/tql-cte.result
+71 -76
View File
@@ -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
| | |
+---------------+----------------------------------------------------------------------------------------------------------------------------------------------------------------------+
tests/cases/standalone/optimizer/join_filter_pushdown_edge.result
+34 -37
View File
@@ -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
tests/cases/standalone/optimizer/lateral_join_guard.result
+6 -7
View File
@@ -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
| | |
tests/cases/standalone/optimizer/rewrite_set_comparison.result
+8 -10
View File
@@ -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 |
tests/cases/standalone/optimizer/scalar_subquery_prepass.result
+14 -16
View File
@@ -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
| | |
+---------------+----------------------------------------------------------------------------------------------------------------------------------+
tests/cases/standalone/optimizer/time_index_filter_pushdown.result
+5 -6
View File
@@ -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