Clean up Exchange code a bit

Author: JoshRosen

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/UnsafeRowConverter.scala

@@ -17,6 +17,8 @@
 
 package org.apache.spark.sql.catalyst.expressions
 
+import org.apache.spark.sql.catalyst.util.DateUtils
+import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.catalyst.util.ObjectPool
 import org.apache.spark.sql.types._
 import org.apache.spark.unsafe.PlatformDependent
@@ -101,6 +103,19 @@ class UnsafeRowConverter(fieldTypes: Array[DataType]) {
 
 }
 
+object UnsafeRowConverter {
+  def supportsSchema(schema: StructType): Boolean = {
+    schema.forall { field =>
+      try {
+        UnsafeColumnWriter.forType(field.dataType)
+        true
+      } catch {
+        case e: UnsupportedOperationException => false
+      }
+    }
+  }
+}
+
 /**
  * Function for writing a column into an UnsafeRow.
  */

sql/core/src/main/scala/org/apache/spark/sql/execution/Exchange.scala

@@ -17,8 +17,6 @@
 
 package org.apache.spark.sql.execution
 
-import scala.util.control.NonFatal
-
 import org.apache.spark.annotation.DeveloperApi
 import org.apache.spark.rdd.{RDD, ShuffledRDD}
 import org.apache.spark.serializer.Serializer
@@ -304,16 +302,8 @@ private[sql] case class EnsureRequirements(sqlContext: SQLContext) extends Rule[
         }
 
         val withSort = if (needSort) {
-          // TODO(josh): this is a hack. Need a better way to determine whether UnsafeRow
-          // supports the given schema.
-          val supportsUnsafeRowConversion: Boolean = try {
-            new UnsafeRowConverter(withShuffle.schema.map(_.dataType).toArray)
-            true
-          } catch {
-            case NonFatal(e) =>
-              false
-          }
-          if (sqlContext.conf.unsafeEnabled && supportsUnsafeRowConversion) {
+          if (sqlContext.conf.unsafeEnabled
+              && UnsafeRowConverter.supportsSchema(withShuffle.schema)) {
             UnsafeExternalSort(rowOrdering, global = false, withShuffle)
           } else if (sqlContext.conf.externalSortEnabled) {
             ExternalSort(rowOrdering, global = false, withShuffle)