xsys created SPARK-40409:
----------------------------
Summary: IncompatibleSchemaException when BYTE stored from
DataFrame to Avro is read using spark-sql
Key: SPARK-40409
URL: https://issues.apache.org/jira/browse/SPARK-40409
Project: Spark
Issue Type: Bug
Components: Input/Output
Affects Versions: 3.2.1
Reporter: xsys
h3. Describe the bug
We are trying to store a BYTE {{"-128"}} to a table created via Spark
DataFrame. The table is created with the Avro file format. We encounter no
errors while creating the table and inserting the aforementioned BYTE value.
However, performing a SELECT query on the table through spark-sql results in an
{{IncompatibleSchemaException}} as shown below:
{code:java}
2022-09-09 21:15:03,248 ERROR executor.Executor: Exception in task 0.0 in stage
0.0 (TID 0)
org.apache.spark.sql.avro.IncompatibleSchemaException: Cannot convert Avro type
{"type":"record","name":"topLevelRecord","fields"$
[{"name":"c1","type":["int","null"]}]} to SQL type STRUCT<`c1`: TINYINT>{code}
h3. Step to reproduce
On Spark 3.2.1 (commit {{{}4f25b3f712{}}}), using {{spark-shell}} with the Avro
package:
{code:java}
./bin/spark-shell --packages org.apache.spark:spark-avro_2.12:3.2.1{code}
Execute the following:
{code:java}
import org.apache.spark.sql.{Row, SparkSession}
import org.apache.spark.sql.types._
val rdd = sc.parallelize(Seq(Row(("-128").toByte)))
val schema = new StructType().add(StructField("c1", ByteType, true))
val df = spark.createDataFrame(rdd, schema)
df.show(false)
df.write.mode("overwrite").format("avro").saveAsTable("byte_avro"){code}
On Spark 3.2.1 (commit {{{}4f25b3f712{}}}), using {{spark-sql}} with the Avro
package:
{code:java}
./bin/spark-sql --packages org.apache.spark:spark-avro_2.12:3.2.1{code}
Execute the following:
{code:java}
spark-sql> select * from byte_avro;{code}
h3. Expected behavior
We expect the output of the {{SELECT}} query to be {{{}-128{}}}. Additionally,
we expect the data type to be preserved (it is changed from BYTE/TINYINT to
INT, hence the mismatch). We tried other formats like ORC and the outcome is
consistent with this expectation. Here are the logs from our attempt at doing
the same with ORC:
{code:java}
scala> df.write.mode("overwrite").format("orc").saveAsTable("byte_orc")
2022-09-09 21:38:28,880 WARN conf.HiveConf: HiveConf of name
hive.stats.jdbc.timeout does not exist
2022-09-09 21:38:28,880 WARN conf.HiveConf: HiveConf of name
hive.stats.retries.wait does not exist
2022-09-09 21:38:34,642 WARN session.SessionState: METASTORE_FILTER_HOOK will
be ignored, since hive.security.authorization.manage
r is set to instance of HiveAuthorizerFactory.
2022-09-09 21:38:34,716 WARN conf.HiveConf: HiveConf of name
hive.internal.ss.authz.settings.applied.marker does not exist
2022-09-09 21:38:34,716 WARN conf.HiveConf: HiveConf of name
hive.stats.jdbc.timeout does not exist
2022-09-09 21:38:34,716 WARN conf.HiveConf: HiveConf of name
hive.stats.retries.wait does not exist
scala> spark.sql("select * from byte_orc;")
res2: org.apache.spark.sql.DataFrame = [c1: tinyint]
scala> spark.sql("select * from byte_orc;").show(false)
+----+
|c1 |
+----+
|-128|
+----+
{code}
h3. Root Cause
h4.
[AvroSerializer|https://github.com/apache/spark/blob/v3.2.1/external/avro/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala#L114-L119]
{code:java}
(catalystType, avroType.getType) match {
case (NullType, NULL) =>
(getter, ordinal) => null
case (BooleanType, BOOLEAN) =>
(getter, ordinal) => getter.getBoolean(ordinal)
case (ByteType, INT) =>
(getter, ordinal) => getter.getByte(ordinal).toInt
case (ShortType, INT) =>
(getter, ordinal) => getter.getShort(ordinal).toInt
case (IntegerType, INT) =>
(getter, ordinal) => getter.getInt(ordinal){code}
h4.
[AvroDeserializer|https://github.com/apache/spark/blob/v3.2.1/external/avro/src/main/scala/org/apache/spark/sql/avro/AvroDeserializer.scala#L121-L130]
{code:java}
(avroType.getType, catalystType) match {
case (NULL, NullType) => (updater, ordinal, _) =>
updater.setNullAt(ordinal)
// TODO: we can avoid boxing if future version of avro provide primitive
accessors.
case (BOOLEAN, BooleanType) => (updater, ordinal, value) =>
updater.setBoolean(ordinal, value.asInstanceOf[Boolean])
case (INT, IntegerType) => (updater, ordinal, value) =>
updater.setInt(ordinal, value.asInstanceOf[Int])
case (INT, DateType) => (updater, ordinal, value) =>
updater.setInt(ordinal, dateRebaseFunc(value.asInstanceOf[Int]))
{code}
AvroSerializer converts Spark's ByteType into Avro's INT. Further, Spark's
AvroDeserializer expects Avro's INT to map to Spark's IntegerType. The mismatch
between user-specified ByteType & the type AvroDeserializer expects
(IntegerType) is the root cause of this issue.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
