You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tw...@apache.org on 2016/08/18 15:44:48 UTC
flink git commit: [FLINK-4189] [table] Introduce symbols for internal
use
Repository: flink
Updated Branches:
refs/heads/master e9a067229 -> cc7450d1d
[FLINK-4189] [table] Introduce symbols for internal use
This closes #2380.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/cc7450d1
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/cc7450d1
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/cc7450d1
Branch: refs/heads/master
Commit: cc7450d1d1ee3b5189d92ba8642c7c6cffdf056a
Parents: e9a0672
Author: twalthr <tw...@apache.org>
Authored: Tue Aug 16 14:55:38 2016 +0200
Committer: twalthr <tw...@apache.org>
Committed: Thu Aug 18 17:38:56 2016 +0200
----------------------------------------------------------------------
docs/apis/table.md | 37 +++++
.../flink/api/scala/table/expressionDsl.scala | 16 ++-
.../flink/api/table/codegen/CodeGenUtils.scala | 6 +-
.../flink/api/table/codegen/CodeGenerator.scala | 37 +++--
.../api/table/codegen/GeneratedExpression.scala | 27 ----
.../api/table/codegen/GeneratedFunction.scala | 23 ----
.../api/table/codegen/calls/MethodCallGen.scala | 44 ++++++
.../codegen/calls/MethodCallGenerator.scala | 41 ------
.../table/codegen/calls/ScalarFunctions.scala | 23 +++-
.../table/codegen/calls/ScalarOperators.scala | 9 +-
.../api/table/codegen/calls/TrimCallGen.scala | 52 +++++++
.../table/codegen/calls/TrimCallGenerator.scala | 53 -------
.../flink/api/table/codegen/generated.scala | 37 +++++
.../table/expressions/ExpressionParser.scala | 65 +++++----
.../api/table/expressions/ExpressionUtils.scala | 58 ++++++++
.../table/expressions/stringExpressions.scala | 32 +++--
.../flink/api/table/expressions/symbols.scala | 132 ++++++++++++++++++
.../flink/api/table/expressions/time.scala | 137 +++++++++++++++++++
.../api/table/plan/RexNodeTranslator.scala | 4 +-
.../api/table/validate/FunctionCatalog.scala | 5 +-
.../api/java/batch/table/ExpressionsITCase.java | 5 +-
.../table/expressions/ScalarFunctionsTest.scala | 129 ++++++++++++++++-
22 files changed, 757 insertions(+), 215 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/cc7450d1/docs/apis/table.md
----------------------------------------------------------------------
diff --git a/docs/apis/table.md b/docs/apis/table.md
index c2806c6..21a7dbe 100644
--- a/docs/apis/table.md
+++ b/docs/apis/table.md
@@ -979,6 +979,10 @@ atom = ( "(" , expression , ")" ) | literal | nullLiteral | fieldReference ;
nullLiteral = "Null(" , dataType , ")" ;
+timeIntervalUnit = "YEAR" | "YEAR_TO_MONTH" | "MONTH" | "DAY" | "DAY_TO_HOUR" | "DAY_TO_MINUTE" | "DAY_TO_SECOND" | "HOUR" | "HOUR_TO_MINUTE" | "HOUR_TO_SECOND" | "MINUTE" | "MINUTE_TO_SECOND" | "SECOND" ;
+
+timePointUnit = "YEAR" | "MONTH" | "DAY" | "HOUR" | "MINUTE" | "SECOND" | "QUARTER" | "WEEK" | "MILLISECOND" | "MICROSECOND" ;
+
{% endhighlight %}
Here, `literal` is a valid Java literal, `fieldReference` specifies a column in the data, and `functionIdentifier` specifies a supported scalar function. The
@@ -1430,6 +1434,17 @@ STRING.toTimestamp()
</td>
</tr>
+ <tr>
+ <td>
+ {% highlight java %}
+TEMPORAL.extract(TIMEINTERVALUNIT)
+{% endhighlight %}
+ </td>
+ <td>
+ <p>Extracts parts of a time point or time interval. Returns the part as a long value. E.g. <code>"2006-06-05".toDate.extract(DAY)</code> leads to 5.</p>
+ </td>
+ </tr>
+
</tbody>
</table>
@@ -1660,6 +1675,17 @@ STRING.toTimestamp
</td>
</tr>
+ <tr>
+ <td>
+ {% highlight scala %}
+TEMPORAL.extract(TimeIntervalUnit)
+{% endhighlight %}
+ </td>
+ <td>
+ <p>Extracts parts of a time point or time interval. Returns the part as a long value. E.g. <code>"2006-06-05".toDate.extract(TimeIntervalUnit.DAY)</code> leads to 5.</p>
+ </td>
+ </tr>
+
</tbody>
</table>
</div>
@@ -1892,6 +1918,17 @@ TIMESTAMP VARCHAR
</td>
</tr>
+ <tr>
+ <td>
+ {% highlight sql %}
+EXTRACT(TIMEINTERVALUNIT FROM TEMPORAL)
+{% endhighlight %}
+ </td>
+ <td>
+ <p>Extracts parts of a time point or time interval. Returns the part as a long value. E.g. <code>EXTRACT(DAY FROM DATE '2006-06-05')</code> leads to 5.</p>
+ </td>
+ </tr>
+
</tbody>
</table>
</div>
http://git-wip-us.apache.org/repos/asf/flink/blob/cc7450d1/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/scala/table/expressionDsl.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/scala/table/expressionDsl.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/scala/table/expressionDsl.scala
index 9ca9c8a..92c61a3 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/scala/table/expressionDsl.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/scala/table/expressionDsl.scala
@@ -22,6 +22,7 @@ import java.sql.{Date, Time, Timestamp}
import org.apache.calcite.avatica.util.DateTimeUtils._
import org.apache.flink.api.common.typeinfo.{SqlTimeTypeInfo, TypeInformation}
import org.apache.flink.api.table.expressions.ExpressionUtils.{toMilliInterval, toMonthInterval}
+import org.apache.flink.api.table.expressions.TimeIntervalUnit.TimeIntervalUnit
import org.apache.flink.api.table.expressions._
import scala.language.implicitConversions
@@ -168,11 +169,11 @@ trait ImplicitExpressionOperations {
removeTrailing: Boolean = true,
character: Expression = TrimConstants.TRIM_DEFAULT_CHAR) = {
if (removeLeading && removeTrailing) {
- Trim(TrimConstants.TRIM_BOTH, character, expr)
+ Trim(TrimMode.BOTH, character, expr)
} else if (removeLeading) {
- Trim(TrimConstants.TRIM_LEADING, character, expr)
+ Trim(TrimMode.LEADING, character, expr)
} else if (removeTrailing) {
- Trim(TrimConstants.TRIM_TRAILING, character, expr)
+ Trim(TrimMode.TRAILING, character, expr)
} else {
expr
}
@@ -230,6 +231,13 @@ trait ImplicitExpressionOperations {
*/
def toTimestamp = Cast(expr, SqlTimeTypeInfo.TIMESTAMP)
+ /**
+ * Extracts parts of a time point or time interval. Returns the part as a long value.
+ *
+ * e.g. "2006-06-05".toDate.extract(DAY) leads to 5
+ */
+ def extract(timeIntervalUnit: TimeIntervalUnit) = Extract(timeIntervalUnit, expr)
+
// interval types
/**
@@ -291,7 +299,7 @@ trait ImplicitExpressionConversions {
def expr = e
}
- implicit class SymbolExpression(s: Symbol) extends ImplicitExpressionOperations {
+ implicit class UnresolvedFieldExpression(s: Symbol) extends ImplicitExpressionOperations {
def expr = UnresolvedFieldReference(s.name)
}
http://git-wip-us.apache.org/repos/asf/flink/blob/cc7450d1/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/CodeGenUtils.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/CodeGenUtils.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/CodeGenUtils.scala
index 18bf49f..170af54 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/CodeGenUtils.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/CodeGenUtils.scala
@@ -96,7 +96,7 @@ object CodeGenUtils {
def primitiveDefaultValue(tpe: TypeInformation[_]): String = tpe match {
case INT_TYPE_INFO => "-1"
- case LONG_TYPE_INFO => "-1"
+ case LONG_TYPE_INFO => "-1L"
case SHORT_TYPE_INFO => "-1"
case BYTE_TYPE_INFO => "-1"
case FLOAT_TYPE_INFO => "-1.0f"
@@ -143,6 +143,10 @@ object CodeGenUtils {
s"${qualifyMethod(BuiltInMethod.TIMESTAMP_TO_LONG.method)}($resultTerm)"
}
+ def compareEnum(term: String, enum: Enum[_]): Boolean =
+ term == qualifyEnum(enum)
+
+
// ----------------------------------------------------------------------------------------------
def requireNumeric(genExpr: GeneratedExpression) =
http://git-wip-us.apache.org/repos/asf/flink/blob/cc7450d1/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/CodeGenerator.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/CodeGenerator.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/CodeGenerator.scala
index fca3fae..4a3865f 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/CodeGenerator.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/CodeGenerator.scala
@@ -21,13 +21,13 @@ package org.apache.flink.api.table.codegen
import java.math.{BigDecimal => JBigDecimal}
import org.apache.calcite.rex._
+import org.apache.calcite.sql.SqlOperator
import org.apache.calcite.sql.`type`.SqlTypeName._
import org.apache.calcite.sql.fun.SqlStdOperatorTable._
-import org.apache.calcite.sql.{SqlLiteral, SqlOperator}
import org.apache.flink.api.common.functions.{FlatJoinFunction, FlatMapFunction, Function, MapFunction}
import org.apache.flink.api.common.typeinfo.{AtomicType, SqlTimeTypeInfo, TypeInformation}
import org.apache.flink.api.common.typeutils.CompositeType
-import org.apache.flink.api.java.typeutils.{PojoTypeInfo, TupleTypeInfo}
+import org.apache.flink.api.java.typeutils.{GenericTypeInfo, PojoTypeInfo, TupleTypeInfo}
import org.apache.flink.api.scala.typeutils.CaseClassTypeInfo
import org.apache.flink.api.table.codegen.CodeGenUtils._
import org.apache.flink.api.table.codegen.Indenter.toISC
@@ -543,8 +543,10 @@ class CodeGenerator(
}
// non-null values
literal.getType.getSqlTypeName match {
+
case BOOLEAN =>
generateNonNullLiteral(resultType, literal.getValue3.toString)
+
case TINYINT =>
val decimal = BigDecimal(value.asInstanceOf[JBigDecimal])
if (decimal.isValidByte) {
@@ -553,6 +555,7 @@ class CodeGenerator(
else {
throw new CodeGenException("Decimal can not be converted to byte.")
}
+
case SMALLINT =>
val decimal = BigDecimal(value.asInstanceOf[JBigDecimal])
if (decimal.isValidShort) {
@@ -561,6 +564,7 @@ class CodeGenerator(
else {
throw new CodeGenException("Decimal can not be converted to short.")
}
+
case INTEGER =>
val decimal = BigDecimal(value.asInstanceOf[JBigDecimal])
if (decimal.isValidInt) {
@@ -569,6 +573,7 @@ class CodeGenerator(
else {
throw new CodeGenException("Decimal can not be converted to integer.")
}
+
case BIGINT =>
val decimal = BigDecimal(value.asInstanceOf[JBigDecimal])
if (decimal.isValidLong) {
@@ -577,6 +582,7 @@ class CodeGenerator(
else {
throw new CodeGenException("Decimal can not be converted to long.")
}
+
case FLOAT =>
val floatValue = value.asInstanceOf[JBigDecimal].floatValue()
floatValue match {
@@ -587,6 +593,7 @@ class CodeGenerator(
generateNonNullLiteral(resultType, "java.lang.Float.POSITIVE_INFINITY")
case _ => generateNonNullLiteral(resultType, floatValue.toString + "f")
}
+
case DOUBLE =>
val doubleValue = value.asInstanceOf[JBigDecimal].doubleValue()
doubleValue match {
@@ -600,25 +607,22 @@ class CodeGenerator(
case DECIMAL =>
val decimalField = addReusableDecimal(value.asInstanceOf[JBigDecimal])
generateNonNullLiteral(resultType, decimalField)
+
case VARCHAR | CHAR =>
generateNonNullLiteral(resultType, "\"" + value.toString + "\"")
- case SYMBOL =>
- val symbolOrdinal =
- if (classOf[Enum[_]].isAssignableFrom(value.getClass) ) {
- value.asInstanceOf[Enum[_]].ordinal()
- } else {
- value.asInstanceOf[SqlLiteral.SqlSymbol].ordinal()
- }
-
- generateNonNullLiteral(resultType, symbolOrdinal.toString)
+ case SYMBOL =>
+ generateSymbol(value.asInstanceOf[Enum[_]])
case DATE =>
generateNonNullLiteral(resultType, value.toString)
+
case TIME =>
generateNonNullLiteral(resultType, value.toString)
+
case TIMESTAMP =>
generateNonNullLiteral(resultType, value.toString + "L")
+
case INTERVAL_YEAR_MONTH =>
val decimal = BigDecimal(value.asInstanceOf[JBigDecimal])
if (decimal.isValidInt) {
@@ -626,6 +630,7 @@ class CodeGenerator(
} else {
throw new CodeGenException("Decimal can not be converted to interval of months.")
}
+
case INTERVAL_DAY_TIME =>
val decimal = BigDecimal(value.asInstanceOf[JBigDecimal])
if (decimal.isValidLong) {
@@ -688,7 +693,7 @@ class CodeGenerator(
requireNumeric(right)
generateArithmeticOperator("*", nullCheck, resultType, left, right)
- case DIVIDE if isNumeric(resultType) =>
+ case DIVIDE | DIVIDE_INTEGER if isNumeric(resultType) =>
val left = operands.head
val right = operands(1)
requireNumeric(left)
@@ -1009,6 +1014,14 @@ class CodeGenerator(
GeneratedExpression(resultTerm, nullTerm, resultCode, literalType)
}
+ private[flink] def generateSymbol(enum: Enum[_]): GeneratedExpression = {
+ GeneratedExpression(
+ qualifyEnum(enum),
+ "false",
+ "",
+ new GenericTypeInfo(enum.getDeclaringClass))
+ }
+
/**
* Converts the external boxed format to an internal mostly primitive field representation.
* Wrapper types can autoboxed to their corresponding primitive type (Integer -> int). External
http://git-wip-us.apache.org/repos/asf/flink/blob/cc7450d1/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/GeneratedExpression.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/GeneratedExpression.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/GeneratedExpression.scala
deleted file mode 100644
index 7c20e38..0000000
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/GeneratedExpression.scala
+++ /dev/null
@@ -1,27 +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.flink.api.table.codegen
-
-import org.apache.flink.api.common.typeinfo.TypeInformation
-
-case class GeneratedExpression(
- resultTerm: String,
- nullTerm: String,
- code: String,
- resultType: TypeInformation[_])
http://git-wip-us.apache.org/repos/asf/flink/blob/cc7450d1/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/GeneratedFunction.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/GeneratedFunction.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/GeneratedFunction.scala
deleted file mode 100644
index 575e7ab..0000000
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/GeneratedFunction.scala
+++ /dev/null
@@ -1,23 +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.flink.api.table.codegen
-
-import org.apache.flink.api.common.typeinfo.TypeInformation
-
-case class GeneratedFunction[T](name: String, returnType: TypeInformation[Any], code: String)
http://git-wip-us.apache.org/repos/asf/flink/blob/cc7450d1/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/calls/MethodCallGen.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/calls/MethodCallGen.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/calls/MethodCallGen.scala
new file mode 100644
index 0000000..376f54a
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/calls/MethodCallGen.scala
@@ -0,0 +1,44 @@
+/*
+ * 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.flink.api.table.codegen.calls
+
+import java.lang.reflect.Method
+
+import org.apache.flink.api.common.typeinfo.TypeInformation
+import org.apache.flink.api.table.codegen.CodeGenUtils.qualifyMethod
+import org.apache.flink.api.table.codegen.calls.CallGenerator.generateCallIfArgsNotNull
+import org.apache.flink.api.table.codegen.{CodeGenerator, GeneratedExpression}
+
+/**
+ * Generates a function call by using a [[java.lang.reflect.Method]].
+ */
+class MethodCallGen(returnType: TypeInformation[_], method: Method) extends CallGenerator {
+
+ override def generate(
+ codeGenerator: CodeGenerator,
+ operands: Seq[GeneratedExpression])
+ : GeneratedExpression = {
+ generateCallIfArgsNotNull(codeGenerator.nullCheck, returnType, operands) {
+ (terms) =>
+ s"""
+ |${qualifyMethod(method)}(${terms.mkString(", ")})
+ |""".stripMargin
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/cc7450d1/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/calls/MethodCallGenerator.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/calls/MethodCallGenerator.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/calls/MethodCallGenerator.scala
deleted file mode 100644
index 72e922d..0000000
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/calls/MethodCallGenerator.scala
+++ /dev/null
@@ -1,41 +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.flink.api.table.codegen.calls
-
-import java.lang.reflect.Method
-
-import org.apache.flink.api.common.typeinfo.TypeInformation
-import org.apache.flink.api.table.codegen.calls.CallGenerator.generateCallIfArgsNotNull
-import org.apache.flink.api.table.codegen.{CodeGenerator, GeneratedExpression}
-
-class MethodCallGenerator(returnType: TypeInformation[_], method: Method) extends CallGenerator {
-
- override def generate(
- codeGenerator: CodeGenerator,
- operands: Seq[GeneratedExpression])
- : GeneratedExpression = {
- generateCallIfArgsNotNull(codeGenerator.nullCheck, returnType, operands) {
- (operandResultTerms) =>
- s"""
- |${method.getDeclaringClass.getCanonicalName}.
- | ${method.getName}(${operandResultTerms.mkString(", ")})
- """.stripMargin
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/cc7450d1/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/calls/ScalarFunctions.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/calls/ScalarFunctions.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/calls/ScalarFunctions.scala
index 9c5fc22..44cb6d2 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/calls/ScalarFunctions.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/calls/ScalarFunctions.scala
@@ -20,11 +20,14 @@ package org.apache.flink.api.table.codegen.calls
import java.lang.reflect.Method
+import org.apache.calcite.avatica.util.TimeUnitRange
import org.apache.calcite.sql.SqlOperator
import org.apache.calcite.sql.fun.SqlStdOperatorTable._
+import org.apache.calcite.sql.fun.SqlTrimFunction
import org.apache.calcite.util.BuiltInMethod
import org.apache.flink.api.common.typeinfo.BasicTypeInfo._
import org.apache.flink.api.common.typeinfo.{BasicTypeInfo, TypeInformation}
+import org.apache.flink.api.java.typeutils.GenericTypeInfo
import org.apache.flink.api.table.functions.utils.ScalarSqlFunction
import scala.collection.mutable
@@ -55,8 +58,8 @@ object ScalarFunctions {
addSqlFunction(
TRIM,
- Seq(INT_TYPE_INFO, STRING_TYPE_INFO, STRING_TYPE_INFO),
- new TrimCallGenerator())
+ Seq(new GenericTypeInfo(classOf[SqlTrimFunction.Flag]), STRING_TYPE_INFO, STRING_TYPE_INFO),
+ new TrimCallGen())
addSqlFunctionMethod(
CHAR_LENGTH,
@@ -111,7 +114,7 @@ object ScalarFunctions {
BuiltInMethod.SIMILAR.method)
// ----------------------------------------------------------------------------------------------
- // arithmetic functions
+ // Arithmetic functions
// ----------------------------------------------------------------------------------------------
addSqlFunctionMethod(
@@ -169,6 +172,16 @@ object ScalarFunctions {
new FloorCeilCallGen(BuiltInMethod.CEIL.method))
// ----------------------------------------------------------------------------------------------
+ // Temporal functions
+ // ----------------------------------------------------------------------------------------------
+
+ addSqlFunctionMethod(
+ EXTRACT_DATE,
+ Seq(new GenericTypeInfo(classOf[TimeUnitRange]), LONG_TYPE_INFO),
+ LONG_TYPE_INFO,
+ BuiltInMethod.UNIX_DATE_EXTRACT.method)
+
+ // ----------------------------------------------------------------------------------------------
/**
* Returns a [[CallGenerator]] that generates all required code for calling the given
@@ -214,7 +227,7 @@ object ScalarFunctions {
returnType: TypeInformation[_],
method: Method)
: Unit = {
- sqlFunctions((sqlOperator, operandTypes)) = new MethodCallGenerator(returnType, method)
+ sqlFunctions((sqlOperator, operandTypes)) = new MethodCallGen(returnType, method)
}
private def addSqlFunctionNotMethod(
@@ -223,7 +236,7 @@ object ScalarFunctions {
method: Method)
: Unit = {
sqlFunctions((sqlOperator, operandTypes)) =
- new NotCallGenerator(new MethodCallGenerator(BOOLEAN_TYPE_INFO, method))
+ new NotCallGenerator(new MethodCallGen(BOOLEAN_TYPE_INFO, method))
}
private def addSqlFunction(
http://git-wip-us.apache.org/repos/asf/flink/blob/cc7450d1/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/calls/ScalarOperators.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/calls/ScalarOperators.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/calls/ScalarOperators.scala
index 81187a5..afe69ed 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/calls/ScalarOperators.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/calls/ScalarOperators.scala
@@ -544,7 +544,14 @@ object ScalarOperators {
(INT_TYPE_INFO, IntervalTypeInfo.INTERVAL_MONTHS) |
(LONG_TYPE_INFO, IntervalTypeInfo.INTERVAL_MILLIS) |
(IntervalTypeInfo.INTERVAL_MONTHS, INT_TYPE_INFO) |
- (IntervalTypeInfo.INTERVAL_MONTHS, LONG_TYPE_INFO) =>
+ (IntervalTypeInfo.INTERVAL_MILLIS, LONG_TYPE_INFO) =>
+ internalExprCasting(operand, targetType)
+
+ // internal reinterpretation of temporal types
+ // Date, Time, Interval Months -> Long
+ case (SqlTimeTypeInfo.DATE, LONG_TYPE_INFO)
+ | (SqlTimeTypeInfo.TIME, LONG_TYPE_INFO)
+ | (IntervalTypeInfo.INTERVAL_MONTHS, LONG_TYPE_INFO) =>
internalExprCasting(operand, targetType)
case (from, to) =>
http://git-wip-us.apache.org/repos/asf/flink/blob/cc7450d1/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/calls/TrimCallGen.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/calls/TrimCallGen.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/calls/TrimCallGen.scala
new file mode 100644
index 0000000..678016b
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/calls/TrimCallGen.scala
@@ -0,0 +1,52 @@
+/*
+ * 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.flink.api.table.codegen.calls
+
+import org.apache.calcite.sql.fun.SqlTrimFunction.Flag.{BOTH, LEADING, TRAILING}
+import org.apache.calcite.util.BuiltInMethod
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo.STRING_TYPE_INFO
+import org.apache.flink.api.table.codegen.CodeGenUtils._
+import org.apache.flink.api.table.codegen.calls.CallGenerator._
+import org.apache.flink.api.table.codegen.{CodeGenerator, GeneratedExpression}
+
+/**
+ * Generates a TRIM function call.
+ *
+ * First operand: trim mode (see [[org.apache.calcite.sql.fun.SqlTrimFunction.Flag]])
+ * Second operand: String to be removed
+ * Third operand: String to be trimmed
+ */
+class TrimCallGen extends CallGenerator {
+
+ override def generate(
+ codeGenerator: CodeGenerator,
+ operands: Seq[GeneratedExpression])
+ : GeneratedExpression = {
+ generateCallIfArgsNotNull(codeGenerator.nullCheck, STRING_TYPE_INFO, operands) {
+ (terms) =>
+ val leading = compareEnum(terms.head, BOTH) || compareEnum(terms.head, LEADING)
+ val trailing = compareEnum(terms.head, BOTH) || compareEnum(terms.head, TRAILING)
+ s"""
+ |${qualifyMethod(BuiltInMethod.TRIM.method)}(
+ | $leading, $trailing, ${terms(1)}, ${terms(2)})
+ |""".stripMargin
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/cc7450d1/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/calls/TrimCallGenerator.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/calls/TrimCallGenerator.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/calls/TrimCallGenerator.scala
deleted file mode 100644
index 3168703..0000000
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/calls/TrimCallGenerator.scala
+++ /dev/null
@@ -1,53 +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.flink.api.table.codegen.calls
-
-import org.apache.calcite.sql.fun.SqlTrimFunction.Flag.{BOTH, LEADING, TRAILING}
-import org.apache.calcite.util.BuiltInMethod
-import org.apache.flink.api.common.typeinfo.BasicTypeInfo.STRING_TYPE_INFO
-import org.apache.flink.api.table.codegen.calls.CallGenerator._
-import org.apache.flink.api.table.codegen.{CodeGenerator, GeneratedExpression}
-
-/**
- * Generates a TRIM function call. The first operand determines the type of trimming
- * (see [[org.apache.calcite.sql.fun.SqlTrimFunction.Flag]]). Second operand determines
- * the String to be removed. Third operand is the String to be trimmed.
- */
-class TrimCallGenerator() extends CallGenerator {
-
- override def generate(
- codeGenerator: CodeGenerator,
- operands: Seq[GeneratedExpression])
- : GeneratedExpression = {
- val method = BuiltInMethod.TRIM.method
- generateCallIfArgsNotNull(codeGenerator.nullCheck, STRING_TYPE_INFO, operands) {
- (operandResultTerms) =>
- s"""
- |${method.getDeclaringClass.getCanonicalName}.${method.getName}(
- | ${operandResultTerms.head} == ${BOTH.ordinal()} ||
- | ${operandResultTerms.head} == ${LEADING.ordinal()},
- | ${operandResultTerms.head} == ${BOTH.ordinal()} ||
- | ${operandResultTerms.head} == ${TRAILING.ordinal()},
- | ${operandResultTerms(1)},
- | ${operandResultTerms(2)})
- |""".stripMargin
- }
- }
-
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/cc7450d1/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/generated.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/generated.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/generated.scala
new file mode 100644
index 0000000..26c6696
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/generated.scala
@@ -0,0 +1,37 @@
+/*
+ * 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.flink.api.table.codegen
+
+import org.apache.flink.api.common.typeinfo.TypeInformation
+
+/**
+ * Describes a generated expression.
+ *
+ * @param resultTerm term to access the result of the expression
+ * @param nullTerm boolean term that indicates if expression is null
+ * @param code code necessary to produce resultTerm and nullTerm
+ * @param resultType type of the resultTerm
+ */
+case class GeneratedExpression(
+ resultTerm: String,
+ nullTerm: String,
+ code: String,
+ resultType: TypeInformation[_])
+
+case class GeneratedFunction[T](name: String, returnType: TypeInformation[Any], code: String)
http://git-wip-us.apache.org/repos/asf/flink/blob/cc7450d1/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/ExpressionParser.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/ExpressionParser.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/ExpressionParser.scala
index 41222c8..1dd480b 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/ExpressionParser.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/ExpressionParser.scala
@@ -21,6 +21,9 @@ import org.apache.calcite.avatica.util.DateTimeUtils.{MILLIS_PER_DAY, MILLIS_PER
import org.apache.flink.api.common.typeinfo.{BasicTypeInfo, SqlTimeTypeInfo, TypeInformation}
import org.apache.flink.api.table.ExpressionParserException
import org.apache.flink.api.table.expressions.ExpressionUtils.{toMilliInterval, toMonthInterval}
+import org.apache.flink.api.table.expressions.TimeIntervalUnit.TimeIntervalUnit
+import org.apache.flink.api.table.expressions.TimePointUnit.TimePointUnit
+import org.apache.flink.api.table.expressions.TrimMode.TrimMode
import org.apache.flink.api.table.typeutils.IntervalTypeInfo
import scala.util.parsing.combinator.{JavaTokenParsers, PackratParsers}
@@ -60,6 +63,8 @@ object ExpressionParser extends JavaTokenParsers with PackratParsers {
lazy val TO_DATE: Keyword = Keyword("toDate")
lazy val TO_TIME: Keyword = Keyword("toTime")
lazy val TO_TIMESTAMP: Keyword = Keyword("toTimestamp")
+ lazy val TRIM: Keyword = Keyword("trim")
+ lazy val EXTRACT: Keyword = Keyword("extract")
lazy val YEAR: Keyword = Keyword("year")
lazy val MONTH: Keyword = Keyword("month")
lazy val DAY: Keyword = Keyword("day")
@@ -73,6 +78,20 @@ object ExpressionParser extends JavaTokenParsers with PackratParsers {
not(SUM) ~ not(IS_NULL) ~ not(IS_NOT_NULL) ~ not(CAST) ~ not(NULL) ~
not(IF) ~> super.ident
+ // symbols
+
+ lazy val timeIntervalUnit: PackratParser[Expression] = TimeIntervalUnit.values map {
+ case unit: TimeIntervalUnit => literal(unit.toString) ^^^ unit.toExpr
+ } reduceLeft(_ | _)
+
+ lazy val timePointUnit: PackratParser[Expression] = TimePointUnit.values map {
+ case unit: TimePointUnit => literal(unit.toString) ^^^ unit.toExpr
+ } reduceLeft(_ | _)
+
+ lazy val trimMode: PackratParser[Expression] = TrimMode.values map {
+ case mode: TrimMode => literal(mode.toString) ^^^ mode.toExpr
+ } reduceLeft(_ | _)
+
// data types
lazy val dataType: PackratParser[TypeInformation[_]] =
@@ -177,20 +196,12 @@ object ExpressionParser extends JavaTokenParsers with PackratParsers {
case e ~ _ ~ _ ~ _ ~ target ~ _ => Alias(e, target.name)
}
- lazy val suffixTrim = composite ~ ".trim(" ~ ("BOTH" | "LEADING" | "TRAILING") ~ "," ~
- expression ~ ")" ^^ {
- case operand ~ _ ~ trimType ~ _ ~ trimCharacter ~ _ =>
- val flag = trimType match {
- case "BOTH" => TrimConstants.TRIM_BOTH
- case "LEADING" => TrimConstants.TRIM_LEADING
- case "TRAILING" => TrimConstants.TRIM_TRAILING
- }
- Trim(flag, trimCharacter, operand)
+ lazy val suffixTrim = composite ~ "." ~ TRIM ~ "(" ~ trimMode ~ "," ~ expression ~ ")" ^^ {
+ case operand ~ _ ~ _ ~ _ ~ mode ~ _ ~ trimCharacter ~ _ => Trim(mode, trimCharacter, operand)
}
- lazy val suffixTrimWithoutArgs = composite <~ ".trim" ~ opt("()") ^^ {
- e =>
- Trim(TrimConstants.TRIM_BOTH, TrimConstants.TRIM_DEFAULT_CHAR, e)
+ lazy val suffixTrimWithoutArgs = composite <~ "." ~ TRIM ~ opt("()") ^^ {
+ e => Trim(TrimMode.BOTH, TrimConstants.TRIM_DEFAULT_CHAR, e)
}
lazy val suffixIf: PackratParser[Expression] =
@@ -198,6 +209,10 @@ object ExpressionParser extends JavaTokenParsers with PackratParsers {
case condition ~ _ ~ _ ~ _ ~ ifTrue ~ _ ~ ifFalse ~ _ => If(condition, ifTrue, ifFalse)
}
+ lazy val suffixExtract = composite ~ "." ~ EXTRACT ~ "(" ~ timeIntervalUnit ~ ")" ^^ {
+ case operand ~ _ ~ _ ~ _ ~ unit ~ _ => Extract(unit, operand)
+ }
+
lazy val suffixFunctionCall =
composite ~ "." ~ functionIdent ~ "(" ~ repsep(expression, ",") ~ ")" ^^ {
case operand ~ _ ~ name ~ _ ~ args ~ _ => Call(name.toUpperCase, operand :: args)
@@ -240,7 +255,7 @@ object ExpressionParser extends JavaTokenParsers with PackratParsers {
suffixTimeInterval | suffixIsNull | suffixIsNotNull | suffixSum | suffixMin | suffixMax |
suffixCount | suffixAvg | suffixCast | suffixAs | suffixTrim | suffixTrimWithoutArgs |
suffixIf | suffixAsc | suffixDesc | suffixToDate | suffixToTimestamp | suffixToTime |
- suffixFunctionCall // function call must always be at the end
+ suffixExtract | suffixFunctionCall // function call must always be at the end
// prefix operators
@@ -275,7 +290,7 @@ object ExpressionParser extends JavaTokenParsers with PackratParsers {
case _ ~ _ ~ e ~ _ ~ target ~ _ => Alias(e, target.name)
}
- lazy val prefixIf: PackratParser[Expression] = composite ~
+ lazy val prefixIf: PackratParser[Expression] =
IF ~ "(" ~ expression ~ "," ~ expression ~ "," ~ expression ~ ")" ^^ {
case _ ~ _ ~ condition ~ _ ~ ifTrue ~ _ ~ ifFalse ~ _ => If(condition, ifTrue, ifFalse)
}
@@ -284,25 +299,21 @@ object ExpressionParser extends JavaTokenParsers with PackratParsers {
case name ~ _ ~ args ~ _ => Call(name.toUpperCase, args)
}
- lazy val prefixTrim = "trim(" ~ ("BOTH" | "LEADING" | "TRAILING") ~ "," ~ expression ~
- "," ~ expression ~ ")" ^^ {
- case _ ~ trimType ~ _ ~ trimCharacter ~ _ ~ operand ~ _ =>
- val flag = trimType match {
- case "BOTH" => TrimConstants.TRIM_BOTH
- case "LEADING" => TrimConstants.TRIM_LEADING
- case "TRAILING" => TrimConstants.TRIM_TRAILING
- }
- Trim(flag, trimCharacter, operand)
+ lazy val prefixTrim = TRIM ~ "(" ~ trimMode ~ "," ~ expression ~ "," ~ expression ~ ")" ^^ {
+ case _ ~ _ ~ mode ~ _ ~ trimCharacter ~ _ ~ operand ~ _ => Trim(mode, trimCharacter, operand)
+ }
+
+ lazy val prefixTrimWithoutArgs = TRIM ~ "(" ~ expression ~ ")" ^^ {
+ case _ ~ _ ~ operand ~ _ => Trim(TrimMode.BOTH, TrimConstants.TRIM_DEFAULT_CHAR, operand)
}
- lazy val prefixTrimWithoutArgs = "trim(" ~ expression ~ ")" ^^ {
- case _ ~ operand ~ _ =>
- Trim(TrimConstants.TRIM_BOTH, TrimConstants.TRIM_DEFAULT_CHAR, operand)
+ lazy val prefixExtract = EXTRACT ~ "(" ~ expression ~ "," ~ timeIntervalUnit ~ ")" ^^ {
+ case _ ~ _ ~ operand ~ _ ~ unit ~ _ => Extract(unit, operand)
}
lazy val prefixed: PackratParser[Expression] =
prefixIsNull | prefixIsNotNull | prefixSum | prefixMin | prefixMax | prefixCount | prefixAvg |
- prefixCast | prefixAs | prefixTrim | prefixTrimWithoutArgs | prefixIf |
+ prefixCast | prefixAs | prefixTrim | prefixTrimWithoutArgs | prefixIf | prefixExtract |
prefixFunctionCall // function call must always be at the end
// suffix/prefix composite
http://git-wip-us.apache.org/repos/asf/flink/blob/cc7450d1/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/ExpressionUtils.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/ExpressionUtils.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/ExpressionUtils.scala
index 89671de..6ac9f58 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/ExpressionUtils.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/ExpressionUtils.scala
@@ -18,6 +18,12 @@
package org.apache.flink.api.table.expressions
+import java.math.BigDecimal
+
+import org.apache.calcite.avatica.util.TimeUnit
+import org.apache.calcite.rel.`type`.RelDataType
+import org.apache.calcite.rex.{RexBuilder, RexNode}
+import org.apache.calcite.sql.fun.SqlStdOperatorTable
import org.apache.flink.api.common.typeinfo.BasicTypeInfo
import org.apache.flink.api.table.typeutils.IntervalTypeInfo
@@ -37,4 +43,56 @@ object ExpressionUtils {
Cast(Mul(expr, Literal(multiplier)), IntervalTypeInfo.INTERVAL_MILLIS)
}
+ // ----------------------------------------------------------------------------------------------
+ // RexNode conversion functions (see org.apache.calcite.sql2rel.StandardConvertletTable)
+ // ----------------------------------------------------------------------------------------------
+
+ /**
+ * Copy of [[org.apache.calcite.sql2rel.StandardConvertletTable#getFactor()]].
+ */
+ private[flink] def getFactor(unit: TimeUnit): BigDecimal = unit match {
+ case TimeUnit.DAY => java.math.BigDecimal.ONE
+ case TimeUnit.HOUR => TimeUnit.DAY.multiplier
+ case TimeUnit.MINUTE => TimeUnit.HOUR.multiplier
+ case TimeUnit.SECOND => TimeUnit.MINUTE.multiplier
+ case TimeUnit.YEAR => java.math.BigDecimal.ONE
+ case TimeUnit.MONTH => TimeUnit.YEAR.multiplier
+ case _ => throw new IllegalArgumentException("Invalid start unit.")
+ }
+
+ /**
+ * Copy of [[org.apache.calcite.sql2rel.StandardConvertletTable#mod()]].
+ */
+ private[flink] def mod(
+ rexBuilder: RexBuilder,
+ resType: RelDataType,
+ res: RexNode,
+ value: BigDecimal)
+ : RexNode = {
+ if (value == BigDecimal.ONE) return res
+ rexBuilder.makeCall(SqlStdOperatorTable.MOD, res, rexBuilder.makeExactLiteral(value, resType))
+ }
+
+ /**
+ * Copy of [[org.apache.calcite.sql2rel.StandardConvertletTable#divide()]].
+ */
+ private[flink] def divide(rexBuilder: RexBuilder, res: RexNode, value: BigDecimal): RexNode = {
+ if (value == BigDecimal.ONE) return res
+ if (value.compareTo(BigDecimal.ONE) < 0 && value.signum == 1) {
+ try {
+ val reciprocal = BigDecimal.ONE.divide(value, BigDecimal.ROUND_UNNECESSARY)
+ return rexBuilder.makeCall(
+ SqlStdOperatorTable.MULTIPLY,
+ res,
+ rexBuilder.makeExactLiteral(reciprocal))
+ } catch {
+ case e: ArithmeticException => // ignore
+ }
+ }
+ rexBuilder.makeCall(
+ SqlStdOperatorTable.DIVIDE_INTEGER,
+ res,
+ rexBuilder.makeExactLiteral(value))
+ }
+
}
http://git-wip-us.apache.org/repos/asf/flink/blob/cc7450d1/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/stringExpressions.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/stringExpressions.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/stringExpressions.scala
index 597ff12..aee7d67 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/stringExpressions.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/stringExpressions.scala
@@ -18,13 +18,12 @@
package org.apache.flink.api.table.expressions
import scala.collection.JavaConversions._
-
import org.apache.calcite.rex.RexNode
import org.apache.calcite.sql.fun.SqlStdOperatorTable
import org.apache.calcite.tools.RelBuilder
-
import org.apache.flink.api.common.typeinfo.BasicTypeInfo._
import org.apache.flink.api.common.typeinfo.TypeInformation
+import org.apache.flink.api.table.expressions.TrimMode.TrimMode
import org.apache.flink.api.table.validate._
/**
@@ -169,22 +168,32 @@ case class SubString(
}
/**
- * Trim `trimString` from `str` according to `trimFlag`:
- * 0 for TRIM_BOTH, 1 for TRIM_LEADING and 2 for TRIM_TRAILING.
+ * Trim `trimString` from `str` according to `trimMode`.
*/
case class Trim(
- trimFlag: Expression,
+ trimMode: Expression,
trimString: Expression,
- str: Expression) extends Expression with InputTypeSpec {
+ str: Expression) extends Expression {
- override private[flink] def children: Seq[Expression] = trimFlag :: trimString :: str :: Nil
+ override private[flink] def children: Seq[Expression] = trimMode :: trimString :: str :: Nil
override private[flink] def resultType: TypeInformation[_] = STRING_TYPE_INFO
- override private[flink] def expectedTypes: Seq[TypeInformation[_]] =
- Seq(INT_TYPE_INFO, STRING_TYPE_INFO, STRING_TYPE_INFO)
+ override private[flink] def validateInput(): ExprValidationResult = {
+ trimMode match {
+ case SymbolExpression(_: TrimMode) =>
+ if (trimString.resultType != STRING_TYPE_INFO) {
+ ValidationFailure(s"String expected for trimString, get ${trimString.resultType}")
+ } else if (str.resultType != STRING_TYPE_INFO) {
+ ValidationFailure(s"String expected for str, get ${str.resultType}")
+ } else {
+ ValidationSuccess
+ }
+ case _ => ValidationFailure("TrimMode symbol expected.")
+ }
+ }
- override def toString: String = s"trim($trimFlag, $trimString, $str)"
+ override def toString: String = s"trim($trimMode, $trimString, $str)"
override private[flink] def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
relBuilder.call(SqlStdOperatorTable.TRIM, children.map(_.toRexNode))
@@ -195,9 +204,6 @@ case class Trim(
* Enumeration of trim flags.
*/
object TrimConstants {
- val TRIM_BOTH = Literal(0)
- val TRIM_LEADING = Literal(1)
- val TRIM_TRAILING = Literal(2)
val TRIM_DEFAULT_CHAR = Literal(" ")
}
http://git-wip-us.apache.org/repos/asf/flink/blob/cc7450d1/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/symbols.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/symbols.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/symbols.scala
new file mode 100644
index 0000000..dfa8820
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/symbols.scala
@@ -0,0 +1,132 @@
+/*
+ * 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.flink.api.table.expressions
+
+import org.apache.calcite.avatica.util.{TimeUnit, TimeUnitRange}
+import org.apache.calcite.rex.RexNode
+import org.apache.calcite.sql.fun.SqlTrimFunction
+import org.apache.calcite.tools.RelBuilder
+import org.apache.flink.api.common.typeinfo.TypeInformation
+
+import scala.language.{existentials, implicitConversions}
+
+/**
+ * General expression class to represent a symbol.
+ */
+case class SymbolExpression(symbol: TableSymbol) extends LeafExpression {
+
+ override private[flink] def resultType: TypeInformation[_] =
+ throw new UnsupportedOperationException("This should not happen. A symbol has no result type.")
+
+ def toExpr = this // triggers implicit conversion
+
+ override private[flink] def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
+ // dirty hack to pass Java enums to Java from Scala
+ val enum = symbol.enum.asInstanceOf[Enum[T] forSome { type T <: Enum[T] }]
+ relBuilder.getRexBuilder.makeFlag(enum)
+ }
+
+ override def toString: String = s"${symbol.symbols}.${symbol.name}"
+
+}
+
+/**
+ * Symbol that wraps a Calcite symbol in form of a Java enum.
+ */
+trait TableSymbol {
+ def symbols: TableSymbols
+ def name: String
+ def enum: Enum[_]
+}
+
+/**
+ * Enumeration of symbols.
+ */
+abstract class TableSymbols extends Enumeration {
+
+ class TableSymbolValue(e: Enum[_]) extends Val(e.name()) with TableSymbol {
+ override def symbols: TableSymbols = TableSymbols.this
+
+ override def enum: Enum[_] = e
+
+ override def name: String = toString()
+ }
+
+ protected final def Value(enum: Enum[_]): TableSymbolValue = new TableSymbolValue(enum)
+
+ implicit def symbolToExpression(symbol: TableSymbolValue): SymbolExpression =
+ SymbolExpression(symbol)
+
+}
+
+/**
+ * Units for working with time intervals.
+ */
+object TimeIntervalUnit extends TableSymbols {
+
+ type TimeIntervalUnit = TableSymbolValue
+
+ val YEAR = Value(TimeUnitRange.YEAR)
+ val YEAR_TO_MONTH = Value(TimeUnitRange.YEAR_TO_MONTH)
+ val MONTH = Value(TimeUnitRange.MONTH)
+ val DAY = Value(TimeUnitRange.DAY)
+ val DAY_TO_HOUR = Value(TimeUnitRange.DAY_TO_HOUR)
+ val DAY_TO_MINUTE = Value(TimeUnitRange.DAY_TO_MINUTE)
+ val DAY_TO_SECOND = Value(TimeUnitRange.DAY_TO_SECOND)
+ val HOUR = Value(TimeUnitRange.HOUR)
+ val HOUR_TO_MINUTE = Value(TimeUnitRange.HOUR_TO_MINUTE)
+ val HOUR_TO_SECOND = Value(TimeUnitRange.HOUR_TO_SECOND)
+ val MINUTE = Value(TimeUnitRange.MINUTE)
+ val MINUTE_TO_SECOND = Value(TimeUnitRange.MINUTE_TO_SECOND)
+ val SECOND = Value(TimeUnitRange.SECOND)
+
+}
+
+/**
+ * Units for working with time points.
+ */
+object TimePointUnit extends TableSymbols {
+
+ type TimePointUnit = TableSymbolValue
+
+ val YEAR = Value(TimeUnit.YEAR)
+ val MONTH = Value(TimeUnit.MONTH)
+ val DAY = Value(TimeUnit.DAY)
+ val HOUR = Value(TimeUnit.HOUR)
+ val MINUTE = Value(TimeUnit.MINUTE)
+ val SECOND = Value(TimeUnit.SECOND)
+ val QUARTER = Value(TimeUnit.QUARTER)
+ val WEEK = Value(TimeUnit.WEEK)
+ val MILLISECOND = Value(TimeUnit.MILLISECOND)
+ val MICROSECOND = Value(TimeUnit.MICROSECOND)
+
+}
+
+/**
+ * Modes for trimming strings.
+ */
+object TrimMode extends TableSymbols {
+
+ type TrimMode = TableSymbolValue
+
+ val BOTH = Value(SqlTrimFunction.Flag.BOTH)
+ val LEADING = Value(SqlTrimFunction.Flag.LEADING)
+ val TRAILING = Value(SqlTrimFunction.Flag.TRAILING)
+
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/cc7450d1/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/time.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/time.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/time.scala
new file mode 100644
index 0000000..a10f4d0
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/time.scala
@@ -0,0 +1,137 @@
+/*
+ * 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.flink.api.table.expressions
+
+import org.apache.calcite.avatica.util.{TimeUnit, TimeUnitRange}
+import org.apache.calcite.rex._
+import org.apache.calcite.sql.`type`.SqlTypeName
+import org.apache.calcite.sql.fun.SqlStdOperatorTable
+import org.apache.calcite.tools.RelBuilder
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo._
+import org.apache.flink.api.common.typeinfo.{SqlTimeTypeInfo, TypeInformation}
+import org.apache.flink.api.table.FlinkRelBuilder
+import org.apache.flink.api.table.expressions.ExpressionUtils.{divide, getFactor, mod}
+import org.apache.flink.api.table.typeutils.{IntervalTypeInfo, TypeCheckUtils}
+import org.apache.flink.api.table.validate.{ExprValidationResult, ValidationFailure, ValidationSuccess}
+
+import scala.collection.JavaConversions._
+
+case class Extract(timeIntervalUnit: Expression, temporal: Expression) extends Expression {
+
+ override private[flink] def children: Seq[Expression] = timeIntervalUnit :: temporal :: Nil
+
+ override private[flink] def resultType: TypeInformation[_] = LONG_TYPE_INFO
+
+ override private[flink] def validateInput(): ExprValidationResult = {
+ if (!TypeCheckUtils.isTemporal(temporal.resultType)) {
+ return ValidationFailure(s"Extract operator requires Temporal input, " +
+ s"but $temporal is of type ${temporal.resultType}")
+ }
+
+ timeIntervalUnit match {
+ case SymbolExpression(TimeIntervalUnit.YEAR)
+ | SymbolExpression(TimeIntervalUnit.MONTH)
+ | SymbolExpression(TimeIntervalUnit.DAY)
+ if temporal.resultType == SqlTimeTypeInfo.DATE
+ || temporal.resultType == SqlTimeTypeInfo.TIMESTAMP
+ || temporal.resultType == IntervalTypeInfo.INTERVAL_MILLIS
+ || temporal.resultType == IntervalTypeInfo.INTERVAL_MONTHS =>
+ ValidationSuccess
+
+ case SymbolExpression(TimeIntervalUnit.HOUR)
+ | SymbolExpression(TimeIntervalUnit.MINUTE)
+ | SymbolExpression(TimeIntervalUnit.SECOND)
+ if temporal.resultType == SqlTimeTypeInfo.TIME
+ || temporal.resultType == SqlTimeTypeInfo.TIMESTAMP
+ || temporal.resultType == IntervalTypeInfo.INTERVAL_MILLIS =>
+ ValidationSuccess
+
+ case _ =>
+ ValidationFailure(s"Extract operator does not support unit '$timeIntervalUnit' for input " +
+ s"of type '${temporal.resultType}'.")
+ }
+ }
+
+ override def toString: String = s"($temporal).extract($timeIntervalUnit)"
+
+ override private[flink] def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
+ // get wrapped Calcite unit
+ val timeUnitRange = timeIntervalUnit
+ .asInstanceOf[SymbolExpression]
+ .symbol
+ .enum
+ .asInstanceOf[TimeUnitRange]
+
+ // convert RexNodes
+ convertExtract(
+ timeIntervalUnit.toRexNode,
+ timeUnitRange,
+ temporal.toRexNode,
+ relBuilder.asInstanceOf[FlinkRelBuilder])
+ }
+
+ /**
+ * Standard conversion of the EXTRACT operator.
+ * Source: [[org.apache.calcite.sql2rel.StandardConvertletTable#convertExtract()]]
+ */
+ private def convertExtract(
+ timeUnitRangeRexNode: RexNode,
+ timeUnitRange: TimeUnitRange,
+ temporal: RexNode,
+ relBuilder: FlinkRelBuilder)
+ : RexNode = {
+ val rexBuilder = relBuilder.getRexBuilder
+ val resultType = relBuilder.getTypeFactory().createTypeFromTypeInfo(LONG_TYPE_INFO)
+ var result = rexBuilder.makeReinterpretCast(
+ resultType,
+ temporal,
+ rexBuilder.makeLiteral(false))
+
+ val unit = timeUnitRange.startUnit
+ val sqlTypeName = temporal.getType.getSqlTypeName
+ unit match {
+ case TimeUnit.YEAR | TimeUnit.MONTH | TimeUnit.DAY =>
+ sqlTypeName match {
+ case SqlTypeName.TIMESTAMP =>
+ result = divide(rexBuilder, result, TimeUnit.DAY.multiplier)
+ return rexBuilder.makeCall(
+ resultType,
+ SqlStdOperatorTable.EXTRACT_DATE,
+ Seq(timeUnitRangeRexNode, result))
+
+ case SqlTypeName.DATE =>
+ return rexBuilder.makeCall(
+ resultType,
+ SqlStdOperatorTable.EXTRACT_DATE,
+ Seq(timeUnitRangeRexNode, result))
+
+ case _ => // do nothing
+ }
+
+ case _ => // do nothing
+ }
+
+ result = mod(rexBuilder, resultType, result, getFactor(unit))
+ result = divide(rexBuilder, result, unit.multiplier)
+ result
+ }
+
+}
+
+
http://git-wip-us.apache.org/repos/asf/flink/blob/cc7450d1/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/RexNodeTranslator.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/RexNodeTranslator.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/RexNodeTranslator.scala
index 670417f..ae8c7c3 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/RexNodeTranslator.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/RexNodeTranslator.scala
@@ -35,10 +35,10 @@ object RexNodeTranslator {
case agg: Aggregation =>
val name = tableEnv.createUniqueAttributeName()
val aggCall = Alias(agg, name)
- val fieldExp = new UnresolvedFieldReference(name)
+ val fieldExp = UnresolvedFieldReference(name)
(fieldExp, List(aggCall))
case n @ Alias(agg: Aggregation, name) =>
- val fieldExp = new UnresolvedFieldReference(name)
+ val fieldExp = UnresolvedFieldReference(name)
(fieldExp, List(n))
case l: LeafExpression =>
(l, Nil)
http://git-wip-us.apache.org/repos/asf/flink/blob/cc7450d1/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/validate/FunctionCatalog.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/validate/FunctionCatalog.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/validate/FunctionCatalog.scala
index f447c47..58b1f69 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/validate/FunctionCatalog.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/validate/FunctionCatalog.scala
@@ -142,7 +142,10 @@ object FunctionCatalog {
"log10" -> classOf[Log10],
"ln" -> classOf[Ln],
"power" -> classOf[Power],
- "mod" -> classOf[Mod]
+ "mod" -> classOf[Mod],
+
+ // temporal functions
+ "extract" -> classOf[Extract]
)
/**
http://git-wip-us.apache.org/repos/asf/flink/blob/cc7450d1/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/batch/table/ExpressionsITCase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/batch/table/ExpressionsITCase.java b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/batch/table/ExpressionsITCase.java
index 9054cab..5476644 100644
--- a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/batch/table/ExpressionsITCase.java
+++ b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/batch/table/ExpressionsITCase.java
@@ -149,11 +149,12 @@ public class ExpressionsITCase extends TableProgramsTestBase {
Table result = table.select(
"(b && true).?('true', 'false')," +
"false.?('true', 'false')," +
- "true.?(true.?(true.?(10, 4), 4), 4)");
+ "true.?(true.?(true.?(10, 4), 4), 4)," +
+ "?((b && true), 'true', 'false')");
DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
List<Row> results = ds.collect();
- String expected = "true,false,10";
+ String expected = "true,false,10,true";
compareResultAsText(results, expected);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/cc7450d1/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/table/expressions/ScalarFunctionsTest.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/table/expressions/ScalarFunctionsTest.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/table/expressions/ScalarFunctionsTest.scala
index b2bac4b..958bebe 100644
--- a/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/table/expressions/ScalarFunctionsTest.scala
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/table/expressions/ScalarFunctionsTest.scala
@@ -18,11 +18,13 @@
package org.apache.flink.api.table.expressions
+import java.sql.{Date, Time, Timestamp}
+
import org.apache.flink.api.common.typeinfo.TypeInformation
import org.apache.flink.api.scala.table._
-import org.apache.flink.api.table.{Row, Types}
import org.apache.flink.api.table.expressions.utils.ExpressionTestBase
import org.apache.flink.api.table.typeutils.RowTypeInfo
+import org.apache.flink.api.table.{Row, Types}
import org.junit.Test
class ScalarFunctionsTest extends ExpressionTestBase {
@@ -466,10 +468,121 @@ class ScalarFunctionsTest extends ExpressionTestBase {
"-1231")
}
+ @Test
+ def testExtract(): Unit = {
+ testAllApis(
+ 'f16.extract(TimeIntervalUnit.YEAR),
+ "f16.extract(YEAR)",
+ "EXTRACT(YEAR FROM f16)",
+ "1996")
+
+ testAllApis(
+ 'f16.extract(TimeIntervalUnit.MONTH),
+ "extract(f16, MONTH)",
+ "EXTRACT(MONTH FROM f16)",
+ "11")
+
+ testAllApis(
+ 'f16.extract(TimeIntervalUnit.DAY),
+ "f16.extract(DAY)",
+ "EXTRACT(DAY FROM f16)",
+ "10")
+
+ testAllApis(
+ 'f18.extract(TimeIntervalUnit.YEAR),
+ "f18.extract(YEAR)",
+ "EXTRACT(YEAR FROM f18)",
+ "1996")
+
+ testAllApis(
+ 'f18.extract(TimeIntervalUnit.MONTH),
+ "f18.extract(MONTH)",
+ "EXTRACT(MONTH FROM f18)",
+ "11")
+
+ testAllApis(
+ 'f18.extract(TimeIntervalUnit.DAY),
+ "f18.extract(DAY)",
+ "EXTRACT(DAY FROM f18)",
+ "10")
+
+ testAllApis(
+ 'f18.extract(TimeIntervalUnit.HOUR),
+ "f18.extract(HOUR)",
+ "EXTRACT(HOUR FROM f18)",
+ "6")
+
+ testAllApis(
+ 'f17.extract(TimeIntervalUnit.HOUR),
+ "f17.extract(HOUR)",
+ "EXTRACT(HOUR FROM f17)",
+ "6")
+
+ testAllApis(
+ 'f18.extract(TimeIntervalUnit.MINUTE),
+ "f18.extract(MINUTE)",
+ "EXTRACT(MINUTE FROM f18)",
+ "55")
+
+ testAllApis(
+ 'f17.extract(TimeIntervalUnit.MINUTE),
+ "f17.extract(MINUTE)",
+ "EXTRACT(MINUTE FROM f17)",
+ "55")
+
+ testAllApis(
+ 'f18.extract(TimeIntervalUnit.SECOND),
+ "f18.extract(SECOND)",
+ "EXTRACT(SECOND FROM f18)",
+ "44")
+
+ testAllApis(
+ 'f17.extract(TimeIntervalUnit.SECOND),
+ "f17.extract(SECOND)",
+ "EXTRACT(SECOND FROM f17)",
+ "44")
+
+ testAllApis(
+ 'f19.extract(TimeIntervalUnit.DAY),
+ "f19.extract(DAY)",
+ "EXTRACT(DAY FROM f19)",
+ "16979")
+
+ testAllApis(
+ 'f19.extract(TimeIntervalUnit.HOUR),
+ "f19.extract(HOUR)",
+ "EXTRACT(HOUR FROM f19)",
+ "7")
+
+ testAllApis(
+ 'f19.extract(TimeIntervalUnit.MINUTE),
+ "f19.extract(MINUTE)",
+ "EXTRACT(MINUTE FROM f19)",
+ "23")
+
+ testAllApis(
+ 'f19.extract(TimeIntervalUnit.SECOND),
+ "f19.extract(SECOND)",
+ "EXTRACT(SECOND FROM f19)",
+ "33")
+
+ testAllApis(
+ 'f20.extract(TimeIntervalUnit.MONTH),
+ "f20.extract(MONTH)",
+ "EXTRACT(MONTH FROM f20)",
+ "1")
+
+ testAllApis(
+ 'f20.extract(TimeIntervalUnit.YEAR),
+ "f20.extract(YEAR)",
+ "EXTRACT(YEAR FROM f20)",
+ "2")
+ }
+
// ----------------------------------------------------------------------------------------------
def testData = {
- val testData = new Row(16)
+ val testData = new Row(21)
testData.setField(0, "This is a test String.")
testData.setField(1, true)
testData.setField(2, 42.toByte)
@@ -486,6 +599,11 @@ class ScalarFunctionsTest extends ExpressionTestBase {
testData.setField(13, -4.6)
testData.setField(14, -3)
testData.setField(15, BigDecimal("-1231.1231231321321321111").bigDecimal)
+ testData.setField(16, Date.valueOf("1996-11-10"))
+ testData.setField(17, Time.valueOf("06:55:44"))
+ testData.setField(18, Timestamp.valueOf("1996-11-10 06:55:44.333"))
+ testData.setField(19, 1467012213000L) // +16979 07:23:33.000
+ testData.setField(20, 25) // +2-01
testData
}
@@ -506,6 +624,11 @@ class ScalarFunctionsTest extends ExpressionTestBase {
Types.FLOAT,
Types.DOUBLE,
Types.INT,
- Types.DECIMAL)).asInstanceOf[TypeInformation[Any]]
+ Types.DECIMAL,
+ Types.DATE,
+ Types.TIME,
+ Types.TIMESTAMP,
+ Types.INTERVAL_MILLIS,
+ Types.INTERVAL_MONTHS)).asInstanceOf[TypeInformation[Any]]
}
}