[SPARK-29783][SQL] Support SQL Standard/ISO_8601 output style for interval type

common/unsafe/src/main/java/org/apache/spark/unsafe/types/CalendarInterval.java

@@ -18,7 +18,6 @@
 package org.apache.spark.unsafe.types;
 
 import java.io.Serializable;
-import java.math.BigDecimal;
 import java.time.Duration;
 import java.time.Period;
 import java.time.temporal.ChronoUnit;
@@ -80,39 +79,8 @@ public int compareTo(CalendarInterval that) {
 
   @Override
   public String toString() {
-    if (months == 0 && days == 0 && microseconds == 0) {
-      return "0 seconds";
-    }
-
-    StringBuilder sb = new StringBuilder();
-
-    if (months != 0) {
-      appendUnit(sb, months / 12, "years");
-      appendUnit(sb, months % 12, "months");
-    }
-
-    appendUnit(sb, days, "days");
-
-    if (microseconds != 0) {
-      long rest = microseconds;
-      appendUnit(sb, rest / MICROS_PER_HOUR, "hours");
-      rest %= MICROS_PER_HOUR;
-      appendUnit(sb, rest / MICROS_PER_MINUTE, "minutes");
-      rest %= MICROS_PER_MINUTE;
-      if (rest != 0) {
-        String s = BigDecimal.valueOf(rest, 6).stripTrailingZeros().toPlainString();
-        sb.append(s).append(" seconds ");
-      }
-    }
-
-    sb.setLength(sb.length() - 1);
-    return sb.toString();
-  }
-
-  private void appendUnit(StringBuilder sb, long value, String unit) {
-    if (value != 0) {
-      sb.append(value).append(' ').append(unit).append(' ');
-    }
+    return "CalendarInterval(months= " + months + ", days = " + days + ", microsecond = " +
+      microseconds + ")";
   }
 
   /**

common/unsafe/src/test/java/org/apache/spark/unsafe/types/CalendarIntervalSuite.java

@@ -46,36 +46,6 @@ public void equalsTest() {
     assertEquals(i1, i6);
   }
 
-  @Test
-  public void toStringTest() {
-    CalendarInterval i;
-
-    i = new CalendarInterval(0, 0, 0);
-    assertEquals("0 seconds", i.toString());
-
-    i = new CalendarInterval(34, 0, 0);
-    assertEquals("2 years 10 months", i.toString());
-
-    i = new CalendarInterval(-34, 0, 0);
-    assertEquals("-2 years -10 months", i.toString());
-
-    i = new CalendarInterval(0, 31, 0);
-    assertEquals("31 days", i.toString());
-
-    i = new CalendarInterval(0, -31, 0);
-    assertEquals("-31 days", i.toString());
-
-    i = new CalendarInterval(0, 0, 3 * MICROS_PER_HOUR + 13 * MICROS_PER_MINUTE + 123);
-    assertEquals("3 hours 13 minutes 0.000123 seconds", i.toString());
-
-    i = new CalendarInterval(0, 0, -3 * MICROS_PER_HOUR - 13 * MICROS_PER_MINUTE - 123);
-    assertEquals("-3 hours -13 minutes -0.000123 seconds", i.toString());
-
-    i = new CalendarInterval(34, 31, 3 * MICROS_PER_HOUR + 13 * MICROS_PER_MINUTE + 123);
-    assertEquals("2 years 10 months 31 days 3 hours 13 minutes 0.000123 seconds",
-      i.toString());
-  }
-
   @Test
   public void periodAndDurationTest() {
     CalendarInterval interval = new CalendarInterval(120, -40, 123456);

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

@@ -30,7 +30,9 @@ import org.apache.spark.sql.catalyst.expressions.codegen.Block._
 import org.apache.spark.sql.catalyst.util._
 import org.apache.spark.sql.catalyst.util.DateTimeConstants._
 import org.apache.spark.sql.catalyst.util.DateTimeUtils._
+import org.apache.spark.sql.catalyst.util.IntervalUtils._
 import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.internal.SQLConf.IntervalStyle._
 import org.apache.spark.sql.types._
 import org.apache.spark.unsafe.UTF8StringBuilder
 import org.apache.spark.unsafe.types.{CalendarInterval, UTF8String}
@@ -281,6 +283,14 @@ abstract class CastBase extends UnaryExpression with TimeZoneAwareExpression wit
 
   // UDFToString
   private[this] def castToString(from: DataType): Any => Any = from match {
+    case CalendarIntervalType => SQLConf.get.intervalOutputStyle match {
+      case SQL_STANDARD =>
+        buildCast[CalendarInterval](_, i => UTF8String.fromString(toSqlStandardString(i)))
+      case ISO_8601 =>
+        buildCast[CalendarInterval](_, i => UTF8String.fromString(toIso8601String(i)))
+      case _ =>
+        buildCast[CalendarInterval](_, i => UTF8String.fromString(toMultiUnitsString(i)))
+    }
     case BinaryType => buildCast[Array[Byte]](_, UTF8String.fromBytes)
     case DateType => buildCast[Int](_, d => UTF8String.fromString(dateFormatter.format(d)))
     case TimestampType => buildCast[Long](_,
@@ -985,6 +995,16 @@ abstract class CastBase extends UnaryExpression with TimeZoneAwareExpression wit
           timestampFormatter.getClass)
         (c, evPrim, evNull) => code"""$evPrim = UTF8String.fromString(
           org.apache.spark.sql.catalyst.util.DateTimeUtils.timestampToString($tf, $c));"""
+      case CalendarIntervalType =>
+        val iu = IntervalUtils.getClass.getCanonicalName.stripSuffix("$")
+        SQLConf.get.intervalOutputStyle match {
+        case SQL_STANDARD =>
+          (c, evPrim, _) => code"""$evPrim = UTF8String.fromString($iu.toSqlStandardString($c));"""
+        case ISO_8601 =>
+          (c, evPrim, _) => code"""$evPrim = UTF8String.fromString($iu.toIso8601String($c));"""
+        case _ =>
+          (c, evPrim, _) => code"""$evPrim = UTF8String.fromString($iu.toMultiUnitsString($c));"""
+        }
       case ArrayType(et, _) =>
         (c, evPrim, evNull) => {
           val buffer = ctx.freshVariable("buffer", classOf[UTF8StringBuilder])

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

@@ -409,6 +409,7 @@ case class Literal (value: Any, dataType: DataType) extends LeafExpression {
         DateTimeUtils.getZoneId(SQLConf.get.sessionLocalTimeZone))
       s"TIMESTAMP('${formatter.format(v)}')"
     case (v: Array[Byte], BinaryType) => s"X'${DatatypeConverter.printHexBinary(v)}'"
+    case (v: CalendarInterval, CalendarIntervalType) => IntervalUtils.toMultiUnitsString(v)
     case _ => value.toString
   }
 }

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JSONOptions.scala

@@ -26,6 +26,8 @@ import com.fasterxml.jackson.core.{JsonFactory, JsonParser}
 import org.apache.spark.internal.Logging
 import org.apache.spark.sql.catalyst.util._
 import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.internal.SQLConf.IntervalStyle
+import org.apache.spark.sql.internal.SQLConf.IntervalStyle.IntervalStyle
 
 /**
  * Options for parsing JSON data into Spark SQL rows.
@@ -92,6 +94,9 @@ private[sql] class JSONOptions(
   val timestampFormat: String =
     parameters.getOrElse("timestampFormat", "uuuu-MM-dd'T'HH:mm:ss.SSSXXX")
 
+  val intervalOutputStyle: IntervalStyle = parameters.get("intervalOutputStyle")
+    .map(IntervalStyle.withName).getOrElse(SQLConf.get.intervalOutputStyle)
+
   val multiLine = parameters.get("multiLine").map(_.toBoolean).getOrElse(false)
 
   /**

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonGenerator.scala

@@ -24,8 +24,8 @@ import com.fasterxml.jackson.core._
 import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.catalyst.expressions.SpecializedGetters
 import org.apache.spark.sql.catalyst.util._
+import org.apache.spark.sql.internal.SQLConf.IntervalStyle._
 import org.apache.spark.sql.types._
-
 /**
  * `JackGenerator` can only be initialized with a `StructType`, a `MapType` or an `ArrayType`.
  * Once it is initialized with `StructType`, it can be used to write out a struct or an array of
@@ -119,6 +119,16 @@ private[sql] class JacksonGenerator(
       (row: SpecializedGetters, ordinal: Int) =>
         gen.writeNumber(row.getDouble(ordinal))
 
+    case CalendarIntervalType =>
+      (row: SpecializedGetters, ordinal: Int) => options.intervalOutputStyle match {
+        case SQL_STANDARD =>
+          gen.writeString(IntervalUtils.toSqlStandardString(row.getInterval(ordinal)))
+        case ISO_8601 =>
+          gen.writeString(IntervalUtils.toIso8601String(row.getInterval(ordinal)))
+        case _ =>
+          gen.writeString(IntervalUtils.toMultiUnitsString(row.getInterval(ordinal)))
+      }
+
     case StringType =>
       (row: SpecializedGetters, ordinal: Int) =>
         gen.writeString(row.getUTF8String(ordinal).toString)
@@ -214,10 +224,21 @@ private[sql] class JacksonGenerator(
   private def writeMapData(
       map: MapData, mapType: MapType, fieldWriter: ValueWriter): Unit = {
     val keyArray = map.keyArray()
+    val keyString = mapType.keyType match {
+      case CalendarIntervalType => options.intervalOutputStyle match {
+        case SQL_STANDARD =>
+          (i: Int) => IntervalUtils.toSqlStandardString(keyArray.getInterval(i))
+        case ISO_8601 =>
+          (i: Int) => IntervalUtils.toIso8601String(keyArray.getInterval(i))
+        case _ =>
+          (i: Int) => IntervalUtils.toMultiUnitsString(keyArray.getInterval(i))
+      }
+      case _ => (i: Int) => keyArray.get(i, mapType.keyType).toString
+    }
     val valueArray = map.valueArray()
     var i = 0
     while (i < map.numElements()) {
-      gen.writeFieldName(keyArray.get(i, mapType.keyType).toString)
+      gen.writeFieldName(keyString(i))
       if (!valueArray.isNullAt(i)) {
         fieldWriter.apply(valueArray, i)
       } else {

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/IntervalUtils.scala

@@ -17,6 +17,7 @@
 
 package org.apache.spark.sql.catalyst.util
 
+import java.math.BigDecimal
 import java.util.concurrent.TimeUnit
 
 import scala.util.control.NonFatal
@@ -424,6 +425,111 @@ object IntervalUtils {
     fromDoubles(interval.months / num, interval.days / num, interval.microseconds / num)
   }
 
+  def toMultiUnitsString(interval: CalendarInterval): String = {
+    if (interval.months == 0 && interval.days == 0 && interval.microseconds == 0) {
+      return "0 seconds"
+    }
+    val sb = new StringBuilder
+    if (interval.months != 0) {
+      appendUnit(sb, interval.months / 12, "years")
+      appendUnit(sb, interval.months % 12, "months")
+    }
+    appendUnit(sb, interval.days, "days")
+    if (interval.microseconds != 0) {
+      var rest = interval.microseconds
+      appendUnit(sb, rest / MICROS_PER_HOUR, "hours")
+      rest %= MICROS_PER_HOUR
+      appendUnit(sb, rest / MICROS_PER_MINUTE, "minutes")
+      rest %= MICROS_PER_MINUTE
+      if (rest != 0) {
+        val s = BigDecimal.valueOf(rest, 6).stripTrailingZeros.toPlainString
+        sb.append(s).append(" seconds ")
+      }
+    }
+    sb.setLength(sb.length - 1)
+    sb.toString
+  }
+
+  private def appendUnit(sb: StringBuilder, value: Long, unit: String): Unit = {
+    if (value != 0) sb.append(value).append(' ').append(unit).append(' ')
+  }
+
+  def toSqlStandardString(interval: CalendarInterval): String = {
+    val yearMonthPart = if (interval.months < 0) {
+      val ma = math.abs(interval.months)
+      "-" + ma / 12 + "-" + ma % 12
+    } else if (interval.months > 0) {
+      "+" + interval.months / 12 + "-" + interval.months % 12
+    } else {
+      ""
+    }
+
+    val dayPart = if (interval.days < 0) {
+      interval.days.toString
+    } else if (interval.days > 0) {
+      "+" + interval.days
+    } else {
+      ""
+    }
+
+    val timePart = if (interval.microseconds != 0) {
+      val sign = if (interval.microseconds > 0) "+" else "-"
+      val sb = new StringBuilder(sign)
+      var rest = math.abs(interval.microseconds)
+      sb.append(rest / MICROS_PER_HOUR)
+      sb.append(':')
+      rest = rest % MICROS_PER_HOUR
+      val minutes = rest / MICROS_PER_MINUTE;
+      if (minutes < 10) {
+        sb.append(0)
+      }
+      sb.append(minutes)
+      sb.append(':')
+      rest %= MICROS_PER_MINUTE
+      val bd = BigDecimal.valueOf(rest, 6)
+      if (bd.compareTo(new BigDecimal(10)) < 0) {
+        sb.append(0)
+      }
+      val s = bd.stripTrailingZeros().toPlainString
+      sb.append(s)
+      sb.toString()
+    } else {
+      ""
+    }
+
+    val intervalList = Seq(yearMonthPart, dayPart, timePart).filter(_.nonEmpty)
+    if (intervalList.nonEmpty) intervalList.mkString(" ") else "0"
+  }
+
+  def toIso8601String(interval: CalendarInterval): String = {
+    val sb = new StringBuilder("P")
+
+    val year = interval.months / 12
+    if (year != 0) sb.append(year + "Y")
+    val month = interval.months % 12
+    if (month != 0) sb.append(month + "M")
+
+    if (interval.days != 0) sb.append(interval.days + "D")
+
+    if (interval.microseconds != 0) {
+      sb.append('T')
+      var rest = interval.microseconds
+      val hour = rest / MICROS_PER_HOUR
+      if (hour != 0) sb.append(hour + "H")
+      rest %= MICROS_PER_HOUR
+      val minute = rest / MICROS_PER_MINUTE
+      if (minute != 0) sb.append(minute + "M")
+      rest %= MICROS_PER_MINUTE
+      if (rest != 0) {
+        val bd = BigDecimal.valueOf(rest, 6)
+        sb.append(bd.stripTrailingZeros().toPlainString + "S")
+      }
+    } else if (interval.days == 0 && interval.months == 0) {
+      sb.append("T0S")
+    }
+    sb.toString()
+  }
+
   private object ParseState extends Enumeration {
     type ParseState = Value
 

sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala

@@ -37,7 +37,6 @@ import org.apache.spark.sql.catalyst.expressions.CodegenObjectFactoryMode
 import org.apache.spark.sql.catalyst.expressions.codegen.CodeGenerator
 import org.apache.spark.sql.catalyst.plans.logical.HintErrorHandler
 import org.apache.spark.sql.connector.catalog.CatalogManager.SESSION_CATALOG_NAME
-import org.apache.spark.sql.internal.SQLConf.StoreAssignmentPolicy
 import org.apache.spark.unsafe.array.ByteArrayMethods
 import org.apache.spark.util.Utils
 
@@ -1774,6 +1773,21 @@ object SQLConf {
     .booleanConf
     .createWithDefault(false)
 
+  object IntervalStyle extends Enumeration {
+    type IntervalStyle = Value
+    val SQL_STANDARD, ISO_8601, MULTI_UNITS = Value
+  }
+
+  val INTERVAL_STYLE = buildConf("spark.sql.intervalOutputStyle")
+    .doc("Display format for interval values. The value SQL_STANDARD will produce output" +
+      " matching SQL standard interval literals. The value ISO_8601 will produce output matching" +
+      " the ISO 8601 standard. The value MULTI_UNITS (which is the default) will produce output" +
+      " in form of value unit pairs, i.e. '3 year 2 months 10 days'")
+    .stringConf
+    .transform(_.toUpperCase(Locale.ROOT))
+    .checkValues(IntervalStyle.values.map(_.toString))
+    .createWithDefault(IntervalStyle.MULTI_UNITS.toString)
+
   val SORT_BEFORE_REPARTITION =
     buildConf("spark.sql.execution.sortBeforeRepartition")
       .internal()
@@ -2502,6 +2516,8 @@ class SQLConf extends Serializable with Logging {
   def storeAssignmentPolicy: StoreAssignmentPolicy.Value =
     StoreAssignmentPolicy.withName(getConf(STORE_ASSIGNMENT_POLICY))
 
+  def intervalOutputStyle: IntervalStyle.Value = IntervalStyle.withName(getConf(INTERVAL_STYLE))
+
   def ansiEnabled: Boolean = getConf(ANSI_ENABLED)
 
   def usePostgreSQLDialect: Boolean = getConf(DIALECT) == Dialect.POSTGRESQL.toString()