[GH-1918] Spark 4 support

.github/workflows/docs.yml

@@ -48,7 +48,7 @@ jobs:
       - name: Compile JavaDoc
         run: mvn -q clean install -DskipTests && mkdir -p docs/api/javadoc/spark && cp -r spark/common/target/apidocs/* docs/api/javadoc/spark/
       - name: Compile ScalaDoc
-        run: mvn scala:doc -pl !common,!snowflake,!flink && mkdir -p docs/api/scaladoc/spark && cp -r spark/common/target/site/scaladocs/* docs/api/scaladoc/spark
+        run: mvn generate-sources scala:doc -pl !common,!snowflake,!flink && mkdir -p docs/api/scaladoc/spark && cp -r spark/common/target/site/scaladocs/* docs/api/scaladoc/spark
       - uses: actions/setup-python@v5
         with:
           python-version: 3.x

.github/workflows/java.yml

@@ -62,6 +62,9 @@ jobs:
       fail-fast: true
       matrix:
         include:
+          - spark: 4.0.0
+            scala: 2.13.8
+            jdk: '17'
           - spark: 3.5.0
             scala: 2.13.8
             jdk: '11'
@@ -105,6 +108,11 @@ jobs:
               export SPARK_HOME=$(python -c "import pyspark; print(pyspark.__path__[0])")
           fi
 
+          if [ "${SPARK_VERSION}" == "4.0.0" ]; then
+              pip install pyspark==4.0.0 pandas shapely apache-sedona pyarrow
+              export SPARK_HOME=$(python -c "import pyspark; print(pyspark.__path__[0])")
+          fi
+
           mvn -q clean install -Dspark=${SPARK_COMPAT_VERSION} -Dscala=${SCALA_VERSION:0:4} -Dspark.version=${SPARK_VERSION} ${SKIP_TESTS}
       - run: mkdir staging
       - run: cp spark-shaded/target/sedona-*.jar staging

.github/workflows/python.yml

@@ -60,33 +60,46 @@ jobs:
     strategy:
       matrix:
         include:
+          - spark: '4.0.0'
+            scala: '2.13.8'
+            java: '17'
+            python: '3.10'
           - spark: '3.5.0'
             scala: '2.12.8'
+            java: '11'
             python: '3.10'
             shapely: '1'
           - spark: '3.5.0'
             scala: '2.12.8'
+            java: '11'
             python: '3.10'
           - spark: '3.5.0'
             scala: '2.12.8'
+            java: '11'
             python: '3.9'
           - spark: '3.5.0'
             scala: '2.12.8'
+            java: '11'
             python: '3.8'
           - spark: '3.4.0'
             scala: '2.12.8'
+            java: '11'
             python: '3.10'
           - spark: '3.4.0'
             scala: '2.12.8'
+            java: '11'
             python: '3.9'
           - spark: '3.4.0'
             scala: '2.12.8'
+            java: '11'
             python: '3.8'
           - spark: '3.4.0'
             scala: '2.12.8'
+            java: '11'
             python: '3.8'
           - spark: '3.4.0'
             scala: '2.12.8'
+            java: '11'
             python: '3.8'
             shapely: '1'
 
@@ -97,7 +110,7 @@ jobs:
       - uses: actions/setup-java@v4
         with:
           distribution: 'zulu'
-          java-version: '11'
+          java-version: '${{ matrix.java }}'
       - uses: actions/setup-python@v5
         with:
           python-version: ${{ matrix.python }}
@@ -123,12 +136,17 @@ jobs:
           SPARK_VERSION: ${{ matrix.spark }}
           PYTHON_VERSION: ${{ matrix.python }}
           SHAPELY_VERSION: ${{ matrix.shapely }}
+          PANDAS_VERSION: ${{ matrix.pandas }}
         run: |
           cd python
           if [ "${SHAPELY_VERSION}" == "1" ]; then
             echo "Patching Pipfile to use Shapely 1.x"
             sed -i 's/^shapely.*$/shapely="<2.0.0"/g' Pipfile
           fi
+          if [ "${PANDAS_VERSION}" == "1" ]; then
+            echo "Patching Pipfile to use Pandas 1.x"
+            sed -i 's/^pandas.*$/pandas="<2.0.0"/g' Pipfile
+          fi
           export PIPENV_CUSTOM_VENV_NAME=python-${PYTHON_VERSION}
           pipenv --python ${PYTHON_VERSION}
           pipenv install pyspark==${SPARK_VERSION}

pom.xml

@@ -76,14 +76,14 @@
         <spatial4j.version>0.8</spatial4j.version>
 
         <jt-jiffle.version>1.1.24</jt-jiffle.version>
-        <antlr-runtime.version>4.9.3</antlr-runtime.version>
         <janino-version>3.1.9</janino-version>
 
         <!-- Actual scala, spark and log4j version will be set by activated profiles.
              Setting a default value helps IDE:s that can't make sense of profiles. -->
         <scala.compat.version>2.12</scala.compat.version>
         <spark.version>3.4.0</spark.version>
         <spark.compat.version>3.4</spark.compat.version>
+        <spark.major.version>3</spark.major.version>
         <log4j.version>2.19.0</log4j.version>
         <graphframe.version>0.8.3-spark3.4</graphframe.version>
 
@@ -736,6 +736,28 @@
                 <skip.deploy.common.modules>true</skip.deploy.common.modules>
             </properties>
         </profile>
+        <profile>
+            <id>sedona-spark-4.0</id>
+            <activation>
+                <property>
+                    <name>spark</name>
+                    <value>4.0</value>
+                </property>
+            </activation>
+            <properties>
+                <spark.version>4.0.0</spark.version>
+                <spark.compat.version>4.0</spark.compat.version>
+                <spark.major.version>4</spark.major.version>
+                <hadoop.version>3.4.1</hadoop.version>
+                <log4j.version>2.24.3</log4j.version>
+                <slf4j.version>2.0.16</slf4j.version>
+                <graphframe.version>0.8.3-spark3.5</graphframe.version>
+                <scala.version>2.13.12</scala.version>
+                <scala.compat.version>2.13</scala.compat.version>
+                <!-- Skip deploying parent module. it will be deployed with sedona-spark-3.3 -->
+                <skip.deploy.common.modules>true</skip.deploy.common.modules>
+            </properties>
+        </profile>
         <profile>
             <id>scala2.13</id>
             <activation>

python/Pipfile

@@ -17,7 +17,7 @@ matplotlib = "*" # implicit dependency of esda
 scipy = "<=1.10.0" # prevent incompatibility with pysal 4.7.0, which is what is resolved to when shapely >2 is specified
 
 [packages]
-pandas="<=1.5.3"
+pandas=">=2.0.0"
 numpy="<2"
 geopandas="*"
 # https://stackoverflow.com/questions/78949093/how-to-resolve-attributeerror-module-fiona-has-no-attribute-path

python/sedona/spark/core/jvm/config.py

@@ -23,7 +23,6 @@
 from re import findall
 from typing import Optional, Tuple
 
-from py4j.protocol import Py4JJavaError
 from pyspark.sql import SparkSession
 
 from sedona.spark.utils.decorators import classproperty
@@ -189,7 +188,7 @@ def get_spark_java_config(
 
         try:
             used_jar_files = java_spark_conf.get(value)
-        except Py4JJavaError:
+        except Exception:
             error_message = f"Didn't find the value of {value} from SparkConf"
             logging.info(error_message)
 

python/tests/sql/test_dataframe_api.py

@@ -19,6 +19,7 @@
 import concurrent.futures
 from typing import Callable, Tuple
 
+import pyspark
 import pytest
 from pyspark.sql import Row
 from pyspark.sql import functions as f
@@ -1760,6 +1761,9 @@ def run_spatial_query():
         os.getenv("SPARK_REMOTE") is not None,
         reason="Checkpoint dir is not available in Spark Connect",
     )
+    @pytest.mark.skipif(
+        pyspark.__version__ >= "4", reason="DBSCAN is not supported yet on Spark 4"
+    )
     def test_dbscan(self):
         df = self.spark.createDataFrame([{"id": 1, "x": 2, "y": 3}]).withColumn(
             "geometry", f.expr("ST_Point(x, y)")

python/tests/stats/test_dbscan.py

@@ -15,6 +15,7 @@
 # specific language governing permissions and limitations
 # under the License.
 
+import pyspark
 import pyspark.sql.functions as f
 import pytest
 from sklearn.cluster import DBSCAN as sklearnDBSCAN
@@ -25,6 +26,9 @@
 from sedona.spark.stats import dbscan
 
 
+@pytest.mark.skipif(
+    pyspark.__version__ >= "4", reason="DBSCAN is not supported yet on Spark 4"
+)
 class TestDBScan(TestBase):
 
     @pytest.fixture

python/tests/test_base.py

@@ -44,26 +44,29 @@ def spark(self):
 
             builder = SedonaContext.builder().appName("SedonaSparkTest")
             if SPARK_REMOTE:
-                builder = (
-                    builder.remote(SPARK_REMOTE)
-                    .config(
+                builder = builder.remote(SPARK_REMOTE).config(
+                    "spark.sql.extensions",
+                    "org.apache.sedona.sql.SedonaSqlExtensions",
+                )
+
+                # Connect is packaged with Spark 4+
+                if pyspark.__version__ < "4":
+                    builder = builder.config(
                         "spark.jars.packages",
                         f"org.apache.spark:spark-connect_2.12:{pyspark.__version__}",
                     )
-                    .config(
-                        "spark.sql.extensions",
-                        "org.apache.sedona.sql.SedonaSqlExtensions",
-                    )
-                    .config(
-                        "spark.sedona.stac.load.itemsLimitMax",
-                        "20",
-                    )
-                )
             else:
-                builder = builder.master("local[*]").config(
+                builder = builder.master("local[*]")
+
+            builder = (
+                builder.config(
                     "spark.sedona.stac.load.itemsLimitMax",
                     "20",
                 )
+                # Pandas on PySpark doesn't work with ANSI mode, which is enabled by default
+                # in Spark 4
+                .config("spark.sql.ansi.enabled", "false")
+            )
 
             # Allows the Sedona .jar to be explicitly set by the caller (e.g, to run
             # pytest against a freshly-built development version of Sedona)

spark/common/pom.xml

@@ -268,6 +268,25 @@
 	<build>
         <sourceDirectory>src/main/java</sourceDirectory>
         <plugins>
+            <plugin>
+                <groupId>org.codehaus.mojo</groupId>
+                <artifactId>build-helper-maven-plugin</artifactId>
+                <version>3.2.0</version>
+                <executions>
+                    <execution>
+                        <id>add-source</id>
+                        <phase>generate-sources</phase>
+                        <goals>
+                            <goal>add-source</goal>
+                        </goals>
+                        <configuration>
+                            <sources>
+                                <source>src/main/scala-spark-${spark.major.version}</source>
+                            </sources>
+                        </configuration>
+                    </execution>
+                </executions>
+            </plugin>
             <plugin>
                 <groupId>net.alchim31.maven</groupId>
                 <artifactId>scala-maven-plugin</artifactId>
@@ -290,4 +309,23 @@
             </plugin>
         </plugins>
 	</build>
+    <profiles>
+        <profile>
+            <id>sedona-spark-4.0</id>
+            <activation>
+                <property>
+                    <name>spark</name>
+                    <value>4.0</value>
+                </property>
+            </activation>
+            <dependencies>
+                <dependency>
+                    <groupId>org.apache.spark</groupId>
+                    <artifactId>spark-sql-api_${scala.compat.version}</artifactId>
+                    <version>${spark.version}</version>
+                    <scope>provided</scope>
+                </dependency>
+            </dependencies>
+        </profile>
+    </profiles>
 </project>

spark/common/src/main/scala/org/apache/spark/sql/sedona_sql/expressions/DataFrameAPI.scala → spark/common/src/main/scala-spark-3/org/apache/spark/sql/sedona_sql/DataFrameShims.scala

@@ -16,57 +16,59 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.spark.sql.sedona_sql.expressions
+package org.apache.spark.sql.sedona_sql
 
 import scala.reflect.ClassTag
 
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.{Column, DataFrame, SparkSession}
+import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.catalyst.expressions.{Expression, Literal}
-import org.apache.spark.sql.Column
 import org.apache.spark.sql.expressions.UserDefinedAggregateFunction
 import org.apache.spark.sql.execution.aggregate.ScalaUDAF
+import org.apache.spark.sql.functions.lit
+import org.apache.spark.sql.types.StructType
 
-trait DataFrameAPI {
-  protected def wrapExpression[E <: Expression: ClassTag](args: Any*): Column = {
-    val exprArgs = args.map(_ match {
-      case c: Column => c.expr
-      case s: String => Column(s).expr
-      case e: Expression => e
-      case x: Any => Literal(x)
-      case null => Literal(null)
-    })
-    val expressionConstructor =
-      implicitly[ClassTag[E]].runtimeClass.getConstructor(classOf[Seq[Expression]])
-    val expressionInstance = expressionConstructor.newInstance(exprArgs).asInstanceOf[E]
-    Column(expressionInstance)
+object DataFrameShims {
+
+  private[sedona_sql] def wrapExpression[E <: Expression: ClassTag](args: Any*): Column = {
+    wrapVarArgExpression[E](args)
   }
 
-  protected def wrapVarArgExpression[E <: Expression: ClassTag](arg: Seq[Any]): Column = {
+  private[sedona_sql] def wrapVarArgExpression[E <: Expression: ClassTag](arg: Seq[Any]): Column = {
+    val runtimeClass = implicitly[ClassTag[E]].runtimeClass
     val exprArgs = arg.map(_ match {
       case c: Column => c.expr
       case s: String => Column(s).expr
       case e: Expression => e
       case x: Any => Literal(x)
       case null => Literal(null)
     })
-    val expressionConstructor =
-      implicitly[ClassTag[E]].runtimeClass.getConstructor(classOf[Seq[Expression]])
+    val expressionConstructor = runtimeClass.getConstructor(classOf[Seq[Expression]])
     val expressionInstance = expressionConstructor.newInstance(exprArgs).asInstanceOf[E]
     Column(expressionInstance)
   }
 
-  protected def wrapAggregator[A <: UserDefinedAggregateFunction: ClassTag](arg: Any*): Column = {
+  private[sedona_sql] def wrapAggregator[A <: UserDefinedAggregateFunction: ClassTag](arg: Any*): Column = {
+    val runtimeClass = implicitly[ClassTag[A]].runtimeClass
     val exprArgs = arg.map(_ match {
       case c: Column => c.expr
       case s: String => Column(s).expr
       case e: Expression => e
       case x: Any => Literal(x)
       case null => Literal(null)
     })
-    val aggregatorClass = implicitly[ClassTag[A]].runtimeClass
-    val aggregatorConstructor = aggregatorClass.getConstructor()
+    val aggregatorConstructor = runtimeClass.getConstructor()
     val aggregatorInstance =
       aggregatorConstructor.newInstance().asInstanceOf[UserDefinedAggregateFunction]
     val scalaAggregator = ScalaUDAF(exprArgs, aggregatorInstance)
     Column(scalaAggregator)
   }
+
+  private[sedona_sql] def createDataFrame(
+      sparkSession: SparkSession,
+      rdd: RDD[InternalRow],
+      schema: StructType): DataFrame = {
+    sparkSession.internalCreateDataFrame(rdd, schema)
+  }
 }