Upgrade to Spark 3.3.2

Upgrade spark version to 3.3.2. Updates one test for a fix in 3.3.2.

Resolves #1479

Closes #1644

Signed-off-by: Allison Portis <[email protected]>
GitOrigin-RevId: c7d6751d28558b41a1d4c72c1301c367fd505339
(cherry picked from commit b16b1bc)

Author: allisonport-db

.github/workflows/test.yaml

@@ -41,7 +41,7 @@ jobs:
           pyenv install 3.7.4
           pyenv global system 3.7.4
           pipenv --python 3.7 install
-          pipenv run pip install pyspark==3.3.1
+          pipenv run pip install pyspark==3.3.2
           pipenv run pip install flake8==3.5.0 pypandoc==1.3.3
           pipenv run pip install importlib_metadata==3.10.0
           pipenv run pip install mypy==0.910

Dockerfile

@@ -18,7 +18,7 @@ FROM python:3.7.3-stretch
 
 RUN apt-get update && apt-get -y install openjdk-8-jdk
 
-RUN pip install pyspark==3.3.1
+RUN pip install pyspark==3.3.2
 
 RUN pip install mypy==0.910
 

build.sbt

@@ -17,7 +17,7 @@
 import java.nio.file.Files
 import TestParallelization._
 
-val sparkVersion = "3.3.1"
+val sparkVersion = "3.3.2"
 val scala212 = "2.12.15"
 val scala213 = "2.13.5"
 val default_scala_version = scala212

core/src/test/scala/org/apache/spark/sql/delta/DeltaTimeTravelSuite.scala

@@ -738,20 +738,18 @@ class DeltaTimeTravelSuite extends QueryTest
   }
 
 
-  // This is currently set up to fail once the bug is fixed (should be when upgrading to Spark 3.4)
-  // See https://github.com/delta-io/delta/issues/1479
-  test("SPARK-41154: Incorrect relation caching for queries with time travel spec") {
+  test("SPARK-41154: Correct relation caching for queries with time travel spec") {
     val tblName = "tab"
     withTable(tblName) {
       sql(s"CREATE TABLE $tblName USING DELTA AS SELECT 1 as c")
       sql(s"INSERT INTO $tblName SELECT 2 as c")
-      assert(
+      checkAnswer(
         sql(s"""
-              |SELECT * FROM $tblName VERSION AS OF '0'
-              |UNION ALL
-              |SELECT * FROM $tblName VERSION AS OF '1'
-              |""".stripMargin
-        ).collect() === Array(Row(1), Row(1))) // this should be Array(Row(1), Row(1), Row(2))
+          |SELECT * FROM $tblName VERSION AS OF '0'
+          |UNION ALL
+          |SELECT * FROM $tblName VERSION AS OF '1'
+          |""".stripMargin),
+        Row(1) :: Row(1) :: Row(2) :: Nil)
     }
   }
 }

examples/scala/build.sbt

@@ -35,8 +35,12 @@ def getMajorMinor(version: String): (Int, Int) = {
   }
 }
 val lookupSparkVersion: PartialFunction[(Int, Int), String] = {
-  // versions 2.1.0 and above
-  case (major, minor) if (major == 2  && minor >= 1) || major >= 3 => "3.3.0"
+  // versions 2.3.0 and above
+  case (major, minor) if (major == 2  && minor >= 3) || major >= 3 => "3.3.2"
+  // versions 2.2.x
+  case (major, minor) if major == 2  && minor == 2 => "3.3.1"
+  // versions 2.1.x
+  case (major, minor) if major == 2  && minor == 1 => "3.3.0"
   // versions 1.0.0 to 2.0.x
   case (major, minor) if major == 1 || (major == 2 && minor == 0) => "3.2.1"
   // versions 0.7.x to 0.8.x