Fixes for spark 2.3.2 upgrade

build.sbt

@@ -8,7 +8,7 @@ organization := "org.biodatageeks"
 
 scalaVersion := "2.11.8"
 
-val DEFAULT_SPARK_2_VERSION = "2.2.2"
+val DEFAULT_SPARK_2_VERSION = "2.3.2"
 val DEFAULT_HADOOP_VERSION = "2.6.5"
 
 
@@ -22,7 +22,7 @@ libraryDependencies +=  "org.apache.spark" % "spark-sql_2.11" % sparkVersion
 libraryDependencies +=  "org.apache.spark" %% "spark-hive" % sparkVersion
 libraryDependencies +=  "org.apache.spark" %% "spark-hive-thriftserver" % sparkVersion
 
-libraryDependencies += "com.holdenkarau" % "spark-testing-base_2.11" % "2.2.0_0.7.4" % "test" excludeAll ExclusionRule(organization = "javax.servlet") excludeAll (ExclusionRule("org.apache.hadoop"))
+libraryDependencies += "com.holdenkarau" % "spark-testing-base_2.11" % "2.3.1_0.10.0" % "test" excludeAll ExclusionRule(organization = "javax.servlet") excludeAll (ExclusionRule("org.apache.hadoop"))
 
 libraryDependencies += "org.apache.spark" %% "spark-hive"       % "2.0.0" % "test"
 

src/main/scala/org/biodatageeks/catalyst/utvf/GenomicInterval.scala

