Build scala 2.13 spark 3.3

README-template.md

@@ -177,7 +177,7 @@ runtime configurations to
 the connector. For example, Bigtable project and instance ID or settings for
 timestamp and timeout configurations.
 For a full list of configurations, refer to
-[BigtableSparkConf.scala](spark-bigtable_2.12/src/main/scala/com/google/cloud/spark/bigtable/datasources/BigtableSparkConf.scala),
+[BigtableSparkConf.scala](spark-bigtable_2.13/src/main/scala/com/google/cloud/spark/bigtable/datasources/BigtableSparkConf.scala),
 where these configs are defined.
 
 ### Bigtable emulator support

pom.xml

@@ -21,7 +21,7 @@
   <groupId>com.google.cloud.spark.bigtable</groupId>
   <artifactId>spark-bigtable-connector</artifactId>
   <packaging>pom</packaging>
-  <version>0.2.1</version>  <!-- ${NEXT_VERSION_FLAG} -->
+  <version>0.2.2</version>  <!-- ${NEXT_VERSION_FLAG} -->
   <name>Spark Bigtable Connector Build Parent</name>
   <description>Parent project for all the Spark Bigtable Connector artifacts</description>
   <url>https://github.com/GoogleCloudDataproc/spark-bigtable-connector</url>
@@ -52,8 +52,8 @@
   </scm>
 
   <modules>
-    <module>spark-bigtable_2.12</module>
-    <module>spark-bigtable_2.12-it</module>
+    <module>spark-bigtable_2.13</module>
+    <module>spark-bigtable_2.13-it</module>
   </modules>
 
   <distributionManagement>
@@ -71,8 +71,8 @@
     <gpg.skip>true</gpg.skip>
     <nexus.remote.skip>false</nexus.remote.skip>
 
-    <scala.version>2.12.14</scala.version>
-    <scala.binary.version>2.12</scala.binary.version>
+    <scala.version>2.13.14</scala.version>
+    <scala.binary.version>2.13</scala.binary.version>
     <spark.version>3.3.3</spark.version>
     <bigtable.java.version>2.42.0</bigtable.java.version>
     <bigtable.java.emulator.version>0.175.0</bigtable.java.emulator.version>

spark-bigtable_2.12-it/pom.xml → spark-bigtable_2.13-it/pom.xml

@@ -26,9 +26,9 @@
   </parent>
 
   <groupId>com.google.cloud.spark.bigtable</groupId>
-  <artifactId>spark-bigtable_2.12-it</artifactId>
+  <artifactId>spark-bigtable_2.13-it</artifactId>
   <name>Google Bigtable - Spark Connector Integration Tests</name>
-  <version>0.2.1</version>  <!-- ${NEXT_VERSION_FLAG} -->
+  <version>0.2.2</version>  <!-- ${NEXT_VERSION_FLAG} -->
 
   <dependencies>
     <dependency>
@@ -51,8 +51,20 @@
 
     <dependency>
       <groupId>com.google.cloud.spark.bigtable</groupId>
-      <artifactId>spark-bigtable_2.12</artifactId>
-      <version>0.2.1</version>  <!-- ${NEXT_VERSION_FLAG} -->
+      <artifactId>spark-bigtable_2.13</artifactId>
+      <version>0.2.2</version>  <!-- ${NEXT_VERSION_FLAG} -->
+      <exclusions>
+        <!-- Exclude google-cloud-bigtable dependency -->
+        <exclusion>
+          <groupId>com.google.cloud</groupId>
+          <artifactId>google-cloud-bigtable</artifactId>
+        </exclusion>
+        <!-- Exclude io.grpc dependencies -->
+        <exclusion>
+          <groupId>io.grpc</groupId>
+          <artifactId>*</artifactId>
+        </exclusion>
+      </exclusions>
     </dependency>
 
     <dependency>

spark-bigtable_2.12-it/src/test/java/com/google/cloud/spark/bigtable/ErrorHandlingIntegrationTest.java → spark-bigtable_2.13-it/src/test/java/com/google/cloud/spark/bigtable/ErrorHandlingIntegrationTest.java

