[NSE-359] [NSE-273] Introduce shim layer to fix compatibility issues for gazelle on spark 3.1 & 3.2

arrow-data-source/common/src/main/scala/com/intel/oap/sql/execution/RowToArrowColumnarExec.scala

@@ -310,4 +310,8 @@ case class RowToArrowColumnarExec(child: SparkPlan) extends UnaryExecNode {
       }
     }
   }
+
+  // For spark 3.2.
+  protected def withNewChildInternal(newChild: SparkPlan): RowToArrowColumnarExec =
+    copy(child = newChild)
 }

arrow-data-source/parquet/pom.xml

@@ -22,6 +22,12 @@
             <artifactId>spark-arrow-datasource-standard</artifactId>
             <version>${project.version}</version>
         </dependency>
+        <dependency>
+            <groupId>com.intel.oap</groupId>
+            <artifactId>spark-sql-columnar-shims-common</artifactId>
+            <version>${project.version}</version>
+            <scope>provided</scope>
+        </dependency>
     </dependencies>
 
 </project>

arrow-data-source/parquet/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFileFormat.scala

@@ -24,6 +24,7 @@ import scala.collection.JavaConverters._
 import scala.util.{Failure, Try}
 
 import com.intel.oap.spark.sql.execution.datasources.arrow.ArrowFileFormat
+import com.intel.oap.sql.shims.SparkShimLoader
 import org.apache.hadoop.conf.Configuration
 import org.apache.hadoop.fs.{FileStatus, Path}
 import org.apache.hadoop.mapreduce.{Job, JobID, OutputCommitter, TaskAttemptContext, TaskAttemptID, TaskID, TaskType}
@@ -274,6 +275,7 @@ class ParquetFileFormat
     val pushDownStringStartWith = sqlConf.parquetFilterPushDownStringStartWith
     val pushDownInFilterThreshold = sqlConf.parquetFilterPushDownInFilterThreshold
     val isCaseSensitive = sqlConf.caseSensitiveAnalysis
+    val parquetOptions = new ParquetOptions(options, sparkSession.sessionState.conf)
 
     (file: PartitionedFile) => {
       assert(file.partitionValues.numFields == partitionSchema.size)
@@ -292,11 +294,17 @@ class ParquetFileFormat
 
       lazy val footerFileMetaData =
         ParquetFileReader.readFooter(sharedConf, filePath, SKIP_ROW_GROUPS).getFileMetaData
+
+      val datetimeRebaseMode =
+        SparkShimLoader.getSparkShims.getDatetimeRebaseMode(footerFileMetaData, parquetOptions)
+
       // Try to push down filters when filter push-down is enabled.
       val pushed = if (enableParquetFilterPushDown) {
         val parquetSchema = footerFileMetaData.getSchema
-        val parquetFilters = new ParquetFilters(parquetSchema, pushDownDate, pushDownTimestamp,
-          pushDownDecimal, pushDownStringStartWith, pushDownInFilterThreshold, isCaseSensitive)
+        val parquetFilters =
+          SparkShimLoader.getSparkShims.newParquetFilters(parquetSchema: MessageType,
+            pushDownDate, pushDownTimestamp, pushDownDecimal, pushDownStringStartWith,
+            pushDownInFilterThreshold, isCaseSensitive, datetimeRebaseMode)
         filters
           // Collects all converted Parquet filter predicates. Notice that not all predicates can be
           // converted (`ParquetFilters.createFilter` returns an `Option`). That's why a `flatMap`
@@ -322,10 +330,6 @@ class ParquetFileFormat
           None
         }
 
-      val datetimeRebaseMode = DataSourceUtils.datetimeRebaseMode(
-        footerFileMetaData.getKeyValueMetaData.get,
-        SQLConf.get.getConf(SQLConf.LEGACY_PARQUET_REBASE_MODE_IN_READ))
-
       val attemptId = new TaskAttemptID(new TaskID(new JobID(), TaskType.MAP, 0), 0)
       val hadoopAttemptContext =
         new TaskAttemptContextImpl(broadcastedHadoopConf.value.value, attemptId)

arrow-data-source/pom.xml

@@ -42,7 +42,7 @@
   </pluginRepositories>
 
   <dependencies>
-  <dependency>
+    <dependency>
        <groupId>javax.servlet</groupId>
        <artifactId>javax.servlet-api</artifactId>
        <version>3.1.0</version>

arrow-data-source/standard/pom.xml

@@ -18,6 +18,12 @@
             <artifactId>spark-arrow-datasource-common</artifactId>
             <version>${project.version}</version>
         </dependency>
+        <dependency>
+            <groupId>com.intel.oap</groupId>
+            <artifactId>spark-sql-columnar-shims-common</artifactId>
+            <version>${project.version}</version>
+            <scope>provided</scope>
+        </dependency>
     </dependencies>
 
     <build>

arrow-data-source/standard/src/main/scala/com/intel/oap/spark/sql/ArrowWriteExtension.scala

@@ -137,6 +137,10 @@ object ArrowWriteExtension {
   private case class ColumnarToFakeRowLogicAdaptor(child: LogicalPlan)
       extends OrderPreservingUnaryNode {
     override def output: Seq[Attribute] = child.output
+
+    // For spark 3.2.
+    protected def withNewChildInternal(newChild: LogicalPlan): ColumnarToFakeRowLogicAdaptor =
+      copy(child = newChild)
   }
 
   private case class ColumnarToFakeRowAdaptor(child: SparkPlan) extends ColumnarToRowTransition {
@@ -149,6 +153,10 @@ object ArrowWriteExtension {
     }
 
     override def output: Seq[Attribute] = child.output
+
+    // For spark 3.2.
+    protected def withNewChildInternal(newChild: SparkPlan): ColumnarToFakeRowAdaptor =
+      copy(child = newChild)
   }
 
   case class SimpleStrategy() extends Strategy {

arrow-data-source/standard/src/main/scala/com/intel/oap/spark/sql/execution/datasources/arrow/ArrowFileFormat.scala

@@ -94,6 +94,11 @@ class ArrowFileFormat extends FileFormat with DataSourceRegister with Serializab
           override def close(): Unit = {
             writeQueue.close()
           }
+
+          // Do NOT add override keyword for compatibility on spark 3.1.
+          def path(): String = {
+            path
+          }
         }
       }
     }

