(kyuubi) 02/05: Fix Spark 4.3 CI failures

[ch...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

pan3793 pushed a commit to branch spark-4.3
in repository https://gitbox.apache.org/repos/asf/kyuubi.git

commit 0f64c63a0f508ece19db85e29320016859dde7b6
Author: Cheng Pan <ch...@apache.org>
AuthorDate: Mon Aug 31 23:49:46 2026 +0800

    Fix Spark 4.3 CI failures
    
    Assisted-by: Codex
---
 .../execution/arrow/KyuubiArrowConverters.scala    | 38 ++++++++++++++++++++--
 .../hive/beeline/common/log/InPlaceUpdate.java     |  4 +--
 2 files changed, 37 insertions(+), 5 deletions(-)

diff --git a/externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/spark/sql/execution/arrow/KyuubiArrowConverters.scala b/externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/spark/sql/execution/arrow/KyuubiArrowConverters.scala
index 9f690357b8..6a7e34bbd5 100644
--- a/externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/spark/sql/execution/arrow/KyuubiArrowConverters.scala
+++ b/externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/spark/sql/execution/arrow/KyuubiArrowConverters.scala
@@ -27,6 +27,7 @@ import org.apache.arrow.vector._
 import org.apache.arrow.vector.compression.NoCompressionCodec
 import org.apache.arrow.vector.ipc.{ArrowStreamWriter, ReadChannel, WriteChannel}
 import org.apache.arrow.vector.ipc.message.{IpcOption, MessageSerializer}
+import org.apache.arrow.vector.types.pojo.Schema
 import org.apache.spark.TaskContext
 import org.apache.spark.internal.Logging
 import org.apache.spark.sql.catalyst.{InternalRow, SQLConfHelper}
@@ -36,6 +37,8 @@ import org.apache.spark.sql.types._
 import org.apache.spark.sql.util.ArrowUtils
 import org.apache.spark.util.Utils
 
+import org.apache.kyuubi.util.reflect.DynMethods
+
 object KyuubiArrowConverters extends SQLConfHelper with Logging {
 
   type Batch = (Array[Byte], Long)
@@ -45,6 +48,37 @@ object KyuubiArrowConverters extends SQLConfHelper with Logging {
   private val ZstdCompressionCodecClassName =
     "org.apache.arrow.compression.ZstdCompressionCodec"
 
+  private lazy val toArrowSchemaMethod: DynMethods.BoundMethod =
+    DynMethods.builder("toArrowSchema")
+      .impl( // SPARK-58005 (4.3.0): added losslessInternalTypes to toArrowSchema
+        ArrowUtils.getClass,
+        classOf[StructType],
+        classOf[String],
+        java.lang.Boolean.TYPE,
+        java.lang.Boolean.TYPE,
+        java.lang.Boolean.TYPE)
+      .impl( // Spark < 4.3.0
+        ArrowUtils.getClass,
+        classOf[StructType],
+        classOf[String],
+        java.lang.Boolean.TYPE,
+        java.lang.Boolean.TYPE)
+      .build()
+      .bind(ArrowUtils)
+
+  private def toArrowSchema(
+      schema: StructType,
+      timeZoneId: String,
+      errorOnDuplicatedFieldNames: Boolean,
+      largeVarTypes: Boolean): Schema = {
+    toArrowSchemaMethod.invoke[Schema](
+      schema,
+      timeZoneId,
+      errorOnDuplicatedFieldNames,
+      largeVarTypes,
+      false)
+  }
+
   // Mirror Spark's SparkSession#enableHiveSupport: check the capability on each request instead
   // of caching, because the codec is a session-level config that can change at runtime.
   private def arrowCompressionAvailable: Boolean = {
@@ -87,7 +121,7 @@ object KyuubiArrowConverters extends SQLConfHelper with Logging {
       "slice",
       0,
       Long.MaxValue)
-    val arrowSchema = ArrowUtils.toArrowSchema(schema, timeZoneId, true, false)
+    val arrowSchema = toArrowSchema(schema, timeZoneId, true, false)
     vectorSchemaRoot = VectorSchemaRoot.create(arrowSchema, sliceAllocator)
     try {
       val recordBatch = MessageSerializer.deserializeRecordBatch(
@@ -290,7 +324,7 @@ object KyuubiArrowConverters extends SQLConfHelper with Logging {
       zstdLevel: Int)
     extends Iterator[Array[Byte]] {
 
-    protected val arrowSchema = ArrowUtils.toArrowSchema(schema, timeZoneId, true, false)
+    protected val arrowSchema = toArrowSchema(schema, timeZoneId, true, false)
     // Validate the codec before allocating Arrow buffers, so an unsupported codec fails fast.
     private val compressionEnabled = codecName match {
       case null | "none" =>
diff --git a/kyuubi-hive-beeline/src/main/java/org/apache/hive/beeline/common/log/InPlaceUpdate.java b/kyuubi-hive-beeline/src/main/java/org/apache/hive/beeline/common/log/InPlaceUpdate.java
index 4f7d562c72..fbed17e038 100644
--- a/kyuubi-hive-beeline/src/main/java/org/apache/hive/beeline/common/log/InPlaceUpdate.java
+++ b/kyuubi-hive-beeline/src/main/java/org/apache/hive/beeline/common/log/InPlaceUpdate.java
@@ -25,7 +25,6 @@ import java.io.PrintStream;
 import java.io.StringWriter;
 import java.text.DecimalFormat;
 import java.util.List;
-import javax.annotation.Nullable;
 import org.apache.commons.lang3.StringUtils;
 import org.fusesource.jansi.Ansi;
 
@@ -161,9 +160,8 @@ public class InPlaceUpdate {
         Lists.transform(
             monitor.rows(),
             new Function<List<String>, String>() {
-              @Nullable
               @Override
-              public String apply(@Nullable List<String> row) {
+              public String apply(List<String> row) {
                 return String.format(VERTEX_FORMAT, row.toArray());
               }
             });