You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by xueyumusic <gi...@git.apache.org> on 2018/06/20 13:34:49 UTC
[GitHub] flink pull request #6188: [FLINK-6846] add timestampAdd tableApi
GitHub user xueyumusic opened a pull request:
https://github.com/apache/flink/pull/6188
[FLINK-6846] add timestampAdd tableApi
## What is the purpose of the change
This PR add timestampAdd TableApi
## Brief change log
add timestampAdd expression parser and node construction
## Verifying this change
unit test
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): no
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: no
## Documentation
- Does this pull request introduce a new feature? yes
- If yes, how is the feature documented? docs
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/xueyumusic/flink timeStampAdd-tableApi
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/flink/pull/6188.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #6188
----
commit 17511d36c39c86ce342a27583a474991713d0062
Author: xueyu <27...@...>
Date: 2018-06-20T13:30:02Z
timestampAdd tableApi
----
---
[GitHub] flink pull request #6188: [FLINK-6846][Table API] add timestampAdd tableApi
Posted by hequn8128 <gi...@git.apache.org>.
Github user hequn8128 commented on a diff in the pull request:
https://github.com/apache/flink/pull/6188#discussion_r197614206
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/expressions/time.scala ---
@@ -328,6 +330,42 @@ case class TemporalOverlaps(
}
}
+ /**
+ * Standard conversion of the TIMESTAMPADD operator.
+ * Source: [[org.apache.calcite.sql2rel.StandardConvertletTable#TimestampAddConvertlet]]
+ */
+case class TimestampAdd(
+ unit: Expression,
+ count: Expression,
+ timestamp: Expression)
+ extends Expression {
+
+ override private[flink] def children = unit :: count :: timestamp :: Nil
+
+ override private[flink] def toRexNode(implicit relBuilder: RelBuilder) = {
--- End diff --
toRexNode(implicit relBuilder: RelBuilder) => toRexNode(implicit relBuilder: RelBuilder): RexNode
---
[GitHub] flink pull request #6188: [FLINK-6846][Table API] add timestampAdd tableApi
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/6188#discussion_r199485393
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/scala/expressionDsl.scala ---
@@ -1029,6 +1029,29 @@ object temporalOverlaps {
TemporalOverlaps(leftTimePoint, leftTemporal, rightTimePoint, rightTemporal)
}
}
+/**
+ * Adds a (signed) integer interval to a timestamp. The unit for the interval is given
+ * by the unit argument, which should be one of the following values: "SECOND", "MINUTE",
+ * "HOUR", "DAY", "WEEK", "MONTH", "QUARTER" or "YEAR".
+ *
+ * e.g. timestampAdd("WEEK", 1, '2003-01-02'.toDate) leads to "2003-01-09".
+ */
+object timestampAdd {
+
+ /**
+ * Adds a (signed) integer interval to a timestamp. The unit for the interval is given
+ * by the unit argument, which should be one of the following values: "SECOND", "MINUTE",
+ * "HOUR", "DAY", "WEEK", "MONTH", "QUARTER" or "YEAR".
+ *
+ * e.g. timestampAdd("WEEK", 1, '2003-01-02'.toDate) leads to "2003-01-09".
+ */
+ def apply(
+ unit: Expression,
--- End diff --
I agree with @walterddr but let's wait for @twalthr who added the `quarter` function
---
[GitHub] flink pull request #6188: [FLINK-6846][Table API] add timestampAdd tableApi
Posted by hequn8128 <gi...@git.apache.org>.
Github user hequn8128 commented on a diff in the pull request:
https://github.com/apache/flink/pull/6188#discussion_r199349619
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/scala/expressionDsl.scala ---
@@ -1029,6 +1029,29 @@ object temporalOverlaps {
TemporalOverlaps(leftTimePoint, leftTemporal, rightTimePoint, rightTemporal)
}
}
+/**
+ * Adds a (signed) integer interval to a timestamp. The unit for the interval is given
+ * by the unit argument, which should be one of the following values: "SECOND", "MINUTE",
+ * "HOUR", "DAY", "WEEK", "MONTH", "QUARTER" or "YEAR".
+ *
+ * e.g. timestampAdd("WEEK", 1, '2003-01-02'.toDate) leads to "2003-01-09".
+ */
+object timestampAdd {
+
+ /**
+ * Adds a (signed) integer interval to a timestamp. The unit for the interval is given
+ * by the unit argument, which should be one of the following values: "SECOND", "MINUTE",
+ * "HOUR", "DAY", "WEEK", "MONTH", "QUARTER" or "YEAR".
+ *
+ * e.g. timestampAdd("WEEK", 1, '2003-01-02'.toDate) leads to "2003-01-09".
+ */
+ def apply(
+ unit: Expression,
--- End diff --
+1 for better consistency. It is good to follow the Table-api style.
---
[GitHub] flink pull request #6188: [FLINK-6846][Table API] add timestampAdd tableApi
Posted by xueyumusic <gi...@git.apache.org>.
Github user xueyumusic commented on a diff in the pull request:
https://github.com/apache/flink/pull/6188#discussion_r197674277
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/expressions/time.scala ---
@@ -349,6 +349,17 @@ case class TimestampAdd(
if (!TypeCheckUtils.isString(unit.resultType)) {
return ValidationFailure(s"TimestampAdd operator requires unit to be of type " +
s"String Literal, but get ${unit.resultType}.")
+ } else {
+ val unitStr = unit.toString()
+ if (!sqlTsiArray.contains(unitStr) &&
+ !sqlTsiArray.map(item => item.split("_").last).contains(unitStr)) {
+ return ValidationFailure(s"TimestampAdd operator requires unit to be one of (YEAR, " +
--- End diff --
fix this, add ")", thank you!
---
[GitHub] flink issue #6188: [FLINK-6846][Table API] add timestampAdd tableApi
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on the issue:
https://github.com/apache/flink/pull/6188
Hmm, I think there is still a case for `timestampAdd`. The problem are time intervals with variable length such as `MONTH`, `QUARTER`, and `YEAR`. All of these cannot be defined by milliseconds, because they depend on the context. So `timstampAdd(ts, 1.month)` returns a different result than `ts + 1.month`.
---
[GitHub] flink pull request #6188: [FLINK-6846][Table API] add timestampAdd tableApi
Posted by hequn8128 <gi...@git.apache.org>.
Github user hequn8128 commented on a diff in the pull request:
https://github.com/apache/flink/pull/6188#discussion_r197614223
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/expressions/time.scala ---
@@ -328,6 +330,42 @@ case class TemporalOverlaps(
}
}
+ /**
+ * Standard conversion of the TIMESTAMPADD operator.
+ * Source: [[org.apache.calcite.sql2rel.StandardConvertletTable#TimestampAddConvertlet]]
+ */
+case class TimestampAdd(
+ unit: Expression,
+ count: Expression,
+ timestamp: Expression)
+ extends Expression {
+
+ override private[flink] def children = unit :: count :: timestamp :: Nil
+
+ override private[flink] def toRexNode(implicit relBuilder: RelBuilder) = {
+ var timeUnit : Option[TimeUnit] = None
+ if (unit.isInstanceOf[Literal]) {
+ var unitValue= unit.asInstanceOf[Literal].value.toString()
+ val sqlTsiArray = Array("SQL_TSI_YEAR", "SQL_TSI_QUARTER", "SQL_TSI_MONTH", "SQL_TSI_WEEK",
+ "SQL_TSI_DAY", "SQL_TSI_HOUR", "SQL_TSI_MINUTE", "SQL_TSI_SECOND")
+ if (sqlTsiArray.contains(unitValue)) {
+ unitValue = unitValue.split("_").last
+ }
+ timeUnit = Some(TimeUnit.valueOf(unitValue))
+ }
+
+ relBuilder.call(SqlStdOperatorTable.DATETIME_PLUS, timestamp.toRexNode,
+ relBuilder.call(SqlStdOperatorTable.MULTIPLY,
+ relBuilder.getRexBuilder.makeIntervalLiteral(timeUnit.get.multiplier,
+ new SqlIntervalQualifier(timeUnit.get,null,SqlParserPos.ZERO)),
+ count.toRexNode))
+ }
+
+ override def toString: String = s"timestampAdd(${children.mkString(", ")})"
+
+ override private[flink] def resultType = STRING_TYPE_INFO
--- End diff --
def resultType => def resultType: TypeInformation[_]
---
[GitHub] flink pull request #6188: [FLINK-6846][Table API] add timestampAdd tableApi
Posted by walterddr <gi...@git.apache.org>.
Github user walterddr commented on a diff in the pull request:
https://github.com/apache/flink/pull/6188#discussion_r199352237
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/scala/expressionDsl.scala ---
@@ -1029,6 +1029,29 @@ object temporalOverlaps {
TemporalOverlaps(leftTimePoint, leftTemporal, rightTimePoint, rightTemporal)
}
}
+/**
+ * Adds a (signed) integer interval to a timestamp. The unit for the interval is given
+ * by the unit argument, which should be one of the following values: "SECOND", "MINUTE",
+ * "HOUR", "DAY", "WEEK", "MONTH", "QUARTER" or "YEAR".
+ *
+ * e.g. timestampAdd("WEEK", 1, '2003-01-02'.toDate) leads to "2003-01-09".
+ */
+object timestampAdd {
+
+ /**
+ * Adds a (signed) integer interval to a timestamp. The unit for the interval is given
+ * by the unit argument, which should be one of the following values: "SECOND", "MINUTE",
+ * "HOUR", "DAY", "WEEK", "MONTH", "QUARTER" or "YEAR".
+ *
+ * e.g. timestampAdd("WEEK", 1, '2003-01-02'.toDate) leads to "2003-01-09".
+ */
+ def apply(
+ unit: Expression,
--- End diff --
+1 for this approach that directly specifies the interval literals.
Regarding Quarter. It seems like a very old implementation and we should probably use `"1970-01-01".toDate.extract(TimeIntervalUnit.QUARTER)` to make it consistent with all other time unit extractions. What do you guys think?
I just tried it out by modifying the `Extract` method and it seems working perfectly.
---
[GitHub] flink pull request #6188: [FLINK-6846][Table API] add timestampAdd tableApi
Posted by hequn8128 <gi...@git.apache.org>.
Github user hequn8128 commented on a diff in the pull request:
https://github.com/apache/flink/pull/6188#discussion_r197640442
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/expressions/time.scala ---
@@ -328,6 +330,71 @@ case class TemporalOverlaps(
}
}
+ /**
+ * Standard conversion of the TIMESTAMPADD operator.
+ * Source: [[org.apache.calcite.sql2rel.StandardConvertletTable#TimestampAddConvertlet]]
+ */
+case class TimestampAdd(
+ unit: Expression,
+ count: Expression,
+ timestamp: Expression)
+ extends Expression {
+
+ private[flink] final val sqlTsiArray = Array("SQL_TSI_YEAR", "SQL_TSI_QUARTER", "SQL_TSI_MONTH",
+ "SQL_TSI_WEEK", "SQL_TSI_DAY", "SQL_TSI_HOUR", "SQL_TSI_MINUTE", "SQL_TSI_SECOND")
+
+ override private[flink] def children: Seq[Expression] = unit :: count :: timestamp :: Nil
+
+ override private[flink] def validateInput(): ValidationResult = {
+ if (!TypeCheckUtils.isString(unit.resultType)) {
--- End diff --
1. Not only a string, but also should be a valid unit, i.e., must be YEAR MONTH...
2. Check count is a type of Integer or Long
3. Add ValidationTest, you can refer to the tests in `ScalarFunctionsValidationTest`
---
[GitHub] flink issue #6188: [FLINK-6846][Table API] add timestampAdd tableApi
Posted by twalthr <gi...@git.apache.org>.
Github user twalthr commented on the issue:
https://github.com/apache/flink/pull/6188
@fhueske `MONTH`, `QUARTER`, and `YEAR` are not converted into milliseconds but treated as interval of months. In general, Flink supports two types of intervals one for millis and one for months. So given that we support `TIMESTAMP + INTERVAL_MONTHS` the behavior should be the same as a timestamp add function.
---
[GitHub] flink pull request #6188: [FLINK-6846][Table API] add timestampAdd tableApi
Posted by hequn8128 <gi...@git.apache.org>.
Github user hequn8128 commented on a diff in the pull request:
https://github.com/apache/flink/pull/6188#discussion_r197662959
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/expressions/time.scala ---
@@ -349,6 +349,17 @@ case class TimestampAdd(
if (!TypeCheckUtils.isString(unit.resultType)) {
return ValidationFailure(s"TimestampAdd operator requires unit to be of type " +
s"String Literal, but get ${unit.resultType}.")
+ } else {
+ val unitStr = unit.toString()
+ if (!sqlTsiArray.contains(unitStr) &&
+ !sqlTsiArray.map(item => item.split("_").last).contains(unitStr)) {
+ return ValidationFailure(s"TimestampAdd operator requires unit to be one of (YEAR, " +
--- End diff --
Remove the "(" from "(YEAR", or add a ")" behind
---
[GitHub] flink pull request #6188: [FLINK-6846][Table API] add timestampAdd tableApi
Posted by xueyumusic <gi...@git.apache.org>.
Github user xueyumusic commented on a diff in the pull request:
https://github.com/apache/flink/pull/6188#discussion_r199040249
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/scala/expressionDsl.scala ---
@@ -1029,6 +1029,29 @@ object temporalOverlaps {
TemporalOverlaps(leftTimePoint, leftTemporal, rightTimePoint, rightTemporal)
}
}
+/**
+ * Adds a (signed) integer interval to a timestamp. The unit for the interval is given
+ * by the unit argument, which should be one of the following values: "SECOND", "MINUTE",
+ * "HOUR", "DAY", "WEEK", "MONTH", "QUARTER" or "YEAR".
+ *
+ * e.g. timestampAdd("WEEK", 1, '2003-01-02'.toDate) leads to "2003-01-09".
+ */
+object timestampAdd {
+
+ /**
+ * Adds a (signed) integer interval to a timestamp. The unit for the interval is given
+ * by the unit argument, which should be one of the following values: "SECOND", "MINUTE",
+ * "HOUR", "DAY", "WEEK", "MONTH", "QUARTER" or "YEAR".
+ *
+ * e.g. timestampAdd("WEEK", 1, '2003-01-02'.toDate) leads to "2003-01-09".
+ */
+ def apply(
+ unit: Expression,
--- End diff --
thanks for your review and advice, @fhueske , yes, At the beginning I had thought these two kinds of function signature, and chose the current one following the sql signature. I think your suggestions is reasonable, it follows the table api style and makes the code concise.
There are two problems that I found are: one is that `1.quarter` has been realized in other meaning (Quarter table api) and another is `1.week` seems to be not existed right now (maybe we can add).Thank you
---
[GitHub] flink issue #6188: [FLINK-6846][Table API] add timestampAdd tableApi
Posted by xueyumusic <gi...@git.apache.org>.
Github user xueyumusic commented on the issue:
https://github.com/apache/flink/pull/6188
Travis error is "The job exceeded the maximum time limit for jobs, and has been terminated". It seems connector module test and not related to this PR.
---
[GitHub] flink pull request #6188: [FLINK-6846][Table API] add timestampAdd tableApi
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/6188#discussion_r199190279
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/scala/expressionDsl.scala ---
@@ -1029,6 +1029,29 @@ object temporalOverlaps {
TemporalOverlaps(leftTimePoint, leftTemporal, rightTimePoint, rightTemporal)
}
}
+/**
+ * Adds a (signed) integer interval to a timestamp. The unit for the interval is given
+ * by the unit argument, which should be one of the following values: "SECOND", "MINUTE",
+ * "HOUR", "DAY", "WEEK", "MONTH", "QUARTER" or "YEAR".
+ *
+ * e.g. timestampAdd("WEEK", 1, '2003-01-02'.toDate) leads to "2003-01-09".
+ */
+object timestampAdd {
+
+ /**
+ * Adds a (signed) integer interval to a timestamp. The unit for the interval is given
+ * by the unit argument, which should be one of the following values: "SECOND", "MINUTE",
+ * "HOUR", "DAY", "WEEK", "MONTH", "QUARTER" or "YEAR".
+ *
+ * e.g. timestampAdd("WEEK", 1, '2003-01-02'.toDate) leads to "2003-01-09".
+ */
+ def apply(
+ unit: Expression,
--- End diff --
I think we can and should add `week`.
`quarter` is a bit more tricky... IMO, the current method is not well aligned with the other methods (`hour`, `day`, etc.) and it would make sense to change it for better consistency. However, such as change would break the API. What do you think @twalthr?
---
[GitHub] flink pull request #6188: [FLINK-6846][Table API] add timestampAdd tableApi
Posted by walterddr <gi...@git.apache.org>.
Github user walterddr commented on a diff in the pull request:
https://github.com/apache/flink/pull/6188#discussion_r197613157
--- Diff: docs/dev/table/tableApi.md ---
@@ -2804,6 +2804,17 @@ dateFormat(TIMESTAMP, STRING)
</td>
</tr>
+ <tr>
+ <td>
+ {% highlight java %}
+timestampAdd(unit, interval, timestamp)
+{% endhighlight %}
+ </td>
+ <td>
+ <p>Adds a (signed) integer interval to a timestamp. The unit for the interval is given by the unit argument, which should be one of the following values: <code>"SECOND"</code>, <code>"MINUTE"</code>, <code>"HOUR"</code>, <code>"DAY"</code>, <code>"WEEK"</code>, <code>"MONTH"</code>, <code>"QUARTER"</code>, or <code>"YEAR"</code>. E.g. <code>timestampAdd("WEEK", 1, '2003-01-02'.toDate)</code> leads to <code>2003-01-09</code>.</p>
--- End diff --
Let's make it consistent with `sql.md`, I believe there's no `"` for the agument values
---
[GitHub] flink issue #6188: [FLINK-6846][Table API] add timestampAdd tableApi
Posted by xueyumusic <gi...@git.apache.org>.
Github user xueyumusic commented on the issue:
https://github.com/apache/flink/pull/6188
The first time to submit pr and could someone review it and give some advice, thanks :)
---
[GitHub] flink pull request #6188: [FLINK-6846][Table API] add timestampAdd tableApi
Posted by walterddr <gi...@git.apache.org>.
Github user walterddr commented on a diff in the pull request:
https://github.com/apache/flink/pull/6188#discussion_r197614566
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/expressions/time.scala ---
@@ -328,6 +330,42 @@ case class TemporalOverlaps(
}
}
+ /**
+ * Standard conversion of the TIMESTAMPADD operator.
+ * Source: [[org.apache.calcite.sql2rel.StandardConvertletTable#TimestampAddConvertlet]]
+ */
+case class TimestampAdd(
+ unit: Expression,
+ count: Expression,
+ timestamp: Expression)
+ extends Expression {
+
+ override private[flink] def children = unit :: count :: timestamp :: Nil
+
+ override private[flink] def toRexNode(implicit relBuilder: RelBuilder) = {
+ var timeUnit : Option[TimeUnit] = None
+ if (unit.isInstanceOf[Literal]) {
+ var unitValue= unit.asInstanceOf[Literal].value.toString()
+ val sqlTsiArray = Array("SQL_TSI_YEAR", "SQL_TSI_QUARTER", "SQL_TSI_MONTH", "SQL_TSI_WEEK",
+ "SQL_TSI_DAY", "SQL_TSI_HOUR", "SQL_TSI_MINUTE", "SQL_TSI_SECOND")
+ if (sqlTsiArray.contains(unitValue)) {
+ unitValue = unitValue.split("_").last
+ }
+ timeUnit = Some(TimeUnit.valueOf(unitValue))
+ }
+
+ relBuilder.call(SqlStdOperatorTable.DATETIME_PLUS, timestamp.toRexNode,
+ relBuilder.call(SqlStdOperatorTable.MULTIPLY,
+ relBuilder.getRexBuilder.makeIntervalLiteral(timeUnit.get.multiplier,
--- End diff --
I think this part can probably be further optimize if `count` is also a Literal, this way the MULTIPLY operator is not necessary and it can directly convert to `IntervalSqlType`. what do you think?
---
[GitHub] flink pull request #6188: [FLINK-6846][Table API] add timestampAdd tableApi
Posted by walterddr <gi...@git.apache.org>.
Github user walterddr commented on a diff in the pull request:
https://github.com/apache/flink/pull/6188#discussion_r197622239
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/expressions/time.scala ---
@@ -328,6 +330,42 @@ case class TemporalOverlaps(
}
}
+ /**
+ * Standard conversion of the TIMESTAMPADD operator.
+ * Source: [[org.apache.calcite.sql2rel.StandardConvertletTable#TimestampAddConvertlet]]
+ */
+case class TimestampAdd(
+ unit: Expression,
+ count: Expression,
+ timestamp: Expression)
+ extends Expression {
+
+ override private[flink] def children = unit :: count :: timestamp :: Nil
+
+ override private[flink] def toRexNode(implicit relBuilder: RelBuilder) = {
+ var timeUnit : Option[TimeUnit] = None
+ if (unit.isInstanceOf[Literal]) {
+ var unitValue= unit.asInstanceOf[Literal].value.toString()
+ val sqlTsiArray = Array("SQL_TSI_YEAR", "SQL_TSI_QUARTER", "SQL_TSI_MONTH", "SQL_TSI_WEEK",
+ "SQL_TSI_DAY", "SQL_TSI_HOUR", "SQL_TSI_MINUTE", "SQL_TSI_SECOND")
+ if (sqlTsiArray.contains(unitValue)) {
+ unitValue = unitValue.split("_").last
+ }
+ timeUnit = Some(TimeUnit.valueOf(unitValue))
+ }
+
+ relBuilder.call(SqlStdOperatorTable.DATETIME_PLUS, timestamp.toRexNode,
+ relBuilder.call(SqlStdOperatorTable.MULTIPLY,
+ relBuilder.getRexBuilder.makeIntervalLiteral(timeUnit.get.multiplier,
--- End diff --
no. what I mean is something like:
```
count match {
case literal: Literal => //...generate a new IntervalSqlType based on unit and count
case other: _ => //... Use your current approach using SQL SqlStdOperatorTable.MULTIPLY
}
```
You can move the logic to validate `unit` as a `Literal` to `validateInput` function like @hequn8128 mentioned.
---
[GitHub] flink pull request #6188: [FLINK-6846][Table API] add timestampAdd tableApi
Posted by hequn8128 <gi...@git.apache.org>.
Github user hequn8128 commented on a diff in the pull request:
https://github.com/apache/flink/pull/6188#discussion_r197614217
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/expressions/time.scala ---
@@ -328,6 +330,42 @@ case class TemporalOverlaps(
}
}
+ /**
+ * Standard conversion of the TIMESTAMPADD operator.
+ * Source: [[org.apache.calcite.sql2rel.StandardConvertletTable#TimestampAddConvertlet]]
+ */
+case class TimestampAdd(
+ unit: Expression,
+ count: Expression,
+ timestamp: Expression)
+ extends Expression {
+
+ override private[flink] def children = unit :: count :: timestamp :: Nil
+
+ override private[flink] def toRexNode(implicit relBuilder: RelBuilder) = {
+ var timeUnit : Option[TimeUnit] = None
+ if (unit.isInstanceOf[Literal]) {
+ var unitValue= unit.asInstanceOf[Literal].value.toString()
+ val sqlTsiArray = Array("SQL_TSI_YEAR", "SQL_TSI_QUARTER", "SQL_TSI_MONTH", "SQL_TSI_WEEK",
+ "SQL_TSI_DAY", "SQL_TSI_HOUR", "SQL_TSI_MINUTE", "SQL_TSI_SECOND")
+ if (sqlTsiArray.contains(unitValue)) {
+ unitValue = unitValue.split("_").last
+ }
+ timeUnit = Some(TimeUnit.valueOf(unitValue))
+ }
+
+ relBuilder.call(SqlStdOperatorTable.DATETIME_PLUS, timestamp.toRexNode,
+ relBuilder.call(SqlStdOperatorTable.MULTIPLY,
+ relBuilder.getRexBuilder.makeIntervalLiteral(timeUnit.get.multiplier,
+ new SqlIntervalQualifier(timeUnit.get,null,SqlParserPos.ZERO)),
--- End diff --
(timeUnit.get,null,SqlParserPos.ZERO) => (timeUnit.get, null, SqlParserPos.ZERO)
add a space after the comma
---
[GitHub] flink pull request #6188: [FLINK-6846][Table API] add timestampAdd tableApi
Posted by walterddr <gi...@git.apache.org>.
Github user walterddr commented on a diff in the pull request:
https://github.com/apache/flink/pull/6188#discussion_r197613915
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/expressions/time.scala ---
@@ -328,6 +330,42 @@ case class TemporalOverlaps(
}
}
+ /**
+ * Standard conversion of the TIMESTAMPADD operator.
+ * Source: [[org.apache.calcite.sql2rel.StandardConvertletTable#TimestampAddConvertlet]]
+ */
+case class TimestampAdd(
+ unit: Expression,
+ count: Expression,
+ timestamp: Expression)
+ extends Expression {
+
+ override private[flink] def children = unit :: count :: timestamp :: Nil
+
+ override private[flink] def toRexNode(implicit relBuilder: RelBuilder) = {
+ var timeUnit : Option[TimeUnit] = None
+ if (unit.isInstanceOf[Literal]) {
--- End diff --
I think if we use pattern matching here it will be much cleaner, something like.
```
unit match {
case literal: Literal => //...
case _ => //...
}
```
That also remind me we should probably override `validateInput` function as well.
---
[GitHub] flink pull request #6188: [FLINK-6846][Table API] add timestampAdd tableApi
Posted by xueyumusic <gi...@git.apache.org>.
Github user xueyumusic commented on a diff in the pull request:
https://github.com/apache/flink/pull/6188#discussion_r197616443
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/expressions/time.scala ---
@@ -328,6 +330,42 @@ case class TemporalOverlaps(
}
}
+ /**
+ * Standard conversion of the TIMESTAMPADD operator.
+ * Source: [[org.apache.calcite.sql2rel.StandardConvertletTable#TimestampAddConvertlet]]
+ */
+case class TimestampAdd(
+ unit: Expression,
+ count: Expression,
+ timestamp: Expression)
+ extends Expression {
+
+ override private[flink] def children = unit :: count :: timestamp :: Nil
+
+ override private[flink] def toRexNode(implicit relBuilder: RelBuilder) = {
+ var timeUnit : Option[TimeUnit] = None
+ if (unit.isInstanceOf[Literal]) {
+ var unitValue= unit.asInstanceOf[Literal].value.toString()
+ val sqlTsiArray = Array("SQL_TSI_YEAR", "SQL_TSI_QUARTER", "SQL_TSI_MONTH", "SQL_TSI_WEEK",
+ "SQL_TSI_DAY", "SQL_TSI_HOUR", "SQL_TSI_MINUTE", "SQL_TSI_SECOND")
+ if (sqlTsiArray.contains(unitValue)) {
+ unitValue = unitValue.split("_").last
+ }
+ timeUnit = Some(TimeUnit.valueOf(unitValue))
+ }
+
+ relBuilder.call(SqlStdOperatorTable.DATETIME_PLUS, timestamp.toRexNode,
+ relBuilder.call(SqlStdOperatorTable.MULTIPLY,
+ relBuilder.getRexBuilder.makeIntervalLiteral(timeUnit.get.multiplier,
--- End diff --
Hi, @walterddr , thanks, do you mean that support both `timestampAdd("DAY", 1 ,"2016-06-15".toDate)` and `timestampAdd("INTERVAL 1 DAY", "2016-06-15".toDate)` two kinds of api. I considered both api format at the beginning and chose only one after. If could support both, that will be better.
---
[GitHub] flink pull request #6188: [FLINK-6846][Table API] add timestampAdd tableApi
Posted by xueyumusic <gi...@git.apache.org>.
Github user xueyumusic commented on a diff in the pull request:
https://github.com/apache/flink/pull/6188#discussion_r197642129
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/expressions/time.scala ---
@@ -328,6 +330,71 @@ case class TemporalOverlaps(
}
}
+ /**
+ * Standard conversion of the TIMESTAMPADD operator.
+ * Source: [[org.apache.calcite.sql2rel.StandardConvertletTable#TimestampAddConvertlet]]
+ */
+case class TimestampAdd(
+ unit: Expression,
+ count: Expression,
+ timestamp: Expression)
+ extends Expression {
+
+ private[flink] final val sqlTsiArray = Array("SQL_TSI_YEAR", "SQL_TSI_QUARTER", "SQL_TSI_MONTH",
+ "SQL_TSI_WEEK", "SQL_TSI_DAY", "SQL_TSI_HOUR", "SQL_TSI_MINUTE", "SQL_TSI_SECOND")
+
+ override private[flink] def children: Seq[Expression] = unit :: count :: timestamp :: Nil
+
+ override private[flink] def validateInput(): ValidationResult = {
+ if (!TypeCheckUtils.isString(unit.resultType)) {
--- End diff --
valid input more strict, and add validation test, thank you
---
[GitHub] flink pull request #6188: [FLINK-6846][Table API] add timestampAdd tableApi
Posted by twalthr <gi...@git.apache.org>.
Github user twalthr commented on a diff in the pull request:
https://github.com/apache/flink/pull/6188#discussion_r199783102
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/scala/expressionDsl.scala ---
@@ -1029,6 +1029,29 @@ object temporalOverlaps {
TemporalOverlaps(leftTimePoint, leftTemporal, rightTimePoint, rightTemporal)
}
}
+/**
+ * Adds a (signed) integer interval to a timestamp. The unit for the interval is given
+ * by the unit argument, which should be one of the following values: "SECOND", "MINUTE",
+ * "HOUR", "DAY", "WEEK", "MONTH", "QUARTER" or "YEAR".
+ *
+ * e.g. timestampAdd("WEEK", 1, '2003-01-02'.toDate) leads to "2003-01-09".
+ */
+object timestampAdd {
+
+ /**
+ * Adds a (signed) integer interval to a timestamp. The unit for the interval is given
+ * by the unit argument, which should be one of the following values: "SECOND", "MINUTE",
+ * "HOUR", "DAY", "WEEK", "MONTH", "QUARTER" or "YEAR".
+ *
+ * e.g. timestampAdd("WEEK", 1, '2003-01-02'.toDate) leads to "2003-01-09".
+ */
+ def apply(
+ unit: Expression,
--- End diff --
I agree, we should use `quarter` for expressing a time and not for a special extraction function that can also be achieved with `.extract(...)`.
---
[GitHub] flink issue #6188: [FLINK-6846][Table API] add timestampAdd tableApi
Posted by hequn8128 <gi...@git.apache.org>.
Github user hequn8128 commented on the issue:
https://github.com/apache/flink/pull/6188
@xueyumusic Great that you step out to contribute to flink, I will take a look this weekend :-)
---
[GitHub] flink pull request #6188: [FLINK-6846][Table API] add timestampAdd tableApi
Posted by walterddr <gi...@git.apache.org>.
Github user walterddr commented on a diff in the pull request:
https://github.com/apache/flink/pull/6188#discussion_r197613995
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/expressions/time.scala ---
@@ -328,6 +330,42 @@ case class TemporalOverlaps(
}
}
+ /**
+ * Standard conversion of the TIMESTAMPADD operator.
+ * Source: [[org.apache.calcite.sql2rel.StandardConvertletTable#TimestampAddConvertlet]]
+ */
+case class TimestampAdd(
+ unit: Expression,
+ count: Expression,
+ timestamp: Expression)
+ extends Expression {
+
+ override private[flink] def children = unit :: count :: timestamp :: Nil
+
+ override private[flink] def toRexNode(implicit relBuilder: RelBuilder) = {
+ var timeUnit : Option[TimeUnit] = None
+ if (unit.isInstanceOf[Literal]) {
+ var unitValue= unit.asInstanceOf[Literal].value.toString()
+ val sqlTsiArray = Array("SQL_TSI_YEAR", "SQL_TSI_QUARTER", "SQL_TSI_MONTH", "SQL_TSI_WEEK",
--- End diff --
I think defining these in a static private field would be more appropriate.
---
[GitHub] flink pull request #6188: [FLINK-6846][Table API] add timestampAdd tableApi
Posted by hequn8128 <gi...@git.apache.org>.
Github user hequn8128 commented on a diff in the pull request:
https://github.com/apache/flink/pull/6188#discussion_r197614200
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/expressions/time.scala ---
@@ -328,6 +330,42 @@ case class TemporalOverlaps(
}
}
+ /**
+ * Standard conversion of the TIMESTAMPADD operator.
+ * Source: [[org.apache.calcite.sql2rel.StandardConvertletTable#TimestampAddConvertlet]]
+ */
+case class TimestampAdd(
+ unit: Expression,
+ count: Expression,
+ timestamp: Expression)
+ extends Expression {
+
+ override private[flink] def children = unit :: count :: timestamp :: Nil
--- End diff --
def children => def children: Seq[Expression]
---
[GitHub] flink issue #6188: [FLINK-6846][Table API] add timestampAdd tableApi
Posted by xueyumusic <gi...@git.apache.org>.
Github user xueyumusic commented on the issue:
https://github.com/apache/flink/pull/6188
Thanks very much for your reivew and guidance, @walterddr @hequn8128 , I have made some modification, please have a review when you are free, thanks!
---
[GitHub] flink pull request #6188: [FLINK-6846][Table API] add timestampAdd tableApi
Posted by hequn8128 <gi...@git.apache.org>.
Github user hequn8128 commented on a diff in the pull request:
https://github.com/apache/flink/pull/6188#discussion_r197640444
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/expressions/time.scala ---
@@ -328,6 +330,71 @@ case class TemporalOverlaps(
}
}
+ /**
+ * Standard conversion of the TIMESTAMPADD operator.
+ * Source: [[org.apache.calcite.sql2rel.StandardConvertletTable#TimestampAddConvertlet]]
+ */
+case class TimestampAdd(
+ unit: Expression,
+ count: Expression,
+ timestamp: Expression)
+ extends Expression {
+
+ private[flink] final val sqlTsiArray = Array("SQL_TSI_YEAR", "SQL_TSI_QUARTER", "SQL_TSI_MONTH",
+ "SQL_TSI_WEEK", "SQL_TSI_DAY", "SQL_TSI_HOUR", "SQL_TSI_MINUTE", "SQL_TSI_SECOND")
+
+ override private[flink] def children: Seq[Expression] = unit :: count :: timestamp :: Nil
+
+ override private[flink] def validateInput(): ValidationResult = {
+ if (!TypeCheckUtils.isString(unit.resultType)) {
+ return ValidationFailure(s"TimestampAdd operator requires unit to be of type " +
+ s"String Literal, but get ${unit.resultType}.")
+ }
+ if (!TypeCheckUtils.isTimePoint(timestamp.resultType)) {
+ return ValidationFailure(s"TimestampAdd operator requires timestamp to be of type " +
+ s"SqlTimeTypeInfo, but get ${timestamp.resultType}.")
+ }
+ ValidationSuccess
+ }
+
+ override private[flink] def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
+ val timeUnit = unit match {
+ case Literal(value: String, STRING_TYPE_INFO) =>
+ if (sqlTsiArray.contains(value)) {
+ Some(TimeUnit.valueOf(value.split("_").last))
+ } else {
+ Some(TimeUnit.valueOf(value))
+ }
+ case _ => None
+ }
+
+ val interval = count match {
+ case Literal(value: Int, INT_TYPE_INFO) =>
+ makeInterval(value.toLong, timeUnit)
+ case Literal(value: Long, LONG_TYPE_INFO) =>
+ makeInterval(value, timeUnit)
+ case _ =>
+ relBuilder.call(SqlStdOperatorTable.MULTIPLY,
+ relBuilder.getRexBuilder.makeIntervalLiteral(timeUnit.get.multiplier,
+ new SqlIntervalQualifier(timeUnit.get, null, SqlParserPos.ZERO)),
+ count.toRexNode)
+ }
+
+ relBuilder.call(SqlStdOperatorTable.DATETIME_PLUS, timestamp.toRexNode, interval)
+ }
+
+ override def toString: String = s"timestampAdd(${children.mkString(", ")})"
+
+ override private[flink] def resultType: TypeInformation[_] = STRING_TYPE_INFO
+
+ private[flink] def makeInterval(value: Long, timeUnit: Option[TimeUnit])
+ (implicit relBuilder: RelBuilder) = {
+ val countWithUnit = timeUnit.get.multiplier.multiply(java.math.BigDecimal.valueOf(value))
+ relBuilder.getRexBuilder.makeIntervalLiteral(countWithUnit,
--- End diff --
indent with two spaces
---
[GitHub] flink issue #6188: [FLINK-6846][Table API] add timestampAdd tableApi
Posted by xueyumusic <gi...@git.apache.org>.
Github user xueyumusic commented on the issue:
https://github.com/apache/flink/pull/6188
@twalthr I looked around and realized that current `+` expression could support TimeInterval (TimePoint) addition, thus the timestampAdd api is duplicated and seems have no necessary.
so I only made these changes:
1. add `1.week` expression
2. modify `quarter` as `1.quarter`
3. modify Extract to support extracting QUARTER and WEEK, as suggested and verified by @walterddr
Please have a review, @twalthr @fhueske @walterddr @hequn8128 thanks for all of you,
---
[GitHub] flink pull request #6188: [FLINK-6846][Table API] add timestampAdd tableApi
Posted by hequn8128 <gi...@git.apache.org>.
Github user hequn8128 commented on a diff in the pull request:
https://github.com/apache/flink/pull/6188#discussion_r197614210
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/expressions/time.scala ---
@@ -328,6 +330,42 @@ case class TemporalOverlaps(
}
}
+ /**
+ * Standard conversion of the TIMESTAMPADD operator.
+ * Source: [[org.apache.calcite.sql2rel.StandardConvertletTable#TimestampAddConvertlet]]
+ */
+case class TimestampAdd(
+ unit: Expression,
+ count: Expression,
+ timestamp: Expression)
+ extends Expression {
+
+ override private[flink] def children = unit :: count :: timestamp :: Nil
+
+ override private[flink] def toRexNode(implicit relBuilder: RelBuilder) = {
+ var timeUnit : Option[TimeUnit] = None
+ if (unit.isInstanceOf[Literal]) {
+ var unitValue= unit.asInstanceOf[Literal].value.toString()
--- End diff --
1. "var unitValue=" => "var unitValue ="
2. indent with two spaces instead of four and same for other lines
---