pin click<8.3.0

Author: pang-wu

.github/workflows/raydp.yml

@@ -74,8 +74,8 @@ jobs:
         run: |
           python -m pip install --upgrade pip
           pip install wheel
-          pip install "numpy<1.24" "click<8.3.0"
-          pip install "pydantic<2.0"
+          pip install "numpy<1.24"
+          pip install "pydantic<2.0" "click<8.3.0"
           SUBVERSION=$(python -c 'import sys; print(sys.version_info[1])')
           if [ "$(uname -s)" == "Linux" ]
           then

core/raydp-main/src/main/scala/org/apache/spark/sql/raydp/ObjectStoreWriter.scala

@@ -93,8 +93,8 @@ class ObjectStoreWriter(@transient val df: DataFrame) extends Serializable {
       batchSize = 0
     }
     val schema = df.schema
-    val arrowSchema = SparkShimLoader.getSparkShims.toArrowSchema(
-      schema, timeZoneId, sparkSession)
+    val arrowSchemaJson = SparkShimLoader.getSparkShims.toArrowSchema(
+      schema, timeZoneId, sparkSession).toJson
 
     val objectIds = df.queryExecution.toRdd.mapPartitions{ iter =>
       val queue = ObjectRefHolder.getQueue(uuid)
@@ -105,6 +105,9 @@ class ObjectStoreWriter(@transient val df: DataFrame) extends Serializable {
       } else {
         Iterator(iter)
       }
+
+      // Reconstruct arrow schema from JSON on executor
+      val arrowSchema = Schema.fromJSON(arrowSchemaJson)
       val allocator = ArrowUtils.rootAllocator.newChildAllocator(
         s"ray object store writer", 0, Long.MaxValue)
       val root = VectorSchemaRoot.create(arrowSchema, allocator)
@@ -229,7 +232,7 @@ object ObjectStoreWriter {
     val rdd = SparkShimLoader.getSparkShims.toArrowBatchRDD(df)
     rdd.persist(storageLevel)
     rdd.count()
-    var executorIds = df.sqlContext.sparkContext.getExecutorIds.toArray
+    val executorIds = df.sparkSession.sparkContext.getExecutorIds.toArray
     val numExecutors = executorIds.length
     val appMasterHandle = Ray.getActor(RayAppMaster.ACTOR_NAME)
                              .get.asInstanceOf[ActorHandle[RayAppMaster]]