Bump to Spark 3.2.2 (#170)

mwiewior · web-flow · commit 9fb185fe49dc · 2022-11-04T10:41:24.000+01:00
* Bumping to Spark 3.2.2

* Page update

* Fixing SPARK-35132

* Bumping netty to 4.1.68.Final
diff --git a/build.sbt b/build.sbt
@@ -3,7 +3,7 @@ import sbtassembly.AssemblyPlugin.autoImport.ShadeRule
 import scala.util.Properties
 
 name := """sequila"""
-val DEFAULT_SPARK_3_VERSION = "3.1.2"
+val DEFAULT_SPARK_3_VERSION = "3.2.2"
 lazy val sparkVersion = Properties.envOrElse("SPARK_VERSION", DEFAULT_SPARK_3_VERSION)
 
 version := s"${sys.env.getOrElse("VERSION", "0.1.0")}"
@@ -18,6 +18,16 @@ val DEFAULT_HADOOP_VERSION = "3.1.2"
 
 lazy val hadoopVersion = Properties.envOrElse("SPARK_HADOOP_VERSION", DEFAULT_HADOOP_VERSION)
 
+val nettyVersion = "4.1.68.Final"
+dependencyOverrides += "io.netty" % "netty-all" % nettyVersion
+dependencyOverrides += "io.netty" % "netty-buffer" % nettyVersion
+dependencyOverrides += "io.netty" % "netty-codec" % nettyVersion
+dependencyOverrides += "io.netty" % "netty-common" % nettyVersion
+dependencyOverrides += "io.netty" % "netty-handler" % nettyVersion
+dependencyOverrides += "io.netty" % "netty-resolver" % nettyVersion
+dependencyOverrides += "io.netty" % "netty-transport" % nettyVersion
+dependencyOverrides += "io.netty" % "netty-transport-native-epoll" % nettyVersion
+dependencyOverrides += "io.netty" % "netty-transport-native-unix-common" % nettyVersion
 dependencyOverrides += "com.google.guava" % "guava" % "15.0"
 dependencyOverrides += "org.apache.orc" % "orc-core" % "1.6.9"
 dependencyOverrides += "org.apache.logging.log4j" % "log4j-core" % "2.3"
@@ -30,7 +40,7 @@ libraryDependencies += "org.apache.spark" %% "spark-core" % sparkVersion
 libraryDependencies += "org.apache.spark" %% "spark-sql" % sparkVersion
 libraryDependencies += "com.github.mrpowers" %% "spark-fast-tests" % "0.21.3"
 libraryDependencies += "com.github.mrpowers" %% "spark-daria" % "0.38.2"
-libraryDependencies += "com.holdenkarau" %% "spark-testing-base" % "3.1.2_1.1.0" % "test" excludeAll ExclusionRule(organization = "javax.servlet") excludeAll (ExclusionRule("org.apache.hadoop"))
+libraryDependencies += "com.holdenkarau" %% "spark-testing-base" % "3.2.0_1.2.0" % "test" excludeAll ExclusionRule(organization = "javax.servlet") excludeAll (ExclusionRule("org.apache.hadoop"))
 libraryDependencies += "org.bdgenomics.adam" %% "adam-core-spark3" % "0.36.0" excludeAll (ExclusionRule("org.seqdoop"))
 libraryDependencies += "org.bdgenomics.adam" %% "adam-apis-spark3" % "0.36.0" excludeAll (ExclusionRule("org.seqdoop"))
 libraryDependencies += "org.bdgenomics.adam" %% "adam-cli-spark3" % "0.36.0" excludeAll (ExclusionRule("org.seqdoop"))
@@ -156,7 +166,8 @@ publishTo := {
   if (!version.value.toLowerCase.contains("snapshot")) {
     sonatypePublishToBundle.value
   } else {
-    val nexus = "http://zsibio.ii.pw.edu.pl/nexus/repository/"
+    val nexus = "https://zsibio.ii.pw.edu.pl/nexus/repository/"
     Some("snapshots" at nexus + "maven-snapshots")
   }
-}
+}
+ThisBuild / useCoursier := true
diff --git a/page/content/en/_index.html b/page/content/en/_index.html
@@ -82,14 +82,14 @@ <h2 class="text-center">Getting started</h2>
 	<div class="text-left">
 	{{< highlight bash>}}
 # ensure you have Apache Spark and GraalVM installed