native-sql-engine/core/pom.xml

@@ -44,6 +44,33 @@
     <nativesql.build_protobuf>${build_protobuf}</nativesql.build_protobuf>
     <nativesql.build_jemalloc>${build_jemalloc}</nativesql.build_jemalloc>
   </properties>
+
+  <profiles>
+    <profile>
+      <id>spark-3.1.1</id>
+      <activation>
+        <activeByDefault>true</activeByDefault>
+      </activation>
+      <dependencies>
+        <dependency>
+          <groupId>com.intel.oap</groupId>
+          <artifactId>spark-sql-columnar-shims-spark311</artifactId>
+          <version>${project.version}</version>
+        </dependency>
+      </dependencies>
+    </profile>
+    <profile>
+      <id>spark-3.2.0</id>
+      <dependencies>
+        <dependency>
+          <groupId>com.intel.oap</groupId>
+          <artifactId>spark-sql-columnar-shims-spark320</artifactId>
+          <version>${project.version}</version>
+        </dependency>
+      </dependencies>
+    </profile>
+  </profiles>
+
   <dependencies>
     <!-- Prevent our dummy JAR from being included in Spark distributions or uploaded to YARN -->
     <dependency>
@@ -166,19 +193,19 @@
     <dependency>
       <groupId>com.fasterxml.jackson.core</groupId>
       <artifactId>jackson-core</artifactId>
-      <version>2.10.0</version>
+      <version>${jackson.version}</version>
       <scope>test</scope>
     </dependency>
     <dependency>
       <groupId>com.fasterxml.jackson.core</groupId>
       <artifactId>jackson-annotations</artifactId>
-      <version>2.10.0</version>
+      <version>${jackson.version}</version>
       <scope>test</scope>
     </dependency>
     <dependency>
       <groupId>com.fasterxml.jackson.core</groupId>
       <artifactId>jackson-databind</artifactId>
-      <version>2.10.0</version>
+      <version>${jackson.version}</version>
       <scope>test</scope>
     </dependency>
     <dependency>
@@ -299,7 +326,7 @@
       <groupId>com.intel.oap</groupId>
       <artifactId>spark-sql-columnar-shims-common</artifactId>
       <version>${project.version}</version>
-      <scope>compile</scope>
+      <scope>provided</scope>
     </dependency>
     <dependency>
       <groupId>org.apache.logging.log4j</groupId>

