Column name being parsed instead of Column Value when querying with Spark using aws-wrapper

[cyonglun]

Describe the bug

Usage of aws-wrapper in SparkSession seems to have issues during read.
Reverting to usage of non aws-wrapper (e.g. com. mysql. cj. jdbc. Driver) and retrieving the IAM auth token manually allowed the read to succeed.

Expected Behavior

SparkSession read should be able to properly encode the column values based on the stated class type.

What plugins are used? What other connection properties were set?

iam

Current Behavior

A simple MySQL query of select id, row_number() over(order by (select null)) as rno from table_name; failed to execute. This is despite the datatype of the ResultSet being correct (rno is DECIMAL)

SparkSqlReader: Schema JSON: {
  "type" : "struct",
  "fields" : [ {
    "name" : "rno",
    "type" : "decimal(38,0)",
    "nullable" : true,
    "metadata" : {
      "isSigned" : true,
      "scale" : 0
    }
  }]
}

The exception seems to indicate that the column name rno is being parsed instead of the column value:

WARN TaskSetManager: Lost task 0.0 in stage 0.0 (TID 0) ([2406:da18:7a9:9701:dcfb:3e3c:9391:425] executor 12): java.sql.SQLDataException: Cannot determine value type from string 'rno'
	at com.mysql.cj.jdbc.exceptions.SQLError.createSQLException(SQLError.java:106)
	at com.mysql.cj.jdbc.exceptions.SQLError.createSQLException(SQLError.java:89)
	at com.mysql.cj.jdbc.exceptions.SQLError.createSQLException(SQLError.java:81)
	at com.mysql.cj.jdbc.exceptions.SQLError.createSQLException(SQLError.java:55)
	at com.mysql.cj.jdbc.exceptions.SQLError.createSQLException(SQLError.java:65)
	at com.mysql.cj.jdbc.exceptions.SQLExceptionsMapping.translateException(SQLExceptionsMapping.java:88)
	at com.mysql.cj.jdbc.result.ResultSetImpl.getBigDecimal(ResultSetImpl.java:670)
	at software.amazon.jdbc.wrapper.ResultSetWrapper.lambda$getBigDecimal$15(ResultSetWrapper.java:235)
	at software.amazon.jdbc.plugin.DefaultConnectionPlugin.execute(DefaultConnectionPlugin.java:128)
	at software.amazon.jdbc.ConnectionPluginManager.lambda$execute$5(ConnectionPluginManager.java:348)
	at software.amazon.jdbc.ConnectionPluginManager.lambda$null$0(ConnectionPluginManager.java:270)
	at software.amazon.jdbc.ConnectionPluginManager.executeWithTelemetry(ConnectionPluginManager.java:246)
	at software.amazon.jdbc.ConnectionPluginManager.lambda$makePluginChainFunc$1(ConnectionPluginManager.java:270)
	at software.amazon.jdbc.ConnectionPluginManager.executeWithSubscribedPlugins(ConnectionPluginManager.java:236)
	at software.amazon.jdbc.ConnectionPluginManager.execute(ConnectionPluginManager.java:345)
	at software.amazon.jdbc.util.WrapperUtils.executeWithPlugins(WrapperUtils.java:244)
	at software.amazon.jdbc.wrapper.ResultSetWrapper.getBigDecimal(ResultSetWrapper.java:229)
	at org.apache.spark.sql.execution.datasources.jdbc.JdbcUtils$.$anonfun$makeGetter$3(JdbcUtils.scala:416)
	at org.apache.spark.sql.execution.datasources.jdbc.JdbcUtils$.$anonfun$makeGetter$3$adapted(JdbcUtils.scala:414)
	at org.apache.spark.sql.execution.datasources.jdbc.JdbcUtils$$anon$1.getNext(JdbcUtils.scala:358)
	at org.apache.spark.sql.execution.datasources.jdbc.JdbcUtils$$anon$1.getNext(JdbcUtils.scala:340)
	at org.apache.spark.util.NextIterator.hasNext(NextIterator.scala:73)
	at org.apache.spark.InterruptibleIterator.hasNext(InterruptibleIterator.scala:37)
	at org.apache.spark.util.CompletionIterator.hasNext(CompletionIterator.scala:31)
	at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.hashAgg_doAggregateWithKeys_0$(Unknown Source)
	at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.processNext(Unknown Source)
	at org.apache.spark.sql.execution.UnsafeRowInterceptor.hasNext(UnsafeRowInterceptor.java:24)
	at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.hasNext(Unknown Source)
	at org.apache.spark.sql.execution.WholeStageCodegenEvaluatorFactory$WholeStageCodegenPartitionEvaluator$$anon$1.hasNext(WholeStageCodegenEvaluatorFactory.scala:43)
	at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:460)
	at org.apache.spark.shuffle.sort.UnsafeShuffleWriter.start(UnsafeShuffleWriter.java:229)
	at org.apache.spark.shuffle.DirectShuffleWriteProcessor.doWrite(DirectShuffleWriteProcessor.scala:44)
	at org.apache.spark.shuffle.ShuffleWriteProcessor.write(ShuffleWriteProcessor.scala:69)
	at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:104)
	at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:54)
	at org.apache.spark.TaskContext.runTaskWithListeners(TaskContext.scala:174)
	at org.apache.spark.scheduler.Task.run(Task.scala:152)
	at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$4(Executor.scala:632)
	at org.apache.spark.util.SparkErrorUtils.tryWithSafeFinally(SparkErrorUtils.scala:64)
	at org.apache.spark.util.SparkErrorUtils.tryWithSafeFinally$(SparkErrorUtils.scala:61)
	at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:96)
	at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:635)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
	at java.base/java.lang.Thread.run(Thread.java:840)
