| Property Name | Default | Meaning |
|---|
- spark.yarn.keytab |
+ spark.kerberos.keytab |
(none) |
The full path to the file that contains the keytab for the principal specified above. This keytab
@@ -477,7 +477,7 @@ providers can be disabled individually by setting `spark.security.credentials.{s
|
- spark.yarn.principal |
+ spark.kerberos.principal |
(none) |
Principal to be used to login to KDC, while running on secure clusters. Equivalent to the
diff --git a/docs/sparkr.md b/docs/sparkr.md
index b4248e8bb21de..55e8f15da17ca 100644
--- a/docs/sparkr.md
+++ b/docs/sparkr.md
@@ -70,12 +70,12 @@ The following Spark driver properties can be set in `sparkConfig` with `sparkR.s
| --master |
- spark.yarn.keytab |
+ spark.kerberos.keytab |
Application Properties |
--keytab |
- spark.yarn.principal |
+ spark.kerberos.principal |
Application Properties |
--principal |
diff --git a/docs/sql-programming-guide.md b/docs/sql-programming-guide.md
index c72fa3d75d67f..6de9de90c62c3 100644
--- a/docs/sql-programming-guide.md
+++ b/docs/sql-programming-guide.md
@@ -1004,6 +1004,17 @@ Configuration of Parquet can be done using the `setConf` method on `SparkSession
+
+ spark.sql.parquet.writeLegacyFormat |
+ false |
+
+ If true, data will be written in a way of Spark 1.4 and earlier. For example, decimal values
+ will be written in Apache Parquet's fixed-length byte array format, which other systems such as
+ Apache Hive and Apache Impala use. If false, the newer format in Parquet will be used. For
+ example, decimals will be written in int-based format. If Parquet output is intended for use
+ with systems that do not support this newer format, set to true.
+ |
+
## ORC Files
diff --git a/external/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaDataConsumer.scala b/external/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaDataConsumer.scala
index ceb9e318b283b..7b1314bc8c3c0 100644
--- a/external/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaDataConsumer.scala
+++ b/external/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaDataConsumer.scala
@@ -134,6 +134,8 @@ private[kafka010] case class InternalKafkaConsumer(
/** Reset the internal pre-fetched data. */
def reset(): Unit = {
_records = ju.Collections.emptyListIterator()
+ _nextOffsetInFetchedData = UNKNOWN_OFFSET
+ _offsetAfterPoll = UNKNOWN_OFFSET
}
/**
@@ -361,8 +363,9 @@ private[kafka010] case class InternalKafkaConsumer(
if (offset < fetchedData.offsetAfterPoll) {
// Offsets in [offset, fetchedData.offsetAfterPoll) are invisible. Return a record to ask
// the next call to start from `fetchedData.offsetAfterPoll`.
+ val nextOffsetToFetch = fetchedData.offsetAfterPoll
fetchedData.reset()
- return fetchedRecord.withRecord(null, fetchedData.offsetAfterPoll)
+ return fetchedRecord.withRecord(null, nextOffsetToFetch)
} else {
// Fetch records from Kafka and update `fetchedData`.
fetchData(offset, pollTimeoutMs)
diff --git a/external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaMicroBatchSourceSuite.scala b/external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaMicroBatchSourceSuite.scala
index e5f008804ee5b..39c2cde7de40d 100644
--- a/external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaMicroBatchSourceSuite.scala
+++ b/external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaMicroBatchSourceSuite.scala
@@ -874,6 +874,58 @@ abstract class KafkaMicroBatchSourceSuiteBase extends KafkaSourceSuiteBase {
)
}
}
+
+ test("SPARK-25495: FetchedData.reset should reset all fields") {
+ val topic = newTopic()
+ val topicPartition = new TopicPartition(topic, 0)
+ testUtils.createTopic(topic, partitions = 1)
+
+ val ds = spark
+ .readStream
+ .format("kafka")
+ .option("kafka.bootstrap.servers", testUtils.brokerAddress)
+ .option("kafka.metadata.max.age.ms", "1")
+ .option("kafka.isolation.level", "read_committed")
+ .option("subscribe", topic)
+ .option("startingOffsets", "earliest")
+ .load()
+ .select($"value".as[String])
+
+ testUtils.withTranscationalProducer { producer =>
+ producer.beginTransaction()
+ (0 to 3).foreach { i =>
+ producer.send(new ProducerRecord[String, String](topic, i.toString)).get()
+ }
+ producer.commitTransaction()
+ }
+ testUtils.waitUntilOffsetAppears(topicPartition, 5)
+
+ val q = ds.writeStream.foreachBatch { (ds, epochId) =>
+ if (epochId == 0) {
+ // Send more message before the tasks of the current batch start reading the current batch
+ // data, so that the executors will prefetch messages in the next batch and drop them. In
+ // this case, if we forget to reset `FetchedData._nextOffsetInFetchedData` or
+ // `FetchedData._offsetAfterPoll` (See SPARK-25495), the next batch will see incorrect
+ // values and return wrong results hence fail the test.
+ testUtils.withTranscationalProducer { producer =>
+ producer.beginTransaction()
+ (4 to 7).foreach { i =>
+ producer.send(new ProducerRecord[String, String](topic, i.toString)).get()
+ }
+ producer.commitTransaction()
+ }
+ testUtils.waitUntilOffsetAppears(topicPartition, 10)
+ checkDatasetUnorderly(ds, (0 to 3).map(_.toString): _*)
+ } else {
+ checkDatasetUnorderly(ds, (4 to 7).map(_.toString): _*)
+ }
+ }.start()
+ try {
+ q.processAllAvailable()
+ } finally {
+ q.stop()
+ }
+ }
}
diff --git a/python/pyspark/sql/functions.py b/python/pyspark/sql/functions.py
index 6da5237d18de4..1c3d9725b285b 100644
--- a/python/pyspark/sql/functions.py
+++ b/python/pyspark/sql/functions.py
@@ -2295,7 +2295,9 @@ def to_json(col, options={}):
into a JSON string. Throws an exception, in the case of an unsupported type.
:param col: name of column containing a struct, an array or a map.
- :param options: options to control converting. accepts the same options as the JSON datasource
+ :param options: options to control converting. accepts the same options as the JSON datasource.
+ Additionally the function supports the `pretty` option which enables
+ pretty JSON generation.
>>> from pyspark.sql import Row
>>> from pyspark.sql.types import *
diff --git a/python/pyspark/sql/session.py b/python/pyspark/sql/session.py
index 9877d6251388b..263cbc56f4799 100644
--- a/python/pyspark/sql/session.py
+++ b/python/pyspark/sql/session.py
@@ -141,8 +141,9 @@ def enableHiveSupport(self):
return self.config("spark.sql.catalogImplementation", "hive")
def _sparkContext(self, sc):
- self._sc = sc
- return self
+ with self._lock:
+ self._sc = sc
+ return self
@since(2.0)
def getOrCreate(self):
diff --git a/python/pyspark/sql/tests.py b/python/pyspark/sql/tests.py
index 95e88772ca5de..64a7ceb3fea96 100644
--- a/python/pyspark/sql/tests.py
+++ b/python/pyspark/sql/tests.py
@@ -5846,7 +5846,8 @@ def test_positional_assignment_conf(self):
import pandas as pd
from pyspark.sql.functions import pandas_udf, PandasUDFType
- with self.sql_conf({"spark.sql.execution.pandas.groupedMap.assignColumnsByPosition": True}):
+ with self.sql_conf({
+ "spark.sql.legacy.execution.pandas.groupedMap.assignColumnsByName": False}):
@pandas_udf("a string, b float", PandasUDFType.GROUPED_MAP)
def foo(_):
diff --git a/python/pyspark/worker.py b/python/pyspark/worker.py
index 974344f01d923..8c59f1f999f18 100644
--- a/python/pyspark/worker.py
+++ b/python/pyspark/worker.py
@@ -97,8 +97,9 @@ def verify_result_length(*a):
def wrap_grouped_map_pandas_udf(f, return_type, argspec, runner_conf):
- assign_cols_by_pos = runner_conf.get(
- "spark.sql.execution.pandas.groupedMap.assignColumnsByPosition", False)
+ assign_cols_by_name = runner_conf.get(
+ "spark.sql.legacy.execution.pandas.groupedMap.assignColumnsByName", "true")
+ assign_cols_by_name = assign_cols_by_name.lower() == "true"
def wrapped(key_series, value_series):
import pandas as pd
@@ -119,7 +120,7 @@ def wrapped(key_series, value_series):
"Expected: {} Actual: {}".format(len(return_type), len(result.columns)))
# Assign result columns by schema name if user labeled with strings, else use position
- if not assign_cols_by_pos and any(isinstance(name, basestring) for name in result.columns):
+ if assign_cols_by_name and any(isinstance(name, basestring) for name in result.columns):
return [(result[field.name], to_arrow_type(field.dataType)) for field in return_type]
else:
return [(result[result.columns[i]], to_arrow_type(field.dataType))
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/DecimalPrecision.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/DecimalPrecision.scala
index e511f8064e28a..82692334544e2 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/DecimalPrecision.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/DecimalPrecision.scala
@@ -290,11 +290,13 @@ object DecimalPrecision extends TypeCoercionRule {
// potentially loosing 11 digits of the fractional part. Using only the precision needed
// by the Literal, instead, the result would be DECIMAL(38 + 1 + 1, 18), which would
// become DECIMAL(38, 16), safely having a much lower precision loss.
- case (l: Literal, r) if r.dataType.isInstanceOf[DecimalType]
- && l.dataType.isInstanceOf[IntegralType] =>
+ case (l: Literal, r) if r.dataType.isInstanceOf[DecimalType] &&
+ l.dataType.isInstanceOf[IntegralType] &&
+ SQLConf.get.literalPickMinimumPrecision =>
b.makeCopy(Array(Cast(l, DecimalType.fromLiteral(l)), r))
- case (l, r: Literal) if l.dataType.isInstanceOf[DecimalType]
- && r.dataType.isInstanceOf[IntegralType] =>
+ case (l, r: Literal) if l.dataType.isInstanceOf[DecimalType] &&
+ r.dataType.isInstanceOf[IntegralType] &&
+ SQLConf.get.literalPickMinimumPrecision =>
b.makeCopy(Array(l, Cast(r, DecimalType.fromLiteral(r))))
// Promote integers inside a binary expression with fixed-precision decimals to decimals,
// and fixed-precision decimals in an expression with floats / doubles to doubles
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/FunctionRegistry.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/FunctionRegistry.scala
index 8b69a47036962..7dafebff79874 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/FunctionRegistry.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/FunctionRegistry.scala
@@ -300,15 +300,6 @@ object FunctionRegistry {
expression[CollectList]("collect_list"),
expression[CollectSet]("collect_set"),
expression[CountMinSketchAgg]("count_min_sketch"),
- expression[RegrCount]("regr_count"),
- expression[RegrSXX]("regr_sxx"),
- expression[RegrSYY]("regr_syy"),
- expression[RegrAvgX]("regr_avgx"),
- expression[RegrAvgY]("regr_avgy"),
- expression[RegrSXY]("regr_sxy"),
- expression[RegrSlope]("regr_slope"),
- expression[RegrR2]("regr_r2"),
- expression[RegrIntercept]("regr_intercept"),
// string functions
expression[Ascii]("ascii"),
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/AttributeSet.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/AttributeSet.scala
index 7420b6b57d8e1..a7e09eee617e9 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/AttributeSet.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/AttributeSet.scala
@@ -17,6 +17,8 @@
package org.apache.spark.sql.catalyst.expressions
+import scala.collection.mutable
+
protected class AttributeEquals(val a: Attribute) {
override def hashCode(): Int = a match {
@@ -39,10 +41,13 @@ object AttributeSet {
/** Constructs a new [[AttributeSet]] given a sequence of [[Expression Expressions]]. */
def apply(baseSet: Iterable[Expression]): AttributeSet = {
- new AttributeSet(
- baseSet
- .flatMap(_.references)
- .map(new AttributeEquals(_)).toSet)
+ fromAttributeSets(baseSet.map(_.references))
+ }
+
+ /** Constructs a new [[AttributeSet]] given a sequence of [[AttributeSet]]s. */
+ def fromAttributeSets(sets: Iterable[AttributeSet]): AttributeSet = {
+ val baseSet = sets.foldLeft(new mutable.LinkedHashSet[AttributeEquals]())( _ ++= _.baseSet)
+ new AttributeSet(baseSet.toSet)
}
}
@@ -94,8 +99,14 @@ class AttributeSet private (val baseSet: Set[AttributeEquals])
* Returns a new [[AttributeSet]] that does not contain any of the [[Attribute Attributes]] found
* in `other`.
*/
- def --(other: Traversable[NamedExpression]): AttributeSet =
- new AttributeSet(baseSet -- other.map(a => new AttributeEquals(a.toAttribute)))
+ def --(other: Traversable[NamedExpression]): AttributeSet = {
+ other match {
+ case otherSet: AttributeSet =>
+ new AttributeSet(baseSet -- otherSet.baseSet)
+ case _ =>
+ new AttributeSet(baseSet -- other.map(a => new AttributeEquals(a.toAttribute)))
+ }
+ }
/**
* Returns a new [[AttributeSet]] that contains all of the [[Attribute Attributes]] found
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Expression.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Expression.scala
index 773aefc0ac1f9..c215735ab1c98 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Expression.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Expression.scala
@@ -85,7 +85,7 @@ abstract class Expression extends TreeNode[Expression] {
def nullable: Boolean
- def references: AttributeSet = AttributeSet(children.flatMap(_.references.iterator))
+ def references: AttributeSet = AttributeSet.fromAttributeSets(children.map(_.references))
/** Returns the result of evaluating this expression on a given input Row */
def eval(input: InternalRow = null): Any
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/regression.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/regression.scala
deleted file mode 100644
index d8f4505588ff2..0000000000000
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/regression.scala
+++ /dev/null
@@ -1,190 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.spark.sql.catalyst.expressions.aggregate
-
-import org.apache.spark.sql.catalyst.dsl.expressions._
-import org.apache.spark.sql.catalyst.expressions._
-import org.apache.spark.sql.types.{AbstractDataType, DoubleType}
-
-/**
- * Base trait for all regression functions.
- */
-trait RegrLike extends AggregateFunction with ImplicitCastInputTypes {
- def y: Expression
- def x: Expression
-
- override def children: Seq[Expression] = Seq(y, x)
- override def inputTypes: Seq[AbstractDataType] = Seq(DoubleType, DoubleType)
-
- protected def updateIfNotNull(exprs: Seq[Expression]): Seq[Expression] = {
- assert(aggBufferAttributes.length == exprs.length)
- val nullableChildren = children.filter(_.nullable)
- if (nullableChildren.isEmpty) {
- exprs
- } else {
- exprs.zip(aggBufferAttributes).map { case (e, a) =>
- If(nullableChildren.map(IsNull).reduce(Or), a, e)
- }
- }
- }
-}
-
-
-@ExpressionDescription(
- usage = "_FUNC_(y, x) - Returns the number of non-null pairs.",
- since = "2.4.0")
-case class RegrCount(y: Expression, x: Expression)
- extends CountLike with RegrLike {
-
- override lazy val updateExpressions: Seq[Expression] = updateIfNotNull(Seq(count + 1L))
-
- override def prettyName: String = "regr_count"
-}
-
-
-@ExpressionDescription(
- usage = "_FUNC_(y, x) - Returns SUM(x*x)-SUM(x)*SUM(x)/N. Any pair with a NULL is ignored.",
- since = "2.4.0")
-case class RegrSXX(y: Expression, x: Expression)
- extends CentralMomentAgg(x) with RegrLike {
-
- override protected def momentOrder = 2
-
- override lazy val updateExpressions: Seq[Expression] = updateIfNotNull(updateExpressionsDef)
-
- override val evaluateExpression: Expression = {
- If(n === Literal(0.0), Literal.create(null, DoubleType), m2)
- }
-
- override def prettyName: String = "regr_sxx"
-}
-
-
-@ExpressionDescription(
- usage = "_FUNC_(y, x) - Returns SUM(y*y)-SUM(y)*SUM(y)/N. Any pair with a NULL is ignored.",
- since = "2.4.0")
-case class RegrSYY(y: Expression, x: Expression)
- extends CentralMomentAgg(y) with RegrLike {
-
- override protected def momentOrder = 2
-
- override lazy val updateExpressions: Seq[Expression] = updateIfNotNull(updateExpressionsDef)
-
- override val evaluateExpression: Expression = {
- If(n === Literal(0.0), Literal.create(null, DoubleType), m2)
- }
-
- override def prettyName: String = "regr_syy"
-}
-
-
-@ExpressionDescription(
- usage = "_FUNC_(y, x) - Returns the average of x. Any pair with a NULL is ignored.",
- since = "2.4.0")
-case class RegrAvgX(y: Expression, x: Expression)
- extends AverageLike(x) with RegrLike {
-
- override lazy val updateExpressions: Seq[Expression] = updateIfNotNull(updateExpressionsDef)
-
- override def prettyName: String = "regr_avgx"
-}
-
-
-@ExpressionDescription(
- usage = "_FUNC_(y, x) - Returns the average of y. Any pair with a NULL is ignored.",
- since = "2.4.0")
-case class RegrAvgY(y: Expression, x: Expression)
- extends AverageLike(y) with RegrLike {
-
- override lazy val updateExpressions: Seq[Expression] = updateIfNotNull(updateExpressionsDef)
-
- override def prettyName: String = "regr_avgy"
-}
-
-// scalastyle:off line.size.limit
-@ExpressionDescription(
- usage = "_FUNC_(y, x) - Returns the covariance of y and x multiplied for the number of items in the dataset. Any pair with a NULL is ignored.",
- since = "2.4.0")
-// scalastyle:on line.size.limit
-case class RegrSXY(y: Expression, x: Expression)
- extends Covariance(y, x) with RegrLike {
-
- override lazy val updateExpressions: Seq[Expression] = updateIfNotNull(updateExpressionsDef)
-
- override val evaluateExpression: Expression = {
- If(n === Literal(0.0), Literal.create(null, DoubleType), ck)
- }
-
- override def prettyName: String = "regr_sxy"
-}
-
-
-// scalastyle:off line.size.limit
-@ExpressionDescription(
- usage = "_FUNC_(y, x) - Returns the slope of the linear regression line. Any pair with a NULL is ignored.",
- since = "2.4.0")
-// scalastyle:on line.size.limit
-case class RegrSlope(y: Expression, x: Expression)
- extends PearsonCorrelation(y, x) with RegrLike {
-
- override lazy val updateExpressions: Seq[Expression] = updateIfNotNull(updateExpressionsDef)
-
- override val evaluateExpression: Expression = {
- If(n < Literal(2.0) || yMk === Literal(0.0), Literal.create(null, DoubleType), ck / yMk)
- }
-
- override def prettyName: String = "regr_slope"
-}
-
-
-// scalastyle:off line.size.limit
-@ExpressionDescription(
- usage = "_FUNC_(y, x) - Returns the coefficient of determination (also called R-squared or goodness of fit) for the regression line. Any pair with a NULL is ignored.",
- since = "2.4.0")
-// scalastyle:on line.size.limit
-case class RegrR2(y: Expression, x: Expression)
- extends PearsonCorrelation(y, x) with RegrLike {
-
- override lazy val updateExpressions: Seq[Expression] = updateIfNotNull(updateExpressionsDef)
-
- override val evaluateExpression: Expression = {
- If(n < Literal(2.0) || yMk === Literal(0.0), Literal.create(null, DoubleType),
- If(xMk === Literal(0.0), Literal(1.0), ck * ck / yMk / xMk))
- }
-
- override def prettyName: String = "regr_r2"
-}
-
-
-// scalastyle:off line.size.limit
-@ExpressionDescription(
- usage = "_FUNC_(y, x) - Returns the y-intercept of the linear regression line. Any pair with a NULL is ignored.",
- since = "2.4.0")
-// scalastyle:on line.size.limit
-case class RegrIntercept(y: Expression, x: Expression)
- extends PearsonCorrelation(y, x) with RegrLike {
-
- override lazy val updateExpressions: Seq[Expression] = updateIfNotNull(updateExpressionsDef)
-
- override val evaluateExpression: Expression = {
- If(n === Literal(0.0) || yMk === Literal(0.0), Literal.create(null, DoubleType),
- xAvg - (ck / yMk) * yAvg)
- }
-
- override def prettyName: String = "regr_intercept"
-}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/collectionOperations.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/collectionOperations.scala
index 85bc1cdb43051..9cc7dbadd923a 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/collectionOperations.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/collectionOperations.scala
@@ -3088,11 +3088,24 @@ case class ArrayRemove(left: Expression, right: Expression)
override def dataType: DataType = left.dataType
override def inputTypes: Seq[AbstractDataType] = {
- val elementType = left.dataType match {
- case t: ArrayType => t.elementType
- case _ => AnyDataType
+ (left.dataType, right.dataType) match {
+ case (ArrayType(e1, hasNull), e2) =>
+ TypeCoercion.findTightestCommonType(e1, e2) match {
+ case Some(dt) => Seq(ArrayType(dt, hasNull), dt)
+ case _ => Seq.empty
+ }
+ case _ => Seq.empty
+ }
+ }
+
+ override def checkInputDataTypes(): TypeCheckResult = {
+ (left.dataType, right.dataType) match {
+ case (ArrayType(e1, _), e2) if e1.sameType(e2) =>
+ TypeUtils.checkForOrderingExpr(e2, s"function $prettyName")
+ case _ => TypeCheckResult.TypeCheckFailure(s"Input to function $prettyName should have " +
+ s"been ${ArrayType.simpleString} followed by a value with same element type, but it's " +
+ s"[${left.dataType.catalogString}, ${right.dataType.catalogString}].")
}
- Seq(ArrayType, elementType)
}
private def elementType: DataType = left.dataType.asInstanceOf[ArrayType].elementType
@@ -3100,14 +3113,6 @@ case class ArrayRemove(left: Expression, right: Expression)
@transient private lazy val ordering: Ordering[Any] =
TypeUtils.getInterpretedOrdering(right.dataType)
- override def checkInputDataTypes(): TypeCheckResult = {
- super.checkInputDataTypes() match {
- case f: TypeCheckResult.TypeCheckFailure => f
- case TypeCheckResult.TypeCheckSuccess =>
- TypeUtils.checkForOrderingExpr(right.dataType, s"function $prettyName")
- }
- }
-
override def nullSafeEval(arr: Any, value: Any): Any = {
val newArray = new Array[Any](arr.asInstanceOf[ArrayData].numElements())
var pos = 0
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JSONOptions.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JSONOptions.scala
index 47eeb70e00427..64152e04928d2 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JSONOptions.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JSONOptions.scala
@@ -113,6 +113,11 @@ private[sql] class JSONOptions(
}
val lineSeparatorInWrite: String = lineSeparator.getOrElse("\n")
+ /**
+ * Generating JSON strings in pretty representation if the parameter is enabled.
+ */
+ val pretty: Boolean = parameters.get("pretty").map(_.toBoolean).getOrElse(false)
+
/** Sets config options on a Jackson [[JsonFactory]]. */
def setJacksonOptions(factory: JsonFactory): Unit = {
factory.configure(JsonParser.Feature.ALLOW_COMMENTS, allowComments)
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonGenerator.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonGenerator.scala
index 9b86d865622dc..d02a2be8ddad6 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonGenerator.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonGenerator.scala
@@ -70,7 +70,10 @@ private[sql] class JacksonGenerator(
s"Initial type ${dataType.catalogString} must be a ${MapType.simpleString}")
}
- private val gen = new JsonFactory().createGenerator(writer).setRootValueSeparator(null)
+ private val gen = {
+ val generator = new JsonFactory().createGenerator(writer).setRootValueSeparator(null)
+ if (options.pretty) generator.useDefaultPrettyPrinter() else generator
+ }
private val lineSeparator: String = options.lineSeparatorInWrite
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala
index 7c461895c5e52..07a653f3b5d48 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala
@@ -532,12 +532,12 @@ object ColumnPruning extends Rule[LogicalPlan] {
def apply(plan: LogicalPlan): LogicalPlan = removeProjectBeforeFilter(plan transform {
// Prunes the unused columns from project list of Project/Aggregate/Expand
- case p @ Project(_, p2: Project) if (p2.outputSet -- p.references).nonEmpty =>
+ case p @ Project(_, p2: Project) if !p2.outputSet.subsetOf(p.references) =>
p.copy(child = p2.copy(projectList = p2.projectList.filter(p.references.contains)))
- case p @ Project(_, a: Aggregate) if (a.outputSet -- p.references).nonEmpty =>
+ case p @ Project(_, a: Aggregate) if !a.outputSet.subsetOf(p.references) =>
p.copy(
child = a.copy(aggregateExpressions = a.aggregateExpressions.filter(p.references.contains)))
- case a @ Project(_, e @ Expand(_, _, grandChild)) if (e.outputSet -- a.references).nonEmpty =>
+ case a @ Project(_, e @ Expand(_, _, grandChild)) if !e.outputSet.subsetOf(a.references) =>
val newOutput = e.output.filter(a.references.contains(_))
val newProjects = e.projections.map { proj =>
proj.zip(e.output).filter { case (_, a) =>
@@ -547,18 +547,18 @@ object ColumnPruning extends Rule[LogicalPlan] {
a.copy(child = Expand(newProjects, newOutput, grandChild))
// Prunes the unused columns from child of `DeserializeToObject`
- case d @ DeserializeToObject(_, _, child) if (child.outputSet -- d.references).nonEmpty =>
+ case d @ DeserializeToObject(_, _, child) if !child.outputSet.subsetOf(d.references) =>
d.copy(child = prunedChild(child, d.references))
// Prunes the unused columns from child of Aggregate/Expand/Generate/ScriptTransformation
- case a @ Aggregate(_, _, child) if (child.outputSet -- a.references).nonEmpty =>
+ case a @ Aggregate(_, _, child) if !child.outputSet.subsetOf(a.references) =>
a.copy(child = prunedChild(child, a.references))
- case f @ FlatMapGroupsInPandas(_, _, _, child) if (child.outputSet -- f.references).nonEmpty =>
+ case f @ FlatMapGroupsInPandas(_, _, _, child) if !child.outputSet.subsetOf(f.references) =>
f.copy(child = prunedChild(child, f.references))
- case e @ Expand(_, _, child) if (child.outputSet -- e.references).nonEmpty =>
+ case e @ Expand(_, _, child) if !child.outputSet.subsetOf(e.references) =>
e.copy(child = prunedChild(child, e.references))
case s @ ScriptTransformation(_, _, _, child, _)
- if (child.outputSet -- s.references).nonEmpty =>
+ if !child.outputSet.subsetOf(s.references) =>
s.copy(child = prunedChild(child, s.references))
// prune unrequired references
@@ -579,7 +579,7 @@ object ColumnPruning extends Rule[LogicalPlan] {
case p @ Project(_, _: Distinct) => p
// Eliminate unneeded attributes from children of Union.
case p @ Project(_, u: Union) =>
- if ((u.outputSet -- p.references).nonEmpty) {
+ if (!u.outputSet.subsetOf(p.references)) {
val firstChild = u.children.head
val newOutput = prunedChild(firstChild, p.references).output
// pruning the columns of all children based on the pruned first child.
@@ -595,7 +595,7 @@ object ColumnPruning extends Rule[LogicalPlan] {
}
// Prune unnecessary window expressions
- case p @ Project(_, w: Window) if (w.windowOutputSet -- p.references).nonEmpty =>
+ case p @ Project(_, w: Window) if !w.windowOutputSet.subsetOf(p.references) =>
p.copy(child = w.copy(
windowExpressions = w.windowExpressions.filter(p.references.contains)))
@@ -611,7 +611,7 @@ object ColumnPruning extends Rule[LogicalPlan] {
// for all other logical plans that inherits the output from it's children
case p @ Project(_, child) =>
val required = child.references ++ p.references
- if ((child.inputSet -- required).nonEmpty) {
+ if (!child.inputSet.subsetOf(required)) {
val newChildren = child.children.map(c => prunedChild(c, required))
p.copy(child = child.withNewChildren(newChildren))
} else {
@@ -621,7 +621,7 @@ object ColumnPruning extends Rule[LogicalPlan] {
/** Applies a projection only when the child is producing unnecessary attributes */
private def prunedChild(c: LogicalPlan, allReferences: AttributeSet) =
- if ((c.outputSet -- allReferences.filter(c.outputSet.contains)).nonEmpty) {
+ if (!c.outputSet.subsetOf(allReferences)) {
Project(c.output.filter(allReferences.contains), c)
} else {
c
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/QueryPlan.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/QueryPlan.scala
index b1ffdca091461..ca0cea6ba7de3 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/QueryPlan.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/QueryPlan.scala
@@ -42,7 +42,7 @@ abstract class QueryPlan[PlanType <: QueryPlan[PlanType]] extends TreeNode[PlanT
* All Attributes that appear in expressions from this operator. Note that this set does not
* include attributes that are implicitly referenced by being passed through to the output tuple.
*/
- def references: AttributeSet = AttributeSet(expressions.flatMap(_.references))
+ def references: AttributeSet = AttributeSet.fromAttributeSets(expressions.map(_.references))
/**
* The set of all attributes that are input to this operator by its children.
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/CaseInsensitiveMap.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/CaseInsensitiveMap.scala
index bb2c5926ae9bb..288a4f34a447e 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/CaseInsensitiveMap.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/CaseInsensitiveMap.scala
@@ -42,7 +42,11 @@ class CaseInsensitiveMap[T] private (val originalMap: Map[String, T]) extends Ma
override def iterator: Iterator[(String, T)] = keyLowerCasedMap.iterator
override def -(key: String): Map[String, T] = {
- new CaseInsensitiveMap(originalMap.filterKeys(!_.equalsIgnoreCase(key)))
+ new CaseInsensitiveMap(originalMap.filter(!_._1.equalsIgnoreCase(key)))
+ }
+
+ override def filterKeys(p: (String) => Boolean): Map[String, T] = {
+ new CaseInsensitiveMap(originalMap.filter(kv => p(kv._1)))
}
}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
index 0e0a01def357e..f6c98805bfb15 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
@@ -451,8 +451,11 @@ object SQLConf {
.createWithDefault(10)
val PARQUET_WRITE_LEGACY_FORMAT = buildConf("spark.sql.parquet.writeLegacyFormat")
- .doc("Whether to be compatible with the legacy Parquet format adopted by Spark 1.4 and prior " +
- "versions, when converting Parquet schema to Spark SQL schema and vice versa.")
+ .doc("If true, data will be written in a way of Spark 1.4 and earlier. For example, decimal " +
+ "values will be written in Apache Parquet's fixed-length byte array format, which other " +
+ "systems such as Apache Hive and Apache Impala use. If false, the newer format in Parquet " +
+ "will be used. For example, decimals will be written in int-based format. If Parquet " +
+ "output is intended for use with systems that do not support this newer format, set to true.")
.booleanConf
.createWithDefault(false)
@@ -1295,15 +1298,15 @@ object SQLConf {
.booleanConf
.createWithDefault(true)
- val PANDAS_GROUPED_MAP_ASSIGN_COLUMNS_BY_POSITION =
- buildConf("spark.sql.execution.pandas.groupedMap.assignColumnsByPosition")
+ val PANDAS_GROUPED_MAP_ASSIGN_COLUMNS_BY_NAME =
+ buildConf("spark.sql.legacy.execution.pandas.groupedMap.assignColumnsByName")
.internal()
- .doc("When true, a grouped map Pandas UDF will assign columns from the returned " +
- "Pandas DataFrame based on position, regardless of column label type. When false, " +
- "columns will be looked up by name if labeled with a string and fallback to use " +
- "position if not. This configuration will be deprecated in future releases.")
+ .doc("When true, columns will be looked up by name if labeled with a string and fallback " +
+ "to use position if not. When false, a grouped map Pandas UDF will assign columns from " +
+ "the returned Pandas DataFrame based on position, regardless of column label type. " +
+ "This configuration will be deprecated in future releases.")
.booleanConf
- .createWithDefault(false)
+ .createWithDefault(true)
val REPLACE_EXCEPT_WITH_FILTER = buildConf("spark.sql.optimizer.replaceExceptWithFilter")
.internal()
@@ -1328,6 +1331,15 @@ object SQLConf {
.booleanConf
.createWithDefault(true)
+ val LITERAL_PICK_MINIMUM_PRECISION =
+ buildConf("spark.sql.legacy.literal.pickMinimumPrecision")
+ .internal()
+ .doc("When integral literal is used in decimal operations, pick a minimum precision " +
+ "required by the literal if this config is true, to make the resulting precision and/or " +
+ "scale smaller. This can reduce the possibility of precision lose and/or overflow.")
+ .booleanConf
+ .createWithDefault(true)
+
val SQL_OPTIONS_REDACTION_PATTERN =
buildConf("spark.sql.redaction.options.regex")
.doc("Regex to decide which keys in a Spark SQL command's options map contain sensitive " +
@@ -1915,13 +1927,15 @@ class SQLConf extends Serializable with Logging {
def pandasRespectSessionTimeZone: Boolean = getConf(PANDAS_RESPECT_SESSION_LOCAL_TIMEZONE)
- def pandasGroupedMapAssignColumnssByPosition: Boolean =
- getConf(SQLConf.PANDAS_GROUPED_MAP_ASSIGN_COLUMNS_BY_POSITION)
+ def pandasGroupedMapAssignColumnsByName: Boolean =
+ getConf(SQLConf.PANDAS_GROUPED_MAP_ASSIGN_COLUMNS_BY_NAME)
def replaceExceptWithFilter: Boolean = getConf(REPLACE_EXCEPT_WITH_FILTER)
def decimalOperationsAllowPrecisionLoss: Boolean = getConf(DECIMAL_OPERATIONS_ALLOW_PREC_LOSS)
+ def literalPickMinimumPrecision: Boolean = getConf(LITERAL_PICK_MINIMUM_PRECISION)
+
def continuousStreamingExecutorQueueSize: Int = getConf(CONTINUOUS_STREAMING_EXECUTOR_QUEUE_SIZE)
def continuousStreamingExecutorPollIntervalMs: Long =
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/plans/PlanTest.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/plans/PlanTest.scala
index 67740c3166471..3081ff935f043 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/plans/PlanTest.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/plans/PlanTest.scala
@@ -22,7 +22,6 @@ import org.scalatest.Suite
import org.scalatest.Tag
import org.apache.spark.SparkFunSuite
-import org.apache.spark.sql.AnalysisException
import org.apache.spark.sql.catalyst.analysis.SimpleAnalyzer
import org.apache.spark.sql.catalyst.expressions._
import org.apache.spark.sql.catalyst.expressions.CodegenObjectFactoryMode
@@ -57,7 +56,7 @@ trait CodegenInterpretedPlanTest extends PlanTest {
* Provides helper methods for comparing plans, but without the overhead of
* mandating a FunSuite.
*/
-trait PlanTestBase extends PredicateHelper { self: Suite =>
+trait PlanTestBase extends PredicateHelper with SQLHelper { self: Suite =>
// TODO(gatorsmile): remove this from PlanTest and all the analyzer rules
protected def conf = SQLConf.get
@@ -174,32 +173,4 @@ trait PlanTestBase extends PredicateHelper { self: Suite =>
plan1 == plan2
}
}
-
- /**
- * Sets all SQL configurations specified in `pairs`, calls `f`, and then restores all SQL
- * configurations.
- */
- protected def withSQLConf(pairs: (String, String)*)(f: => Unit): Unit = {
- val conf = SQLConf.get
- val (keys, values) = pairs.unzip
- val currentValues = keys.map { key =>
- if (conf.contains(key)) {
- Some(conf.getConfString(key))
- } else {
- None
- }
- }
- (keys, values).zipped.foreach { (k, v) =>
- if (SQLConf.staticConfKeys.contains(k)) {
- throw new AnalysisException(s"Cannot modify the value of a static config: $k")
- }
- conf.setConfString(k, v)
- }
- try f finally {
- keys.zip(currentValues).foreach {
- case (key, Some(value)) => conf.setConfString(key, value)
- case (key, None) => conf.unsetConf(key)
- }
- }
- }
}
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/plans/SQLHelper.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/plans/SQLHelper.scala
new file mode 100644
index 0000000000000..4d869d79ad594
--- /dev/null
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/plans/SQLHelper.scala
@@ -0,0 +1,64 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.catalyst.plans
+
+import java.io.File
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.util.Utils
+
+trait SQLHelper {
+
+ /**
+ * Sets all SQL configurations specified in `pairs`, calls `f`, and then restores all SQL
+ * configurations.
+ */
+ protected def withSQLConf(pairs: (String, String)*)(f: => Unit): Unit = {
+ val conf = SQLConf.get
+ val (keys, values) = pairs.unzip
+ val currentValues = keys.map { key =>
+ if (conf.contains(key)) {
+ Some(conf.getConfString(key))
+ } else {
+ None
+ }
+ }
+ (keys, values).zipped.foreach { (k, v) =>
+ if (SQLConf.staticConfKeys.contains(k)) {
+ throw new AnalysisException(s"Cannot modify the value of a static config: $k")
+ }
+ conf.setConfString(k, v)
+ }
+ try f finally {
+ keys.zip(currentValues).foreach {
+ case (key, Some(value)) => conf.setConfString(key, value)
+ case (key, None) => conf.unsetConf(key)
+ }
+ }
+ }
+
+ /**
+ * Generates a temporary path without creating the actual file/directory, then pass it to `f`. If
+ * a file/directory is created there by `f`, it will be delete after `f` returns.
+ */
+ protected def withTempPath(f: File => Unit): Unit = {
+ val path = Utils.createTempDir()
+ path.delete()
+ try f(path) finally Utils.deleteRecursively(path)
+ }
+}
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/util/CaseInsensitiveMapSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/util/CaseInsensitiveMapSuite.scala
new file mode 100644
index 0000000000000..03eed4aaa750b
--- /dev/null
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/util/CaseInsensitiveMapSuite.scala
@@ -0,0 +1,53 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.util
+
+import org.apache.spark.{SparkConf, SparkFunSuite}
+import org.apache.spark.serializer.JavaSerializer
+
+class CaseInsensitiveMapSuite extends SparkFunSuite {
+ private def shouldBeSerializable(m: Map[String, String]): Unit = {
+ new JavaSerializer(new SparkConf()).newInstance().serialize(m)
+ }
+
+ test("Keys are case insensitive") {
+ val m = CaseInsensitiveMap(Map("a" -> "b", "foO" -> "bar"))
+ assert(m("FOO") == "bar")
+ assert(m("fOo") == "bar")
+ assert(m("A") == "b")
+ shouldBeSerializable(m)
+ }
+
+ test("CaseInsensitiveMap should be serializable after '-' operator") {
+ val m = CaseInsensitiveMap(Map("a" -> "b", "foo" -> "bar")) - "a"
+ assert(m == Map("foo" -> "bar"))
+ shouldBeSerializable(m)
+ }
+
+ test("CaseInsensitiveMap should be serializable after '+' operator") {
+ val m = CaseInsensitiveMap(Map("a" -> "b", "foo" -> "bar")) + ("x" -> "y")
+ assert(m == Map("a" -> "b", "foo" -> "bar", "x" -> "y"))
+ shouldBeSerializable(m)
+ }
+
+ test("CaseInsensitiveMap should be serializable after 'filterKeys' method") {
+ val m = CaseInsensitiveMap(Map("a" -> "b", "foo" -> "bar")).filterKeys(_ == "foo")
+ assert(m == Map("foo" -> "bar"))
+ shouldBeSerializable(m)
+ }
+}
diff --git a/sql/core/benchmarks/SortBenchmark-results.txt b/sql/core/benchmarks/SortBenchmark-results.txt
new file mode 100644
index 0000000000000..0d00a0c89d02d
--- /dev/null
+++ b/sql/core/benchmarks/SortBenchmark-results.txt
@@ -0,0 +1,17 @@
+================================================================================================
+radix sort
+================================================================================================
+
+Java HotSpot(TM) 64-Bit Server VM 1.8.0_162-b12 on Mac OS X 10.13.6
+Intel(R) Core(TM) i7-7820HQ CPU @ 2.90GHz
+
+radix sort 25000000: Best/Avg Time(ms) Rate(M/s) Per Row(ns) Relative
+------------------------------------------------------------------------------------------------
+reference TimSort key prefix array 11770 / 11960 2.1 470.8 1.0X
+reference Arrays.sort 2106 / 2128 11.9 84.3 5.6X
+radix sort one byte 93 / 100 269.7 3.7 126.9X
+radix sort two bytes 171 / 179 146.0 6.9 68.7X
+radix sort eight bytes 659 / 664 37.9 26.4 17.9X
+radix sort key prefix array 1024 / 1053 24.4 41.0 11.5X
+
+
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/arrow/ArrowUtils.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/arrow/ArrowUtils.scala
index 533097ac399e9..b1e8fb39ac9de 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/arrow/ArrowUtils.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/arrow/ArrowUtils.scala
@@ -131,11 +131,8 @@ object ArrowUtils {
} else {
Nil
}
- val pandasColsByPosition = if (conf.pandasGroupedMapAssignColumnssByPosition) {
- Seq(SQLConf.PANDAS_GROUPED_MAP_ASSIGN_COLUMNS_BY_POSITION.key -> "true")
- } else {
- Nil
- }
- Map(timeZoneConf ++ pandasColsByPosition: _*)
+ val pandasColsByName = Seq(SQLConf.PANDAS_GROUPED_MAP_ASSIGN_COLUMNS_BY_NAME.key ->
+ conf.pandasGroupedMapAssignColumnsByName.toString)
+ Map(timeZoneConf ++ pandasColsByName: _*)
}
}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/functions.scala b/sql/core/src/main/scala/org/apache/spark/sql/functions.scala
index 10b67d7a1ca54..4c58e77df485e 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/functions.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/functions.scala
@@ -3619,6 +3619,8 @@ object functions {
* @param e a column containing a struct, an array or a map.
* @param options options to control how the struct column is converted into a json string.
* accepts the same options and the json data source.
+ * Additionally the function supports the `pretty` option which enables
+ * pretty JSON generation.
*
* @group collection_funcs
* @since 2.1.0
@@ -3635,6 +3637,8 @@ object functions {
* @param e a column containing a struct, an array or a map.
* @param options options to control how the struct column is converted into a json string.
* accepts the same options and the json data source.
+ * Additionally the function supports the `pretty` option which enables
+ * pretty JSON generation.
*
* @group collection_funcs
* @since 2.1.0
diff --git a/sql/core/src/test/resources/sql-tests/inputs/udaf-regrfunctions.sql b/sql/core/src/test/resources/sql-tests/inputs/udaf-regrfunctions.sql
deleted file mode 100644
index 92c7e26e3add2..0000000000000
--- a/sql/core/src/test/resources/sql-tests/inputs/udaf-regrfunctions.sql
+++ /dev/null
@@ -1,56 +0,0 @@
---
--- Licensed to the Apache Software Foundation (ASF) under one or more
--- contributor license agreements. See the NOTICE file distributed with
--- this work for additional information regarding copyright ownership.
--- The ASF licenses this file to You under the Apache License, Version 2.0
--- (the "License"); you may not use this file except in compliance with
--- the License. You may obtain a copy of the License at
---
--- http://www.apache.org/licenses/LICENSE-2.0
---
--- Unless required by applicable law or agreed to in writing, software
--- distributed under the License is distributed on an "AS IS" BASIS,
--- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
--- See the License for the specific language governing permissions and
--- limitations under the License.
---
-
-CREATE OR REPLACE TEMPORARY VIEW t1 AS SELECT * FROM VALUES
- (101, 1, 1, 1),
- (201, 2, 1, 1),
- (301, 3, 1, 1),
- (401, 4, 1, 11),
- (501, 5, 1, null),
- (601, 6, null, 1),
- (701, 6, null, null),
- (102, 1, 2, 2),
- (202, 2, 1, 2),
- (302, 3, 2, 1),
- (402, 4, 2, 12),
- (502, 5, 2, null),
- (602, 6, null, 2),
- (702, 6, null, null),
- (103, 1, 3, 3),
- (203, 2, 1, 3),
- (303, 3, 3, 1),
- (403, 4, 3, 13),
- (503, 5, 3, null),
- (603, 6, null, 3),
- (703, 6, null, null),
- (104, 1, 4, 4),
- (204, 2, 1, 4),
- (304, 3, 4, 1),
- (404, 4, 4, 14),
- (504, 5, 4, null),
- (604, 6, null, 4),
- (704, 6, null, null),
- (800, 7, 1, 1)
-as t1(id, px, y, x);
-
-select px, var_pop(x), var_pop(y), corr(y,x), covar_samp(y,x), covar_pop(y,x), regr_count(y,x),
- regr_slope(y,x), regr_intercept(y,x), regr_r2(y,x), regr_sxx(y,x), regr_syy(y,x), regr_sxy(y,x),
- regr_avgx(y,x), regr_avgy(y,x), regr_count(y,x)
-from t1 group by px order by px;
-
-
-select id, regr_count(y,x) over (partition by px) from t1 order by id;
diff --git a/sql/core/src/test/resources/sql-tests/results/udaf-regrfunctions.sql.out b/sql/core/src/test/resources/sql-tests/results/udaf-regrfunctions.sql.out
deleted file mode 100644
index d7d009a64bf84..0000000000000
--- a/sql/core/src/test/resources/sql-tests/results/udaf-regrfunctions.sql.out
+++ /dev/null
@@ -1,93 +0,0 @@
--- Automatically generated by SQLQueryTestSuite
--- Number of queries: 3
-
-
--- !query 0
-CREATE OR REPLACE TEMPORARY VIEW t1 AS SELECT * FROM VALUES
- (101, 1, 1, 1),
- (201, 2, 1, 1),
- (301, 3, 1, 1),
- (401, 4, 1, 11),
- (501, 5, 1, null),
- (601, 6, null, 1),
- (701, 6, null, null),
- (102, 1, 2, 2),
- (202, 2, 1, 2),
- (302, 3, 2, 1),
- (402, 4, 2, 12),
- (502, 5, 2, null),
- (602, 6, null, 2),
- (702, 6, null, null),
- (103, 1, 3, 3),
- (203, 2, 1, 3),
- (303, 3, 3, 1),
- (403, 4, 3, 13),
- (503, 5, 3, null),
- (603, 6, null, 3),
- (703, 6, null, null),
- (104, 1, 4, 4),
- (204, 2, 1, 4),
- (304, 3, 4, 1),
- (404, 4, 4, 14),
- (504, 5, 4, null),
- (604, 6, null, 4),
- (704, 6, null, null),
- (800, 7, 1, 1)
-as t1(id, px, y, x)
--- !query 0 schema
-struct<>
--- !query 0 output
-
-
-
--- !query 1
-select px, var_pop(x), var_pop(y), corr(y,x), covar_samp(y,x), covar_pop(y,x), regr_count(y,x),
- regr_slope(y,x), regr_intercept(y,x), regr_r2(y,x), regr_sxx(y,x), regr_syy(y,x), regr_sxy(y,x),
- regr_avgx(y,x), regr_avgy(y,x), regr_count(y,x)
-from t1 group by px order by px
--- !query 1 schema
-struct