[CARMEL-4270] UpdateCommand and DeleteCommand should skip to optimize subquery in their condition (delta-io#16)

Author: LantaoJin

build.sbt

@@ -176,8 +176,6 @@ packageBin in Compile := spPackage.value
 
 sparkComponents := Seq("sql")
 
-publishArtifact in Test := true
-
 /********************
  * Release settings *
  ********************/

src/main/scala/org/apache/spark/sql/delta/commands/DeleteCommand.scala

@@ -25,7 +25,7 @@ import org.apache.spark.sql.{Column, Dataset, Row, SparkSession}
 import org.apache.spark.sql.catalyst.analysis.EliminateSubqueryAliases
 import org.apache.spark.sql.catalyst.expressions.{EqualNullSafe, Expression, InputFileName, Literal, Not}
 import org.apache.spark.sql.catalyst.plans.QueryPlan
-import org.apache.spark.sql.catalyst.plans.logical.{DeltaDelete, LogicalPlan}
+import org.apache.spark.sql.catalyst.plans.logical.{DeltaDelete, LogicalPlan, SkipOptimizingSubquery}
 import org.apache.spark.sql.execution.SQLExecution
 import org.apache.spark.sql.execution.command.RunnableCommand
 import org.apache.spark.sql.execution.metric.{SQLMetric, SQLMetrics}
@@ -47,7 +47,7 @@ case class DeleteCommand(
     tahoeFileIndex: TahoeFileIndex,
     target: LogicalPlan,
     condition: Option[Expression])
-  extends RunnableCommand with DeltaCommand {
+  extends RunnableCommand with DeltaCommand with SkipOptimizingSubquery {
 
   override def innerChildren: Seq[QueryPlan[_]] = Seq(target)
 

src/main/scala/org/apache/spark/sql/delta/commands/UpdateCommand.scala

@@ -24,7 +24,7 @@ import org.apache.spark.SparkContext
 import org.apache.spark.sql.{Column, Dataset, Row, SparkSession}
 import org.apache.spark.sql.catalyst.expressions.{Alias, Expression, If, Literal}
 import org.apache.spark.sql.catalyst.plans.QueryPlan
-import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, SupportsSubquery}
+import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, SkipOptimizingSubquery, SupportsSubquery}
 import org.apache.spark.sql.execution.SQLExecution
 import org.apache.spark.sql.execution.command.RunnableCommand
 import org.apache.spark.sql.execution.metric.{SQLMetric, SQLMetrics}
@@ -46,7 +46,7 @@ case class UpdateCommand(
     target: LogicalPlan,
     updateExpressions: Seq[Expression],
     condition: Option[Expression])
-  extends RunnableCommand with DeltaCommand with SupportsSubquery {
+  extends RunnableCommand with DeltaCommand with SupportsSubquery with SkipOptimizingSubquery {
 
   override def innerChildren: Seq[QueryPlan[_]] = Seq(target)