[CORE]Fix the bug that don't pushdown mysql datasource when there are…

… some spark-provided functions in subquery #64 This PR solved the first question about Issues[61]#61, don't `pushdown `MySQL `datasource `when there are some functions provided by spark in subquery.
Qihoo360 · Nov 12, 2019 · d25e615 · d25e615
1 parent e2daaaf
commit d25e615
Showing 1 changed file with 18 additions and 14 deletions.
diff --git a/...in/scala/org/apache/spark/sql/xsql/execution/datasources/mysql/MysqlSpecialStrategy.scala b/...in/scala/org/apache/spark/sql/xsql/execution/datasources/mysql/MysqlSpecialStrategy.scala
@@ -474,25 +474,29 @@ class TransmitOriginalQuery(session: SparkSession) extends Rule[LogicalPlan] {
       "rand",
       "cast",
       "like")
-    plan.expressions.forall {
-      case a: Alias =>
-        if (a.child.isInstanceOf[AttributeReference] || a.child.isInstanceOf[ScalarSubquery]
+
+    @inline def functionNotPushdown(plan: LogicalPlan): Boolean = {
+      plan.expressions.exists {
+        case a: Alias =>
+          if (a.child.isInstanceOf[AttributeReference] || a.child.isInstanceOf[ScalarSubquery]
             || a.child.isInstanceOf[BinaryArithmetic]) {
-          true
-        } else {
-          val func = if (a.child.isInstanceOf[AggregateExpression]) {
-            a.child.asInstanceOf[AggregateExpression].aggregateFunction
-          } else {
-            a.child
-          }
-          if (!pushdownFunctions.contains(func.prettyName)) {
             false
           } else {
-            true
+            val func = if (a.child.isInstanceOf[AggregateExpression]) {
+              a.child.asInstanceOf[AggregateExpression].aggregateFunction
+            } else {
+              a.child
+            }
+            if (pushdownFunctions.contains(func.prettyName)) {
+              false
+            } else {
+              true
+            }
           }
-        }
-      case _ => true
+        case _ => false
+      }
     }
+    plan.find(functionNotPushdown).isEmpty
   }
 
   /**