Skip to content

Commit

Permalink
another fix
Browse files Browse the repository at this point in the history
  • Loading branch information
@cloud-fan
cloud-fan committed Sep 5, 2018
1 parent 2b2c894 commit 9e060a4
Show file tree
Hide file tree
Showing 3 changed files with 55 additions and 53 deletions.
50 changes: 25 additions & 25 deletions sql/core/src/main/scala/org/apache/spark/sql/UDFRegistration.scala
View file
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,7 @@ import org.apache.spark.sql.catalyst.analysis.FunctionRegistry
import org.apache.spark.sql.catalyst.expressions.{Expression, ScalaUDF}
import org.apache.spark.sql.execution.aggregate.ScalaUDAF
import org.apache.spark.sql.execution.python.UserDefinedPythonFunction
import org.apache.spark.sql.expressions.{UserDefinedAggregateFunction, UserDefinedFunction}
import org.apache.spark.sql.expressions.{SparkUserDefinedFunction, UserDefinedAggregateFunction, UserDefinedFunction}
import org.apache.spark.sql.types.DataType
import org.apache.spark.util.Utils

Expand Down Expand Up @@ -131,7 +131,7 @@ class UDFRegistration private[sql] (functionRegistry: FunctionRegistry) extends
| ". Expected: $x; Found: " + e.length)
| }
| functionRegistry.createOrReplaceTempFunction(name, builder)
| val udf = UserDefinedFunction.create(func, dataType, inputSchemas).withName(name)
| val udf = SparkUserDefinedFunction.create(func, dataType, inputSchemas).withName(name)
| if (nullable) udf else udf.asNonNullable()
|}""".stripMargin)
}
Expand Down Expand Up @@ -177,7 +177,7 @@ class UDFRegistration private[sql] (functionRegistry: FunctionRegistry) extends
". Expected: 0; Found: " + e.length)
}
functionRegistry.createOrReplaceTempFunction(name, builder)
val udf = UserDefinedFunction.create(func, dataType, inputSchemas).withName(name)
val udf = SparkUserDefinedFunction.create(func, dataType, inputSchemas).withName(name)
if (nullable) udf else udf.asNonNullable()
}

Expand All @@ -197,7 +197,7 @@ class UDFRegistration private[sql] (functionRegistry: FunctionRegistry) extends
". Expected: 1; Found: " + e.length)
}
functionRegistry.createOrReplaceTempFunction(name, builder)
val udf = UserDefinedFunction.create(func, dataType, inputSchemas).withName(name)
val udf = SparkUserDefinedFunction.create(func, dataType, inputSchemas).withName(name)
if (nullable) udf else udf.asNonNullable()
}

Expand All @@ -217,7 +217,7 @@ class UDFRegistration private[sql] (functionRegistry: FunctionRegistry) extends
". Expected: 2; Found: " + e.length)
}
functionRegistry.createOrReplaceTempFunction(name, builder)
val udf = UserDefinedFunction.create(func, dataType, inputSchemas).withName(name)
val udf = SparkUserDefinedFunction.create(func, dataType, inputSchemas).withName(name)
if (nullable) udf else udf.asNonNullable()
}

Expand All @@ -237,7 +237,7 @@ class UDFRegistration private[sql] (functionRegistry: FunctionRegistry) extends
". Expected: 3; Found: " + e.length)
}
functionRegistry.createOrReplaceTempFunction(name, builder)
val udf = UserDefinedFunction.create(func, dataType, inputSchemas).withName(name)
val udf = SparkUserDefinedFunction.create(func, dataType, inputSchemas).withName(name)
if (nullable) udf else udf.asNonNullable()
}

Expand All @@ -257,7 +257,7 @@ class UDFRegistration private[sql] (functionRegistry: FunctionRegistry) extends
". Expected: 4; Found: " + e.length)
}
functionRegistry.createOrReplaceTempFunction(name, builder)
val udf = UserDefinedFunction.create(func, dataType, inputSchemas).withName(name)
val udf = SparkUserDefinedFunction.create(func, dataType, inputSchemas).withName(name)
if (nullable) udf else udf.asNonNullable()
}

Expand All @@ -277,7 +277,7 @@ class UDFRegistration private[sql] (functionRegistry: FunctionRegistry) extends
". Expected: 5; Found: " + e.length)
}
functionRegistry.createOrReplaceTempFunction(name, builder)
val udf = UserDefinedFunction.create(func, dataType, inputSchemas).withName(name)
val udf = SparkUserDefinedFunction.create(func, dataType, inputSchemas).withName(name)
if (nullable) udf else udf.asNonNullable()
}

Expand All @@ -297,7 +297,7 @@ class UDFRegistration private[sql] (functionRegistry: FunctionRegistry) extends
". Expected: 6; Found: " + e.length)
}
functionRegistry.createOrReplaceTempFunction(name, builder)
val udf = UserDefinedFunction.create(func, dataType, inputSchemas).withName(name)
val udf = SparkUserDefinedFunction.create(func, dataType, inputSchemas).withName(name)
if (nullable) udf else udf.asNonNullable()
}

Expand All @@ -317,7 +317,7 @@ class UDFRegistration private[sql] (functionRegistry: FunctionRegistry) extends
". Expected: 7; Found: " + e.length)
}
functionRegistry.createOrReplaceTempFunction(name, builder)
val udf = UserDefinedFunction.create(func, dataType, inputSchemas).withName(name)
val udf = SparkUserDefinedFunction.create(func, dataType, inputSchemas).withName(name)
if (nullable) udf else udf.asNonNullable()
}

Expand All @@ -337,7 +337,7 @@ class UDFRegistration private[sql] (functionRegistry: FunctionRegistry) extends
". Expected: 8; Found: " + e.length)
}
functionRegistry.createOrReplaceTempFunction(name, builder)
val udf = UserDefinedFunction.create(func, dataType, inputSchemas).withName(name)
val udf = SparkUserDefinedFunction.create(func, dataType, inputSchemas).withName(name)
if (nullable) udf else udf.asNonNullable()
}

Expand All @@ -357,7 +357,7 @@ class UDFRegistration private[sql] (functionRegistry: FunctionRegistry) extends
". Expected: 9; Found: " + e.length)
}
functionRegistry.createOrReplaceTempFunction(name, builder)
val udf = UserDefinedFunction.create(func, dataType, inputSchemas).withName(name)
val udf = SparkUserDefinedFunction.create(func, dataType, inputSchemas).withName(name)
if (nullable) udf else udf.asNonNullable()
}

Expand All @@ -377,7 +377,7 @@ class UDFRegistration private[sql] (functionRegistry: FunctionRegistry) extends
". Expected: 10; Found: " + e.length)
}
functionRegistry.createOrReplaceTempFunction(name, builder)
val udf = UserDefinedFunction.create(func, dataType, inputSchemas).withName(name)
val udf = SparkUserDefinedFunction.create(func, dataType, inputSchemas).withName(name)
if (nullable) udf else udf.asNonNullable()
}

Expand All @@ -397,7 +397,7 @@ class UDFRegistration private[sql] (functionRegistry: FunctionRegistry) extends
". Expected: 11; Found: " + e.length)
}
functionRegistry.createOrReplaceTempFunction(name, builder)
val udf = UserDefinedFunction.create(func, dataType, inputSchemas).withName(name)
val udf = SparkUserDefinedFunction.create(func, dataType, inputSchemas).withName(name)
if (nullable) udf else udf.asNonNullable()
}

Expand All @@ -417,7 +417,7 @@ class UDFRegistration private[sql] (functionRegistry: FunctionRegistry) extends
". Expected: 12; Found: " + e.length)
}
functionRegistry.createOrReplaceTempFunction(name, builder)
val udf = UserDefinedFunction.create(func, dataType, inputSchemas).withName(name)
val udf = SparkUserDefinedFunction.create(func, dataType, inputSchemas).withName(name)
if (nullable) udf else udf.asNonNullable()
}

Expand All @@ -437,7 +437,7 @@ class UDFRegistration private[sql] (functionRegistry: FunctionRegistry) extends
". Expected: 13; Found: " + e.length)
}
functionRegistry.createOrReplaceTempFunction(name, builder)
val udf = UserDefinedFunction.create(func, dataType, inputSchemas).withName(name)
val udf = SparkUserDefinedFunction.create(func, dataType, inputSchemas).withName(name)
if (nullable) udf else udf.asNonNullable()
}

Expand All @@ -457,7 +457,7 @@ class UDFRegistration private[sql] (functionRegistry: FunctionRegistry) extends
". Expected: 14; Found: " + e.length)
}
functionRegistry.createOrReplaceTempFunction(name, builder)
val udf = UserDefinedFunction.create(func, dataType, inputSchemas).withName(name)
val udf = SparkUserDefinedFunction.create(func, dataType, inputSchemas).withName(name)
if (nullable) udf else udf.asNonNullable()
}

Expand All @@ -477,7 +477,7 @@ class UDFRegistration private[sql] (functionRegistry: FunctionRegistry) extends
". Expected: 15; Found: " + e.length)
}
functionRegistry.createOrReplaceTempFunction(name, builder)
val udf = UserDefinedFunction.create(func, dataType, inputSchemas).withName(name)
val udf = SparkUserDefinedFunction.create(func, dataType, inputSchemas).withName(name)
if (nullable) udf else udf.asNonNullable()
}

Expand All @@ -497,7 +497,7 @@ class UDFRegistration private[sql] (functionRegistry: FunctionRegistry) extends
". Expected: 16; Found: " + e.length)
}
functionRegistry.createOrReplaceTempFunction(name, builder)
val udf = UserDefinedFunction.create(func, dataType, inputSchemas).withName(name)
val udf = SparkUserDefinedFunction.create(func, dataType, inputSchemas).withName(name)
if (nullable) udf else udf.asNonNullable()
}

Expand All @@ -517,7 +517,7 @@ class UDFRegistration private[sql] (functionRegistry: FunctionRegistry) extends
". Expected: 17; Found: " + e.length)
}
functionRegistry.createOrReplaceTempFunction(name, builder)
val udf = UserDefinedFunction.create(func, dataType, inputSchemas).withName(name)
val udf = SparkUserDefinedFunction.create(func, dataType, inputSchemas).withName(name)
if (nullable) udf else udf.asNonNullable()
}

Expand All @@ -537,7 +537,7 @@ class UDFRegistration private[sql] (functionRegistry: FunctionRegistry) extends
". Expected: 18; Found: " + e.length)
}
functionRegistry.createOrReplaceTempFunction(name, builder)
val udf = UserDefinedFunction.create(func, dataType, inputSchemas).withName(name)
val udf = SparkUserDefinedFunction.create(func, dataType, inputSchemas).withName(name)
if (nullable) udf else udf.asNonNullable()
}

Expand All @@ -557,7 +557,7 @@ class UDFRegistration private[sql] (functionRegistry: FunctionRegistry) extends
". Expected: 19; Found: " + e.length)
}
functionRegistry.createOrReplaceTempFunction(name, builder)
val udf = UserDefinedFunction.create(func, dataType, inputSchemas).withName(name)
val udf = SparkUserDefinedFunction.create(func, dataType, inputSchemas).withName(name)
if (nullable) udf else udf.asNonNullable()
}

Expand All @@ -577,7 +577,7 @@ class UDFRegistration private[sql] (functionRegistry: FunctionRegistry) extends
". Expected: 20; Found: " + e.length)
}
functionRegistry.createOrReplaceTempFunction(name, builder)
val udf = UserDefinedFunction.create(func, dataType, inputSchemas).withName(name)
val udf = SparkUserDefinedFunction.create(func, dataType, inputSchemas).withName(name)
if (nullable) udf else udf.asNonNullable()
}

Expand All @@ -597,7 +597,7 @@ class UDFRegistration private[sql] (functionRegistry: FunctionRegistry) extends
". Expected: 21; Found: " + e.length)
}
functionRegistry.createOrReplaceTempFunction(name, builder)
val udf = UserDefinedFunction.create(func, dataType, inputSchemas).withName(name)
val udf = SparkUserDefinedFunction.create(func, dataType, inputSchemas).withName(name)
if (nullable) udf else udf.asNonNullable()
}

Expand All @@ -617,7 +617,7 @@ class UDFRegistration private[sql] (functionRegistry: FunctionRegistry) extends
". Expected: 22; Found: " + e.length)
}
functionRegistry.createOrReplaceTempFunction(name, builder)
val udf = UserDefinedFunction.create(func, dataType, inputSchemas).withName(name)
val udf = SparkUserDefinedFunction.create(func, dataType, inputSchemas).withName(name)
if (nullable) udf else udf.asNonNullable()
}

Expand Down
6 changes: 4 additions & 2 deletions sql/core/src/main/scala/org/apache/spark/sql/expressions/UserDefinedFunction.scala
View file
Original file line number Diff line number Diff line change
Expand Up @@ -49,7 +49,7 @@ case class UserDefinedFunction protected[sql] (

// This is a `var` instead of in the constructor for backward compatibility of this case class.
// TODO: revisit this case class in Spark 3.0, and narrow down the public surface.
private var nullableTypes: Option[Seq[Boolean]] = None
private[sql] var nullableTypes: Option[Seq[Boolean]] = None

/**
* Returns true when the UDF can return a nullable value.
Expand Down Expand Up @@ -139,7 +139,9 @@ case class UserDefinedFunction protected[sql] (
}
}

private[sql] object UserDefinedFunction {
// We have to use a name different than `UserDefinedFunction` here, to avoid breaking the binary
// compatibility of the auto-generate UserDefinedFunction object.
private[sql] object SparkUserDefinedFunction {

def create(
f: AnyRef,
Expand Down
Loading

0 comments on commit 9e060a4

Please sign in to comment.