-sdk install spark 3.1.2
+sdk install spark 3.2.2
 sdk install java 21.3.0.r11-grl
 # set Apache Spark and JDK using sdkman
-sdk use spark 3.1.2
+sdk use spark 3.2.2
 # use GraalVM for best performance
 sdk use java 21.3.0.r11-grl
 # in case you prefer to use a Python interface
-pip install pysequila==0.3.3
+pip install pysequila==0.4.0
 # download sample data
 mkdir -p data
 #BAM
@@ -130,7 +130,7 @@ <h2 class="text-center">Getting started</h2>
 		{{< highlight bash>}}
 pyspark --master local[1] \
 		--driver-memory 4g \
-		--packages org.biodatageeks:sequila_2.12:1.0.0
+		--packages org.biodatageeks:sequila_2.12:1.1.0
 		{{< / highlight >}}
 		{{< highlight python>}}
 from pysequila import SequilaSession
@@ -187,7 +187,7 @@ <h2 class="text-center">Getting started</h2>
 		{{< highlight bash>}}
 pyspark --master local[1] \
 --driver-memory 4g \
---packages org.biodatageeks:sequila_2.12:1.0.0
+--packages org.biodatageeks:sequila_2.12:1.1.0
 		{{< / highlight >}}
 {{< highlight python>}}
 from pysequila import SequilaSession
@@ -256,7 +256,7 @@ <h2 class="text-center">Getting started</h2>
 		{{< highlight bash>}}
 pyspark --master local[1] \
 --driver-memory 4g \
---packages org.biodatageeks:sequila_2.12:1.0.0
+--packages org.biodatageeks:sequila_2.12:1.1.0
 		{{< / highlight >}}
 		{{< highlight python>}}
 targets_df = ss.read\
diff --git a/project/plugins.sbt b/project/plugins.sbt
@@ -9,3 +9,5 @@ addSbtPlugin("com.julianpeeters" % "sbt-avrohugger" % "2.0.0-RC19")
 addSbtPlugin("org.xerial.sbt" % "sbt-sonatype" % "3.9.4")
 addSbtPlugin("com.jsuereth" % "sbt-pgp" % "2.0.1")
 addSbtPlugin("org.scalastyle" %% "scalastyle-sbt-plugin" % "1.0.0")
+
+addSbtPlugin("net.virtual-void" % "sbt-dependency-graph" % "0.10.0-RC1")
diff --git a/src/main/scala/org/biodatageeks/sequila/datasources/BAM/SequilaDataSourceStrategy.scala b/src/main/scala/org/biodatageeks/sequila/datasources/BAM/SequilaDataSourceStrategy.scala
@@ -102,6 +102,7 @@ case class SequilaDataSourceStrategy(spark: SparkSession) extends Strategy
         l.output.toStructType,
         Set.empty,
         Set.empty,
+        None,
         toCatalystRDD(l, baseRelation.buildScan()),
         baseRelation,
         None) :: Nil
@@ -215,6 +216,7 @@ case class SequilaDataSourceStrategy(spark: SparkSession) extends Strategy
         projects.map(_.toAttribute).toStructType,
         Set.empty,
         Set.empty,
+        None,
         scanBuilder(requestedColumns, candidatePredicates, pushedFilters),
         relation.relation,
         relation.catalogTable.map(_.identifier))
@@ -229,6 +231,7 @@ case class SequilaDataSourceStrategy(spark: SparkSession) extends Strategy
         requestedColumns.toStructType,
         Set.empty,
         Set.empty,
+        None,
         scanBuilder(requestedColumns, candidatePredicates, pushedFilters),
         relation.relation,
         relation.catalogTable.map(_.identifier))
diff --git a/src/main/scala/org/biodatageeks/sequila/pileup/PileupStrategy.scala b/src/main/scala/org/biodatageeks/sequila/pileup/PileupStrategy.scala
@@ -119,4 +119,5 @@ case class PileupPlan [T<:BDGAlignInputFormat](plan:LogicalPlan, spark:SparkSess
     conf
   }
 
+  override protected def withNewChildrenInternal(newChildren: IndexedSeq[SparkPlan]): SparkPlan = this
 }
diff --git a/src/main/scala/org/biodatageeks/sequila/rangejoins/methods/IntervalTree/IntervalTreeJoinOptim.scala b/src/main/scala/org/biodatageeks/sequila/rangejoins/methods/IntervalTree/IntervalTreeJoinOptim.scala
@@ -100,4 +100,6 @@ case class IntervalTreeJoinOptim(left: SparkPlan,
     }
 
   }
