Allow EnforceSingleRowNode for predicate extractor

Assuming we have a daily ingested table that is partitioned on ds,
a filter like `ds = (SELECT '2020-07-01')` is converted into an
INNER JOIN, but this value is not passed to the other side of Join,
which leads to full table scan.

This commit will enable this value being treated as predicate, and
thus we only need to read this one partition.

presto-main/src/main/java/com/facebook/presto/sql/planner/EffectivePredicateExtractor.java

@@ -26,6 +26,7 @@
 import com.facebook.presto.spi.relation.VariableReferenceExpression;
 import com.facebook.presto.sql.planner.optimizations.JoinNodeUtils;
 import com.facebook.presto.sql.planner.plan.AssignUniqueId;
+import com.facebook.presto.sql.planner.plan.EnforceSingleRowNode;
 import com.facebook.presto.sql.planner.plan.ExchangeNode;
 import com.facebook.presto.sql.planner.plan.InternalPlanVisitor;
 import com.facebook.presto.sql.planner.plan.JoinNode;
@@ -160,6 +161,15 @@ public Expression visitExchange(ExchangeNode node, Void context)
             });
         }
 
+        @Override
+        public Expression visitEnforceSingleRow(EnforceSingleRowNode node, Void context)
+        {
+            if (node.getSource() instanceof ProjectNode) {
+                return node.getSource().accept(this, context);
+            }
+            return TRUE_LITERAL;
+        }
+
         @Override
         public Expression visitProject(ProjectNode node, Void context)
         {

presto-main/src/main/java/com/facebook/presto/sql/planner/RowExpressionPredicateExtractor.java

@@ -31,6 +31,7 @@
 import com.facebook.presto.spi.relation.RowExpression;
 import com.facebook.presto.spi.relation.VariableReferenceExpression;
 import com.facebook.presto.sql.planner.plan.AssignUniqueId;
+import com.facebook.presto.sql.planner.plan.EnforceSingleRowNode;
 import com.facebook.presto.sql.planner.plan.ExchangeNode;
 import com.facebook.presto.sql.planner.plan.InternalPlanVisitor;
 import com.facebook.presto.sql.planner.plan.JoinNode;
@@ -156,6 +157,15 @@ public RowExpression visitExchange(ExchangeNode node, Void context)
             });
         }
 
+        @Override
+        public RowExpression visitEnforceSingleRow(EnforceSingleRowNode node, Void context)
+        {
+            if (node.getSource() instanceof ProjectNode) {
+                return node.getSource().accept(this, context);
+            }
+            return TRUE_CONSTANT;
+        }
+
         @Override
         public RowExpression visitProject(ProjectNode node, Void context)
         {

presto-main/src/test/java/com/facebook/presto/sql/planner/TestLogicalPlanner.java

@@ -471,6 +471,19 @@ public void testPushDownJoinConditionConjunctsToInnerSideBasedOnInheritedPredica
                                                                 "REGION_REGIONKEY", "regionkey")))))));
     }
 
+    @Test
+    public void testScalarSubqueryJoinFilterPushdown()
+    {
+        assertPlan(
+                "SELECT * FROM orders WHERE orderkey = (SELECT 1)",
+                anyTree(
+                        join(INNER, ImmutableList.of(),
+                                filter("orderkey = BIGINT '1'",
+                                        tableScan("orders", ImmutableMap.of("orderkey", "orderkey"))),
+                                anyTree(
+                                        project(ImmutableMap.of("orderkey", expression("1")), any())))));
+    }
+
     @Test
     public void testSameScalarSubqueryIsAppliedOnlyOnce()
     {
@@ -1103,9 +1116,9 @@ public void testJoinNullFilters()
                                 LEFT,
                                 ImmutableList.of(equiJoinClause("NATION_REGIONKEY", "REGION_REGIONKEY")),
                                 anyTree(
-                                            tableScan(
-                                                    "nation",
-                                                    ImmutableMap.of("NATION_REGIONKEY", "regionkey"))),
+                                        tableScan(
+                                                "nation",
+                                                ImmutableMap.of("NATION_REGIONKEY", "regionkey"))),
                                 anyTree(
                                         filter("region_REGIONKEY IS NOT NULL",
                                                 tableScan(
@@ -1128,6 +1141,6 @@ public void testJoinNullFilters()
                                         tableScan(
                                                 "region",
                                                 ImmutableMap.of(
-                                                "REGION_REGIONKEY", "regionkey"))))));
+                                                        "REGION_REGIONKEY", "regionkey"))))));
     }
 }