Caused by: com.mysql.cj.exceptions.DataConversionException: Cannot determine value type from string 'rno'
	at com.mysql.cj.result.AbstractNumericValueFactory.createFromBytes(AbstractNumericValueFactory.java:56)
	at com.mysql.cj.protocol.a.MysqlTextValueDecoder.decodeByteArray(MysqlTextValueDecoder.java:151)
	at com.mysql.cj.protocol.result.AbstractResultsetRow.decodeAndCreateReturnValue(AbstractResultsetRow.java:127)
	at com.mysql.cj.protocol.result.AbstractResultsetRow.getValueFromBytes(AbstractResultsetRow.java:235)
	at com.mysql.cj.protocol.a.result.ByteArrayRow.getValue(ByteArrayRow.java:82)
	... 39 more

Reproduction Steps

Perform a load and show using SparkSession when running on EMR (serverless).

var query = "(select  id, row_number() over(order by (select null)) as rno from table_name) dataset";

var read =
        sparkSession
            .read()
            .format("jdbc")
            .option("url", jdbcUrl) // in format "jdbc:aws-wrapper:mysql://{HOST}:{PORT}/{DATABASE}"
            .option("user", username)
            .option("dbtable", query)
            .option("driver","software.amazon.jdbc.Driver")
            .option("lowerBound", 1)
            .option("fetchsize", 2000000)
            .option("upperBound", 100)
            .option("numPartitions" 1)
            .option("partitionColumn", "rno")
            .option("wrapperPlugins", "iam") // no password, using iam auth
            .option("wrapperDialect", DialectCodes.AURORA_MYSQL)
            .option("wrapperTargetDriverDialect", TargetDriverDialectCodes.MYSQL_CONNECTOR_J);

var result = read.load();

result.show(); // throws DataConversionException

Possible Solution

No response

Additional Information/Context

emr-7.6.0

The AWS Advanced JDBC Driver version used

2.5.6

JDK version used

17

Operating System and version

n/a

[sergiyvamz]

Hi @cyonglun

Thank you for reaching out with this issue. Did you have a chance to run your application with vanilla com.mysql.cj.jdbc.Driver (no AWS Wrapper)? Do you observe the same issue with such configuration?

Thank you

[cyonglun]

Hi @sergiyvamz

Yea I did test with the driver com.mysql.cj.jdbc.Driver and the query did not encounter this issue.

Thanks you

[mobiawa]

Same Issue here.
I've tried several options or driver versions but nothing seems to work except switching to com.mysql.cj.jdbc.Driver

[etcl]

The problem is not on the wrapper, is on the dialect in spark

https://github.com/apache/spark/blob/master/sql/core/src/main/scala/org/apache/spark/sql/jdbc/MySQLDialect.scala#L40

probably the best option will be a pull request to spark to allows the MySQLDialect to handle the prefix jdbc:aws-wrapper:mysql but as a hotfix, what I did is create a wrapper of the MySQLDialect with the aws-wrapper prefix an register on the spark session

