Skip to content

plan: enhance predicate pushdown over join #7645

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 14 commits into from
Sep 17, 2018
Merged

plan: enhance predicate pushdown over join #7645

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
14 commits
Select commit Hold shift + click to select a range
5817f27
plan: enhance predicate pushdown over join
eurekaka Sep 5, 2018
b9ac61e
add example in comments
eurekaka Sep 6, 2018
7e1f616
remove unnecessary extractOnConditions call, and add param to indicate
eurekaka Sep 6, 2018
45c8a9c
derive join conditions for outer join as well
eurekaka Sep 6, 2018
e9c35a1
format change
eurekaka Sep 6, 2018
8a96dd2
add comments for functions
eurekaka Sep 6, 2018
8dd8476
update explain result of q21 caused by propagate constant for semi join
eurekaka Sep 6, 2018
7498f1c
add unit test
eurekaka Sep 6, 2018
b87c33c
Merge branch 'master' into ppd_join_dnf
eurekaka Sep 10, 2018
4d9756b
Merge branch 'master' into ppd_join_dnf
eurekaka Sep 10, 2018
215bab1
address comments
eurekaka Sep 7, 2018
be3f2ac
Merge branch 'master' into ppd_join_dnf
eurekaka Sep 12, 2018
cf0d988
Merge branch 'master' into ppd_join_dnf
zz-jason Sep 14, 2018
7514148
Merge branch 'master' into ppd_join_dnf
winoros Sep 17, 2018
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
52 changes: 27 additions & 25 deletions cmd/explaintest/r/tpch.result
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -446,25 +446,27 @@ id count task operator info
Sort_22 768.91 root shipping.supp_nation:asc, shipping.cust_nation:asc, shipping.l_year:asc
└─Projection_24 768.91 root shipping.supp_nation, shipping.cust_nation, shipping.l_year, 14_col_0
└─HashAgg_27 768.91 root group by:shipping.cust_nation, shipping.l_year, shipping.supp_nation, funcs:sum(shipping.volume), firstrow(shipping.supp_nation), firstrow(shipping.cust_nation), firstrow(shipping.l_year)
└─Projection_28 91321768.29 root n1.n_name, n2.n_name, extract("YEAR", tpch.lineitem.l_shipdate), mul(tpch.lineitem.l_extendedprice, minus(1, tpch.lineitem.l_discount))
└─HashLeftJoin_32 91321768.29 root inner join, inner:TableReader_80, equal:[eq(tpch.customer.c_nationkey, n2.n_nationkey)], other cond:or(and(eq(n1.n_name, "JAPAN"), eq(n2.n_name, "INDIA")), and(eq(n1.n_name, "INDIA"), eq(n2.n_name, "JAPAN")))
├─HashLeftJoin_37 91321768.29 root inner join, inner:TableReader_78, equal:[eq(tpch.orders.o_custkey, tpch.customer.c_custkey)]
│ ├─IndexJoin_42 91321768.29 root inner join, inner:TableReader_41, outer key:tpch.lineitem.l_orderkey, inner key:tpch.orders.o_orderkey
│ │ ├─HashLeftJoin_61 91321768.29 root inner join, inner:TableReader_74, equal:[eq(tpch.supplier.s_nationkey, n1.n_nationkey)]
│ │ │ ├─HashLeftJoin_66 91321768.29 root inner join, inner:TableReader_72, equal:[eq(tpch.lineitem.l_suppkey, tpch.supplier.s_suppkey)]
│ │ │ │ ├─TableReader_70 91321768.29 root data:Selection_69
│ │ │ │ │ └─Selection_69 91321768.29 cop ge(tpch.lineitem.l_shipdate, 1995-01-01 00:00:00.000000), le(tpch.lineitem.l_shipdate, 1996-12-31 00:00:00.000000)
│ │ │ │ │ └─TableScan_68 300005811.00 cop table:lineitem, range:[-inf,+inf], keep order:false
│ │ │ │ └─TableReader_72 500000.00 root data:TableScan_71
│ │ │ │ └─TableScan_71 500000.00 cop table:supplier, range:[-inf,+inf], keep order:false
│ │ │ └─TableReader_74 25.00 root data:TableScan_73
│ │ │ └─TableScan_73 25.00 cop table:n1, range:[-inf,+inf], keep order:false
│ │ └─TableReader_41 1.00 root data:TableScan_40
│ │ └─TableScan_40 1.00 cop table:orders, range: decided by [tpch.lineitem.l_orderkey], keep order:false
│ └─TableReader_78 7500000.00 root data:TableScan_77
│ └─TableScan_77 7500000.00 cop table:customer, range:[-inf,+inf], keep order:false
└─TableReader_80 25.00 root data:TableScan_79
└─TableScan_79 25.00 cop table:n2, range:[-inf,+inf], keep order:false
└─Projection_28 584459.32 root n1.n_name, n2.n_name, extract("YEAR", tpch.lineitem.l_shipdate), mul(tpch.lineitem.l_extendedprice, minus(1, tpch.lineitem.l_discount))
└─HashLeftJoin_33 584459.32 root inner join, inner:TableReader_85, equal:[eq(tpch.customer.c_nationkey, n2.n_nationkey)], other cond:or(and(eq(n1.n_name, "JAPAN"), eq(n2.n_name, "INDIA")), and(eq(n1.n_name, "INDIA"), eq(n2.n_name, "JAPAN")))
├─IndexJoin_37 7305741.46 root inner join, inner:TableReader_36, outer key:tpch.orders.o_custkey, inner key:tpch.customer.c_custkey
│ ├─IndexJoin_43 7305741.46 root inner join, inner:TableReader_42, outer key:tpch.lineitem.l_orderkey, inner key:tpch.orders.o_orderkey
│ │ ├─HashLeftJoin_64 7305741.46 root inner join, inner:TableReader_78, equal:[eq(tpch.supplier.s_nationkey, n1.n_nationkey)]
│ │ │ ├─HashLeftJoin_69 91321768.29 root inner join, inner:TableReader_75, equal:[eq(tpch.lineitem.l_suppkey, tpch.supplier.s_suppkey)]
│ │ │ │ ├─TableReader_73 91321768.29 root data:Selection_72
│ │ │ │ │ └─Selection_72 91321768.29 cop ge(tpch.lineitem.l_shipdate, 1995-01-01 00:00:00.000000), le(tpch.lineitem.l_shipdate, 1996-12-31 00:00:00.000000)
│ │ │ │ │ └─TableScan_71 300005811.00 cop table:lineitem, range:[-inf,+inf], keep order:false
│ │ │ │ └─TableReader_75 500000.00 root data:TableScan_74
│ │ │ │ └─TableScan_74 500000.00 cop table:supplier, range:[-inf,+inf], keep order:false
│ │ │ └─TableReader_78 2.00 root data:Selection_77
│ │ │ └─Selection_77 2.00 cop or(eq(n1.n_name, "JAPAN"), eq(n1.n_name, "INDIA"))
│ │ │ └─TableScan_76 25.00 cop table:n1, range:[-inf,+inf], keep order:false
│ │ └─TableReader_42 1.00 root data:TableScan_41
│ │ └─TableScan_41 1.00 cop table:orders, range: decided by [tpch.lineitem.l_orderkey], keep order:false
│ └─TableReader_36 1.00 root data:TableScan_35
│ └─TableScan_35 1.00 cop table:customer, range: decided by [tpch.orders.o_custkey], keep order:false
└─TableReader_85 2.00 root data:Selection_84
└─Selection_84 2.00 cop or(eq(n2.n_name, "INDIA"), eq(n2.n_name, "JAPAN"))
└─TableScan_83 25.00 cop table:n2, range:[-inf,+inf], keep order:false
/*
Q8 National Market Share Query
This query determines how the market share of a given nation within a given region has changed over two years for
Expand Down Expand Up @@ -1086,12 +1088,12 @@ and l_shipinstruct = 'DELIVER IN PERSON'
);
id count task operator info
StreamAgg_13 1.00 root funcs:sum(mul(tpch.lineitem.l_extendedprice, minus(1, tpch.lineitem.l_discount)))
└─IndexJoin_29 10667730.87 root inner join, inner:TableReader_28, outer key:tpch.lineitem.l_partkey, inner key:tpch.part.p_partkey, other cond:or(and(and(eq(tpch.part.p_brand, "Brand#52"), in(tpch.part.p_container, "SM CASE", "SM BOX", "SM PACK", "SM PKG")), and(ge(tpch.lineitem.l_quantity, 4), and(le(tpch.lineitem.l_quantity, 14), le(tpch.part.p_size, 5)))), or(and(and(eq(tpch.part.p_brand, "Brand#11"), in(tpch.part.p_container, "MED BAG", "MED BOX", "MED PKG", "MED PACK")), and(ge(tpch.lineitem.l_quantity, 18), and(le(tpch.lineitem.l_quantity, 28), le(tpch.part.p_size, 10)))), and(and(eq(tpch.part.p_brand, "Brand#51"), in(tpch.part.p_container, "LG CASE", "LG BOX", "LG PACK", "LG PKG")), and(ge(tpch.lineitem.l_quantity, 29), and(le(tpch.lineitem.l_quantity, 39), le(tpch.part.p_size, 15))))))
├─TableReader_34 10667730.87 root data:Selection_33
│ └─Selection_33 10667730.87 cop eq(tpch.lineitem.l_shipinstruct, "DELIVER IN PERSON"), in(tpch.lineitem.l_shipmode, "AIR", "AIR REG")
└─IndexJoin_29 6286493.79 root inner join, inner:TableReader_28, outer key:tpch.lineitem.l_partkey, inner key:tpch.part.p_partkey, other cond:or(and(and(eq(tpch.part.p_brand, "Brand#52"), in(tpch.part.p_container, "SM CASE", "SM BOX", "SM PACK", "SM PKG")), and(ge(tpch.lineitem.l_quantity, 4), and(le(tpch.lineitem.l_quantity, 14), le(tpch.part.p_size, 5)))), or(and(and(eq(tpch.part.p_brand, "Brand#11"), in(tpch.part.p_container, "MED BAG", "MED BOX", "MED PKG", "MED PACK")), and(ge(tpch.lineitem.l_quantity, 18), and(le(tpch.lineitem.l_quantity, 28), le(tpch.part.p_size, 10)))), and(and(eq(tpch.part.p_brand, "Brand#51"), in(tpch.part.p_container, "LG CASE", "LG BOX", "LG PACK", "LG PKG")), and(ge(tpch.lineitem.l_quantity, 29), and(le(tpch.lineitem.l_quantity, 39), le(tpch.part.p_size, 15))))))
├─TableReader_34 6286493.79 root data:Selection_33
│ └─Selection_33 6286493.79 cop eq(tpch.lineitem.l_shipinstruct, "DELIVER IN PERSON"), in(tpch.lineitem.l_shipmode, "AIR", "AIR REG"), or(and(ge(tpch.lineitem.l_quantity, 4), le(tpch.lineitem.l_quantity, 14)), or(and(ge(tpch.lineitem.l_quantity, 18), le(tpch.lineitem.l_quantity, 28)), and(ge(tpch.lineitem.l_quantity, 29), le(tpch.lineitem.l_quantity, 39))))
│ └─TableScan_32 300005811.00 cop table:lineitem, range:[-inf,+inf], keep order:false
└─TableReader_28 10000000.00 root data:Selection_27
└─Selection_27 10000000.00 cop ge(tpch.part.p_size, 1)
└─TableReader_28 8000000.00 root data:Selection_27
└─Selection_27 8000000.00 cop ge(tpch.part.p_size, 1), or(and(eq(tpch.part.p_brand, "Brand#52"), and(in(tpch.part.p_container, "SM CASE", "SM BOX", "SM PACK", "SM PKG"), le(tpch.part.p_size, 5))), or(and(eq(tpch.part.p_brand, "Brand#11"), and(in(tpch.part.p_container, "MED BAG", "MED BOX", "MED PKG", "MED PACK"), le(tpch.part.p_size, 10))), and(eq(tpch.part.p_brand, "Brand#51"), and(in(tpch.part.p_container, "LG CASE", "LG BOX", "LG PACK", "LG PKG"), le(tpch.part.p_size, 15)))))
└─TableScan_26 1.00 cop table:part, range: decided by [tpch.lineitem.l_partkey], keep order:false
/*
Q20 Potential Part Promotion Query
Expand Down Expand Up @@ -1217,7 +1219,7 @@ Projection_25 100.00 root tpch.supplier.s_name, 17_col_0
└─HashAgg_31 320000.00 root group by:tpch.supplier.s_name, funcs:count(1), firstrow(tpch.supplier.s_name)
└─Selection_32 3786715.90 root not(16_aux_0)
└─IndexJoin_38 4733394.87 root left outer semi join, inner:IndexLookUp_37, outer key:l1.l_orderkey, inner key:l3.l_orderkey, other cond:ne(l3.l_suppkey, l1.l_suppkey)
├─IndexJoin_82 4733394.87 root semi join, inner:IndexLookUp_81, outer key:l1.l_orderkey, inner key:l2.l_orderkey, other cond:ne(l2.l_suppkey, l1.l_suppkey)
├─IndexJoin_82 4733394.87 root semi join, inner:IndexLookUp_81, outer key:l1.l_orderkey, inner key:l2.l_orderkey, other cond:ne(l2.l_suppkey, l1.l_suppkey), ne(l2.l_suppkey, tpch.supplier.s_suppkey)
│ ├─HashLeftJoin_88 5916743.59 root inner join, inner:TableReader_117, equal:[eq(tpch.supplier.s_nationkey, tpch.nation.n_nationkey)]
│ │ ├─HashLeftJoin_93 147918589.81 root inner join, inner:TableReader_114, equal:[eq(l1.l_suppkey, tpch.supplier.s_suppkey)]
│ │ │ ├─IndexJoin_100 147918589.81 root inner join, inner:IndexLookUp_99, outer key:tpch.orders.o_orderkey, inner key:l1.l_orderkey
Expand Down
40 changes: 40 additions & 0 deletions expression/util.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -372,6 +372,46 @@ func extractFiltersFromDNF(ctx sessionctx.Context, dnfFunc *ScalarFunction) ([]E
return extractedExpr, ComposeDNFCondition(ctx, newDNFItems...)
}

// DeriveRelaxedFiltersFromDNF given a DNF expression, derive a relaxed DNF expression which only contains columns
// in specified schema; the derived expression is a superset of original expression, i.e, any tuple satisfying
// the original expression must satisfy the derived expression. Return nil when the derived expression is univeral set.
// A running example is: for schema of t1, `(t1.a=1 and t2.a=1) or (t1.a=2 and t2.a=2)` would be derived as
// `t1.a=1 or t1.a=2`, while `t1.a=1 or t2.a=1` would get nil.
func DeriveRelaxedFiltersFromDNF(expr Expression, schema *Schema) Expression {
sf, ok := expr.(*ScalarFunction)
if !ok || sf.FuncName.L != ast.LogicOr {
return nil
}
ctx := sf.GetCtx()
dnfItems := FlattenDNFConditions(sf)
newDNFItems := make([]Expression, 0, len(dnfItems))
for _, dnfItem := range dnfItems {
cnfItems := SplitCNFItems(dnfItem)
newCNFItems := make([]Expression, 0, len(cnfItems))
for _, cnfItem := range cnfItems {
if itemSF, ok := cnfItem.(*ScalarFunction); ok && itemSF.FuncName.L == ast.LogicOr {
relaxedCNFItem := DeriveRelaxedFiltersFromDNF(cnfItem, schema)
if relaxedCNFItem != nil {
newCNFItems = append(newCNFItems, relaxedCNFItem)
}
// If relaxed expression for embedded DNF is univeral set, just drop this CNF item
continue
}
// This cnfItem must be simple expression now
// If it cannot be fully covered by schema, just drop this CNF item
if ExprFromSchema(cnfItem, schema) {
newCNFItems = append(newCNFItems, cnfItem)
}
}
// If this DNF item involves no column of specified schema, the relaxed expression must be universal set
if len(newCNFItems) == 0 {
return nil
}
newDNFItems = append(newDNFItems, ComposeCNFCondition(ctx, newCNFItems...))
}
return ComposeDNFCondition(ctx, newDNFItems...)
}

// GetRowLen gets the length if the func is row, returns 1 if not row.
func GetRowLen(e Expression) int {
if f, ok := e.(*ScalarFunction); ok && f.FuncName.L == ast.RowFunc {
Expand Down
24 changes: 21 additions & 3 deletions plan/logical_plan_builder.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -175,9 +175,12 @@ func (b *planBuilder) buildResultSetNode(node ast.ResultSetNode) (p LogicalPlan,
}
}

func extractOnCondition(conditions []expression.Expression, left LogicalPlan, right LogicalPlan) (
eqCond []*expression.ScalarFunction, leftCond []expression.Expression, rightCond []expression.Expression,
otherCond []expression.Expression) {
// extractOnCondition divide conditions in CNF of join node into 4 groups.
// These conditions can be where conditions, join conditions, or collection of both.
// If deriveLeft/deriveRight is set, we would try to derive more conditions for left/right plan.
func extractOnCondition(conditions []expression.Expression, left LogicalPlan, right LogicalPlan,
deriveLeft bool, deriveRight bool) (eqCond []*expression.ScalarFunction, leftCond []expression.Expression,
rightCond []expression.Expression, otherCond []expression.Expression) {
for _, expr := range conditions {
binop, ok := expr.(*expression.ScalarFunction)
if ok && binop.FuncName.L == ast.EQ {
Expand Down Expand Up @@ -210,6 +213,21 @@ func extractOnCondition(conditions []expression.Expression, left LogicalPlan, ri
} else if allFromLeft {
leftCond = append(leftCond, expr)
} else {
// Relax expr to two supersets: leftRelaxedCond and rightRelaxedCond, the expression now is
// `expr AND leftRelaxedCond AND rightRelaxedCond`. Motivation is to push filters down to
// children as much as possible.
if deriveLeft {
leftRelaxedCond := expression.DeriveRelaxedFiltersFromDNF(expr, left.Schema())
if leftRelaxedCond != nil {
leftCond = append(leftCond, leftRelaxedCond)
}
}
if deriveRight {
rightRelaxedCond := expression.DeriveRelaxedFiltersFromDNF(expr, right.Schema())
if rightRelaxedCond != nil {
rightCond = append(rightCond, rightRelaxedCond)
}
}
otherCond = append(otherCond, expr)
}
}
Expand Down
154 changes: 154 additions & 0 deletions plan/logical_plan_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@
package plan

import (
"fmt"
"sort"
"testing"

Expand Down Expand Up @@ -439,6 +440,159 @@ func (s *testPlanSuite) TestPredicatePushDown(c *C) {
}
}

func (s *testPlanSuite) TestJoinPredicatePushDown(c *C) {
defer testleak.AfterTest(c)()
tests := []struct {
sql string
left string
right string
}{
// issue #7628, inner join
{
sql: "select * from t as t1 join t as t2 on t1.b = t2.b where t1.a > t2.a",
left: "[]",
right: "[]",
},
{
sql: "select * from t as t1 join t as t2 on t1.b = t2.b where t1.a=1 or t2.a=1",
left: "[]",
right: "[]",
},
{
sql: "select * from t as t1 join t as t2 on t1.b = t2.b where (t1.a=1 and t2.a=1) or (t1.a=2 and t2.a=2)",
left: "[or(eq(t1.a, 1), eq(t1.a, 2))]",
right: "[or(eq(t2.a, 1), eq(t2.a, 2))]",
},
{
sql: "select * from t as t1 join t as t2 on t1.b = t2.b where (t1.c=1 and (t1.a=3 or t2.a=3)) or (t1.a=2 and t2.a=2)",
left: "[or(eq(t1.c, 1), eq(t1.a, 2))]",
right: "[]",
},
{
sql: "select * from t as t1 join t as t2 on t1.b = t2.b where (t1.c=1 and ((t1.a=3 and t2.a=3) or (t1.a=4 and t2.a=4)))",
left: "[eq(t1.c, 1) or(eq(t1.a, 3), eq(t1.a, 4))]",
right: "[or(eq(t2.a, 3), eq(t2.a, 4))]",
},
{
sql: "select * from t as t1 join t as t2 on t1.b = t2.b where (t1.a>1 and t1.a < 3 and t2.a=1) or (t1.a=2 and t2.a=2)",
left: "[or(and(gt(t1.a, 1), lt(t1.a, 3)), eq(t1.a, 2))]",
right: "[or(eq(t2.a, 1), eq(t2.a, 2))]",
},
{
sql: "select * from t as t1 join t as t2 on t1.b = t2.b and ((t1.a=1 and t2.a=1) or (t1.a=2 and t2.a=2))",
left: "[or(eq(t1.a, 1), eq(t1.a, 2))]",
right: "[or(eq(t2.a, 1), eq(t2.a, 2))]",
},
// issue #7628, left join
{
sql: "select * from t as t1 left join t as t2 on t1.b = t2.b and ((t1.a=1 and t2.a=1) or (t1.a=2 and t2.a=2))",
left: "[]",
right: "[or(eq(t2.a, 1), eq(t2.a, 2))]",
},
{
sql: "select * from t as t1 left join t as t2 on t1.b = t2.b and t1.a > t2.a",
left: "[]",
right: "[]",
},
{
sql: "select * from t as t1 left join t as t2 on t1.b = t2.b and (t1.a=1 or t2.a=1)",
left: "[]",
right: "[]",
},
{
sql: "select * from t as t1 left join t as t2 on t1.b = t2.b and ((t1.c=1 and (t1.a=3 or t2.a=3)) or (t1.a=2 and t2.a=2))",
left: "[]",
right: "[]",
},
{
sql: "select * from t as t1 left join t as t2 on t1.b = t2.b and ((t2.c=1 and (t1.a=3 or t2.a=3)) or (t1.a=2 and t2.a=2))",
left: "[]",
right: "[or(eq(t2.c, 1), eq(t2.a, 2))]",
},
{
sql: "select * from t as t1 left join t as t2 on t1.b = t2.b and ((t1.c=1 and ((t1.a=3 and t2.a=3) or (t1.a=4 and t2.a=4))) or (t1.a=2 and t2.a=2))",
left: "[]",
right: "[or(or(eq(t2.a, 3), eq(t2.a, 4)), eq(t2.a, 2))]",
},
}
for _, ca := range tests {
comment := Commentf("for %s", ca.sql)
stmt, err := s.ParseOneStmt(ca.sql, "", "")
c.Assert(err, IsNil, comment)
p, err := BuildLogicalPlan(s.ctx, stmt, s.is)
c.Assert(err, IsNil, comment)
p, err = logicalOptimize(flagPredicatePushDown|flagDecorrelate|flagPrunColumns, p.(LogicalPlan))
c.Assert(err, IsNil, comment)
proj, ok := p.(*LogicalProjection)
c.Assert(ok, IsTrue, comment)
join, ok := proj.children[0].(*LogicalJoin)
c.Assert(ok, IsTrue, comment)
leftPlan, ok := join.children[0].(*DataSource)
c.Assert(ok, IsTrue, comment)
rightPlan, ok := join.children[1].(*DataSource)
c.Assert(ok, IsTrue, comment)
leftCond := fmt.Sprintf("%s", leftPlan.pushedDownConds)
rightCond := fmt.Sprintf("%s", rightPlan.pushedDownConds)
c.Assert(leftCond, Equals, ca.left, comment)
c.Assert(rightCond, Equals, ca.right, comment)
}
}

func (s *testPlanSuite) TestOuterWherePredicatePushDown(c *C) {
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We can use SimpleRewriter so don't need to build actual plan in test.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hmm, we need to test the predicates are correctly derived and pushed to specific child in PredicatePushDown as well, so I prefer building logical plan.

defer testleak.AfterTest(c)()
tests := []struct {
sql string
sel string
left string
right string
}{
// issue #7628, left join with where condition
{
sql: "select * from t as t1 left join t as t2 on t1.b = t2.b where (t1.a=1 and t2.a=1) or (t1.a=2 and t2.a=2)",
sel: "[or(and(eq(t1.a, 1), eq(t2.a, 1)), and(eq(t1.a, 2), eq(t2.a, 2)))]",
left: "[or(eq(t1.a, 1), eq(t1.a, 2))]",
right: "[]",
},
{
sql: "select * from t as t1 left join t as t2 on t1.b = t2.b where (t1.c=1 and (t1.a=3 or t2.a=3)) or (t1.a=2 and t2.a=2)",
sel: "[or(and(eq(t1.c, 1), or(eq(t1.a, 3), eq(t2.a, 3))), and(eq(t1.a, 2), eq(t2.a, 2)))]",
left: "[or(eq(t1.c, 1), eq(t1.a, 2))]",
right: "[]",
},
{
sql: "select * from t as t1 left join t as t2 on t1.b = t2.b where (t1.c=1 and ((t1.a=3 and t2.a=3) or (t1.a=4 and t2.a=4))) or (t1.a=2 and t2.a=2)",
sel: "[or(and(eq(t1.c, 1), or(and(eq(t1.a, 3), eq(t2.a, 3)), and(eq(t1.a, 4), eq(t2.a, 4)))), and(eq(t1.a, 2), eq(t2.a, 2)))]",
left: "[or(and(eq(t1.c, 1), or(eq(t1.a, 3), eq(t1.a, 4))), eq(t1.a, 2))]",
right: "[]",
},
}
for _, ca := range tests {
comment := Commentf("for %s", ca.sql)
stmt, err := s.ParseOneStmt(ca.sql, "", "")
c.Assert(err, IsNil, comment)
p, err := BuildLogicalPlan(s.ctx, stmt, s.is)
c.Assert(err, IsNil, comment)
p, err = logicalOptimize(flagPredicatePushDown|flagDecorrelate|flagPrunColumns, p.(LogicalPlan))
c.Assert(err, IsNil, comment)
proj, ok := p.(*LogicalProjection)
c.Assert(ok, IsTrue, comment)
selection, ok := proj.children[0].(*LogicalSelection)
c.Assert(ok, IsTrue, comment)
selCond := fmt.Sprintf("%s", selection.Conditions)
c.Assert(selCond, Equals, ca.sel, comment)
join, ok := selection.children[0].(*LogicalJoin)
c.Assert(ok, IsTrue, comment)
leftPlan, ok := join.children[0].(*DataSource)
c.Assert(ok, IsTrue, comment)
rightPlan, ok := join.children[1].(*DataSource)
c.Assert(ok, IsTrue, comment)
leftCond := fmt.Sprintf("%s", leftPlan.pushedDownConds)
rightCond := fmt.Sprintf("%s", rightPlan.pushedDownConds)
c.Assert(leftCond, Equals, ca.left, comment)
c.Assert(rightCond, Equals, ca.right, comment)
}
}

func newPartitionInfoSchema(definitions []model.PartitionDefinition) infoschema.InfoSchema {
tableInfo := *MockTable()
cols := make([]*model.ColumnInfo, 0, len(tableInfo.Columns))
Expand Down
Loading