SPARKC-647 UDTValue performance fix - reuse CassandraRowMetadata instead of building it for each row (b2.5) (#1309)

* UDTValue to have CassandraRowMetadata in constructor similar to japi.UDTValue

* preserve compatibility for UDTValue case class constructor/apply/unapply by introducing a new field metaData_ that can keep the precalculated CassandraRowMetadata

(cherry picked from commit 39e55cc)

* lazy val metadata

(cherry picked from commit 3a62237)

* change case class default constructor, add unapply method

Co-authored-by: Igor Kamyshnikov <[email protected]>
Co-authored-by: Igor Kamyshnikov <[email protected]>

Author: gowa

connector/src/main/scala/org/apache/spark/sql/cassandra/CassandraSQLRow.scala

@@ -85,7 +85,7 @@ object CassandraSQLRow {
       case set: Set[_] => set.map(toSparkSqlType).toSeq
       case list: List[_] => list.map(toSparkSqlType)
       case map: Map[_, _] => map map { case(k, v) => (toSparkSqlType(k), toSparkSqlType(v))}
-      case udt: UDTValue => UDTValue(udt.columnNames, udt.columnValues.map(toSparkSqlType))
+      case udt: UDTValue => UDTValue(udt.metaData, udt.columnValues.map(toSparkSqlType))
       case tupleValue: TupleValue => TupleValue(tupleValue.values.map(toSparkSqlType): _*)
       case dateRange: DateRange => dateRange.toString
       case _ => value.asInstanceOf[AnyRef]
@@ -106,7 +106,7 @@ object CassandraSQLRow {
       case set: Set[_] => set.map(toUnsafeSqlType).toSeq
       case list: List[_] => list.map(toUnsafeSqlType)
       case map: Map[_, _] => map map { case(k, v) => (toUnsafeSqlType(k), toUnsafeSqlType(v))}
-      case udt: UDTValue => UDTValue(udt.columnNames, udt.columnValues.map(toUnsafeSqlType))
+      case udt: UDTValue => UDTValue(udt.metaData, udt.columnValues.map(toUnsafeSqlType))
       case tupleValue: TupleValue => TupleValue(tupleValue.values.map(toUnsafeSqlType): _*)
       case dateRange: DateRange => UTF8String.fromString(dateRange.toString)
       case instant: Instant => java.sql.Timestamp.from(instant)

driver/src/main/scala/com/datastax/spark/connector/UDTValue.scala

@@ -7,12 +7,19 @@ import com.datastax.spark.connector.util.DriverUtil.toName
 import scala.collection.JavaConversions._
 import scala.reflect.runtime.universe._
 
-final case class UDTValue(columnNames: IndexedSeq[String], columnValues: IndexedSeq[AnyRef])
+final case class UDTValue(metaData: CassandraRowMetadata, columnValues: IndexedSeq[AnyRef])
   extends ScalaGettableData {
+
+  def this(columnNames: IndexedSeq[String], columnValues: IndexedSeq[AnyRef]) =
+    this(CassandraRowMetadata.fromColumnNames(columnNames), columnValues)
+
+  def columnNames: IndexedSeq[String] = metaData.columnNames
+
   override def productArity: Int = columnValues.size
   override def productElement(i: Int) = columnValues(i)
 
-  override def metaData = CassandraRowMetadata.fromColumnNames(columnNames)
+  def unapply(t: UDTValue): Some[(IndexedSeq[String],IndexedSeq[AnyRef])] =
+    Some((t.metaData.columnNames,t.columnValues))
 }
 
 object UDTValue {
@@ -35,4 +42,7 @@ object UDTValue {
       case x: UDTValue => x
     }
   }
+
+  def apply(columnNames: IndexedSeq[String], columnValues: IndexedSeq[AnyRef]): UDTValue =
+    new UDTValue(columnNames, columnValues)
 }

driver/src/main/scala/com/datastax/spark/connector/types/UserDefinedType.scala

@@ -7,7 +7,7 @@ import com.datastax.oss.driver.api.core.data.{UdtValue => DriverUDTValue}
 import com.datastax.spark.connector.cql.{FieldDef, StructDef}
 import com.datastax.spark.connector.types.ColumnType.fromDriverType
 import com.datastax.spark.connector.types.TypeAdapters.ValueByNameAdapter
-import com.datastax.spark.connector.{ColumnName, UDTValue}
+import com.datastax.spark.connector.{CassandraRowMetadata, ColumnName, UDTValue}
 
 import scala.collection.JavaConversions._
 import scala.reflect.runtime.universe._
@@ -37,6 +37,7 @@ case class UserDefinedType(
   def cqlTypeName = name
 
   val fieldConvereters = columnTypes.map(_.converterToCassandra)
+  private lazy val metadata = CassandraRowMetadata.fromColumnNames(columnNames)
 
   private lazy val valueByNameConverter = scala.util.Try(TypeConverter.forType[ValueByNameAdapter]).toOption
 
@@ -51,7 +52,7 @@ case class UserDefinedType(
             val columnValue = columnConverter.convert(udtValue.getRaw(columnName))
             columnValue
           }
-        new UDTValue(columnNames, columnValues)
+        new UDTValue(metadata, columnValues)
       case value if valueByNameConverter.exists(_.convertPF.isDefinedAt(value)) =>
         val valuesByName = valueByNameConverter.get.convert(value)
         val columnValues =
@@ -61,14 +62,14 @@ case class UserDefinedType(
             val columnValue = columnConverter.convert(valuesByName.getByName(columnName))
             columnValue
           }
-        new UDTValue(columnNames, columnValues)
+        new UDTValue(metadata, columnValues)
     }
   }
 
   override type ValueRepr = UDTValue
 
   override def newInstance(columnValues: Any*): UDTValue = {
-    UDTValue(columnNames, columnValues.map(_.asInstanceOf[AnyRef]).toIndexedSeq)
+    UDTValue(metadata, columnValues.map(_.asInstanceOf[AnyRef]).toIndexedSeq)
   }
 }