native-sql-engine/core/src/main/scala/com/intel/oap/execution/CoalesceBatchesExec.scala

@@ -147,6 +147,10 @@ case class CoalesceBatchesExec(child: SparkPlan) extends UnaryExecNode {
       new CloseableColumnBatchIterator(res)
     }
   }
+
+  // For spark 3.2.
+  protected def withNewChildInternal(newChild: SparkPlan): CoalesceBatchesExec =
+    copy(child = newChild)
 }
 
 object CoalesceBatchesExec {

native-sql-engine/core/src/main/scala/com/intel/oap/execution/ColumnarBasicPhysicalOperators.scala

@@ -100,8 +100,11 @@ case class ColumnarConditionProjectExec(
     }
   }
 
-  def isNullIntolerant(expr: Expression): Boolean = expr match {
-    case e: NullIntolerant => e.children.forall(isNullIntolerant)
+  // In spark 3.2, PredicateHelper has already introduced isNullIntolerant with completely same
+  // code. If we use the same method name, override keyword is required. But in spark3.1, no
+  // method is overridden. So we use an independent method name.
+  def isNullIntolerantInternal(expr: Expression): Boolean = expr match {
+    case e: NullIntolerant => e.children.forall(isNullIntolerantInternal)
     case _ => false
   }
 
@@ -110,7 +113,7 @@ case class ColumnarConditionProjectExec(
 
   val notNullAttributes = if (condition != null) {
     val (notNullPreds, otherPreds) = splitConjunctivePredicates(condition).partition {
-      case IsNotNull(a) => isNullIntolerant(a) && a.references.subsetOf(child.outputSet)
+      case IsNotNull(a) => isNullIntolerantInternal(a) && a.references.subsetOf(child.outputSet)
       case _ => false
     }
     notNullPreds.flatMap(_.references).distinct.map(_.exprId)
@@ -267,6 +270,9 @@ case class ColumnarConditionProjectExec(
     }
   }
 
+  // For spark 3.2.
+  protected def withNewChildInternal(newChild: SparkPlan): ColumnarConditionProjectExec =
+    copy(child = newChild)
 }
 
 case class ColumnarUnionExec(children: Seq[SparkPlan]) extends SparkPlan {
@@ -308,6 +314,10 @@ case class ColumnarUnionExec(children: Seq[SparkPlan]) extends SparkPlan {
       : org.apache.spark.rdd.RDD[org.apache.spark.sql.catalyst.InternalRow] = {
     throw new UnsupportedOperationException(s"This operator doesn't support doExecute().")
   }
+
+  // For spark 3.2.
+  protected def withNewChildrenInternal(newChildren: IndexedSeq[SparkPlan]): ColumnarUnionExec =
+    copy(children = newChildren)
 }
 
 //TODO(): consolidate locallimit and globallimit
@@ -380,6 +390,10 @@ case class ColumnarLocalLimitExec(limit: Int, child: SparkPlan) extends LimitExe
     throw new UnsupportedOperationException(s"This operator doesn't support doExecute().")
   }
 
+  protected def withNewChildInternal(newChild: SparkPlan):
+  ColumnarLocalLimitExec =
+    copy(child = newChild)
+
 }
 
 case class ColumnarGlobalLimitExec(limit: Int, child: SparkPlan) extends LimitExec {
@@ -451,4 +465,8 @@ case class ColumnarGlobalLimitExec(limit: Int, child: SparkPlan) extends LimitEx
       : org.apache.spark.rdd.RDD[org.apache.spark.sql.catalyst.InternalRow] = {
     throw new UnsupportedOperationException(s"This operator doesn't support doExecute().")
   }
+
+  protected def withNewChildInternal(newChild: SparkPlan):
+  ColumnarGlobalLimitExec =
+    copy(child = newChild)
 }

native-sql-engine/core/src/main/scala/com/intel/oap/execution/ColumnarBroadcastHashJoinExec.scala

@@ -21,9 +21,11 @@ import com.google.common.collect.Lists
 import com.intel.oap.GazellePluginConfig
 import com.intel.oap.expression._
 import com.intel.oap.vectorized.{ExpressionEvaluator, _}
+import com.intel.oap.sql.shims.SparkShimLoader
 import org.apache.arrow.gandiva.expression._
 import org.apache.arrow.vector.types.pojo.{ArrowType, Field}
 import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.SQLContext
 import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.catalyst.expressions._
 import org.apache.spark.sql.catalyst.plans._