+
+  override protected def withNewChildrenInternal(newLeft: SparkPlan, newRight: SparkPlan): SparkPlan =  copy(left = newLeft, right = newRight)
 }
diff --git a/src/main/scala/org/biodatageeks/sequila/rangejoins/methods/IntervalTree/IntervalTreeJoinOptimChromosome.scala b/src/main/scala/org/biodatageeks/sequila/rangejoins/methods/IntervalTree/IntervalTreeJoinOptimChromosome.scala
@@ -115,4 +115,6 @@ case class IntervalTreeJoinOptimChromosome(left: SparkPlan,
       .stats
       .sizeInBytes != Long.MaxValue)
   }
+
+  override protected def withNewChildrenInternal(newLeft: SparkPlan, newRight: SparkPlan): SparkPlan  =  copy(left = newLeft, right = newRight)
 }
diff --git a/src/main/scala/org/biodatageeks/sequila/rangejoins/methods/genApp/IntervalTreeJoin.scala b/src/main/scala/org/biodatageeks/sequila/rangejoins/methods/genApp/IntervalTreeJoin.scala
@@ -62,4 +62,6 @@ case class IntervalTreeJoin(left: SparkPlan,
     }
 
   }
+
+  override protected def withNewChildrenInternal(newLeft: SparkPlan, newRight: SparkPlan): SparkPlan =  copy(left = newLeft, right = newRight)
 }
diff --git a/src/main/scala/org/biodatageeks/sequila/rangejoins/methods/genApp/IntervalTreeJoinChromosome.scala b/src/main/scala/org/biodatageeks/sequila/rangejoins/methods/genApp/IntervalTreeJoinChromosome.scala
@@ -65,4 +65,6 @@ IntervalTreeJoinChromosome(left: SparkPlan,
     }
 
   }
+
+  override protected def withNewChildrenInternal(newLeft: SparkPlan, newRight: SparkPlan): SparkPlan =  copy(left = newLeft, right = newRight)
 }
diff --git a/src/main/scala/org/biodatageeks/sequila/utvf/GenomicIntervalStrategy.scala b/src/main/scala/org/biodatageeks/sequila/utvf/GenomicIntervalStrategy.scala
@@ -31,4 +31,6 @@ case class GenomicIntervalPlan(plan: LogicalPlan, spark: SparkSession,interval:G
       )
   }
   def children: Seq[SparkPlan] = Nil
+
+  override protected def withNewChildrenInternal(newChildren: IndexedSeq[SparkPlan]): SparkPlan = this
 }
