[SPARK-22000][SQL] Address missing Upcast in JavaTypeInference.deserializerFor

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/JavaTypeInference.scala

@@ -211,7 +211,8 @@ object JavaTypeInference {
                 c == classOf[java.lang.Double] ||
                 c == classOf[java.lang.Float] ||
                 c == classOf[java.lang.Byte] ||
-                c == classOf[java.lang.Boolean] =>
+                c == classOf[java.lang.Boolean] ||
+                c == classOf[java.lang.String] =>
         StaticInvoke(
           c,
           ObjectType(c),
@@ -235,9 +236,6 @@ object JavaTypeInference {
           path :: Nil,
           returnNullable = false)
 
-      case c if c == classOf[java.lang.String] =>
-        Invoke(path, "toString", ObjectType(classOf[String]))
-
       case c if c == classOf[java.math.BigDecimal] =>
         Invoke(path, "toJavaBigDecimal", ObjectType(classOf[java.math.BigDecimal]))
 

sql/core/src/test/java/test/org/apache/spark/sql/JavaBeanDeserializationSuite.java

@@ -20,6 +20,15 @@
 import java.io.Serializable;
 import java.util.*;
 
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.catalyst.expressions.GenericInternalRow;
+import org.apache.spark.sql.catalyst.expressions.GenericRow;
+import org.apache.spark.sql.catalyst.expressions.UnsafeRow;
+import org.apache.spark.sql.catalyst.expressions.UnsafeRowConverterSuite;
+import org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter;
+import org.apache.spark.sql.types.DataTypes;
+import org.apache.spark.sql.types.StructType;
+import org.apache.spark.unsafe.types.UTF8String;
 import org.junit.*;
 
 import org.apache.spark.sql.Dataset;
@@ -115,6 +124,70 @@ public void testBeanWithMapFieldsDeserialization() {
     Assert.assertEquals(records, MAP_RECORDS);
   }
 
+  private static final List<Row> ROWS_SPARK_22000 = new ArrayList<>();
+  private static final List<RecordSpark22000> RECORDS_SPARK_22000 = new ArrayList<>();
+
+  private static Row createRecordSpark22000Row(Long index) {
+    Object[] values = new Object[] {
+      index.shortValue(),
+      index.intValue(),
+      index,
+      index.floatValue(),
+      index.doubleValue(),
+      String.valueOf(index),
+      index % 2 == 0,
+      new java.sql.Timestamp(System.currentTimeMillis())
+    };
+    return new GenericRow(values);
+  }
+
+  private static RecordSpark22000 createRecordSpark22000(Row recordRow) {
+    RecordSpark22000 record = new RecordSpark22000();
+    record.setShortField(String.valueOf(recordRow.getShort(0)));
+    record.setIntField(String.valueOf(recordRow.getInt(1)));
+    record.setLongField(String.valueOf(recordRow.getLong(2)));
+    record.setFloatField(String.valueOf(recordRow.getFloat(3)));
+    record.setDoubleField(String.valueOf(recordRow.getDouble(4)));
+    record.setStringField(recordRow.getString(5));
+    record.setBooleanField(String.valueOf(recordRow.getBoolean(6)));
+    record.setTimestampField(String.valueOf(recordRow.getTimestamp(7).getTime() * 1000));
+    return record;
+  }
+
+  static {
+    for (long idx = 0 ; idx < 5 ; idx++) {
+      Row row = createRecordSpark22000Row(idx);
+      ROWS_SPARK_22000.add(row);
+      RECORDS_SPARK_22000.add(createRecordSpark22000(row));
+    }
+  }
+
+  @Test
+  public void testSpark22000() {
+    // Here we try to convert the fields, from any types to string.
+    // Before applying SPARK-22000, Spark called toString() against variable which type might be primitive.
+    // SPARK-22000 it calls String.valueOf() which finally calls toString() but handles boxing
+    // if the type is primitive.
+    Encoder<RecordSpark22000> encoder = Encoders.bean(RecordSpark22000.class);
+
+    StructType schema = new StructType()
+      .add("shortField", DataTypes.ShortType)
+      .add("intField", DataTypes.IntegerType)
+      .add("longField", DataTypes.LongType)
+      .add("floatField", DataTypes.FloatType)
+      .add("doubleField", DataTypes.DoubleType)
+      .add("stringField", DataTypes.StringType)
+      .add("booleanField", DataTypes.BooleanType)
+      .add("timestampField", DataTypes.TimestampType);
+
+    Dataset<Row> dataFrame = spark.createDataFrame(ROWS_SPARK_22000, schema);
+    Dataset<RecordSpark22000> dataset = dataFrame.as(encoder);
+
+    List<RecordSpark22000> records = dataset.collectAsList();
+
+    Assert.assertEquals(RECORDS_SPARK_22000, records);
+  }
+
   public static class ArrayRecord {
 
     private int id;
@@ -252,4 +325,116 @@ public String toString() {
       return String.format("[%d,%d]", startTime, endTime);
     }
   }
+
+  public static class RecordSpark22000 {
+    private String shortField;
+    private String intField;
+    private String longField;
+    private String floatField;
+    private String doubleField;
+    private String stringField;
+    private String booleanField;
+    private String timestampField;
+
+    public RecordSpark22000() { }
+
+    public String getShortField() {
+      return shortField;
+    }
+
+    public void setShortField(String shortField) {
+      this.shortField = shortField;
+    }
+
+    public String getIntField() {
+      return intField;
+    }
+
+    public void setIntField(String intField) {
+      this.intField = intField;
+    }
+
+    public String getLongField() {
+      return longField;
+    }
+
+    public void setLongField(String longField) {
+      this.longField = longField;
+    }
+
+    public String getFloatField() {
+      return floatField;
+    }
+
+    public void setFloatField(String floatField) {
+      this.floatField = floatField;
+    }
+
+    public String getDoubleField() {
+      return doubleField;
+    }
+
+    public void setDoubleField(String doubleField) {
+      this.doubleField = doubleField;
+    }
+
+    public String getStringField() {
+      return stringField;
+    }
+
+    public void setStringField(String stringField) {
+      this.stringField = stringField;
+    }
+
+    public String getBooleanField() {
+      return booleanField;
+    }
+
+    public void setBooleanField(String booleanField) {
+      this.booleanField = booleanField;
+    }
+
+    public String getTimestampField() {
+      return timestampField;
+    }
+
+    public void setTimestampField(String timestampField) {
+      this.timestampField = timestampField;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+      if (this == o) return true;
+      if (o == null || getClass() != o.getClass()) return false;
+      RecordSpark22000 that = (RecordSpark22000) o;
+      return Objects.equals(shortField, that.shortField) &&
+              Objects.equals(intField, that.intField) &&
+              Objects.equals(longField, that.longField) &&
+              Objects.equals(floatField, that.floatField) &&
+              Objects.equals(doubleField, that.doubleField) &&
+              Objects.equals(stringField, that.stringField) &&
+              Objects.equals(booleanField, that.booleanField) &&
+              Objects.equals(timestampField, that.timestampField);
+    }
+
+    @Override
+    public int hashCode() {
+      return Objects.hash(shortField, intField, longField, floatField, doubleField, stringField,
+              booleanField, timestampField);
+    }
+
+    @Override
+    public String toString() {
+      return com.google.common.base.Objects.toStringHelper(this)
+              .add("shortField", shortField)
+              .add("intField", intField)
+              .add("longField", longField)
+              .add("floatField", floatField)
+              .add("doubleField", doubleField)
+              .add("stringField", stringField)
+              .add("booleanField", booleanField)
+              .add("timestampField", timestampField)
+              .toString();
+    }
+  }
 }