Github user cloud-fan commented on a diff in the pull request:
https://github.com/apache/spark/pull/22814#discussion_r228061155
--- Diff:
external/avro/src/main/scala/org/apache/spark/sql/avro/AvroDataToCatalyst.scala
---
@@ -44,24 +59,74 @@ case class AvroDataToCatalyst(child: Expression,
jsonFormatSchema: String)
@transient private var result: Any = _
+ @transient private lazy val parseMode: ParseMode = {
+ val mode = AvroOptions(options).parseMode
+ if (mode != PermissiveMode && mode != FailFastMode) {
+ throw new AnalysisException(unacceptableModeMessage(mode.name))
+ }
+ mode
+ }
+
+ private def unacceptableModeMessage(name: String): String = {
+ s"from_avro() doesn't support the $name mode. " +
+ s"Acceptable modes are ${PermissiveMode.name} and
${FailFastMode.name}."
+ }
+
+ @transient private lazy val nullResultRow: Any = dataType match {
+ case st: StructType =>
+ val resultRow = new SpecificInternalRow(st.map(_.dataType))
+ for(i <- 0 until st.length) {
+ resultRow.setNullAt(i)
+ }
+ resultRow
+
+ case _ =>
+ null
+ }
+
+
override def nullSafeEval(input: Any): Any = {
val binary = input.asInstanceOf[Array[Byte]]
- decoder = DecoderFactory.get().binaryDecoder(binary, 0, binary.length,
decoder)
- result = reader.read(result, decoder)
- deserializer.deserialize(result)
+ try {
+ decoder = DecoderFactory.get().binaryDecoder(binary, 0,
binary.length, decoder)
+ result = reader.read(result, decoder)
+ deserializer.deserialize(result)
+ } catch {
+ // There could be multiple possible exceptions here, e.g.
java.io.IOException,
+ // AvroRuntimeException, ArrayIndexOutOfBoundsException, etc.
+ // To make it simple, catch all the exceptions here.
+ case e: Exception => parseMode match {
--- End diff --
we should catch `NonFatal` to be safer
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
