mesos://HOST:PORT | Connect to the given Mesos cluster.
The port must be whichever one your is configured to use, which is 5050 by default.
Or, for a Mesos cluster using ZooKeeper, use mesos://zk://....
diff --git a/mllib/src/main/scala/org/apache/spark/ml/regression/GeneralizedLinearRegression.scala b/mllib/src/main/scala/org/apache/spark/ml/regression/GeneralizedLinearRegression.scala
index 33137b0c0fdec..d6093a01c671c 100644
--- a/mllib/src/main/scala/org/apache/spark/ml/regression/GeneralizedLinearRegression.scala
+++ b/mllib/src/main/scala/org/apache/spark/ml/regression/GeneralizedLinearRegression.scala
@@ -1133,7 +1133,8 @@ class GeneralizedLinearRegressionSummary private[regression] (
private[regression] lazy val link: Link = familyLink.link
/** Number of instances in DataFrame predictions. */
- private[regression] lazy val numInstances: Long = predictions.count()
+ @Since("2.2.0")
+ lazy val numInstances: Long = predictions.count()
/** The numeric rank of the fitted linear model. */
@Since("2.0.0")
diff --git a/mllib/src/main/scala/org/apache/spark/ml/regression/LinearRegression.scala b/mllib/src/main/scala/org/apache/spark/ml/regression/LinearRegression.scala
index 45df1d9be647d..f7e3c8fa5b6e6 100644
--- a/mllib/src/main/scala/org/apache/spark/ml/regression/LinearRegression.scala
+++ b/mllib/src/main/scala/org/apache/spark/ml/regression/LinearRegression.scala
@@ -696,7 +696,8 @@ class LinearRegressionSummary private[regression] (
lazy val numInstances: Long = predictions.count()
/** Degrees of freedom */
- private val degreesOfFreedom: Long = if (privateModel.getFitIntercept) {
+ @Since("2.2.0")
+ val degreesOfFreedom: Long = if (privateModel.getFitIntercept) {
numInstances - privateModel.coefficients.size - 1
} else {
numInstances - privateModel.coefficients.size
diff --git a/python/pyspark/broadcast.py b/python/pyspark/broadcast.py
index 74dee1420754a..b1b59f73d6718 100644
--- a/python/pyspark/broadcast.py
+++ b/python/pyspark/broadcast.py
@@ -21,6 +21,7 @@
from tempfile import NamedTemporaryFile
from pyspark.cloudpickle import print_exec
+from pyspark.util import _exception_message
if sys.version < '3':
import cPickle as pickle
@@ -82,7 +83,8 @@ def dump(self, value, f):
except pickle.PickleError:
raise
except Exception as e:
- msg = "Could not serialize broadcast: " + e.__class__.__name__ + ": " + e.message
+ msg = "Could not serialize broadcast: %s: %s" \
+ % (e.__class__.__name__, _exception_message(e))
print_exec(sys.stderr)
raise pickle.PicklingError(msg)
f.close()
diff --git a/python/pyspark/cloudpickle.py b/python/pyspark/cloudpickle.py
index 959fb8b357f99..389bee7eee6e9 100644
--- a/python/pyspark/cloudpickle.py
+++ b/python/pyspark/cloudpickle.py
@@ -56,6 +56,7 @@
import traceback
import weakref
+from pyspark.util import _exception_message
if sys.version < '3':
from pickle import Pickler
@@ -152,13 +153,13 @@ def dump(self, obj):
except pickle.PickleError:
raise
except Exception as e:
- if "'i' format requires" in e.message:
- msg = "Object too large to serialize: " + e.message
+ emsg = _exception_message(e)
+ if "'i' format requires" in emsg:
+ msg = "Object too large to serialize: %s" % emsg
else:
- msg = "Could not serialize object: " + e.__class__.__name__ + ": " + e.message
+ msg = "Could not serialize object: %s: %s" % (e.__class__.__name__, emsg)
print_exec(sys.stderr)
raise pickle.PicklingError(msg)
-
def save_memoryview(self, obj):
"""Fallback to save_string"""
diff --git a/python/pyspark/util.py b/python/pyspark/util.py
new file mode 100644
index 0000000000000..e5d332ce54429
--- /dev/null
+++ b/python/pyspark/util.py
@@ -0,0 +1,45 @@
+# -*- coding: utf-8 -*-
+#
+# 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.
+#
+
+__all__ = []
+
+
+def _exception_message(excp):
+ """Return the message from an exception as either a str or unicode object. Supports both
+ Python 2 and Python 3.
+
+ >>> msg = "Exception message"
+ >>> excp = Exception(msg)
+ >>> msg == _exception_message(excp)
+ True
+
+ >>> msg = u"unicöde"
+ >>> excp = Exception(msg)
+ >>> msg == _exception_message(excp)
+ True
+ """
+ if hasattr(excp, "message"):
+ return excp.message
+ return str(excp)
+
+
+if __name__ == "__main__":
+ import doctest
+ (failure_count, test_count) = doctest.testmod()
+ if failure_count:
+ exit(-1)
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/JavaTypeInference.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/JavaTypeInference.scala
index 9d4617dda555f..86a73a319ec3f 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/JavaTypeInference.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/JavaTypeInference.scala
@@ -204,20 +204,19 @@ object JavaTypeInference {
typeToken.getRawType match {
case c if !inferExternalType(c).isInstanceOf[ObjectType] => getPath
- case c if c == classOf[java.lang.Short] =>
- NewInstance(c, getPath :: Nil, ObjectType(c))
- case c if c == classOf[java.lang.Integer] =>
- NewInstance(c, getPath :: Nil, ObjectType(c))
- case c if c == classOf[java.lang.Long] =>
- NewInstance(c, getPath :: Nil, ObjectType(c))
- case c if c == classOf[java.lang.Double] =>
- NewInstance(c, getPath :: Nil, ObjectType(c))
- case c if c == classOf[java.lang.Byte] =>
- NewInstance(c, getPath :: Nil, ObjectType(c))
- case c if c == classOf[java.lang.Float] =>
- NewInstance(c, getPath :: Nil, ObjectType(c))
- case c if c == classOf[java.lang.Boolean] =>
- NewInstance(c, getPath :: Nil, ObjectType(c))
+ case c if c == classOf[java.lang.Short] ||
+ c == classOf[java.lang.Integer] ||
+ c == classOf[java.lang.Long] ||
+ c == classOf[java.lang.Double] ||
+ c == classOf[java.lang.Float] ||
+ c == classOf[java.lang.Byte] ||
+ c == classOf[java.lang.Boolean] =>
+ StaticInvoke(
+ c,
+ ObjectType(c),
+ "valueOf",
+ getPath :: Nil,
+ propagateNull = true)
case c if c == classOf[java.sql.Date] =>
StaticInvoke(
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/ScalaReflection.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/ScalaReflection.scala
index 198122759e4ad..82710a2a183ab 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/ScalaReflection.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/ScalaReflection.scala
@@ -132,7 +132,7 @@ object ScalaReflection extends ScalaReflection {
def deserializerFor[T : TypeTag]: Expression = {
val tpe = localTypeOf[T]
val clsName = getClassNameFromType(tpe)
- val walkedTypePath = s"""- root class: "${clsName}"""" :: Nil
+ val walkedTypePath = s"""- root class: "$clsName"""" :: Nil
deserializerFor(tpe, None, walkedTypePath)
}
@@ -204,37 +204,37 @@ object ScalaReflection extends ScalaReflection {
case t if t <:< localTypeOf[java.lang.Integer] =>
val boxedType = classOf[java.lang.Integer]
val objectType = ObjectType(boxedType)
- NewInstance(boxedType, getPath :: Nil, objectType)
+ StaticInvoke(boxedType, objectType, "valueOf", getPath :: Nil, propagateNull = true)
case t if t <:< localTypeOf[java.lang.Long] =>
val boxedType = classOf[java.lang.Long]
val objectType = ObjectType(boxedType)
- NewInstance(boxedType, getPath :: Nil, objectType)
+ StaticInvoke(boxedType, objectType, "valueOf", getPath :: Nil, propagateNull = true)
case t if t <:< localTypeOf[java.lang.Double] =>
val boxedType = classOf[java.lang.Double]
val objectType = ObjectType(boxedType)
- NewInstance(boxedType, getPath :: Nil, objectType)
+ StaticInvoke(boxedType, objectType, "valueOf", getPath :: Nil, propagateNull = true)
case t if t <:< localTypeOf[java.lang.Float] =>
val boxedType = classOf[java.lang.Float]
val objectType = ObjectType(boxedType)
- NewInstance(boxedType, getPath :: Nil, objectType)
+ StaticInvoke(boxedType, objectType, "valueOf", getPath :: Nil, propagateNull = true)
case t if t <:< localTypeOf[java.lang.Short] =>
val boxedType = classOf[java.lang.Short]
val objectType = ObjectType(boxedType)
- NewInstance(boxedType, getPath :: Nil, objectType)
+ StaticInvoke(boxedType, objectType, "valueOf", getPath :: Nil, propagateNull = true)
case t if t <:< localTypeOf[java.lang.Byte] =>
val boxedType = classOf[java.lang.Byte]
val objectType = ObjectType(boxedType)
- NewInstance(boxedType, getPath :: Nil, objectType)
+ StaticInvoke(boxedType, objectType, "valueOf", getPath :: Nil, propagateNull = true)
case t if t <:< localTypeOf[java.lang.Boolean] =>
val boxedType = classOf[java.lang.Boolean]
val objectType = ObjectType(boxedType)
- NewInstance(boxedType, getPath :: Nil, objectType)
+ StaticInvoke(boxedType, objectType, "valueOf", getPath :: Nil, propagateNull = true)
case t if t <:< localTypeOf[java.sql.Date] =>
StaticInvoke(
@@ -270,12 +270,14 @@ object ScalaReflection extends ScalaReflection {
case t if t <:< localTypeOf[Array[_]] =>
val TypeRef(_, _, Seq(elementType)) = t
- val Schema(_, elementNullable) = schemaFor(elementType)
+ val Schema(dataType, elementNullable) = schemaFor(elementType)
val className = getClassNameFromType(elementType)
val newTypePath = s"""- array element class: "$className"""" +: walkedTypePath
- val mapFunction: Expression => Expression = p => {
- val converter = deserializerFor(elementType, Some(p), newTypePath)
+ val mapFunction: Expression => Expression = element => {
+ // upcast the array element to the data type the encoder expected.
+ val casted = upCastToExpectedType(element, dataType, newTypePath)
+ val converter = deserializerFor(elementType, Some(casted), newTypePath)
if (elementNullable) {
converter
} else {
@@ -305,12 +307,14 @@ object ScalaReflection extends ScalaReflection {
case t if t <:< localTypeOf[Seq[_]] =>
val TypeRef(_, _, Seq(elementType)) = t
- val Schema(_, elementNullable) = schemaFor(elementType)
+ val Schema(dataType, elementNullable) = schemaFor(elementType)
val className = getClassNameFromType(elementType)
val newTypePath = s"""- array element class: "$className"""" +: walkedTypePath
- val mapFunction: Expression => Expression = p => {
- val converter = deserializerFor(elementType, Some(p), newTypePath)
+ val mapFunction: Expression => Expression = element => {
+ // upcast the array element to the data type the encoder expected.
+ val casted = upCastToExpectedType(element, dataType, newTypePath)
+ val converter = deserializerFor(elementType, Some(casted), newTypePath)
if (elementNullable) {
converter
} else {
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
index b0cdef70297cf..9816b33ae8dff 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
@@ -26,7 +26,7 @@ import org.apache.spark.sql.catalyst.catalog._
import org.apache.spark.sql.catalyst.encoders.OuterScopes
import org.apache.spark.sql.catalyst.expressions._
import org.apache.spark.sql.catalyst.expressions.aggregate._
-import org.apache.spark.sql.catalyst.expressions.objects.{MapObjects, NewInstance, UnresolvedMapObjects}
+import org.apache.spark.sql.catalyst.expressions.objects.{LambdaVariable, MapObjects, NewInstance, UnresolvedMapObjects}
import org.apache.spark.sql.catalyst.expressions.SubExprUtils._
import org.apache.spark.sql.catalyst.optimizer.BooleanSimplification
import org.apache.spark.sql.catalyst.plans._
@@ -2321,7 +2321,11 @@ class Analyzer(
*/
object ResolveUpCast extends Rule[LogicalPlan] {
private def fail(from: Expression, to: DataType, walkedTypePath: Seq[String]) = {
- throw new AnalysisException(s"Cannot up cast ${from.sql} from " +
+ val fromStr = from match {
+ case l: LambdaVariable => "array element"
+ case e => e.sql
+ }
+ throw new AnalysisException(s"Cannot up cast $fromStr from " +
s"${from.dataType.simpleString} to ${to.simpleString} as it may truncate\n" +
"The type path of the target object is:\n" + walkedTypePath.mkString("", "\n", "\n") +
"You can either add an explicit cast to the input data or choose a higher precision " +
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/TypeCoercion.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/TypeCoercion.scala
index 768897dc0713c..e1dd010d37a95 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/TypeCoercion.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/TypeCoercion.scala
@@ -571,6 +571,7 @@ object TypeCoercion {
NaNvl(l, Cast(r, DoubleType))
case NaNvl(l, r) if l.dataType == FloatType && r.dataType == DoubleType =>
NaNvl(Cast(l, DoubleType), r)
+ case NaNvl(l, r) if r.dataType == NullType => NaNvl(l, Cast(r, l.dataType))
}
}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/predicates.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/predicates.scala
index 8acb740f8db8c..5034566132f7a 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/predicates.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/predicates.scala
@@ -92,11 +92,12 @@ trait PredicateHelper {
protected def canEvaluateWithinJoin(expr: Expression): Boolean = expr match {
// Non-deterministic expressions are not allowed as join conditions.
case e if !e.deterministic => false
- case l: ListQuery =>
+ case _: ListQuery | _: Exists =>
// A ListQuery defines the query which we want to search in an IN subquery expression.
// Currently the only way to evaluate an IN subquery is to convert it to a
// LeftSemi/LeftAnti/ExistenceJoin by `RewritePredicateSubquery` rule.
// It cannot be evaluated as part of a Join operator.
+ // An Exists shouldn't be push into a Join operator too.
false
case e: SubqueryExpression =>
// non-correlated subquery will be replaced as literal
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/TypeCoercionSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/TypeCoercionSuite.scala
index 3e0c357b6de42..011d09ff60641 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/TypeCoercionSuite.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/TypeCoercionSuite.scala
@@ -656,14 +656,20 @@ class TypeCoercionSuite extends PlanTest {
test("nanvl casts") {
ruleTest(TypeCoercion.FunctionArgumentConversion,
- NaNvl(Literal.create(1.0, FloatType), Literal.create(1.0, DoubleType)),
- NaNvl(Cast(Literal.create(1.0, FloatType), DoubleType), Literal.create(1.0, DoubleType)))
+ NaNvl(Literal.create(1.0f, FloatType), Literal.create(1.0, DoubleType)),
+ NaNvl(Cast(Literal.create(1.0f, FloatType), DoubleType), Literal.create(1.0, DoubleType)))
ruleTest(TypeCoercion.FunctionArgumentConversion,
- NaNvl(Literal.create(1.0, DoubleType), Literal.create(1.0, FloatType)),
- NaNvl(Literal.create(1.0, DoubleType), Cast(Literal.create(1.0, FloatType), DoubleType)))
+ NaNvl(Literal.create(1.0, DoubleType), Literal.create(1.0f, FloatType)),
+ NaNvl(Literal.create(1.0, DoubleType), Cast(Literal.create(1.0f, FloatType), DoubleType)))
ruleTest(TypeCoercion.FunctionArgumentConversion,
NaNvl(Literal.create(1.0, DoubleType), Literal.create(1.0, DoubleType)),
NaNvl(Literal.create(1.0, DoubleType), Literal.create(1.0, DoubleType)))
+ ruleTest(TypeCoercion.FunctionArgumentConversion,
+ NaNvl(Literal.create(1.0f, FloatType), Literal.create(null, NullType)),
+ NaNvl(Literal.create(1.0f, FloatType), Cast(Literal.create(null, NullType), FloatType)))
+ ruleTest(TypeCoercion.FunctionArgumentConversion,
+ NaNvl(Literal.create(1.0, DoubleType), Literal.create(null, NullType)),
+ NaNvl(Literal.create(1.0, DoubleType), Cast(Literal.create(null, NullType), DoubleType)))
}
test("type coercion for If") {
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/encoders/EncoderResolutionSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/encoders/EncoderResolutionSuite.scala
index e5a3e1fd374dc..630e8a7990e7b 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/encoders/EncoderResolutionSuite.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/encoders/EncoderResolutionSuite.scala
@@ -33,6 +33,8 @@ case class StringIntClass(a: String, b: Int)
case class ComplexClass(a: Long, b: StringLongClass)
+case class PrimitiveArrayClass(arr: Array[Long])
+
case class ArrayClass(arr: Seq[StringIntClass])
case class NestedArrayClass(nestedArr: Array[ArrayClass])
@@ -66,6 +68,27 @@ class EncoderResolutionSuite extends PlanTest {
encoder.resolveAndBind(attrs).fromRow(InternalRow(InternalRow(str, 1.toByte), 2))
}
+ test("real type doesn't match encoder schema but they are compatible: primitive array") {
+ val encoder = ExpressionEncoder[PrimitiveArrayClass]
+ val attrs = Seq('arr.array(IntegerType))
+ val array = new GenericArrayData(Array(1, 2, 3))
+ encoder.resolveAndBind(attrs).fromRow(InternalRow(array))
+ }
+
+ test("the real type is not compatible with encoder schema: primitive array") {
+ val encoder = ExpressionEncoder[PrimitiveArrayClass]
+ val attrs = Seq('arr.array(StringType))
+ assert(intercept[AnalysisException](encoder.resolveAndBind(attrs)).message ==
+ s"""
+ |Cannot up cast array element from string to bigint as it may truncate
+ |The type path of the target object is:
+ |- array element class: "scala.Long"
+ |- field (class: "scala.Array", name: "arr")
+ |- root class: "org.apache.spark.sql.catalyst.encoders.PrimitiveArrayClass"
+ |You can either add an explicit cast to the input data or choose a higher precision type
+ """.stripMargin.trim + " of the field in the target object")
+ }
+
test("real type doesn't match encoder schema but they are compatible: array") {
val encoder = ExpressionEncoder[ArrayClass]
val attrs = Seq('arr.array(new StructType().add("a", "int").add("b", "int").add("c", "int")))
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/DataFrameNaFunctions.scala b/sql/core/src/main/scala/org/apache/spark/sql/DataFrameNaFunctions.scala
index 93d565d9fe904..052d85ad33bd6 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/DataFrameNaFunctions.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/DataFrameNaFunctions.scala
@@ -408,8 +408,7 @@ final class DataFrameNaFunctions private[sql](df: DataFrame) {
val quotedColName = "`" + col.name + "`"
val colValue = col.dataType match {
case DoubleType | FloatType =>
- // nanvl only supports these types
- nanvl(df.col(quotedColName), lit(null).cast(col.dataType))
+ nanvl(df.col(quotedColName), lit(null)) // nanvl only supports these types
case _ => df.col(quotedColName)
}
coalesce(colValue, lit(replacement).cast(col.dataType)).as(col.name)
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/SubquerySuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/SubquerySuite.scala
index 5fe6667ceca18..0f0199cbe2777 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/SubquerySuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/SubquerySuite.scala
@@ -844,4 +844,14 @@ class SubquerySuite extends QueryTest with SharedSQLContext {
Row(0) :: Row(1) :: Nil)
}
}
+
+ test("ListQuery and Exists should work even no correlated references") {
+ checkAnswer(
+ sql("select * from l, r where l.a = r.c AND (r.d in (select d from r) OR l.a >= 1)"),
+ Row(2, 1.0, 2, 3.0) :: Row(2, 1.0, 2, 3.0) :: Row(2, 1.0, 2, 3.0) ::
+ Row(2, 1.0, 2, 3.0) :: Row(3.0, 3.0, 3, 2.0) :: Row(6, null, 6, null) :: Nil)
+ checkAnswer(
+ sql("select * from l, r where l.a = r.c + 1 AND (exists (select * from r) OR l.a = r.c)"),
+ Row(3, 3.0, 2, 3.0) :: Row(3, 3.0, 2, 3.0) :: Nil)
+ }
}
|