@@ -23,12 +23,12 @@
 import com.google.cloud.spark.bigtable.fakeserver.FakeServerBuilder;
 import com.google.cloud.spark.bigtable.fakeserver.FakeTableAdminService;
 import com.google.cloud.spark.bigtable.model.TestRow;
-import com.google.cloud.spark.bigtable.repackaged.com.google.bigtable.admin.v2.CreateTableRequest;
-import com.google.cloud.spark.bigtable.repackaged.com.google.bigtable.v2.MutateRowsRequest;
-import com.google.cloud.spark.bigtable.repackaged.com.google.bigtable.v2.ReadRowsRequest;
-import com.google.cloud.spark.bigtable.repackaged.com.google.bigtable.v2.SampleRowKeysRequest;
-import com.google.cloud.spark.bigtable.repackaged.io.grpc.Server;
-import com.google.cloud.spark.bigtable.repackaged.io.grpc.Status;
+import /*com.google.cloud.spark.bigtable.repackaged.*/com.google.bigtable.admin.v2.CreateTableRequest;
+import /*com.google.cloud.spark.bigtable.repackaged.*/com.google.bigtable.v2.MutateRowsRequest;
+import /*com.google.cloud.spark.bigtable.repackaged.*/com.google.bigtable.v2.ReadRowsRequest;
+import /*com.google.cloud.spark.bigtable.repackaged.*/com.google.bigtable.v2.SampleRowKeysRequest;
+import /*com.google.cloud.spark.bigtable.repackaged.*/io.grpc.Server;
+import /*com.google.cloud.spark.bigtable.repackaged.*/io.grpc.Status;
 import com.google.errorprone.annotations.Keep;
 import java.util.ArrayList;
 import junitparams.JUnitParamsRunner;

spark-bigtable_2.12-it/src/test/java/com/google/cloud/spark/bigtable/OpenLineageIntegrationTest.java → spark-bigtable_2.13-it/src/test/java/com/google/cloud/spark/bigtable/OpenLineageIntegrationTest.java

@@ -3,6 +3,7 @@
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.core.Is.is;
 
+import com.google.api.gax.core.NoCredentialsProvider;
 import com.google.cloud.bigtable.admin.v2.BigtableTableAdminClient;
 import com.google.cloud.bigtable.admin.v2.BigtableTableAdminSettings;
 import com.google.gson.JsonObject;
