From 567f6c609dd5bacc9e09bed41cbdb984a94b6f74 Mon Sep 17 00:00:00 2001
From: Yubi Lee <eubnara@gmail.com>
Date: Wed, 11 Jun 2025 10:54:27 +0900
Subject: [PATCH] PHOENIX-7635 Update Connectors to use Spark 3.5.6

---
 phoenix5-spark3/pom.xml                       | 24 ++++++++++++++-----
 .../connector/writer/PhoenixDataWriter.java   |  7 +++---
 pom.xml                                       |  2 +-
 3 files changed, 23 insertions(+), 10 deletions(-)

diff --git a/phoenix5-spark3/pom.xml b/phoenix5-spark3/pom.xml
index 634cb9de..f93162d0 100644
--- a/phoenix5-spark3/pom.xml
+++ b/phoenix5-spark3/pom.xml
@@ -60,6 +60,12 @@
         </exclusion>
       </exclusions>
     </dependency>
+    <dependency>
+      <groupId>org.apache.spark</groupId>
+      <artifactId>spark-common-utils_${scala.binary.version}</artifactId>
+      <version>${spark.version}</version>
+      <scope>provided</scope>
+    </dependency>
     <dependency>
       <groupId>org.apache.spark</groupId>
       <artifactId>spark-unsafe_${scala.binary.version}</artifactId>
@@ -78,6 +84,12 @@
       <version>${spark.version}</version>
       <scope>provided</scope>
     </dependency>
+    <dependency>
+      <groupId>org.apache.spark</groupId>
+      <artifactId>spark-sql-api_${scala.binary.version}</artifactId>
+      <version>${spark.version}</version>
+      <scope>provided</scope>
+    </dependency>
 
     <!-- Phoenix dependencies -->
     <dependency>
@@ -106,6 +118,12 @@
       <version>${scala.version}</version>
       <scope>provided</scope>
     </dependency>
+    <dependency>
+      <groupId>org.scala-lang</groupId>
+      <artifactId>scala-reflect</artifactId>
+      <version>${scala.version}</version>
+      <scope>provided</scope>
+    </dependency>
 
     <!-- HBase dependencies -->
     <dependency>
@@ -143,12 +161,6 @@
     </dependency>
 
     <!-- Scala Test dependencies -->
-    <dependency>
-      <groupId>org.scala-lang</groupId>
-      <artifactId>scala-reflect</artifactId>
-      <version>${scala.version}</version>
-      <scope>test</scope>
-    </dependency>
     <dependency>
       <groupId>org.scalactic</groupId>
       <artifactId>scalactic_${scala.binary.version}</artifactId>
diff --git a/phoenix5-spark3/src/main/java/org/apache/phoenix/spark/sql/connector/writer/PhoenixDataWriter.java b/phoenix5-spark3/src/main/java/org/apache/phoenix/spark/sql/connector/writer/PhoenixDataWriter.java
index ea5338fe..360a545e 100644
--- a/phoenix5-spark3/src/main/java/org/apache/phoenix/spark/sql/connector/writer/PhoenixDataWriter.java
+++ b/phoenix5-spark3/src/main/java/org/apache/phoenix/spark/sql/connector/writer/PhoenixDataWriter.java
@@ -36,7 +36,7 @@
 import org.apache.spark.sql.Row;
 import org.apache.spark.sql.catalyst.InternalRow;
 import org.apache.spark.sql.catalyst.encoders.ExpressionEncoder;
-import org.apache.spark.sql.catalyst.encoders.RowEncoder$;
+import org.apache.spark.sql.catalyst.types.DataTypeUtils;
 import org.apache.spark.sql.execution.datasources.SparkJdbcUtil;
 import org.apache.spark.sql.execution.datasources.jdbc.PhoenixJdbcDialect$;
 import org.apache.spark.sql.connector.write.DataWriter;
@@ -67,10 +67,11 @@ public class PhoenixDataWriter implements DataWriter<InternalRow> {
         this.schema = options.getSchema();
 
         List<Attribute> attrs = new ArrayList<>();
-        for (AttributeReference ref : scala.collection.JavaConverters.seqAsJavaListConverter(schema.toAttributes()).asJava()) {
+        for (AttributeReference ref : scala.collection.JavaConverters.seqAsJavaListConverter(
+            DataTypeUtils.toAttributes(schema)).asJava()) {
             attrs.add(ref.toAttribute());
         }
-        encoder = RowEncoder$.MODULE$.apply(schema).resolveAndBind( scala.collection.JavaConverters.asScalaIteratorConverter(attrs.iterator()).asScala().toSeq(), SimpleAnalyzer$.MODULE$);
+        encoder = ExpressionEncoder.apply(schema).resolveAndBind( scala.collection.JavaConverters.asScalaIteratorConverter(attrs.iterator()).asScala().toSeq(), SimpleAnalyzer$.MODULE$);
         try {
             this.conn = DriverManager.getConnection(jdbcUrl, connectionProps);
             List<String> colNames =  new ArrayList<>(Arrays.asList(options.getSchema().names()));
diff --git a/pom.xml b/pom.xml
index acb45f16..d038788c 100644
--- a/pom.xml
+++ b/pom.xml
@@ -119,7 +119,7 @@
     <spark.version>2.4.8</spark.version>
     <scala.version.for.spark2>2.11.12</scala.version.for.spark2>
     <scala.binary.version.for.spark2>2.11</scala.binary.version.for.spark2>
-    <spark3.version>3.4.4</spark3.version>
+    <spark3.version>3.5.6</spark3.version>
     <scala.version.for.spark3>2.12.18</scala.version.for.spark3>
     <scala.binary.version.for.spark3>2.12</scala.binary.version.for.spark3>
     <log4j.version>1.2.17</log4j.version>