DRILL-2962: Enhance scalar check to recurse below single-input rels. …

…Bump up Calcite version to 1.1.0-drill-r3. Enable TPCH q11 and remove the q11_1 version.

Add new test class for correlated subqueries (more tests will be added later).

exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/JoinUtils.java

@@ -27,7 +27,7 @@
 import org.apache.calcite.plan.volcano.RelSubset;
 import org.apache.drill.exec.planner.logical.DrillAggregateRel;
 import org.apache.drill.exec.planner.logical.DrillFilterRel;
-
+import org.apache.drill.exec.planner.logical.DrillProjectRel;
 import org.apache.drill.common.exceptions.DrillRuntimeException;
 import org.apache.drill.common.expression.ErrorCollector;
 import org.apache.drill.common.expression.ErrorCollectorImpl;
@@ -41,6 +41,7 @@
 
 import java.util.LinkedList;
 import java.util.List;
+
 import com.google.common.collect.Lists;
 
 public class JoinUtils {
@@ -205,16 +206,25 @@ public static void addLeastRestrictiveCasts(LogicalExpression[] leftExpressions,
     }
   }
 
-  public static boolean isScalarSubquery(RelNode childrel) {
+  /**
+   * Utility method to check if a subquery (represented by its root RelNode) is provably scalar. Currently
+   * only aggregates with no group-by are considered scalar. In the future, this method should be generalized
+   * to include more cases and reconciled with Calcite's notion of scalar.
+   * @param root The root RelNode to be examined
+   * @return True if the root rel or its descendant is scalar, False otherwise
+   */
+  public static boolean isScalarSubquery(RelNode root) {
     DrillAggregateRel agg = null;
-    RelNode currentrel = childrel;
+    RelNode currentrel = root;
     while (agg == null && currentrel != null) {
       if (currentrel instanceof DrillAggregateRel) {
         agg = (DrillAggregateRel)currentrel;
-      } else if (currentrel instanceof DrillFilterRel) {
-        currentrel = currentrel.getInput(0);
       } else if (currentrel instanceof RelSubset) {
         currentrel = ((RelSubset)currentrel).getBest() ;
+      } else if (currentrel.getInputs().size() == 1) {
+        // If the rel is not an aggregate or RelSubset, but is a single-input rel (could be Project,
+        // Filter, Sort etc.), check its input
+        currentrel = currentrel.getInput(0);
       } else {
         break;
       }

exec/java-exec/src/test/java/org/apache/drill/TestCorrelation.java

@@ -0,0 +1,40 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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.
+ */
+package org.apache.drill;
+
+import org.junit.Test;
+
+public class TestCorrelation extends PlanTestBase {
+  static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(TestCorrelation.class);
+
+  @Test  // DRILL-2962
+  public void testScalarAggCorrelatedSubquery() throws Exception {
+    String query = "select count(*) as cnt from cp.`tpch/nation.parquet` n1 "
+      + " where n1.n_nationkey  > (select avg(n2.n_regionkey) * 4 from cp.`tpch/nation.parquet` n2 "
+      + " where n1.n_regionkey = n2.n_nationkey)";
+
+    testBuilder()
+      .sqlQuery(query)
+      .ordered()
+      .baselineColumns("cnt")
+      .baselineValues((long) 17)
+      .build()
+      .run();
+  }
+
+}

exec/java-exec/src/test/java/org/apache/drill/TestTpchDistributed.java

@@ -80,16 +80,10 @@ public void tpch10() throws Exception{
   }
 
   @Test
-  @Ignore // depends on fix for Calcite-695 or an implementation of SqlSingleValueAggFunction in Drill
   public void tpch11() throws Exception{
     testDistributed("queries/tpch/11.sql");
   }
 
-  @Test // slight variant of tpch-11 that does not require SqlSingleValueAggFunction
-  public void tpch11_1() throws Exception{
-    testDistributed("queries/tpch/11_1.sql");
-  }
-
   @Test
   public void tpch12() throws Exception{
     testDistributed("queries/tpch/12.sql");

pom.xml

@@ -1038,7 +1038,7 @@
           <dependency>
             <groupId>org.apache.calcite</groupId>
             <artifactId>calcite-core</artifactId>
-            <version>1.1.0-drill-r2</version>
+            <version>1.1.0-drill-r3</version>
             <exclusions>
               <exclusion>
                 <groupId>org.jgrapht</groupId>