From e3666bd5dc95f3e6374565ef5bc79b7923cfdec4 Mon Sep 17 00:00:00 2001
From: Miguel Molina <miguel@erizocosmi.co>
Date: Fri, 22 Feb 2019 15:13:55 +0100
Subject: [PATCH] add exchanges to SubqueryAlias nodes too

Signed-off-by: Miguel Molina <miguel@erizocosmi.co>
---
 sql/analyzer/parallelize.go      |  4 +++
 sql/analyzer/parallelize_test.go | 52 ++++++++++++++++++++++++++++++++
 2 files changed, 56 insertions(+)

diff --git a/sql/analyzer/parallelize.go b/sql/analyzer/parallelize.go
index 126ab6d1e..595f36923 100644
--- a/sql/analyzer/parallelize.go
+++ b/sql/analyzer/parallelize.go
@@ -83,6 +83,10 @@ func isParallelizable(node sql.Node) bool {
 		case sql.Table:
 			lastWasTable = true
 			tableSeen = true
+		case *plan.SubqueryAlias:
+			lastWasTable = true
+			tableSeen = true
+			return false
 		default:
 			ok = false
 			return false
diff --git a/sql/analyzer/parallelize_test.go b/sql/analyzer/parallelize_test.go
index 8cbeb1d1f..775372c24 100644
--- a/sql/analyzer/parallelize_test.go
+++ b/sql/analyzer/parallelize_test.go
@@ -226,3 +226,55 @@ func TestRemoveRedundantExchanges(t *testing.T) {
 	require.NoError(err)
 	require.Equal(expected, result)
 }
+
+func TestParallelizeSubquery(t *testing.T) {
+	require := require.New(t)
+	table := mem.NewTable("t", nil)
+	rule := getRuleFrom(OnceAfterAll, "parallelize")
+	node := plan.NewProject(
+		[]sql.Expression{
+			expression.NewLiteral(1, sql.Int64),
+		},
+		plan.NewSubqueryAlias("sub", plan.NewProject(
+			nil,
+			plan.NewInnerJoin(
+				plan.NewFilter(
+					expression.NewLiteral(1, sql.Int64),
+					plan.NewResolvedTable(table),
+				),
+				plan.NewFilter(
+					expression.NewLiteral(1, sql.Int64),
+					plan.NewResolvedTable(table),
+				),
+				expression.NewLiteral(1, sql.Int64),
+			),
+		)),
+	)
+
+	expected := plan.NewExchange(
+		2,
+		plan.NewProject(
+			[]sql.Expression{
+				expression.NewLiteral(1, sql.Int64),
+			},
+			plan.NewSubqueryAlias("sub", plan.NewProject(
+				nil,
+				plan.NewInnerJoin(
+					plan.NewFilter(
+						expression.NewLiteral(1, sql.Int64),
+						plan.NewResolvedTable(table),
+					),
+					plan.NewFilter(
+						expression.NewLiteral(1, sql.Int64),
+						plan.NewResolvedTable(table),
+					),
+					expression.NewLiteral(1, sql.Int64),
+				),
+			)),
+		),
+	)
+
+	result, err := rule.Apply(sql.NewEmptyContext(), &Analyzer{Parallelism: 2}, node)
+	require.NoError(err)
+	require.Equal(expected, result)
+}