@@ -35,6 +37,7 @@ import org.apache.spark.sql.catalyst.expressions.codegen.CodegenContext
 import org.apache.spark.sql.catalyst.optimizer.{BuildLeft, BuildRight, BuildSide}
 import org.apache.spark.sql.catalyst.plans.physical.{HashPartitioning, Partitioning, PartitioningCollection}
 import org.apache.spark.sql.execution.metric.SQLMetrics
+import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.util.ArrowUtils
 import org.apache.spark.sql.vectorized.{ColumnVector, ColumnarBatch}
 import org.apache.spark.util.{ExecutorManager, UserAddedJarUtils}
@@ -59,7 +62,7 @@ case class ColumnarBroadcastHashJoinExec(
     nullAware: Boolean = false)
     extends BaseJoinExec
     with ColumnarCodegenSupport
-    with ShuffledJoin {
+    with ColumnarShuffledJoin {
 
   val sparkConf = sparkContext.getConf
   val numaBindingInfo = GazellePluginConfig.getConf.numaBindingInfo
@@ -89,6 +92,9 @@ case class ColumnarBroadcastHashJoinExec(
   }
   buildCheck()
 
+  // A method in ShuffledJoin of spark3.2.
+  def isSkewJoin: Boolean = false
+
   def buildCheck(): Unit = {
     joinType match {
       case _: InnerLike =>
@@ -145,13 +151,13 @@ case class ColumnarBroadcastHashJoinExec(
     throw new UnsupportedOperationException(
       s"ColumnarBroadcastHashJoinExec doesn't support doExecute")
   }
-
   val isNullAwareAntiJoin : Boolean = nullAware
 
-  val broadcastHashJoinOutputPartitioningExpandLimit: Int = sqlContext.getConf(
-    "spark.sql.execution.broadcastHashJoin.outputPartitioningExpandLimit").trim().toInt
-
   override lazy val outputPartitioning: Partitioning = {
+    val broadcastHashJoinOutputPartitioningExpandLimit: Int =
+      SparkShimLoader
+        .getSparkShims
+        .getBroadcastHashJoinOutputPartitioningExpandLimit(this: SparkPlan)
     joinType match {
       case _: InnerLike if broadcastHashJoinOutputPartitioningExpandLimit > 0 =>
         streamedPlan.outputPartitioning match {
@@ -193,7 +199,10 @@ case class ColumnarBroadcastHashJoinExec(
   // Seq("a", "b", "c"), Seq("a", "b", "y"), Seq("a", "x", "c"), Seq("a", "x", "y").
   // The expanded expressions are returned as PartitioningCollection.
   private def expandOutputPartitioning(partitioning: HashPartitioning): PartitioningCollection = {
-    val maxNumCombinations = broadcastHashJoinOutputPartitioningExpandLimit
+    val maxNumCombinations =
+      SparkShimLoader
+        .getSparkShims
+        .getBroadcastHashJoinOutputPartitioningExpandLimit(this: SparkPlan)
     var currentNumCombinations = 0
 
     def generateExprCombinations(current: Seq[Expression],
@@ -640,4 +649,9 @@ case class ColumnarBroadcastHashJoinExec(
     }
 
   }
+
+  // For spark 3.2.
+  protected def withNewChildrenInternal(newLeft: SparkPlan, newRight: SparkPlan):
+  ColumnarBroadcastHashJoinExec =
+    copy(left = newLeft, right = newRight)
 }

native-sql-engine/core/src/main/scala/com/intel/oap/execution/ColumnarCoalesceExec.scala

@@ -69,6 +69,10 @@ case class ColumnarCoalesceExec(numPartitions: Int, child: SparkPlan) extends Un
       child.executeColumnar().coalesce(numPartitions, shuffle = false)
     }
   }
+
+  // For spark 3.2.
+  protected def withNewChildInternal(newChild: SparkPlan): ColumnarCoalesceExec =
+    copy(child = newChild)
 }
 
 object ColumnarCoalesceExec {

native-sql-engine/core/src/main/scala/com/intel/oap/execution/ColumnarExpandExec.scala

@@ -133,4 +133,8 @@ case class ColumnarExpandExec(
       new CloseableColumnBatchIterator(res)
     }
   }
+
+  // For spark 3.2.
+  protected def withNewChildInternal(newChild: SparkPlan): ColumnarExpandExec =
+    copy(child = newChild)
 }