address more comments

Author: gengliangwang

external/avro/src/main/scala/org/apache/spark/sql/avro/AvroDeserializer.scala

@@ -96,6 +96,8 @@ class AvroDeserializer(rootAvroType: Schema, rootCatalystType: DataType) {
           // For backward compatibility, if the Avro type is Long and it is not logical type,
           // the value is processed as timestamp type with millisecond precision.
           updater.setLong(ordinal, value.asInstanceOf[Long] * 1000)
+        case other => throw new IncompatibleSchemaException(
+          s"Cannot convert Avro logical type ${other} to Catalyst Timestamp type.")
       }
 
       case (LONG, DateType) => (updater, ordinal, value) =>

external/avro/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala

@@ -93,13 +93,14 @@ class AvroSerializer(rootCatalystType: DataType, rootAvroType: Schema, nullable:
         (getter, ordinal) => ByteBuffer.wrap(getter.getBinary(ordinal))
       case DateType =>
         (getter, ordinal) => getter.getInt(ordinal) * DateTimeUtils.MILLIS_PER_DAY
-      case TimestampType =>
-        (getter, ordinal) => avroType.getLogicalType match {
-          case _: TimestampMillis => getter.getLong(ordinal) / 1000
-          case _: TimestampMicros => getter.getLong(ordinal)
+      case TimestampType => avroType.getLogicalType match {
+          case _: TimestampMillis => (getter, ordinal) => getter.getLong(ordinal) / 1000
+          case _: TimestampMicros => (getter, ordinal) => getter.getLong(ordinal)
           // For backward compatibility, if the Avro type is Long and it is not logical type,
           // output the timestamp value as with millisecond precision.
-          case null => getter.getLong(ordinal) / 1000
+          case null => (getter, ordinal) => getter.getLong(ordinal) / 1000
+          case other => throw new IncompatibleSchemaException(
+            s"Cannot convert Catalyst Timestamp type to Avro logical type ${other}")
         }
 
       case ArrayType(et, containsNull) =>

external/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala

@@ -36,7 +36,7 @@ import org.apache.spark.sql._
 import org.apache.spark.sql.execution.datasources.DataSource
 import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.test.{SharedSQLContext, SQLTestUtils}
-import org.apache.spark.sql.types.{StructType, _}
+import org.apache.spark.sql.types._
 
 class AvroSuite extends QueryTest with SharedSQLContext with SQLTestUtils {
   val episodesAvro = testFile("episodes.avro")