Skip to content

Commit

Permalink
update try
Browse files Browse the repository at this point in the history
  • Loading branch information
yaooqinn committed Feb 16, 2021
1 parent 290990d commit 2ec60e6
Show file tree
Hide file tree
Showing 4 changed files with 120 additions and 99 deletions.

This file was deleted.

Original file line number Diff line number Diff line change
@@ -0,0 +1,116 @@
/*
* 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.teradata

import java.io.UnsupportedEncodingException
import java.time.DateTimeException

import org.apache.spark.sql.catalyst.{FunctionIdentifier, InternalRow}
import org.apache.spark.sql.catalyst.expressions.{Expression, ExpressionDescription, ExpressionInfo, UnaryExpression}
import org.apache.spark.sql.catalyst.expressions.codegen.{CodegenContext, CodeGenerator, ExprCode}
import org.apache.spark.sql.catalyst.expressions.codegen.Block._
import org.apache.spark.sql.extra.FunctionDescription
import org.apache.spark.sql.types.DataType

// scalastyle:off line.size.limit
// [SPARK-31335][SQL] Add try function support https://github.com/apache/spark/pull/28106
@ExpressionDescription(
usage = """
_FUNC_(expr) - Evaluate an expression and handle certain types of runtime exceptions by returning NULL.
In cases where it is preferable that queries produce NULL instead of failing when corrupt or invalid data is encountered, the TRY function may be useful, especially when ANSI mode is on and the users need null-tolerant on certain columns or outputs.
AnalysisExceptions will not be handled by this, typically runtime exceptions handled by _FUNC_ function are:
* ArightmeticException - e.g. division by zero, numeric value out of range,
* NumberFormatException - e.g. invalid casting,
* IllegalArgumentException - e.g. invalid datetime pattern, missing format argument for string formatting,
* DateTimeException - e.g. invalid datetime values
* UnsupportedEncodingException - e.g. encode or decode string with invalid charset
""",
examples = """
Examples:
> SELECT _FUNC_(1 / 0);
NULL
> SELECT _FUNC_(date_format(timestamp '2019-10-06', 'yyyy-MM-dd uucc'));
NULL
> SELECT _FUNC_((5e36BD + 0.1) + 5e36BD);
NULL
> SELECT _FUNC_(regexp_extract('1a 2b 14m', '\\d+', 1));
NULL
> SELECT _FUNC_(encode('abc', 'utf-88'));
NULL
""",
since = "0.1.0")
// scalastyle:on line.size.limit
case class TryExpression(child: Expression) extends UnaryExpression {

override def nullable: Boolean = true
override def dataType: DataType = child.dataType
override def prettyName: String = "try"

override def eval(input: InternalRow): Any = {
try {
child.eval(input)
} catch {
case e if TryExpression.canSuppress(e) => null
}
}

override def doGenCode(ctx: CodegenContext, ev: ExprCode): ExprCode = {
val eval = child.genCode(ctx)
val javaType = CodeGenerator.javaType(dataType)
ev.copy(code =
code"""
|boolean ${ev.isNull} = false;
|$javaType ${ev.value} = ${CodeGenerator.defaultValue(dataType)};
|try {
| ${eval.code}
| ${ev.isNull} = ${eval.isNull};
| ${ev.value} = ${eval.value};
|} catch (java.lang.Exception e) {
| if (org.apache.spark.sql.catalyst.expressions.teradata.TryExpression.canSuppress(e)) {
| ${ev.isNull} = true;
| } else {
| throw e;
| }
|}
|""".stripMargin)
}
}

object TryExpression {

/**
* Certain runtime exceptions that can be suppressed by [[TryExpression]], those not listed here
* are not handled by try function, e.g. exceptions that related access controls or critical ones
* like [[InterruptedException]] etc, or superclass like [[RuntimeException]] that can suppress
* all of its subclasses.
*/
def canSuppress(e: Throwable): Boolean = e match {
case _: IllegalArgumentException |
_: ArithmeticException |
_: DateTimeException |
_: NumberFormatException |
_: UnsupportedEncodingException => true
case _ => false
}

val fd: FunctionDescription = (
new FunctionIdentifier("try"),
new ExpressionInfo(classOf[TryExpression].getCanonicalName, "try"),
(children: Seq[Expression]) => TryExpression(children.head))
}
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,7 @@
package org.apache.spark.sql.extra

import org.apache.spark.sql.SparkSessionExtensions
import org.apache.spark.sql.catalyst.expressions.teradata.{Char2HexInt, CosineSimilarity, Infinity, IsFinite, IsInfinite, NaN, Try}
import org.apache.spark.sql.catalyst.expressions.teradata._

class TeradataExtensions extends Extensions {
override def apply(extensions: SparkSessionExtensions): Unit = {
Expand All @@ -32,7 +32,7 @@ class TeradataExtensions extends Extensions {
extensions.injectFunction(IsFinite.fd)
extensions.injectFunction(IsInfinite.fd)
extensions.injectFunction(NaN.fd)
extensions.injectFunction(Try.fd)
extensions.injectFunction(TryExpression.fd)
}
}

Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,6 @@
package org.apache.spark.sql.extra

import org.apache.spark.SparkException

import org.apache.spark.sql.{AnalysisException, DataFrame, Row}

class TeradataExtensionsTest extends SparkSessionHelper {
Expand Down Expand Up @@ -72,11 +71,8 @@ class TeradataExtensionsTest extends SparkSessionHelper {
spark.sql("set spark.sql.ansi.enabled=true")
intercept[SparkException](spark.sql("select assert_true(1<0)").collect())

val res1 = spark.sql("select try(assert_true(1<0))")
assert(res1.head().isNullAt(0))
val res2 =
spark.sql("select try(assert_true(3 < b)) from values (1, 2), (2, 3), (4, 5) t(a, b)")
assert(res2.head().isNullAt(0))
intercept[SparkException](spark.sql("select try(assert_true(1<0))").collect())

spark.sql(
"create table abcde using parquet as select cast(a as string)," +
" b from values (interval 1 day , 2)," +
Expand Down

0 comments on commit 2ec60e6

Please sign in to comment.