Merge pull request #3231 from dolthub/angela/joinfilterdrop

Push filters down into join condition

Author: angelamayxie

enginetest/enginetests.go

@@ -148,7 +148,7 @@ func TestSpatialQueriesPrepared(t *testing.T, harness Harness) {
 
 // TestJoinQueries tests join queries against a provided harness.
 func TestJoinQueries(t *testing.T, harness Harness) {
-	harness.Setup(setup.MydbData, setup.MytableData, setup.Pk_tablesData, setup.OthertableData, setup.NiltableData, setup.XyData, setup.FooData)
+	harness.Setup(setup.MydbData, setup.MytableData, setup.Pk_tablesData, setup.OthertableData, setup.NiltableData, setup.XyData, setup.FooData, setup.Comp_index_tablesData)
 	e, err := harness.NewEngine(t)
 	require.NoError(t, err)
 
@@ -263,7 +263,7 @@ func TestQueriesPrepared(t *testing.T, harness Harness) {
 
 // TestJoinQueriesPrepared tests join queries as prepared statements against a provided harness.
 func TestJoinQueriesPrepared(t *testing.T, harness Harness) {
-	harness.Setup(setup.MydbData, setup.MytableData, setup.Pk_tablesData, setup.OthertableData, setup.NiltableData, setup.XyData, setup.FooData)
+	harness.Setup(setup.MydbData, setup.MytableData, setup.Pk_tablesData, setup.OthertableData, setup.NiltableData, setup.XyData, setup.FooData, setup.Comp_index_tablesData)
 	for _, tt := range queries.JoinQueryTests {
 		if tt.SkipPrepared {
 			continue

enginetest/join_planning_tests.go

@@ -425,9 +425,9 @@ var JoinPlanningTests = []joinPlanScript{
 				exp:   []sql.Row{{0, 2}, {2, 1}, {3, 3}},
 			},
 			{
-				// anti join will be cross-join-right, be passed non-nil parent row
+				// anti join will be cross-join-right, then converted to inner join when filters are pushed down, be passed non-nil parent row
 				q:     "select x,a from ab, (select * from xy where x != (select r from rs where r = 1) order by 1) sq where x = 2 and b = 2 order by 1,2;",
-				types: []plan.JoinType{plan.JoinTypeCrossHash, plan.JoinTypeLeftOuter},
+				types: []plan.JoinType{plan.JoinTypeInner, plan.JoinTypeLeftOuter},
 				exp:   []sql.Row{{2, 0}, {2, 1}, {2, 2}},
 			},
 			{

enginetest/queries/integration_plans.go

@@ -789,6 +789,18 @@ on w = 0;`,
 		Query:    "select * from foo.othertable join mydb.othertable on foo.othertable.text = 'a'",
 		Expected: []sql.Row{{"a", 4, "third", 1}, {"a", 4, "second", 2}, {"a", 4, "first", 3}},
 	},
+	{
+		// Regression test ensuring that filters are not dropped after join optimization
+		// https://github.com/dolthub/dolt/issues/9868
+		Query:    "select * from comp_index_t0 c join comp_index_t0 b join comp_index_t0 a on a.v2 = b.pk and b.v2 = c.pk and c.v2 = 1",
+		Expected: []sql.Row{},
+	},
+	{
+		// Regression test ensuring that filters are not dropped after join optimization
+		// https://github.com/dolthub/dolt/issues/9868
+		Query:    "select * from comp_index_t0 a join comp_index_t0 b join comp_index_t0 c on a.v2 = b.pk and b.v2 = c.pk and c.v2 = 5",
+		Expected: []sql.Row{},
+	},
 }
 
 var JoinScriptTests = []ScriptTest{

enginetest/queries/join_queries.go

@@ -115,5 +115,6 @@ var (
 		XyData,
 		Invert_pkData,
 		JoinData,
+		Comp_index_tablesData,
 	}
 )

enginetest/queries/query_plans.go

@@ -391,10 +391,14 @@ func evalSpatialIndexPlanTest(t *testing.T, harness Harness, e QueryEngine, quer
 			if n == nil {
 				return false
 			}
-			if _, ok := n.(*plan.Filter); ok {
+			switch n := n.(type) {
+			case *plan.Filter:
 				hasFilter = true
-			}
-			if _, ok := n.(*plan.IndexedTableAccess); ok {
+			case *plan.JoinNode:
+				if n.Filter != nil {
+					hasFilter = true
+				}
+			case *plan.IndexedTableAccess:
 				hasRightOrder = hasFilter
 				hasIndex = true
 			}

enginetest/queries/tpch_plans.go

@@ -87,7 +87,7 @@ func pushFilters(ctx *sql.Context, a *Analyzer, n sql.Node, scope *plan.Scope, s
 
 			// move filter predicates directly above their respective tables in joins
 			ret, same, err := pushdownAboveTables(n, filters)
-			if same || err != nil {
+			if err != nil {
 				return n, transform.SameTree, err
 			}
 
@@ -96,7 +96,7 @@ func pushFilters(ctx *sql.Context, a *Analyzer, n sql.Node, scope *plan.Scope, s
 				return n, transform.SameTree, fmt.Errorf("pushdown mistakenly converted filter to non-filter: %T", ret)
 			}
 			// remove handled
-			newF := removePushedDownPredicates(ctx, a, retF, filters)
+			newF := updateFilterNode(ctx, a, retF, filters)
 			if newF != nil {
 				same = transform.NewTree
 				ret = newF
@@ -202,7 +202,7 @@ func transformPushdownSubqueryAliasFilters(ctx *sql.Context, a *Analyzer, n sql.
 		return transform.NodeWithCtx(n, filterPushdownChildSelector, func(c transform.Context) (sql.Node, transform.TreeIdentity, error) {
 			switch node := c.Node.(type) {
 			case *plan.Filter:
-				newF := removePushedDownPredicates(ctx, a, node, filters)
+				newF := updateFilterNode(ctx, a, node, filters)
 				if newF == nil {
 					return node, transform.SameTree, nil
 				}
@@ -311,22 +311,38 @@ func pushdownFiltersUnderSubqueryAlias(ctx *sql.Context, a *Analyzer, sa *plan.S
 	return n, transform.NewTree, nil
 }
 
-// removePushedDownPredicates removes all handled filter predicates from the filter given and returns. If all
-// predicates have been handled, it replaces the filter with its child.
-func removePushedDownPredicates(ctx *sql.Context, a *Analyzer, node *plan.Filter, filters *filterSet) sql.Node {
-	if filters.handledCount() == 0 {
-		a.Log("no handled filters, leaving filter untouched")
-		return nil
-	}
-
-	// figure out if the filter's filters were all handled
+// updateFilterNode updates the filter node based on the filter predicates handled. Any handled filter predicates are
+// removed from the filter node. If all filter predicates have been handled and there are no unhandled predicates, the
+// filter node is removed. If there are remaining filter predicates and the immediate child of the filter is a non-outer
+// join, the remaining unhandled filters are pushed into the join node and added to the join filters, and the filter
+// node is removed.
+func updateFilterNode(ctx *sql.Context, a *Analyzer, node *plan.Filter, filters *filterSet) sql.Node {
 	filterExpressions := expression.SplitConjunction(node.Expression)
 	unhandled := subtractExprSet(filterExpressions, filters.handledFilters)
+
 	if len(unhandled) == 0 {
 		a.Log("filter node has no unhandled filters, so it will be removed")
 		return node.Child
 	}
 
+	// push filters into joinChild
+	if joinChild, ok := node.Child.(*plan.JoinNode); ok && !joinChild.Op.IsOuter() {
+		a.Log("pushing filters into join node")
+		if joinChild.Op.IsCross() {
+			return plan.NewInnerJoin(joinChild.Left(), joinChild.Right(), expression.JoinAnd(unhandled...))
+		}
+		if joinChild.Filter != nil {
+			unhandled = append(unhandled, joinChild.Filter)
+		}
+		joinChild.Filter = expression.JoinAnd(unhandled...)
+		return joinChild
+	}
+
+	if filters.handledCount() == 0 {
+		a.Log("no handled filters, leaving filter untouched")
+		return nil
+	}
+
 	if len(unhandled) == len(filterExpressions) {
 		a.Log("no filters removed from filter node")
 		return nil

enginetest/scriptgen/setup/helper.go

@@ -395,15 +395,15 @@ func (j *joinOrderBuilder) hasEqEdge(leftCol, rightCol sql.ColumnId) bool {
 	return false
 }
 
-func (j *joinOrderBuilder) findVertexFromCol(col sql.ColumnId) (vertexIndex, GroupId) {
+func (j *joinOrderBuilder) findVertexFromCol(col sql.ColumnId) (vertexIndex, GroupId, bool) {
 	for i, v := range j.vertices {
 		if t, ok := v.(SourceRel); ok {
 			if t.Group().RelProps.FuncDeps().All().Contains(col) {
-				return vertexIndex(i), t.Group().Id
+				return vertexIndex(i), t.Group().Id, true
 			}
 		}
 	}
-	panic("vertex not found")
+	return 0, 0, false
 }
 
 func (j *joinOrderBuilder) findVertexFromGroup(grp GroupId) vertexIndex {
@@ -421,8 +421,11 @@ func (j *joinOrderBuilder) findVertexFromGroup(grp GroupId) vertexIndex {
 // on an equality filter between two columns.
 func (j *joinOrderBuilder) makeTransitiveEdge(col1, col2 sql.ColumnId) {
 	var vert vertexSet
-	v1, _ := j.findVertexFromCol(col1)
-	v2, _ := j.findVertexFromCol(col2)
+	v1, _, v1found := j.findVertexFromCol(col1)
+	v2, _, v2found := j.findVertexFromCol(col2)
+	if !v1found || !v2found {
+		return
+	}
 	vert = vert.add(v1).add(v2)
 
 	// find edge where the vertices are provided but partitioned