Merge remote-tracking branch 'apache/master' into SPARK-35780-full-range-datetime

Author: linhongliu-db

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

@@ -562,24 +562,24 @@ public UTF8String trim() {
   }
 
   /**
-   * Trims whitespaces ({@literal <=} ASCII 32) from both ends of this string.
+   * Trims whitespace ASCII characters from both ends of this string.
    *
-   * Note that, this method is the same as java's {@link String#trim}, and different from
-   * {@link UTF8String#trim()} which remove only spaces(= ASCII 32) from both ends.
+   * Note that, this method is different from {@link UTF8String#trim()} which removes
+   * only spaces(= ASCII 32) from both ends.
    *
    * @return A UTF8String whose value is this UTF8String, with any leading and trailing white
    * space removed, or this UTF8String if it has no leading or trailing whitespace.
    *
    */
   public UTF8String trimAll() {
     int s = 0;
-    // skip all of the whitespaces (<=0x20) in the left side
+    // skip all of the whitespaces in the left side
     while (s < this.numBytes && Character.isWhitespace(getByte(s))) s++;
     if (s == this.numBytes) {
       // Everything trimmed
       return EMPTY_UTF8;
     }
-    // skip all of the whitespaces (<=0x20) in the right side
+    // skip all of the whitespaces in the right side
     int e = this.numBytes - 1;
     while (e > s && Character.isWhitespace(getByte(e))) e--;
     if (s == 0 && e == numBytes - 1) {

core/src/main/resources/error/error-classes.json

@@ -1,4 +1,8 @@
 {
+  "AMBIGUOUS_FIELD_NAME" : {
+    "message" : [ "Field name %s is ambiguous and has %s matching fields in the struct." ],
+    "sqlState" : "42000"
+  },
   "DIVIDE_BY_ZERO" : {
     "message" : [ "divide by zero" ],
     "sqlState" : "22012"
@@ -7,6 +11,10 @@
     "message" : [ "Found duplicate keys '%s'" ],
     "sqlState" : "23000"
   },
+  "INVALID_FIELD_NAME" : {
+    "message" : [ "Field name %s is invalid: %s is not a struct." ],
+    "sqlState" : "42000"
+  },
   "MISSING_COLUMN" : {
     "message" : [ "cannot resolve '%s' given input columns: [%s]" ],
     "sqlState" : "42000"

core/src/test/scala/org/apache/spark/scheduler/HealthTrackerSuite.scala

@@ -62,11 +62,11 @@ class HealthTrackerSuite extends SparkFunSuite with BeforeAndAfterEach with Mock
   // All executors and hosts used in tests should be in this set, so that [[assertEquivalentToSet]]
   // works.  Its OK if its got extraneous entries
   val allExecutorAndHostIds = {
-    (('A' to 'Z')++ (1 to 100).map(_.toString))
-      .flatMap{ suffix =>
+    ('A' to 'Z')
+      .flatMap { suffix =>
         Seq(s"host$suffix", s"host-$suffix")
       }
-  }.toSet
+  }.toSet ++ (1 to 100).map(_.toString)
 
   /**
    * Its easier to write our tests as if we could directly look at the sets of nodes & executors in

docs/sql-migration-guide.md

@@ -83,7 +83,9 @@ license: |
 
   - In Spark 3.2, `TRANSFORM` operator can support `ArrayType/MapType/StructType` without Hive SerDe, in this mode, we use `StructsToJosn` to convert `ArrayType/MapType/StructType` column to `STRING` and use `JsonToStructs` to parse `STRING` to `ArrayType/MapType/StructType`. In Spark 3.1, Spark just support case `ArrayType/MapType/StructType` column as `STRING` but can't support parse `STRING` to `ArrayType/MapType/StructType` output columns.
 
-  - In Spark 3.2, the unit-to-unit interval literals like `INTERVAL '1-1' YEAR TO MONTH` are converted to ANSI interval types: `YearMonthIntervalType` or `DayTimeIntervalType`. In Spark 3.1 and earlier, such interval literals are converted to `CalendarIntervalType`. To restore the behavior before Spark 3.2, you can set `spark.sql.legacy.interval.enabled` to `true`.
+  - In Spark 3.2, the unit-to-unit interval literals like `INTERVAL '1-1' YEAR TO MONTH` and the unit list interval literals like `INTERVAL '3' DAYS '1' HOUR` are converted to ANSI interval types: `YearMonthIntervalType` or `DayTimeIntervalType`. In Spark 3.1 and earlier, such interval literals are converted to `CalendarIntervalType`. To restore the behavior before Spark 3.2, you can set `spark.sql.legacy.interval.enabled` to `true`.
+
+  - In Spark 3.2, the unit list interval literals can not mix year-month fields (YEAR and MONTH) and day-time fields (WEEK, DAY, ..., MICROSECOND). For example, `INTERVAL 1 day 1 hour` is invalid in Spark 3.2. In Spark 3.1 and earlier, there is no such limitation and the literal returns value of `CalendarIntervalType`. To restore the behavior before Spark 3.2, you can set `spark.sql.legacy.interval.enabled` to `true`.
 
   - In Spark 3.2, Spark supports `DayTimeIntervalType` and `YearMonthIntervalType` as inputs and outputs of `TRANSFORM` clause in Hive `SERDE` mode, the behavior is different between Hive `SERDE` mode and `ROW FORMAT DELIMITED` mode when these two types are used as inputs. In Hive `SERDE` mode, `DayTimeIntervalType` column is converted to `HiveIntervalDayTime`, its string format is `[-]?d h:m:s.n`, but in `ROW FORMAT DELIMITED` mode the format is `INTERVAL '[-]?d h:m:s.n' DAY TO TIME`. In Hive `SERDE` mode, `YearMonthIntervalType` column is converted to `HiveIntervalYearMonth`, its string format is `[-]?y-m`, but in `ROW FORMAT DELIMITED` mode the format is `INTERVAL '[-]?y-m' YEAR TO MONTH`.
 
@@ -157,6 +159,8 @@ license: |
 
 - In Spark 3.0, JSON datasource and JSON function `schema_of_json` infer TimestampType from string values if they match to the pattern defined by the JSON option `timestampFormat`. Since version 3.0.1, the timestamp type inference is disabled by default. Set the JSON option `inferTimestamp` to `true` to enable such type inference.
 
+- In Spark 3.0, when casting string to integral types(tinyint, smallint, int and bigint), datetime types(date, timestamp and interval) and boolean type, the leading and trailing characters (<= ASCII 32) will be trimmed. For example, `cast('\b1\b' as int)` results `1`. Since Spark 3.0.1, only the leading and trailing whitespace ASCII characters will be trimmed. For example, `cast('\t1\t' as int)` results `1` but `cast('\b1\b' as int)` results `NULL`.
+
 ## Upgrading from Spark SQL 2.4 to 3.0
 
 ### Dataset/DataFrame APIs

external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaMicroBatchSourceSuite.scala

@@ -2429,8 +2429,9 @@ class KafkaSourceStressSuite extends KafkaSourceTest {
       (d, running) => {
         Random.nextInt(5) match {
           case 0 => // Add a new topic
-            topics = topics ++ Seq(newStressTopic)
-            AddKafkaData(topics.toSet, d: _*)(message = s"Add topic $newStressTopic",
+            val newTopic = newStressTopic
+            topics = topics ++ Seq(newTopic)
+            AddKafkaData(topics.toSet, d: _*)(message = s"Add topic $newTopic",
               topicAction = (topic, partition) => {
                 if (partition.isEmpty) {
                   testUtils.createTopic(topic, partitions = nextInt(1, 6))

resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/SparkKubernetesClientFactory.scala

@@ -18,11 +18,14 @@ package org.apache.spark.deploy.k8s
 
 import java.io.File
 
+import com.fasterxml.jackson.databind.ObjectMapper
 import com.google.common.base.Charsets
 import com.google.common.io.Files
 import io.fabric8.kubernetes.client.{ConfigBuilder, DefaultKubernetesClient, KubernetesClient}
+import io.fabric8.kubernetes.client.Config.KUBERNETES_REQUEST_RETRY_BACKOFFLIMIT_SYSTEM_PROPERTY
 import io.fabric8.kubernetes.client.Config.autoConfigure
 import io.fabric8.kubernetes.client.utils.HttpClientUtils
+import io.fabric8.kubernetes.client.utils.Utils.getSystemPropertyOrEnvVar
 import okhttp3.Dispatcher
 
 import org.apache.spark.SparkConf
@@ -74,6 +77,11 @@ private[spark] object SparkKubernetesClientFactory extends Logging {
       kubeContext.map("context " + _).getOrElse("current context") +
       " from users K8S config file")
 
+    // if backoff limit is not set then set it to 3
+    if (getSystemPropertyOrEnvVar(KUBERNETES_REQUEST_RETRY_BACKOFFLIMIT_SYSTEM_PROPERTY) == null) {
+      System.setProperty(KUBERNETES_REQUEST_RETRY_BACKOFFLIMIT_SYSTEM_PROPERTY, "3")
+    }
+
     // Start from an auto-configured config with the desired context
     // Fabric 8 uses null to indicate that the users current context should be used so if no
     // explicit setting pass null
@@ -102,6 +110,8 @@ private[spark] object SparkKubernetesClientFactory extends Logging {
     val httpClientWithCustomDispatcher = baseHttpClient.newBuilder()
       .dispatcher(dispatcher)
       .build()
+    logDebug("Kubernetes client config: " +
+      new ObjectMapper().writerWithDefaultPrettyPrinter().writeValueAsString(config))
     new DefaultKubernetesClient(httpClientWithCustomDispatcher, config)
   }
 

sql/catalyst/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBase.g4

@@ -877,7 +877,7 @@ interval
     ;
 
 errorCapturingMultiUnitsInterval
-    : multiUnitsInterval unitToUnitInterval?
+    : body=multiUnitsInterval unitToUnitInterval?
     ;
 
 multiUnitsInterval

sql/catalyst/src/main/scala-2.12/org/apache/spark/sql/catalyst/expressions/AttributeMap.scala

@@ -50,4 +50,6 @@ class AttributeMap[A](val baseMap: Map[ExprId, (Attribute, A)])
   override def iterator: Iterator[(Attribute, A)] = baseMap.valuesIterator
 
   override def -(key: Attribute): Map[Attribute, A] = baseMap.values.toMap - key
+
+  def ++(other: AttributeMap[A]): AttributeMap[A] = new AttributeMap(baseMap ++ other.baseMap)
 }

sql/catalyst/src/main/scala-2.13/org/apache/spark/sql/catalyst/expressions/AttributeMap.scala

@@ -51,4 +51,6 @@ class AttributeMap[A](val baseMap: Map[ExprId, (Attribute, A)])
   override def iterator: Iterator[(Attribute, A)] = baseMap.valuesIterator
 
   override def removed(key: Attribute): Map[Attribute, A] = baseMap.values.toMap - key
+
+  def ++(other: AttributeMap[A]): AttributeMap[A] = new AttributeMap(baseMap ++ other.baseMap)
 }

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

@@ -20,6 +20,7 @@ package org.apache.spark.sql
 import org.apache.spark.{SparkThrowable, SparkThrowableHelper}
 import org.apache.spark.annotation.Stable
 import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.trees.Origin
 
 /**
  * Thrown when a query fails to analyze, usually because the query itself is invalid.
@@ -48,12 +49,11 @@ class AnalysisException protected[sql] (
   def this(
       errorClass: String,
       messageParameters: Array[String],
-      line: Option[Int],
-      startPosition: Option[Int]) =
+      origin: Origin) =
     this(
       SparkThrowableHelper.getMessage(errorClass, messageParameters),
-      line = line,
-      startPosition = startPosition,
+      line = origin.line,
+      startPosition = origin.startPosition,
       errorClass = Some(errorClass),
       messageParameters = messageParameters)