package com.xing.datahub.jdbcexporter

import org.apache.spark.sql.AnalysisException
import org.apache.spark.sql.connector.catalog.Identifier
import org.apache.spark.sql.connector.catalog.index.TableIndex
import org.apache.spark.sql.connector.expressions.NamedReference
import org.apache.spark.sql.execution.datasources.jdbc.JDBCOptions
import org.apache.spark.sql.jdbc._
import org.apache.spark.sql.types._
import org.apache.spark.sql.connector.expressions.Expression

import java.sql.Connection
import java.util

object AwsWrapperMySQLDialect extends JdbcDialect {

  private val wrapper = JdbcDialects.get("jdbc:mysql")

  override def canHandle(url: String): Boolean = url.startsWith("jdbc:aws-wrapper:mysql")

  override def isSupportedFunction(funcName: String): Boolean      = wrapper.isSupportedFunction(funcName)
  override def compileExpression(expr: Expression): Option[String] = wrapper.compileExpression(expr)
  override def getCatalystType(sqlType: Int, typeName: String, size: Int, md: MetadataBuilder): Option[DataType] =
    wrapper.getCatalystType(sqlType, typeName, size, md)
  override def quoteIdentifier(colName: String): String = wrapper.quoteIdentifier(colName)

  override def schemasExists(conn: Connection, options: JDBCOptions, schema: String): Boolean =
    wrapper.schemasExists(conn, options, schema)
  override def listSchemas(conn: Connection, options: JDBCOptions): Array[Array[String]] =
    wrapper.listSchemas(conn, options)
  override def isCascadingTruncateTable(): Option[Boolean] = wrapper.isCascadingTruncateTable()
  override def getUpdateColumnTypeQuery(tableName: String, columnName: String, newDataType: String): String =
    wrapper.getUpdateColumnTypeQuery(tableName, columnName, newDataType)

  override def getRenameColumnQuery(
      tableName: String,
      oldColumnName: String,
      newColumnName: String,
      dbMajorVersion: Int
  ): String = wrapper.getRenameColumnQuery(tableName, oldColumnName, newColumnName, dbMajorVersion)

  override def getUpdateColumnNullabilityQuery(tableName: String, columnName: String, nullable: Boolean): String =
    wrapper.getUpdateColumnNullabilityQuery(tableName, columnName, nullable)
  override def getTableCommentQuery(table: String, comment: String): String =
    wrapper.getTableCommentQuery(table, comment)
  override def getJDBCType(dt: DataType): Option[JdbcType] = wrapper.getJDBCType(dt)
  override def getSchemaCommentQuery(schema: String, comment: String): String =
    wrapper.getSchemaCommentQuery(schema, comment)
  override def removeSchemaCommentQuery(schema: String): String = wrapper.removeSchemaCommentQuery(schema)

  override def createIndex(
      indexName: String,
      tableIdent: Identifier,
      columns: Array[NamedReference],
      columnsProperties: util.Map[NamedReference, util.Map[String, String]],
      properties: util.Map[String, String]
  ): String = wrapper.createIndex(indexName, tableIdent, columns, columnsProperties, properties)
  override def indexExists(conn: Connection, indexName: String, tableIdent: Identifier, options: JDBCOptions): Boolean =
    wrapper.indexExists(conn, indexName, tableIdent, options)
  override def dropIndex(indexName: String, tableIdent: Identifier): String = dropIndex(indexName, tableIdent)
  override def listIndexes(conn: Connection, tableIdent: Identifier, options: JDBCOptions): Array[TableIndex] =
    wrapper.listIndexes(conn, tableIdent, options)
  override def classifyException(message: String, e: Throwable): AnalysisException =
    wrapper.classifyException(message, e)
  override def dropSchema(schema: String, cascade: Boolean): String = wrapper.dropSchema(schema, cascade)
  override def getJdbcSQLQueryBuilder(options: JDBCOptions): JdbcSQLQueryBuilder =
    wrapper.getJdbcSQLQueryBuilder(options)
  override def supportsLimit: Boolean  = wrapper.supportsLimit
  override def supportsOffset: Boolean = wrapper.supportsOffset

}

    JdbcDialects.registerDialect(AwsWrapperMySQLDialect)

[karenc-bq]

Thank you @etcl for the workaround. We will add this to our documentation and reach out to Spark about wrapper adoption.