In the current implementation of Spark when working with arrays of data returned from ClickHouse through the clickhouse-java driver, there is a type conversion error. When the driver returns an array of primitive types (e.g., bytes or integers), Spark attempts to cast them to an array of objects, which is not possible in Java and Scala, as arrays of primitives are not subtypes of arrays of objects. This leads to a ClassCastException.
Steps to reproduce
CREATE TABLE statements for tables involved:
CREATE TABLE example_table (
id UInt32,
data Array(Int32)
) ENGINE = MergeTree()
ORDER BY id;
INSERT INTO example_table VALUES (1, [1, 2, 3]), (2, [4, 5, 6]);
At first we implement custom Clickhouse dialect that handles Array(T) because in native Spark, Array(T) is unsupported type
import scala.util.matching.Regex
import org.apache.spark.sql.jdbc.{JdbcDialect, JdbcType}
import org.apache.spark.sql.execution.datasources.jdbc.{JdbcUtils}
import org.apache.spark.sql.types._
import org.slf4j.LoggerFactory
import java.sql.Types
private object ClickhouseDialectExtension extends JdbcDialect {
private val logger = LoggerFactory.getLogger(getClass)
private val arrayTypePattern: Regex = "^Array\\((.*)\\)$".r
override def canHandle(url: String): Boolean = {
url.startsWith("jdbc:clickhouse")
}
override def getCatalystType(
sqlType: Int,
typeName: String,
size: Int,
md: MetadataBuilder): Option[DataType] = {
sqlType match {
case Types.ARRAY =>
arrayTypePattern.findFirstMatchIn(typeName) match {
case Some(m) =>
val elementType = m.group(1)
JdbcUtils.getCommonJDBCType(elementType).map(dt => ArrayType(dt))
case None => None
}
case _ => None
}
}
override def getJDBCType(dt: DataType): Option[JdbcType] = dt match {
case ArrayType(et, _) =>
logger.debug("Custom mapping applied: Array[T_1] for ArrayType(T_0)")
getJDBCType(et)
.orElse(JdbcUtils.getCommonJDBCType(et))
.map(jdbcType => JdbcType(s"Array(${jdbcType.databaseTypeDefinition})", Types.ARRAY))
case _ => None
}
}
val spark = SparkSession
.builder()
.master("local[*]")
.appName("Spark Clickhouse Dialect Test Session")
.config("spark.jars", jarPaths) // include the JAR file containing the compiled custom dialect
.getOrCreate()
// Register custom Clickhouse dialect
JdbcDialects.registerDialect(ClickhouseDialectExtension)
val df = spark.read
.format("jdbc")
.option("url", s"jdbc:clickhouse://$jdbcHostname:${jdbcPort}/$database")
.option("dbtable", "example_table")
.load()
df.collect() // raises error: java.lang.ClassCastException: [B cannot be cast to [Ljava.lang.Object;
Expected behaviour
The driver should return an array of objects instead of an array of primitives to avoid ClassCastException in Spark.
Describe the bug
In the current implementation of Spark when working with arrays of data returned from ClickHouse through the clickhouse-java driver, there is a type conversion error. When the driver returns an array of primitive types (e.g., bytes or integers), Spark attempts to cast them to an array of objects, which is not possible in Java and Scala, as arrays of primitives are not subtypes of arrays of objects. This leads to a
ClassCastException
.Steps to reproduce
CREATE TABLE
statements for tables involved:At first we implement custom Clickhouse dialect that handles
Array(T)
because in native Spark,Array(T)
is unsupported typeExpected behaviour
The driver should return an array of objects instead of an array of primitives to avoid
ClassCastException
in Spark.Code example
Example code snippet in Spark that demonstrates the issue:
Error log
Configuration
Environment
ClickHouse server
clickhouse-server:latest-alpine