[FLINK-3580] [table] Add QUARTER function

fullcontact · Sep 9, 2016 · 9420a77 · 9420a77
1 parent 8c0d624
commit 9420a77
Show file tree

Hide file tree

Showing 6 changed files with 90 additions and 4 deletions.
diff --git a/docs/dev/table_api.md b/docs/dev/table_api.md
@@ -1520,7 +1520,18 @@ 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>
+ <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>
+
+ <tr>
+  <td>
+  {% highlight java %}
+DATE.quarter()
+{% endhighlight %}
+ </td>
+ <td>
+ <p>Returns the quarter of a year from a SQL date. E.g. <code>'1994-09-27'.toDate.quarter()</code> leads to 3.</p>
  </td>
  </tr>
 
@@ -1531,7 +1542,7 @@ TIMEPOINT.floor(TIMEINTERVALUNIT)
 {% endhighlight %}
  </td>
  <td>
- <p>Rounds a time point down to the given unit. E.g. <code>"12:44:31".toDate.floor(MINUTE)</code> leads to 12:44:00.</p>
+ <p>Rounds a time point down to the given unit. E.g. <code>'12:44:31'.toDate.floor(MINUTE)</code> leads to 12:44:00.</p>
  </td>
  </tr>
 
@@ -1542,7 +1553,7 @@ TIMEPOINT.ceil(TIMEINTERVALUNIT)
 {% endhighlight %}
  </td>
  <td>
- <p>Rounds a time point up to the given unit. E.g. <code>"12:44:31".toTime.floor(MINUTE)</code> leads to 12:45:00.</p>
+ <p>Rounds a time point up to the given unit. E.g. <code>'12:44:31'.toTime.floor(MINUTE)</code> leads to 12:45:00.</p>
  </td>
  </tr>
 
@@ -1909,6 +1920,17 @@ TEMPORAL.extract(TimeIntervalUnit)
  </td>
  </tr>
 
+ <tr>
+  <td>
+  {% highlight scala %}
+DATE.quarter()
+{% endhighlight %}
+ </td>
+ <td>
+ <p>Returns the quarter of a year from a SQL date. E.g. <code>"1994-09-27".toDate.quarter()</code> leads to 3.</p>
+ </td>
+ </tr>
+
  <tr>
   <td>
   {% highlight scala %}

diff --git a/...libraries/flink-table/src/main/scala/org/apache/flink/api/scala/table/expressionDsl.scala b/...libraries/flink-table/src/main/scala/org/apache/flink/api/scala/table/expressionDsl.scala
@@ -276,6 +276,13 @@ trait ImplicitExpressionOperations {
  */
  def extract(timeIntervalUnit: TimeIntervalUnit) = Extract(timeIntervalUnit, expr)
 
+ /**
+ * Returns the quarter of a year from a SQL date.
+ *
+ * e.g. "1994-09-27".toDate.quarter() leads to 3
+ */
+ def quarter() = Quarter(expr)
+
  /**
  * Rounds down a time point to the given unit.
  *

diff --git a/...flink-table/src/main/scala/org/apache/flink/api/table/codegen/calls/ScalarFunctions.scala b/...flink-table/src/main/scala/org/apache/flink/api/table/codegen/calls/ScalarFunctions.scala
@@ -199,6 +199,12 @@ object ScalarFunctions {
  LONG_TYPE_INFO,
  BuiltInMethod.UNIX_DATE_EXTRACT.method)
 
+ addSqlFunctionMethod(
+ EXTRACT_DATE,
+ Seq(new GenericTypeInfo(classOf[TimeUnitRange]), SqlTimeTypeInfo.DATE),
+ LONG_TYPE_INFO,
+ BuiltInMethod.UNIX_DATE_EXTRACT.method)
+
  addSqlFunction(
  FLOOR,
  Seq(SqlTimeTypeInfo.DATE, new GenericTypeInfo(classOf[TimeUnitRange])),

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
@@ -97,6 +97,8 @@ case class Extract(timeIntervalUnit: Expression, temporal: Expression) extends E
  temporal: RexNode,
  relBuilder: FlinkRelBuilder)
  : RexNode = {
+
+ // TODO convert this into Table API expressions to make the code more readable
  val rexBuilder = relBuilder.getRexBuilder
  val resultType = relBuilder.getTypeFactory().createTypeFromTypeInfo(LONG_TYPE_INFO)
  var result = rexBuilder.makeReinterpretCast(
@@ -248,3 +250,30 @@ case class LocalTime() extends CurrentTimePoint(SqlTimeTypeInfo.TIME, local = tr
 
 case class LocalTimestamp() extends CurrentTimePoint(SqlTimeTypeInfo.TIMESTAMP, local = true)
 
+/**
+ * Extracts the quarter of a year from a SQL date.
+ */
+case class Quarter(child: Expression) extends UnaryExpression with InputTypeSpec {
+
+ override private[flink] def expectedTypes: Seq[TypeInformation[_]] = Seq(SqlTimeTypeInfo.DATE)
+
+ override private[flink] def resultType: TypeInformation[_] = LONG_TYPE_INFO
+
+ override def toString: String = s"($child).quarter()"
+
+ override private[flink] def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
+ /**
+ * Standard conversion of the QUARTER operator.
+ * Source: [[org.apache.calcite.sql2rel.StandardConvertletTable#convertQuarter()]]
+ */
+ Plus(
+ Div(
+ Minus(
+ Extract(TimeIntervalUnit.MONTH, child),
+ Literal(1L)),
+ Literal(TimeUnit.QUARTER.multiplier.longValue())),
+ Literal(1L)
+ ).toRexNode
+ }
+}
+
diff --git a/...ries/flink-table/src/main/scala/org/apache/flink/api/table/validate/FunctionCatalog.scala b/...ries/flink-table/src/main/scala/org/apache/flink/api/table/validate/FunctionCatalog.scala
@@ -158,7 +158,8 @@ object FunctionCatalog {
  "currentTime" -> classOf[CurrentTime],
  "currentTimestamp" -> classOf[CurrentTimestamp],
  "localTime" -> classOf[LocalTime],
- "localTimestamp" -> classOf[LocalTimestamp]
+ "localTimestamp" -> classOf[LocalTimestamp],
+ "quarter" -> classOf[Quarter]
 
  // TODO implement function overloading here
  // "floor" -> classOf[TemporalFloor]

diff --git a/...ink-table/src/test/scala/org/apache/flink/api/table/expressions/ScalarFunctionsTest.scala b/...ink-table/src/test/scala/org/apache/flink/api/table/expressions/ScalarFunctionsTest.scala
@@ -828,6 +828,27 @@ class ScalarFunctionsTest extends ExpressionTestBase {
  "true")
  }
 
+ @Test
+ def testQuarter(): Unit = {
+ testAllApis(
+ "1997-01-27".toDate.quarter(),
+ "'1997-01-27'.toDate.quarter()",
+ "QUARTER(DATE '1997-01-27')",
+ "1")
+
+ testAllApis(
+ "1997-04-27".toDate.quarter(),
+ "'1997-04-27'.toDate.quarter()",
+ "QUARTER(DATE '1997-04-27')",
+ "2")
+
+ testAllApis(
+ "1997-12-31".toDate.quarter(),
+ "'1997-12-31'.toDate.quarter()",
+ "QUARTER(DATE '1997-12-31')",
+ "4")
+ }
+
  // ----------------------------------------------------------------------------------------------
 
  def testData = {