diff --git a/src/main/scala/org/biodatageeks/sequila/utvf/ResolveTableValuedFunctionsSeq.scala b/src/main/scala/org/biodatageeks/sequila/utvf/ResolveTableValuedFunctionsSeq.scala
@@ -47,7 +47,7 @@ object ResolveTableValuedFunctionsSeq extends Rule[LogicalPlan] {
     def implicitCast(values: Seq[Expression]): Option[Seq[Expression]] = {
       if (args.length == values.length) {
         val casted = values.zip(args).map { case (value, (_, expectedType)) =>
-          TypeCoercion.ImplicitTypeCasts.implicitCast(value, expectedType)
+          TypeCoercion.implicitCast(value, expectedType)
         }
         if (casted.forall(_.isDefined)) {
           return Some(casted.map(_.get))
@@ -130,7 +130,7 @@ object ResolveTableValuedFunctionsSeq extends Rule[LogicalPlan] {
 
   override def apply(plan: LogicalPlan): LogicalPlan = plan transformUp {
     case u: UnresolvedTableValuedFunction if u.functionArgs.forall(_.resolved) =>
-      val resolvedFunc = builtinFunctions.get(u.functionName.toLowerCase(Locale.ROOT)) match {
+      val resolvedFunc = builtinFunctions.get(u.name.funcName.toLowerCase(Locale.ROOT)) match {
         case Some(tvf) =>
           val resolved = tvf.flatMap { case (argList, resolver) =>
             argList.implicitCast(u.functionArgs) match {
@@ -143,12 +143,12 @@ object ResolveTableValuedFunctionsSeq extends Rule[LogicalPlan] {
           resolved.headOption.getOrElse {
             val argTypes = u.functionArgs.map(_.dataType.typeName).mkString(", ")
             u.failAnalysis(
-              s"""error: table-valued function ${u.functionName} with alternatives:
+              s"""error: table-valued function ${u.name.funcName} with alternatives:
                  |${tvf.keys.map(_.toString).toSeq.sorted.map(x => s" ($x)").mkString("\n")}
                  |cannot be applied to: (${argTypes})""".stripMargin)
           }
         case _ =>
-          u.failAnalysis(s"could not resolve `${u.functionName}` to a table-valued function")
+          u.failAnalysis(s"could not resolve `${u.name.funcName}` to a table-valued function")
       }
 
       // If alias names assigned, add `Project` with the aliases
@@ -157,7 +157,7 @@ object ResolveTableValuedFunctionsSeq extends Rule[LogicalPlan] {
         // Checks if the number of the aliases is equal to expected one
         if (u.output.size != outputAttrs.size) {
           u.failAnalysis(s"Number of given aliases does not match number of output columns. " +
-            s"Function name: ${u.functionName}; number of aliases: " +
+            s"Function name: ${u.name.funcName}; number of aliases: " +
             s"${u.output.size}; number of output columns: ${outputAttrs.size}.")
         }
         val aliases = outputAttrs.zip(u.output).map {
diff --git a/src/main/scala/org/biodatageeks/sequila/utvf/SeQuiLaAnalyzer.scala b/src/main/scala/org/biodatageeks/sequila/utvf/SeQuiLaAnalyzer.scala
@@ -24,7 +24,7 @@ class SeQuiLaAnalyzer(session: SparkSession) extends
     new ResolveSQLOnFile(session) +:
     new FallBackFileSourceV2(session) +:
       new ResolveSessionCatalog(
-        catalogManager, catalog.isTempView, catalog.isTempFunction) +:
+        catalogManager) +:
     ResolveEncodersInScalaAgg+: session.extensions.buildResolutionRules(session)
 
 
@@ -72,9 +72,11 @@ class SeQuiLaAnalyzer(session: SparkSession) extends
         ResolveRelations ::
         ResolveTables ::
         ResolvePartitionSpec ::
+        ResolveAlterTableCommands ::
         AddMetadataColumns ::
+        DeduplicateRelations ::
         ResolveReferences ::
-        ResolveCreateNamedStruct ::
+        ResolveExpressionsWithNamePlaceholders ::
         ResolveDeserializer ::
         ResolveNewInstance ::
         ResolveUpCast ::
@@ -97,21 +99,23 @@ class SeQuiLaAnalyzer(session: SparkSession) extends
         GlobalAggregates ::
         ResolveAggregateFunctions ::
         TimeWindowing ::
+        SessionWindowing ::
         ResolveInlineTables ::
-        ResolveHigherOrderFunctions(v1SessionCatalog) ::
+        ResolveHigherOrderFunctions(catalogManager) ::
         ResolveLambdaVariables ::
         ResolveTimeZone ::
         ResolveRandomSeed ::
         ResolveBinaryArithmetic ::
         ResolveUnion ::
-        TypeCoercion.typeCoercionRules ++
+        typeCoercionRules ++
+          Seq(ResolveWithCTE) ++
           extendedResolutionRules : _*),
+    Batch("Remove TempResolvedColumn", Once, RemoveTempResolvedColumn),
     Batch("Apply Char Padding", Once,
       ApplyCharTypePadding),
     Batch("Post-Hoc Resolution", Once,
-      Seq(ResolveNoopDropTable) ++
+      Seq(ResolveCommandsWithIfExists) ++
         postHocResolutionRules: _*),
-    Batch("Normalize Alter Table", Once, ResolveAlterTableChanges),
     Batch("Remove Unresolved Hints", Once,
       new ResolveHints.RemoveAllHints),
     Batch("Nondeterministic", Once,
@@ -124,6 +128,8 @@ class SeQuiLaAnalyzer(session: SparkSession) extends
     Batch("Subquery", Once,
       UpdateOuterReferences),
     Batch("Cleanup", fixedPoint,
-      CleanupAliases)
+      CleanupAliases),
+    Batch("HandleAnalysisOnlyCommand", Once,
+      HandleAnalysisOnlyCommand)
   )
 }
diff --git a/src/main/scala/org/biodatageeks/sequila/utvf/SequilaSession.scala b/src/main/scala/org/biodatageeks/sequila/utvf/SequilaSession.scala
@@ -5,7 +5,7 @@ import htsjdk.samtools.ValidationStringency
 import org.apache.log4j.Logger
 import org.apache.spark.sql.catalyst.analysis.{Analyzer, SeQuiLaAnalyzer}
 import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
-import org.apache.spark.sql.execution.QueryExecution
+import org.apache.spark.sql.execution.{CommandExecutionMode, QueryExecution}
 import org.apache.spark.sql.execution.datasources.SequilaDataSourceStrategy
 import org.apache.spark.sql.functions.{lit, typedLit}
 import org.apache.spark.sql.internal.SessionState
@@ -120,6 +120,7 @@ case class SequilaSessionState(sparkSession: SparkSession, customAnalyzer: Analy
     sparkSession.sessionState.conf,
     sparkSession.sessionState.experimentalMethods,
     sparkSession.sessionState.functionRegistry,
+    sparkSession.sessionState.tableFunctionRegistry,
     sparkSession.sessionState.udfRegistration,
     () => sparkSession.sessionState.catalog,
     sparkSession.sessionState.sqlParser,
@@ -129,7 +130,7 @@ case class SequilaSessionState(sparkSession: SparkSession, customAnalyzer: Analy
     () => sparkSession.sessionState.streamingQueryManager,
     sparkSession.sessionState.listenerManager,
     () =>sparkSession.sessionState.resourceLoader,
-    executePlan,
+    sparkSession.sessionState.executePlan,
     (sparkSession:SparkSession,sessionState: SessionState) => sessionState.clone(sparkSession),
     sparkSession.sessionState.columnarRules,
     sparkSession.sessionState.queryStagePrepRules

Original file line number	Diff line number	Diff line change
`@@ -119,4 +119,5 @@ case class PileupPlan [T<:BDGAlignInputFormat](plan:LogicalPlan, spark:SparkSess`
`119`	`119`	`conf`
`120`	`120`	`}`
`121`	`121`
	`122`	`+ override protected def withNewChildrenInternal(newChildren: IndexedSeq[SparkPlan]): SparkPlan = this`
`122`	`123`	`}`
Original file line number	Diff line number	Diff line change
`@@ -100,4 +100,6 @@ case class IntervalTreeJoinOptim(left: SparkPlan,`
`100`	`100`	`}`
`101`	`101`
`102`	`102`	`}`
	`103`	`+`
	`104`	`+ override protected def withNewChildrenInternal(newLeft: SparkPlan, newRight: SparkPlan): SparkPlan = copy(left = newLeft, right = newRight)`
`103`	`105`	`}`
Original file line number	Diff line number	Diff line change
`@@ -115,4 +115,6 @@ case class IntervalTreeJoinOptimChromosome(left: SparkPlan,`
`115`	`115`	`.stats`
`116`	`116`	`.sizeInBytes != Long.MaxValue)`
`117`	`117`	`}`
	`118`	`+`
	`119`	`+ override protected def withNewChildrenInternal(newLeft: SparkPlan, newRight: SparkPlan): SparkPlan = copy(left = newLeft, right = newRight)`
`118`	`120`	`}`
Original file line number	Diff line number	Diff line change
`@@ -62,4 +62,6 @@ case class IntervalTreeJoin(left: SparkPlan,`
`62`	`62`	`}`
`63`	`63`
`64`	`64`	`}`
	`65`	`+`
	`66`	`+ override protected def withNewChildrenInternal(newLeft: SparkPlan, newRight: SparkPlan): SparkPlan = copy(left = newLeft, right = newRight)`
`65`	`67`	`}`
Original file line number	Diff line number	Diff line change
`@@ -65,4 +65,6 @@ IntervalTreeJoinChromosome(left: SparkPlan,`
`65`	`65`	`}`
`66`	`66`
`67`	`67`	`}`
	`68`	`+`
	`69`	`+ override protected def withNewChildrenInternal(newLeft: SparkPlan, newRight: SparkPlan): SparkPlan = copy(left = newLeft, right = newRight)`
`68`	`70`	`}`
Original file line number	Diff line number	Diff line change
`@@ -31,4 +31,6 @@ case class GenomicIntervalPlan(plan: LogicalPlan, spark: SparkSession,interval:G`
`31`	`31`	`)`
`32`	`32`	`}`
`33`	`33`	`def children: Seq[SparkPlan] = Nil`
	`34`	`+`
	`35`	`+ override protected def withNewChildrenInternal(newChildren: IndexedSeq[SparkPlan]): SparkPlan = this`
`34`	`36`	`}`