marmbrus's comments.

Author: yhuai

mllib/src/main/scala/org/apache/spark/mllib/util/modelSaveLoad.scala

@@ -110,7 +110,7 @@ private[mllib] object Loader {
       assert(loadedFields.contains(field.name), s"Unable to parse model data." +
         s"  Expected field with name ${field.name} was missing in loaded schema:" +
         s" ${loadedFields.mkString(", ")}")
-      assert(DataType.equalsIgnoreNullability(loadedFields(field.name), field.dataType),
+      assert(loadedFields(field.name).sameType(field.dataType),
         s"Unable to parse model data.  Expected field $field but found field" +
           s" with different type: ${loadedFields(field.name)}")
     }

sql/catalyst/src/main/scala/org/apache/spark/sql/types/dataTypes.scala

@@ -181,7 +181,7 @@ object DataType {
   /**
    * Compares two types, ignoring nullability of ArrayType, MapType, StructType.
    */
-  private[spark] def equalsIgnoreNullability(left: DataType, right: DataType): Boolean = {
+  private[types] def equalsIgnoreNullability(left: DataType, right: DataType): Boolean = {
     (left, right) match {
       case (ArrayType(leftElementType, _), ArrayType(rightElementType, _)) =>
         equalsIgnoreNullability(leftElementType, rightElementType)
@@ -213,7 +213,7 @@ object DataType {
    *   if and only if for all every pair of fields, `to.nullable` is true, or both
    *   of `fromField.nullable` and `toField.nullable` are false.
    */
-  private[spark] def equalsIgnoreCompatibleNullability(from: DataType, to: DataType): Boolean = {
+  private[sql] def equalsIgnoreCompatibleNullability(from: DataType, to: DataType): Boolean = {
     (from, to) match {
       case (ArrayType(fromElement, fn), ArrayType(toElement, tn)) =>
         (tn || !fn) && equalsIgnoreCompatibleNullability(fromElement, toElement)
@@ -235,20 +235,6 @@ object DataType {
       case (fromDataType, toDataType) => fromDataType == toDataType
     }
   }
-
-  /** Sets all nullable/containsNull/valueContainsNull to true. */
-  private[spark] def alwaysNullable(dataType: DataType): DataType = dataType match {
-    case ArrayType(elementType, _) =>
-      ArrayType(alwaysNullable(elementType), containsNull = true)
-    case MapType(keyType, valueType, _) =>
-      MapType(alwaysNullable(keyType), alwaysNullable(valueType), valueContainsNull = true)
-    case StructType(fields) =>
-      val newFields = fields.map { field =>
-        StructField(field.name, alwaysNullable(field.dataType), nullable = true)
-      }
-      StructType(newFields)
-    case other => other
-  }
 }
 
 
@@ -281,6 +267,16 @@ abstract class DataType {
   def prettyJson: String = pretty(render(jsonValue))
 
   def simpleString: String = typeName
+
+  /** Check if `this` and `other` are the same data type when ignoring nullability
+   *  (`StructField.nullable`, `ArrayType.containsNull`, and `MapType.valueContainsNull`).
+   */
+  def sameType(other: DataType): Boolean = DataType.equalsIgnoreNullability(this, other)
+
+  /** Returns the same data type but set all nullability fields are true
+   * (`StructField.nullable`, `ArrayType.containsNull`, and `MapType.valueContainsNull`).
+   */
+  def asNullable: DataType
 }
 
 /**
@@ -296,6 +292,8 @@ class NullType private() extends DataType {
   // this type. Otherwise, the companion object would be of type "NullType$" in byte code.
   // Defined with a private constructor so the companion object is the only possible instantiation.
   override def defaultSize: Int = 1
+
+  override def asNullable: NullType = this
 }
 
 case object NullType extends NullType
@@ -361,6 +359,8 @@ class StringType private() extends NativeType with PrimitiveType {
    * The default size of a value of the StringType is 4096 bytes.
    */
   override def defaultSize: Int = 4096
+
+  override def asNullable: StringType = this
 }
 
 case object StringType extends StringType
@@ -395,6 +395,8 @@ class BinaryType private() extends NativeType with PrimitiveType {
    * The default size of a value of the BinaryType is 4096 bytes.
    */
   override def defaultSize: Int = 4096
+
+  override def asNullable: BinaryType = this
 }
 
 case object BinaryType extends BinaryType
@@ -420,6 +422,8 @@ class BooleanType private() extends NativeType with PrimitiveType {
    * The default size of a value of the BooleanType is 1 byte.
    */
   override def defaultSize: Int = 1
+
+  override def asNullable: BooleanType = this
 }
 
 case object BooleanType extends BooleanType
@@ -450,6 +454,8 @@ class TimestampType private() extends NativeType {
    * The default size of a value of the TimestampType is 12 bytes.
    */
   override def defaultSize: Int = 12
+
+  override def asNullable: TimestampType = this
 }
 
 case object TimestampType extends TimestampType
@@ -478,6 +484,8 @@ class DateType private() extends NativeType {
    * The default size of a value of the DateType is 4 bytes.
    */
   override def defaultSize: Int = 4
+
+  override def asNullable: DateType = this
 }
 
 case object DateType extends DateType
@@ -536,6 +544,8 @@ class LongType private() extends IntegralType {
   override def defaultSize: Int = 8
 
   override def simpleString = "bigint"
+
+  override def asNullable: LongType = this
 }
 
 case object LongType extends LongType
@@ -565,6 +575,8 @@ class IntegerType private() extends IntegralType {
   override def defaultSize: Int = 4
 
   override def simpleString = "int"
+
+  override def asNullable: IntegerType = this
 }
 
 case object IntegerType extends IntegerType
@@ -594,6 +606,8 @@ class ShortType private() extends IntegralType {
   override def defaultSize: Int = 2
 
   override def simpleString = "smallint"
+
+  override def asNullable: ShortType = this
 }
 
 case object ShortType extends ShortType
@@ -623,6 +637,8 @@ class ByteType private() extends IntegralType {
   override def defaultSize: Int = 1
 
   override def simpleString = "tinyint"
+
+  override def asNullable: ByteType = this
 }
 
 case object ByteType extends ByteType
@@ -689,6 +705,8 @@ case class DecimalType(precisionInfo: Option[PrecisionInfo]) extends FractionalT
     case Some(PrecisionInfo(precision, scale)) => s"decimal($precision,$scale)"
     case None => "decimal(10,0)"
   }
+
+  override def asNullable: DecimalType = this
 }
 
 
@@ -747,6 +765,8 @@ class DoubleType private() extends FractionalType {
    * The default size of a value of the DoubleType is 8 bytes.
    */
   override def defaultSize: Int = 8
+
+  override def asNullable: DoubleType = this
 }
 
 case object DoubleType extends DoubleType
@@ -775,6 +795,8 @@ class FloatType private() extends FractionalType {
    * The default size of a value of the FloatType is 4 bytes.
    */
   override def defaultSize: Int = 4
+
+  override def asNullable: FloatType = this
 }
 
 case object FloatType extends FloatType
@@ -823,6 +845,8 @@ case class ArrayType(elementType: DataType, containsNull: Boolean) extends DataT
   override def defaultSize: Int = 100 * elementType.defaultSize
 
   override def simpleString = s"array<${elementType.simpleString}>"
+
+  override def asNullable: ArrayType = ArrayType(elementType.asNullable, containsNull = true)
 }
 
 
@@ -1068,6 +1092,15 @@ case class StructType(fields: Array[StructField]) extends DataType with Seq[Stru
    */
   private[sql] def merge(that: StructType): StructType =
     StructType.merge(this, that).asInstanceOf[StructType]
+
+  override def asNullable: StructType = {
+    val newFields = fields.map {
+      case StructField(name, dataType, nullable, metadata) =>
+        StructField(name, dataType.asNullable, nullable = true, metadata)
+    }
+
+    StructType(newFields)
+  }
 }
 
 
@@ -1120,6 +1153,9 @@ case class MapType(
   override def defaultSize: Int = 100 * (keyType.defaultSize + valueType.defaultSize)
 
   override def simpleString = s"map<${keyType.simpleString},${valueType.simpleString}>"
+
+  override def asNullable: MapType =
+    MapType(keyType.asNullable, valueType.asNullable, valueContainsNull = true)
 }
 
 
@@ -1173,4 +1209,8 @@ abstract class UserDefinedType[UserType] extends DataType with Serializable {
    * The default size of a value of the UserDefinedType is 4096 bytes.
    */
   override def defaultSize: Int = 4096
+
+  override def sameType(other: DataType): Boolean = ???
+
+  override def asNullable: DataType = ???
 }

sql/core/src/main/scala/org/apache/spark/sql/json/JSONRelation.scala

@@ -131,7 +131,7 @@ private[sql] case class JSONRelation(
 
   override def equals(other: Any): Boolean = other match {
     case that: JSONRelation =>
-      (this.path == that.path) && (DataType.equalsIgnoreNullability(this.schema, that.schema))
+      (this.path == that.path) && this.schema.sameType(that.schema)
     case _ => false
   }
 }

sql/core/src/main/scala/org/apache/spark/sql/parquet/ParquetRelation.scala

@@ -175,8 +175,7 @@ private[sql] object ParquetRelation {
     ParquetRelation.enableLogForwarding()
     // This is a hack. We always set nullable/containsNull/valueContainsNull to true
     // for the schema of a parquet data.
-    val schema =
-      DataType.alwaysNullable(StructType.fromAttributes(attributes)).asInstanceOf[StructType]
+    val schema = StructType.fromAttributes(attributes).asNullable
     val newAttributes = schema.toAttributes
     ParquetTypesConverter.writeMetaData(newAttributes, path, conf)
     new ParquetRelation(path.toString, Some(conf), sqlContext) {

sql/core/src/main/scala/org/apache/spark/sql/parquet/ParquetTableOperations.scala

@@ -280,8 +280,7 @@ private[sql] case class InsertIntoParquetTable(
     val conf = ContextUtil.getConfiguration(job)
     // This is a hack. We always set nullable/containsNull/valueContainsNull to true
     // for the schema of a parquet data.
-    val schema =
-      DataType.alwaysNullable(StructType.fromAttributes(relation.output)).asInstanceOf[StructType]
+    val schema = StructType.fromAttributes(relation.output).asNullable
     RowWriteSupport.setSchema(schema.toAttributes, conf)
 
     val fspath = new Path(relation.path)

sql/core/src/main/scala/org/apache/spark/sql/parquet/newParquet.scala

@@ -120,7 +120,7 @@ private[sql] class DefaultSource
       val df =
         sqlContext.createDataFrame(
           data.queryExecution.toRdd,
-          DataType.alwaysNullable(data.schema).asInstanceOf[StructType])
+          data.schema.asNullable)
       val createdRelation =
         createRelation(sqlContext, parameters, df.schema).asInstanceOf[ParquetRelation2]
       createdRelation.insert(df, overwrite = mode == SaveMode.Overwrite)
@@ -185,11 +185,11 @@ private[sql] case class ParquetRelation2(
       val schemaEquality = if (shouldMergeSchemas) {
         shouldMergeSchemas == relation.shouldMergeSchemas
       } else {
-        DataType.equalsIgnoreNullability(schema, relation.schema)
+        schema.sameType(relation.schema)
       }
 
       val maybeSchemaEquality = (maybeMetastoreSchema, relation.maybeMetastoreSchema) match {
-        case (Some(left), Some(right)) => DataType.equalsIgnoreNullability(left, right)
+        case (Some(left), Some(right)) => left.sameType(right)
         case (left, right) => left == right
       }
 

sql/core/src/main/scala/org/apache/spark/sql/sources/rules.scala

@@ -56,7 +56,7 @@ private[sql] object PreInsertCastAndRename extends Rule[LogicalPlan] {
       child: LogicalPlan) = {
     val newChildOutput = expectedOutput.zip(child.output).map {
       case (expected, actual) =>
-        val needCast = !DataType.equalsIgnoreNullability(expected.dataType, actual.dataType)
+        val needCast = !expected.dataType.sameType(actual.dataType)
         // We want to make sure the filed names in the data to be inserted exactly match
         // names in the schema.
         val needRename = expected.name != actual.name

sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveMetastoreCatalog.scala

@@ -634,7 +634,7 @@ private[hive] class HiveMetastoreCatalog(hive: HiveContext) extends Catalog with
         p
       } else if (childOutputDataTypes.size == tableOutputDataTypes.size &&
         childOutputDataTypes.zip(tableOutputDataTypes)
-          .forall { case (left, right) => DataType.equalsIgnoreNullability(left, right) }) {
+          .forall { case (left, right) => left.sameType(right) }) {
         // If both types ignoring nullability of ArrayType, MapType, StructType are the same,
         // use InsertIntoHiveTable instead of InsertIntoTable.
         InsertIntoHiveTable(p.table, p.partition, p.child, p.overwrite)
@@ -682,8 +682,7 @@ private[hive] case class InsertIntoHiveTable(
   override def output = child.output
 
   override lazy val resolved = childrenResolved && child.output.zip(table.output).forall {
-    case (childAttr, tableAttr) =>
-      DataType.equalsIgnoreNullability(childAttr.dataType, tableAttr.dataType)
+    case (childAttr, tableAttr) => childAttr.dataType.sameType(tableAttr.dataType)
   }
 }