@@ -87,7 +88,6 @@ public void testOpenLineageEvents() throws Exception {
       // event data.
       Dataset<Row> outputReadDf = readDataframeFromBigtable(spark, outputCatalog);
       assertDataFramesEqual(outputReadDf, outputDf);
-
       List<JsonObject> jsonObjects = parseEventLog(lineageFile);
       assertThat(jsonObjects.isEmpty(), is(false));
 
@@ -113,7 +113,6 @@ public void testOpenLineageEvents() throws Exception {
   private static SparkSession createSparkSessionWithOL() throws IOException {
     lineageFile = File.createTempFile("openlineage_test_" + System.nanoTime(), ".log");
     lineageFile.deleteOnExit();
-
     spark =
         SparkSession.builder()
             .master("local")

spark-bigtable_2.12-it/src/test/java/com/google/cloud/spark/bigtable/RDDReadWriteIntegrationTests.java → spark-bigtable_2.13-it/src/test/java/com/google/cloud/spark/bigtable/RDDReadWriteIntegrationTests.java

@@ -23,10 +23,14 @@
 import com.google.cloud.bigtable.admin.v2.BigtableTableAdminSettings;
 import com.google.cloud.spark.bigtable.datasources.BigtableSparkConf;
 import com.google.cloud.spark.bigtable.datasources.BigtableSparkConfBuilder;
-import com.google.cloud.spark.bigtable.repackaged.com.google.cloud.bigtable.data.v2.models.Row;
-import com.google.cloud.spark.bigtable.repackaged.com.google.cloud.bigtable.data.v2.models.RowCell;
-import com.google.cloud.spark.bigtable.repackaged.com.google.cloud.bigtable.data.v2.models.RowMutationEntry;
-import com.google.cloud.spark.bigtable.repackaged.io.grpc.Status;
+//import com.google.cloud.spark.bigtable.repackaged.com.google.cloud.bigtable.data.v2.models.Row;
+//import com.google.cloud.spark.bigtable.repackaged.com.google.cloud.bigtable.data.v2.models.RowCell;
+//import com.google.cloud.spark.bigtable.repackaged.com.google.cloud.bigtable.data.v2.models.RowMutationEntry;
+//import com.google.cloud.spark.bigtable.repackaged.io.grpc.Status;
+import /*com.google.cloud.spark.bigtable.repackaged.*/com.google.cloud.bigtable.data.v2.models.Row;
+import /*com.google.cloud.spark.bigtable.repackaged.*/com.google.cloud.bigtable.data.v2.models.RowCell;
+import /*com.google.cloud.spark.bigtable.repackaged.*/com.google.cloud.bigtable.data.v2.models.RowMutationEntry;
+import /*com.google.cloud.spark.bigtable.repackaged.*/io.grpc.Status;
 import java.util.ArrayList;
 import java.util.List;
 import junitparams.JUnitParamsRunner;

spark-bigtable_2.12-it/src/test/java/com/google/cloud/spark/bigtable/ReadWriteIntegrationTest.java → spark-bigtable_2.13-it/src/test/java/com/google/cloud/spark/bigtable/ReadWriteIntegrationTest.java

@@ -20,12 +20,13 @@
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 
+import com.google.api.gax.rpc.NotFoundException;
 import com.google.cloud.bigtable.admin.v2.BigtableTableAdminClient;
 import com.google.cloud.bigtable.admin.v2.BigtableTableAdminSettings;
 import com.google.cloud.spark.bigtable.model.Favorites;
 import com.google.cloud.spark.bigtable.model.TestAvroRow;
 import com.google.cloud.spark.bigtable.model.TestRow;
-import com.google.cloud.spark.bigtable.repackaged.com.google.api.gax.rpc.NotFoundException;
+//import com.google.cloud.spark.bigtable.repackaged.com.google.api.gax.rpc.NotFoundException;
 import java.util.ArrayList;
 import junitparams.JUnitParamsRunner;
 import junitparams.Parameters;

spark-bigtable_2.12/pom.xml → spark-bigtable_2.13/pom.xml

@@ -26,9 +26,9 @@
   </parent>
 
   <groupId>com.google.cloud.spark.bigtable</groupId>
-  <artifactId>spark-bigtable_2.12</artifactId>
+  <artifactId>spark-bigtable_2.13</artifactId>
   <name>Google Bigtable - Apache Spark Connector</name>
-  <version>0.2.1</version>  <!-- ${NEXT_VERSION_FLAG} -->
+  <version>0.2.2</version>  <!-- ${NEXT_VERSION_FLAG} -->
 
   <dependencies>
     <dependency>
@@ -41,7 +41,18 @@
       <artifactId>grpc-google-cloud-bigtable-admin-v2</artifactId>
       <version>${bigtable.java.version}</version>
     </dependency>
-
+    <!-- https://mvnrepository.com/artifact/com.fasterxml.jackson.core/jackson-core -->
+    <dependency>
+      <groupId>com.fasterxml.jackson.core</groupId>
+      <artifactId>jackson-core</artifactId>
+      <version>2.15.2</version>
+    </dependency>
+    <!-- https://mvnrepository.com/artifact/com.fasterxml.jackson.core/jackson-annotations -->
+    <dependency>
+      <groupId>com.fasterxml.jackson.core</groupId>
+      <artifactId>jackson-annotations</artifactId>
+      <version>2.15.2</version>
+    </dependency>
     <!-- To fix the "NoClassDefFoundError: scala/util/parsing/json/JSON$" error with Spark 3.4. -->
     <dependency>
       <groupId>org.scala-lang.modules</groupId>
@@ -220,6 +231,11 @@
               </transformers>
               <artifactSet>
                 <excludes>
+                  <!-- Exclude Bigtable related dependencies -->
+                  <exclude>com.google.cloud:google-cloud-bigtable</exclude>
+                  <!-- Exclude gRPC related dependencies -->
+                  <exclude>com.google.api.grpc:*</exclude>
+                  <!-- Existing exclusions for slf4j and reload4j -->
                   <exclude>org.slf4j:slf4j-reload4j</exclude>
                   <exclude>org.slf4j:slf4j-api</exclude>
                   <exclude>ch.qos.reload4j:reload4j</exclude>
@@ -230,7 +246,7 @@
                   <pattern>io.netty</pattern>
                   <shadedPattern>com.google.cloud.spark.bigtable.repackaged.io.netty</shadedPattern>
                 </relocation>
-                <relocation>
+                <!--<relocation>
                   <pattern>io.grpc</pattern>
                   <shadedPattern>com.google.cloud.spark.bigtable.repackaged.io.grpc</shadedPattern>
                 </relocation>
@@ -239,8 +255,9 @@
                   <shadedPattern>com.google.cloud.spark.bigtable.repackaged.com.google</shadedPattern>
                   <excludes>
                     <exclude>com.google.cloud.spark.bigtable.**</exclude>
+                    <exclude>com.google.cloud.bigtable.**</exclude>
                   </excludes>
-                </relocation>
+                </relocation>-->
                 <relocation>
                   <pattern>io.openlineage.spark.shade</pattern>
                   <shadedPattern>com.google.cloud.spark.bigtable.repackaged.io.openlineage.spark.shade</shadedPattern>

third_party/hbase-spark-connector/hbase-connectors/src/main/scala/com/google/cloud/spark/bigtable/datasources/BigtableTableCatalog.scala

@@ -272,11 +272,9 @@ object BigtableTableCatalog {
     val parameters = params
     val jString = parameters(tableCatalog)
     val map = JSON.parseFull(jString).get.asInstanceOf[Map[String, _]]
-    val tableMeta = map.get(table).get.asInstanceOf[Map[String, _]]
-    val tName = tableMeta.get(tableName).get.asInstanceOf[String]
-    val cIter = map
-      .get(columns)
-      .get
+    val tableMeta = map(table).asInstanceOf[Map[String, _]]
+    val tName = tableMeta(tableName).asInstanceOf[String]
+    val cIter = map(columns)
       .asInstanceOf[Map[String, Map[String, String]]]
       .toIterator
     val schemaMap = mutable.HashMap.empty[String, Field]
@@ -286,7 +284,7 @@ object BigtableTableCatalog {
       val f = Field(
         name,
         column.getOrElse(cf, rowKey),
-        column.get(col).get,
+        column(col),
         column.get(`type`),
         sAvro,
         len

third_party/hbase-spark-connector/hbase-connectors/src/main/scala/com/google/cloud/spark/bigtable/datasources/SchemaConverters.scala

@@ -33,7 +33,7 @@ import java.nio.ByteBuffer
 import java.sql.Timestamp
 import java.util
 import java.util.HashMap
-import scala.collection.JavaConversions._
+import scala.jdk.CollectionConverters._
 
 @InterfaceAudience.Private
 abstract class AvroException(msg: String) extends Exception(msg)
@@ -69,7 +69,7 @@ object SchemaConverters {
       case ENUM    => SchemaType(StringType, nullable = false)
 
       case RECORD =>
-        val fields = avroSchema.getFields.map { f =>
+        val fields = avroSchema.getFields.asScala.toSeq.map { f =>
           val schemaType = toSqlType(f.schema())
           StructField(f.name, schemaType.dataType, schemaType.nullable)
         }
@@ -95,18 +95,18 @@ object SchemaConverters {
         )
 
       case UNION =>
-        if (avroSchema.getTypes.exists(_.getType == NULL)) {
+        if (avroSchema.getTypes.asScala.exists(_.getType == NULL)) {
           // In case of a union with null, eliminate it and make a recursive call
           val remainingUnionTypes =
-            avroSchema.getTypes.filterNot(_.getType == NULL)
+            avroSchema.getTypes.asScala.filterNot(_.getType == NULL)
           if (remainingUnionTypes.size == 1) {
-            toSqlType(remainingUnionTypes.get(0)).copy(nullable = true)
+            toSqlType(remainingUnionTypes.head).copy(nullable = true)
           } else {
-            toSqlType(Schema.createUnion(remainingUnionTypes))
+            toSqlType(Schema.createUnion(remainingUnionTypes.asJava))
               .copy(nullable = true)
           }
         } else
-          avroSchema.getTypes.map(_.getType) match {
+          avroSchema.getTypes.asScala.map(_.getType) match {
             case Seq(t1, t2) if Set(t1, t2) == Set(INT, LONG) =>
               SchemaType(LongType, nullable = false)
             case Seq(t1, t2) if Set(t1, t2) == Set(FLOAT, DOUBLE) =>
@@ -182,7 +182,7 @@ object SchemaConverters {
           }
       case RECORD =>
         val fieldConverters =
-          schema.getFields.map(f => createConverterToSQL(f.schema))
+          schema.getFields.asScala.map(f => createConverterToSQL(f.schema))
         (item: Any) =>
           if (item == null) {
             null
@@ -203,10 +203,10 @@ object SchemaConverters {
             null
           } else {
             try {
-              item.asInstanceOf[GenericData.Array[Any]].map(elementConverter)
+              item.asInstanceOf[GenericData.Array[Any]].asScala.map(elementConverter)
             } catch {
               case e: Throwable =>
-                item.asInstanceOf[util.ArrayList[Any]].map(elementConverter)
+                item.asInstanceOf[util.ArrayList[Any]].asScala.map(elementConverter)
             }
           }
       case MAP =>
@@ -216,20 +216,21 @@ object SchemaConverters {
             null
           } else {
             item
-              .asInstanceOf[HashMap[Any, Any]]
+              .asInstanceOf[util.HashMap[Any, Any]]
+              .asScala
               .map(x => (x._1.toString, valueConverter(x._2)))
               .toMap
           }
       case UNION =>
-        if (schema.getTypes.exists(_.getType == NULL)) {
-          val remainingUnionTypes = schema.getTypes.filterNot(_.getType == NULL)
+        if (schema.getTypes.asScala.exists(_.getType == NULL)) {
+          val remainingUnionTypes = schema.getTypes.asScala.filterNot(_.getType == NULL)
           if (remainingUnionTypes.size == 1) {
-            createConverterToSQL(remainingUnionTypes.get(0))
+            createConverterToSQL(remainingUnionTypes.head)
           } else {
-            createConverterToSQL(Schema.createUnion(remainingUnionTypes))
+            createConverterToSQL(Schema.createUnion(remainingUnionTypes.asJava))
           }
         } else
-          schema.getTypes.map(_.getType) match {
+          schema.getTypes.asScala.map(_.getType) match {
             case Seq(t1, t2) if Set(t1, t2) == Set(INT, LONG) =>
               (item: Any) => {
                 item match {
@@ -413,7 +414,7 @@ object SchemaConverters {
           if (item == null) {
             null
           } else {
-            val javaMap = new HashMap[String, Any]()
+            val javaMap = new util.HashMap[String, Any]()
             item.asInstanceOf[Map[String, Any]].foreach { case (key, value) =>
               javaMap.put(key, valueConverter(value))
             }
@@ -474,7 +475,7 @@ object AvroSerdes {
         val encoder2: BinaryEncoder =
           EncoderFactory.get().directBinaryEncoder(bao2, null)
         writer2.write(gr, encoder2)
-        bao2.toByteArray()
+        bao2.toByteArray
       case _ => throw new Exception(s"unsupported data type ${schema.getType}")
     }
   }