@@ -15,7 +15,7 @@ case class GenomicInterval(
 
   override def newInstance(): GenomicInterval = copy(output = output.map(_.newInstance()))
 
-  override def computeStats(conf: SQLConf): Statistics = {
+  def computeStats(conf: SQLConf): Statistics = {
     val sizeInBytes = IntegerType.defaultSize * 2 //FIXME: Add contigName size
     Statistics( sizeInBytes = sizeInBytes )
   }

src/main/scala/org/biodatageeks/catalyst/utvf/ResolveTableValuedFunctionsSeq.scala

@@ -183,7 +183,7 @@ case class BDGCoverage(tableName:String, sampleId:String, result: String, target
 
   override def newInstance(): BDGCoverage = copy(output = output.map(_.newInstance()))
 
-  override def computeStats(conf: SQLConf): Statistics = {
+  def computeStats(conf: SQLConf): Statistics = {
     val sizeInBytes = LongType.defaultSize * numElements
     Statistics( sizeInBytes = sizeInBytes )
   }

src/main/scala/org/biodatageeks/catalyst/utvf/SeQuiLaAnalyzer.scala

@@ -33,34 +33,36 @@ class SeQuiLaAnalyzer(catalog: SessionCatalog, conf: SQLConf) extends Analyzer(c
       EliminateUnions,
       new SubstituteUnresolvedOrdinals(conf)),
     Batch("Resolution", fixedPoint,
-      ResolveTableValuedFunctionsSeq,
-        ResolveRelations,
-        ResolveReferences,
-        ResolveCreateNamedStruct,
-        ResolveDeserializer,
-        ResolveNewInstance,
-        ResolveUpCast ,
-        ResolveGroupingAnalytics,
-        ResolvePivot,
-        ResolveOrdinalInOrderByAndGroupBy,
-        ResolveAggAliasInGroupBy,
-        ResolveMissingReferences,
-        ExtractGenerator,
-        ResolveGenerate,
-        ResolveFunctions,
-        ResolveAliases,
-        ResolveSubquery,
-        //ResolveSubqueryColumnAliases ::
-        ResolveWindowOrder,
-        ResolveWindowFrame,
-        ResolveNaturalAndUsingJoin,
-        ExtractWindowExpressions,
-        GlobalAggregates,
-        ResolveAggregateFunctions,
-        TimeWindowing,
-        ResolveInlineTables(conf),
-        ResolveTimeZone(conf),
-        TypeCoercion.typeCoercionRules(1)),
+      ResolveTableValuedFunctionsSeq ::
+      ResolveRelations ::
+        ResolveReferences ::
+        ResolveCreateNamedStruct ::
+        ResolveDeserializer ::
+        ResolveNewInstance ::
+        ResolveUpCast ::
+        ResolveGroupingAnalytics ::
+        ResolvePivot ::
+        ResolveOrdinalInOrderByAndGroupBy ::
+        ResolveAggAliasInGroupBy ::
+        ResolveMissingReferences ::
+        ExtractGenerator ::
+        ResolveGenerate ::
+        ResolveFunctions ::
+        ResolveAliases ::
+        ResolveSubquery ::
+        ResolveSubqueryColumnAliases ::
+        ResolveWindowOrder ::
+        ResolveWindowFrame ::
+        ResolveNaturalAndUsingJoin ::
+
+        ExtractWindowExpressions ::
+        GlobalAggregates ::
+        ResolveAggregateFunctions ::
+        TimeWindowing ::
+        ResolveInlineTables(conf) ::
+        ResolveTimeZone(conf) ::
+        TypeCoercion.typeCoercionRules(conf) ++
+          extendedResolutionRules : _*),
     Batch("Post-Hoc Resolution", Once, postHocResolutionRules: _*),
     Batch("SeQuiLa", Once,sequilaOptmazationRules: _*), //SeQuilaOptimization rules
     Batch("View", Once,

src/main/scala/org/biodatageeks/catalyst/utvf/SequilaSession.scala

@@ -8,7 +8,7 @@ import org.apache.spark.sql.catalyst.catalog.SessionCatalog
 import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
 import org.apache.spark.sql.catalyst.rules.Rule
 import org.apache.spark.sql.execution.QueryExecution
-import org.apache.spark.sql.execution.command.{BAMCTASOptimizationRule, BAMIASOptimizationRule}
+//import org.apache.spark.sql.execution.command.{BAMCTASOptimizationRule, BAMIASOptimizationRule}
 import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.internal.{SQLConf, SessionState}
 import org.biodatageeks.preprocessing.coverage.CoverageStrategy
@@ -26,10 +26,10 @@ case class SequilaSession(sparkSession: SparkSession) extends SparkSession(spark
   @transient override lazy val sessionState = SequilaSessionState(sparkSession,sequilaAnalyzer,executePlan)
 
   //new rules
-  sequilaAnalyzer.sequilaOptmazationRules = Seq(
-    new BAMCTASOptimizationRule(sparkSession),
-    new BAMIASOptimizationRule(sparkSession)
-  )
+//  sequilaAnalyzer.sequilaOptmazationRules = Seq(
+//    new BAMCTASOptimizationRule(sparkSession),
+//    new BAMIASOptimizationRule(sparkSession)
+//  )
 
 
 }
@@ -40,14 +40,14 @@ case class SequilaSessionState(sparkSession: SparkSession, customAnalyzer: Analy
     sparkSession.sessionState.experimentalMethods,
     sparkSession.sessionState.functionRegistry,
     sparkSession.sessionState.udfRegistration,
-    sparkSession.sessionState.catalog,
+    () => sparkSession.sessionState.catalog,
     sparkSession.sessionState.sqlParser,
-    customAnalyzer,
-    sparkSession.sessionState.optimizer,
+    () =>customAnalyzer,
+    () =>sparkSession.sessionState.optimizer,
     sparkSession.sessionState.planner,
     sparkSession.sessionState.streamingQueryManager,
     sparkSession.sessionState.listenerManager,
-    sparkSession.sessionState.resourceLoader,
+    () =>sparkSession.sessionState.resourceLoader,
     executePlan,
     (sparkSession:SparkSession,sessionState: SessionState) => sessionState.clone(sparkSession)){
 }

src/main/scala/org/biodatageeks/datasources/BAM/SequilaDataSourceStrategy.scala

@@ -32,7 +32,7 @@ case class SequilaDataSourceStrategy(spark: SparkSession) extends Strategy with
     //optimized strategy for queries like SELECT (distinct )sampleId FROM  BDGAlignmentRelation
     case a: Aggregate if a.schema.length == 1 && a.schema.head.name == BDGInternalParams.SAMPLE_COLUMN_NAME => {
       a.child match {
-        case PhysicalOperation(projects, filters, l@LogicalRelation(t: PrunedFilteredScan, _, _)) => {
+        case PhysicalOperation(projects, filters, l@LogicalRelation(t: PrunedFilteredScan, _, _,false)) => {
           l.catalogTable.get.provider match {
             case Some(BDGInputDataType.BAMInputDataType) => {
               pruneFilterProject(
@@ -48,36 +48,37 @@ case class SequilaDataSourceStrategy(spark: SparkSession) extends Strategy with
       }
     }
 
-    case PhysicalOperation(projects, filters, l @ LogicalRelation(t: CatalystScan, _, _)) =>
+    case PhysicalOperation(projects, filters, l @ LogicalRelation(t: CatalystScan, _, _,false)) =>
       pruneFilterProjectRaw(
         l,
         projects,
         filters,
         (requestedColumns, allPredicates, _) =>
           toCatalystRDD(l, requestedColumns, t.buildScan(requestedColumns, allPredicates))) :: Nil
 
-    case PhysicalOperation(projects, filters, l @ LogicalRelation(t: PrunedFilteredScan, _, _)) => {
+    case PhysicalOperation(projects, filters, l @ LogicalRelation(t: PrunedFilteredScan, _, _,false)) => {
       pruneFilterProject(
         l,
         projects,
         filters,
         (a, f) => toCatalystRDD(l, a, t.buildScan(a.map(_.name).toArray, f))) :: Nil
     }
 
-    case PhysicalOperation(projects, filters, l @ LogicalRelation(t: PrunedScan, _, _)) =>
+    case PhysicalOperation(projects, filters, l @ LogicalRelation(t: PrunedScan, _, _,false)) =>
       pruneFilterProject(
         l,
         projects,
         filters,
         (a, _) => toCatalystRDD(l, a, t.buildScan(a.map(_.name).toArray))) :: Nil
 
-    case l @ LogicalRelation(baseRelation: TableScan, _, _) =>
+    case l @ LogicalRelation(baseRelation: TableScan, _, _,false) =>
       RowDataSourceScanExec(
         l.output,
+        l.output.indices,
+        Set.empty,
+        Set.empty,
         toCatalystRDD(l, baseRelation.buildScan()),
         baseRelation,
-        UnknownPartitioning(0),
-        Map.empty,
         None) :: Nil
 
     case _ => Nil
@@ -186,8 +187,11 @@ case class SequilaDataSourceStrategy(spark: SparkSession) extends Strategy with
 
       val scan = RowDataSourceScanExec(
         projects.map(_.toAttribute),
+        projects.map(_.toAttribute).indices,
+        Set.empty,
+        Set.empty,
         scanBuilder(requestedColumns, candidatePredicates, pushedFilters),
-        relation.relation, UnknownPartitioning(0), metadata,
+        relation.relation,
         relation.catalogTable.map(_.identifier))
       filterCondition.map(execution.FilterExec(_, scan)).getOrElse(scan)
     } else {
@@ -197,8 +201,11 @@ case class SequilaDataSourceStrategy(spark: SparkSession) extends Strategy with
 
       val scan = RowDataSourceScanExec(
         requestedColumns,
+        requestedColumns.indices,
+        Set.empty,
+        Set.empty,
         scanBuilder(requestedColumns, candidatePredicates, pushedFilters),
-        relation.relation, UnknownPartitioning(0), metadata,
+        relation.relation,
         relation.catalogTable.map(_.identifier))
       execution.ProjectExec(
         projects, filterCondition.map(execution.FilterExec(_, scan)).getOrElse(scan))