[CARMEL-7500][CARMEL-5963] CartesianProduct may introduce many concur… (apache#272)

Author: fenzhu

sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala

@@ -1684,6 +1684,15 @@ object SQLConf {
     .booleanConf
     .createWithDefault(true)
 
+  val CROSS_JOINS_ENABLED_FOR_JDBC_RELATION = buildConf(
+    "spark.carmel.sql.crossJoin.forJdbc.enabled")
+    .internal()
+    .doc("When false, we will throw an error if a query contains a cartesian product with " +
+      "jdbc relation in same stage.")
+    .version("2.0.0")
+    .booleanConf
+    .createWithDefault(true)
+
   val GROUP_BY_ORDINAL = buildConf("spark.sql.groupByOrdinal")
     .doc("When true, the ordinal numbers in group by clauses are treated as the position " +
       "in the select list. When false, the ordinal numbers are ignored.")
@@ -5566,6 +5575,9 @@ class SQLConf extends Serializable with Logging with SqlApiConf {
 
   def crossJoinEnabled: Boolean = getConf(SQLConf.CROSS_JOINS_ENABLED)
 
+  def crossJoinForJdbcRelationEnabled: Boolean =
+    getConf(SQLConf.CROSS_JOINS_ENABLED_FOR_JDBC_RELATION)
+
   override def sessionLocalTimeZone: String = getConf(SQLConf.SESSION_LOCAL_TIMEZONE)
 
   def jsonGeneratorIgnoreNullFields: Boolean = getConf(SQLConf.JSON_GENERATOR_IGNORE_NULL_FIELDS)

sql/core/src/main/scala/org/apache/spark/sql/execution/joins/CartesianProductExec.scala

@@ -23,7 +23,9 @@ import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.catalyst.expressions.{Attribute, Expression, JoinedRow, Predicate, UnsafeRow}
 import org.apache.spark.sql.catalyst.expressions.codegen.GenerateUnsafeRowJoiner
 import org.apache.spark.sql.catalyst.plans.{Inner, JoinType}
-import org.apache.spark.sql.execution.{ExternalAppendOnlyUnsafeRowArray, SparkPlan}
+import org.apache.spark.sql.execution.{DataSourceScanExec, ExternalAppendOnlyUnsafeRowArray, LeafExecNode, SparkPlan}
+import org.apache.spark.sql.execution.datasources.jdbc.JDBCRelation
+import org.apache.spark.sql.execution.exchange.Exchange
 import org.apache.spark.sql.execution.metric.SQLMetrics
 import org.apache.spark.util.CompletionIterator
 
@@ -72,6 +74,14 @@ case class CartesianProductExec(
     "numOutputRows" -> SQLMetrics.createMetric(sparkContext, "number of output rows"))
 
   protected override def doExecute(): RDD[InternalRow] = {
+    if (!conf.crossJoinForJdbcRelationEnabled &&
+      (containsJDBCRelationInSameStage(left) || containsJDBCRelationInSameStage(right))) {
+      throw new SparkException(
+        s"""Detected JDBC Relation in CartesianProduct, abort. Refer to
+           | https://wiki.vip.corp.ebay.com/x/QhCrJw#id-07FAQ-DetectedJDBCRelationinCartesianProduct
+           | for more details."""
+          .stripMargin)
+    }
     val numOutputRows = longMetric("numOutputRows")
 
     val leftResults = left.execute().asInstanceOf[RDD[UnsafeRow]]
@@ -102,6 +112,15 @@ case class CartesianProductExec(
     }
   }
 
+  private def containsJDBCRelationInSameStage(plan: SparkPlan): Boolean = {
+    plan match {
+      case scan: DataSourceScanExec if scan.relation.isInstanceOf[JDBCRelation] => true
+      case ex: Exchange => false
+      case _ =>
+        !plan.isInstanceOf[LeafExecNode] && plan.children.exists(containsJDBCRelationInSameStage)
+    }
+  }
+
   override protected def withNewChildrenInternal(
       newLeft: SparkPlan, newRight: SparkPlan): CartesianProductExec =
     copy(left = newLeft, right = newRight)

sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCSuite.scala

@@ -35,9 +35,11 @@ import org.apache.spark.sql.catalyst.parser.CatalystSqlParser
 import org.apache.spark.sql.catalyst.plans.logical.ShowCreateTable
 import org.apache.spark.sql.catalyst.util.{CaseInsensitiveMap, CharVarcharUtils, DateTimeTestUtils}
 import org.apache.spark.sql.execution.{DataSourceScanExec, ExtendedMode, ProjectExec}
+import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanExec
 import org.apache.spark.sql.execution.command.{ExplainCommand, ShowCreateTableCommand}
 import org.apache.spark.sql.execution.datasources.LogicalRelation
 import org.apache.spark.sql.execution.datasources.jdbc.{JDBCOptions, JDBCPartition, JDBCRelation, JdbcUtils}
+import org.apache.spark.sql.execution.joins.CartesianProductExec
 import org.apache.spark.sql.execution.metric.InputOutputMetricsHelper
 import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.sources._
@@ -2057,4 +2059,63 @@ class JDBCSuite extends QueryTest with SharedSparkSession {
     val df = sql("SELECT * FROM composite_name WHERE `last name` = 'smith'")
     assert(df.collect.toSet === Set(Row("smith", 1)))
   }
+
+  test("Check CartesianProducts For JDBC Relation") {
+    withTable("hdfs_foobar") {
+      sql(
+        """create table hdfs_foobar using parquet as
+          |select * from foobar
+          |""".stripMargin)
+
+      sql(
+        """create or replace temporary view all_foobar as
+          |select * from hdfs_foobar
+          |union all select * from foobar
+          |""".stripMargin)
+
+      sql(
+        """create or replace temporary view result as
+          |(select t_name, count(*) as count from all_foobar c
+          |join (select 'mary' as t_name) l
+          |on c.name like l.t_name
+          |group BY 1)
+          |""".stripMargin)
+
+      Seq(true, false).foreach(crossJoinForJdbcRelationEnabled => {
+        withSQLConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+          SQLConf.CROSS_JOINS_ENABLED_FOR_JDBC_RELATION.key ->
+            crossJoinForJdbcRelationEnabled.toString) {
+
+          val df = sql("SELECT * FROM result")
+
+          // scalastyle:off println
+          println(s"optimizedPlan: ${df.queryExecution.optimizedPlan}")
+          println(s"executedPlan: ${df.queryExecution.executedPlan}")
+          // scalastyle:on println
+
+          val plan = df.queryExecution.executedPlan match {
+            case ae: AdaptiveSparkPlanExec => ae.executedPlan
+            case p => p
+          }
+
+          assert(plan.collectFirst({
+            case cartesianProduct: CartesianProductExec => cartesianProduct
+          }).nonEmpty)
+
+          assert(plan.collectFirst({
+            case jdbc: DataSourceScanExec if jdbc.relation.isInstanceOf[JDBCRelation] => jdbc
+          }).nonEmpty)
+
+          if (crossJoinForJdbcRelationEnabled) {
+            df.collect()
+          } else {
+            val message = intercept[SparkException] {
+              df.collect()
+            }.getMessage
+            assert(message.contains(s"Detected JDBC Relation in CartesianProduct"))
+          }
+        }
+      })
+    }
+  }
 }