[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16623446#comment-16623446
]
ASF GitHub Bot commented on FLINK-9713:
---
pnowojski commented on issue #6299: [FLINK-9713][table][sql] Support processing
time versioned joins
URL: https://github.com/apache/flink/pull/6299#issuecomment-423503610
Travis is green - merging
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16623219#comment-16623219
]
ASF GitHub Bot commented on FLINK-9713:
---
pnowojski edited a comment on issue #6299: [FLINK-9713][table][sql] Support
processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#issuecomment-423448194
Thanks for the review @twalthr :) I'm pushing latest update (already
squashed). Will merge it once it's green.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16623218#comment-16623218
]
ASF GitHub Bot commented on FLINK-9713:
---
pnowojski commented on issue #6299: [FLINK-9713][table][sql] Support processing
time versioned joins
URL: https://github.com/apache/flink/pull/6299#issuecomment-423448194
Thanks for the review @twalthr :) I'm pushing latest update. Will merge it
once it's green.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16623217#comment-16623217
]
ASF GitHub Bot commented on FLINK-9713:
---
pnowojski commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r219411672
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/expressions/fieldExpression.scala
##
@@ -191,12 +174,10 @@ case class RowtimeAttribute(expr: Expression) extends
TimeAttribute(expr) {
ValidationSuccess
case WindowReference(_, _) =>
ValidationFailure("Reference to a rowtime or proctime window
required.")
- case TemporalTableReference(_, _) =>
-ValidationSuccess
case any =>
ValidationFailure(
- s"The '.rowtime' expression can only be used for table definitions,
windows, " +
-s" and temporal tables, while [${any}] was found.")
+ s"The '.rowtime' expression can only be used for table definitions,
windows " +
+s"and temporal table definitions, while [$any] was found.")
Review comment:
Yes, you are right :)
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16623216#comment-16623216
]
ASF GitHub Bot commented on FLINK-9713:
---
pnowojski commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r219411313
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/calcite/RelTimeIndicatorConverter.scala
##
@@ -175,13 +173,41 @@ class RelTimeIndicatorConverter(rexBuilder: RexBuilder)
extends RelShuttle {
val rewrittenTemporalJoin = temporalJoin.copy(temporalJoin.getTraitSet,
List(left, right))
-// Materialize all of the time attributes from the right side of temporal
join
-val indicesToMaterialize =
- (left.getRowType.getFieldCount until
rewrittenTemporalJoin.getRowType.getFieldCount).toSet
+val indicesToMaterialize =
gatherIndicesToMaterialize(rewrittenTemporalJoin, left, right)
materializerUtils.projectAndMaterializeFields(rewrittenTemporalJoin,
indicesToMaterialize)
}
+ private def gatherIndicesToMaterialize(
+ temporalJoin: Join,
+ left: RelNode,
+ right: RelNode)
+: Set[Int] = {
+
+// Materialize all of the time attributes from the right side of temporal
join
+var indicesToMaterialize =
+ (left.getRowType.getFieldCount until
temporalJoin.getRowType.getFieldCount).toSet
+
+if (!hasRowtimeAttribute(right.getRowType)) {
Review comment:
No, I forgot. Adding one now.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16623213#comment-16623213
]
ASF GitHub Bot commented on FLINK-9713:
---
pnowojski commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r219410591
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/calcite/RelTimeIndicatorConverter.scala
##
@@ -175,13 +173,41 @@ class RelTimeIndicatorConverter(rexBuilder: RexBuilder)
extends RelShuttle {
val rewrittenTemporalJoin = temporalJoin.copy(temporalJoin.getTraitSet,
List(left, right))
-// Materialize all of the time attributes from the right side of temporal
join
-val indicesToMaterialize =
- (left.getRowType.getFieldCount until
rewrittenTemporalJoin.getRowType.getFieldCount).toSet
+val indicesToMaterialize =
gatherIndicesToMaterialize(rewrittenTemporalJoin, left, right)
materializerUtils.projectAndMaterializeFields(rewrittenTemporalJoin,
indicesToMaterialize)
}
+ private def gatherIndicesToMaterialize(
Review comment:
fixed, but imo in visitor pattern is better to have follow this pattern:
```
public visit(X);
private doSthWith(X);
private doSthElseWith(X);
public visit(Y)
private doSthWith(Y);
private doSthElseWith(Y);
```
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16623198#comment-16623198
]
ASF GitHub Bot commented on FLINK-9713:
---
twalthr commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r219407109
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/calcite/RelTimeIndicatorConverter.scala
##
@@ -175,13 +173,41 @@ class RelTimeIndicatorConverter(rexBuilder: RexBuilder)
extends RelShuttle {
val rewrittenTemporalJoin = temporalJoin.copy(temporalJoin.getTraitSet,
List(left, right))
-// Materialize all of the time attributes from the right side of temporal
join
-val indicesToMaterialize =
- (left.getRowType.getFieldCount until
rewrittenTemporalJoin.getRowType.getFieldCount).toSet
+val indicesToMaterialize =
gatherIndicesToMaterialize(rewrittenTemporalJoin, left, right)
materializerUtils.projectAndMaterializeFields(rewrittenTemporalJoin,
indicesToMaterialize)
}
+ private def gatherIndicesToMaterialize(
+ temporalJoin: Join,
+ left: RelNode,
+ right: RelNode)
+: Set[Int] = {
+
+// Materialize all of the time attributes from the right side of temporal
join
+var indicesToMaterialize =
+ (left.getRowType.getFieldCount until
temporalJoin.getRowType.getFieldCount).toSet
+
+if (!hasRowtimeAttribute(right.getRowType)) {
Review comment:
Did you also add a test for this?
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16623203#comment-16623203
]
ASF GitHub Bot commented on FLINK-9713:
---
twalthr commented on issue #6299: [FLINK-9713][table][sql] Support processing
time versioned joins
URL: https://github.com/apache/flink/pull/6299#issuecomment-423443446
Thank you @pnowojski. I added my last 4 comments. Feel free to merge this PR
after you addressed them. +1 from my side.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16623199#comment-16623199
]
ASF GitHub Bot commented on FLINK-9713:
---
twalthr commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r219407507
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/expressions/fieldExpression.scala
##
@@ -191,12 +174,10 @@ case class RowtimeAttribute(expr: Expression) extends
TimeAttribute(expr) {
ValidationSuccess
case WindowReference(_, _) =>
ValidationFailure("Reference to a rowtime or proctime window
required.")
- case TemporalTableReference(_, _) =>
-ValidationSuccess
case any =>
ValidationFailure(
- s"The '.rowtime' expression can only be used for table definitions,
windows, " +
-s" and temporal tables, while [${any}] was found.")
+ s"The '.rowtime' expression can only be used for table definitions,
windows " +
+s"and temporal table definitions, while [$any] was found.")
Review comment:
I think table definitions are not supported anymore. So we can reset this
message entirely, right?
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16623190#comment-16623190
]
ASF GitHub Bot commented on FLINK-9713:
---
twalthr commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r219405006
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/calcite/RelTimeIndicatorConverter.scala
##
@@ -175,13 +173,41 @@ class RelTimeIndicatorConverter(rexBuilder: RexBuilder)
extends RelShuttle {
val rewrittenTemporalJoin = temporalJoin.copy(temporalJoin.getTraitSet,
List(left, right))
-// Materialize all of the time attributes from the right side of temporal
join
-val indicesToMaterialize =
- (left.getRowType.getFieldCount until
rewrittenTemporalJoin.getRowType.getFieldCount).toSet
+val indicesToMaterialize =
gatherIndicesToMaterialize(rewrittenTemporalJoin, left, right)
materializerUtils.projectAndMaterializeFields(rewrittenTemporalJoin,
indicesToMaterialize)
}
+ private def gatherIndicesToMaterialize(
Review comment:
nit: put private methods below public methods
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16622300#comment-16622300
]
ASF GitHub Bot commented on FLINK-9713:
---
pnowojski commented on issue #6299: [FLINK-9713][table][sql] Support processing
time versioned joins
URL: https://github.com/apache/flink/pull/6299#issuecomment-423247190
Yes, you were right `TemporalTableReference` was not needed after cleaning
up `.rowtime` usages. Applied changes, please check the `fixup` commits :)
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=1664#comment-1664
]
ASF GitHub Bot commented on FLINK-9713:
---
pnowojski commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r219214189
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/calcite/RelTimeIndicatorConverter.scala
##
@@ -166,9 +167,20 @@ class RelTimeIndicatorConverter(rexBuilder: RexBuilder)
extends RelShuttle {
val right = join.getRight.accept(this)
LogicalJoin.create(left, right, join.getCondition, join.getVariablesSet,
join.getJoinType)
-
}
+ def visit(temporalJoin: LogicalTemporalTableJoin): RelNode = {
+val left = temporalJoin.getLeft.accept(this)
+val right = temporalJoin.getRight.accept(this)
+
+val rewrittenTemporalJoin = temporalJoin.copy(temporalJoin.getTraitSet,
List(left, right))
+
+// Materialize all of the time attributes from the right side of temporal
join
+val indicesToMaterialize =
Review comment:
As we discussed, it is probably better to materialise it to make it more
consistent.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=1662#comment-1662
]
ASF GitHub Bot commented on FLINK-9713:
---
pnowojski commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r219213980
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/calcite/RelTimeIndicatorConverter.scala
##
@@ -442,11 +402,63 @@ class RexTimeIndicatorMaterializer(
// materialize function's result and operands
case _ if isTimeIndicatorType(updatedCall.getType) =>
-updatedCall.clone(timestamp, materializedOperands)
+updatedCall.clone(materializerUtils.getTimestamp, materializedOperands)
// materialize function's operands only
case _ =>
updatedCall.clone(updatedCall.getType, materializedOperands)
}
}
}
+
+/**
+ * Helper class for shared logic of materializing time attributes in
[[RelNode]] and [[RexNode]].
+ */
+class RexTimeIndicatorMaterializerUtils(rexBuilder: RexBuilder) {
+
+ private val timestamp = rexBuilder
+.getTypeFactory
+.asInstanceOf[FlinkTypeFactory]
+.createTypeFromTypeInfo(SqlTimeTypeInfo.TIMESTAMP, isNullable = false)
+
+ def getTimestamp: RelDataType = {
+timestamp
+ }
+
+ def projectAndMaterializeFields(input: RelNode, indicesToMaterialize:
Set[Int]) : RelNode = {
+val projects = input.getRowType.getFieldList.map { field =>
+ materializeIfContains(
+new RexInputRef(field.getIndex, field.getType),
+field.getIndex,
+indicesToMaterialize)
+}
+
+LogicalProject.create(
Review comment:
I know, but I think there is no need to complicate the logic here (and in
other similar places), when we can relay on other simply optimisation rules
like merging projections.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16622216#comment-16622216
]
ASF GitHub Bot commented on FLINK-9713:
---
pnowojski commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r219212679
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/table.scala
##
@@ -158,36 +158,60 @@ class Table(
/**
* Creates [[TemporalTableFunction]] backed up by this table as a history
table.
-*
-* @param timeIndicator field for the [[TemporalTableFunction]]. Must
points to a time indicator
-* @param primaryKeyfield for the [[TemporalTableFunction]].
-* @return [[TemporalTableFunction]]
+* Temporal Tables represent a concept of a table that changes over time
and for which
+* Flink keeps track of those changes. [[TemporalTableFunction]] provides a
way how to access
Review comment:
I would keep `access` because we have in the backlog story that would enable
to write following query:
```
SELECT * FROM Rates('2017/06/14 14:23:12`);
```
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16622219#comment-16622219
]
ASF GitHub Bot commented on FLINK-9713:
---
pnowojski commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r219212679
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/table.scala
##
@@ -158,36 +158,60 @@ class Table(
/**
* Creates [[TemporalTableFunction]] backed up by this table as a history
table.
-*
-* @param timeIndicator field for the [[TemporalTableFunction]]. Must
points to a time indicator
-* @param primaryKeyfield for the [[TemporalTableFunction]].
-* @return [[TemporalTableFunction]]
+* Temporal Tables represent a concept of a table that changes over time
and for which
+* Flink keeps track of those changes. [[TemporalTableFunction]] provides a
way how to access
Review comment:
I would keep `access` because we have in the backlog story that would enable
to write following query:
```
SELECT * FROM Rates('2017/06/14 14:23:12`);
```
Otherwise it would almost for sure drift out of sync :)
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16622103#comment-16622103
]
ASF GitHub Bot commented on FLINK-9713:
---
twalthr commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r219174337
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/calcite/RelTimeIndicatorConverter.scala
##
@@ -442,11 +402,63 @@ class RexTimeIndicatorMaterializer(
// materialize function's result and operands
case _ if isTimeIndicatorType(updatedCall.getType) =>
-updatedCall.clone(timestamp, materializedOperands)
+updatedCall.clone(materializerUtils.getTimestamp, materializedOperands)
// materialize function's operands only
case _ =>
updatedCall.clone(updatedCall.getType, materializedOperands)
}
}
}
+
+/**
+ * Helper class for shared logic of materializing time attributes in
[[RelNode]] and [[RexNode]].
+ */
+class RexTimeIndicatorMaterializerUtils(rexBuilder: RexBuilder) {
+
+ private val timestamp = rexBuilder
+.getTypeFactory
+.asInstanceOf[FlinkTypeFactory]
+.createTypeFromTypeInfo(SqlTimeTypeInfo.TIMESTAMP, isNullable = false)
+
+ def getTimestamp: RelDataType = {
+timestamp
+ }
+
+ def projectAndMaterializeFields(input: RelNode, indicesToMaterialize:
Set[Int]) : RelNode = {
+val projects = input.getRowType.getFieldList.map { field =>
+ materializeIfContains(
+new RexInputRef(field.getIndex, field.getType),
+field.getIndex,
+indicesToMaterialize)
+}
+
+LogicalProject.create(
Review comment:
Do we always want to add a project or should we check first if the project
is necessary? We are doing this in `convertAggregate`.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16622102#comment-16622102
]
ASF GitHub Bot commented on FLINK-9713:
---
twalthr commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r219175922
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/calcite/RelTimeIndicatorConverter.scala
##
@@ -166,9 +167,20 @@ class RelTimeIndicatorConverter(rexBuilder: RexBuilder)
extends RelShuttle {
val right = join.getRight.accept(this)
LogicalJoin.create(left, right, join.getCondition, join.getVariablesSet,
join.getJoinType)
-
}
+ def visit(temporalJoin: LogicalTemporalTableJoin): RelNode = {
+val left = temporalJoin.getLeft.accept(this)
+val right = temporalJoin.getRight.accept(this)
+
+val rewrittenTemporalJoin = temporalJoin.copy(temporalJoin.getTraitSet,
List(left, right))
+
+// Materialize all of the time attributes from the right side of temporal
join
+val indicesToMaterialize =
Review comment:
The processing time of the right side must not be materialized.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16622072#comment-16622072
]
ASF GitHub Bot commented on FLINK-9713:
---
twalthr commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r219165974
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/table.scala
##
@@ -158,36 +158,60 @@ class Table(
/**
* Creates [[TemporalTableFunction]] backed up by this table as a history
table.
-*
-* @param timeIndicator field for the [[TemporalTableFunction]]. Must
points to a time indicator
-* @param primaryKeyfield for the [[TemporalTableFunction]].
-* @return [[TemporalTableFunction]]
+* Temporal Tables represent a concept of a table that changes over time
and for which
+* Flink keeps track of those changes. [[TemporalTableFunction]] provides a
way how to access
+* those data.
+*
+* For more information please check Flink's documentation on Temporal
Tables.
+*
+* Currently [[TemporalTableFunction]]s are only supported in streaming.
+*
+* @param timeAttribute Must points to a time indicator. Provides a way to
compare which records
Review comment:
replace "indicator"
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16622073#comment-16622073
]
ASF GitHub Bot commented on FLINK-9713:
---
twalthr commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r219167509
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/expressions/fieldExpression.scala
##
@@ -195,8 +195,9 @@ case class RowtimeAttribute(expr: Expression) extends
TimeAttribute(expr) {
ValidationSuccess
case any =>
ValidationFailure(
- s"The '.rowtime' expression can only be used for table definitions,
windows, " +
-s" and temporal tables, while [${any}] was found.")
+ s"The '.rowtime' expression can only be used for table definitions,
windows " +
Review comment:
Reset the message to the original one?
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16622071#comment-16622071
]
ASF GitHub Bot commented on FLINK-9713:
---
twalthr commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r219165539
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/table.scala
##
@@ -158,36 +158,60 @@ class Table(
/**
* Creates [[TemporalTableFunction]] backed up by this table as a history
table.
-*
-* @param timeIndicator field for the [[TemporalTableFunction]]. Must
points to a time indicator
-* @param primaryKeyfield for the [[TemporalTableFunction]].
-* @return [[TemporalTableFunction]]
+* Temporal Tables represent a concept of a table that changes over time
and for which
+* Flink keeps track of those changes. [[TemporalTableFunction]] provides a
way how to access
Review comment:
"join" instead of "access"? Because we can really access the data in other
operations.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16621673#comment-16621673
]
ASF GitHub Bot commented on FLINK-9713:
---
twalthr commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r219080353
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/table.scala
##
@@ -155,6 +156,50 @@ class Table(
select(withResolvedAggFunctionCall: _*)
}
+ /**
+* Creates [[TemporalTableFunction]] backed up by this table as a history
table.
+*
+* @param timeIndicator field for the [[TemporalTableFunction]]. Must
points to a time indicator
+* @param primaryKeyfield for the [[TemporalTableFunction]].
+* @return [[TemporalTableFunction]]
+*/
+ def createTemporalTableFunction(
+ timeIndicator: String,
+ primaryKey: String): TemporalTableFunction = {
+createTemporalTableFunction(
+ ExpressionParser.parseExpression(timeIndicator),
+ ExpressionParser.parseExpression(primaryKey))
+ }
+
+ /**
+* Creates [[TemporalTableFunction]] backed up by this table as a history
table.
+*
+* @param timeIndicator field for the [[TemporalTableFunction]]. Must
points to a time indicator
+* @param primaryKeyfield for the [[TemporalTableFunction]].
+* @return [[TemporalTableFunction]]
+*/
+ def createTemporalTableFunction(
+ timeIndicator: Expression,
+ primaryKey: Expression): TemporalTableFunction = {
+val temporalTable = new TemporalTable(timeIndicator, primaryKey,
logicalPlan)
+ .validate(tableEnv)
+ .asInstanceOf[TemporalTable]
+
+TemporalTableFunction.create(
+ this,
+ temporalTable.timeIndicator,
+ validatePrimaryKeyExpression(temporalTable.primaryKey))
+ }
+
+ private def validatePrimaryKeyExpression(expression: Expression): String = {
+expression match {
+ case fieldReference: ResolvedFieldReference =>
+fieldReference.name
+ case _ => throw new ValidationException(
+s"Unsupported expression [${expression}] as primary key")
Review comment:
I would still add `top-level`. Because nested field references are not
supported.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16621668#comment-16621668
]
ASF GitHub Bot commented on FLINK-9713:
---
twalthr commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r219078553
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/BatchTableEnvironment.scala
##
@@ -448,48 +448,13 @@ abstract class BatchTableEnvironment(
* @return The optimized [[RelNode]] tree
*/
private[flink] def optimize(relNode: RelNode): RelNode = {
-
-// 0. convert sub-queries before query decorrelation
-val convSubQueryPlan = runHepPlanner(
- HepMatchOrder.BOTTOM_UP, FlinkRuleSets.TABLE_SUBQUERY_RULES, relNode,
relNode.getTraitSet)
-
-// 0. convert table references
-val fullRelNode = runHepPlanner(
- HepMatchOrder.BOTTOM_UP,
- FlinkRuleSets.TABLE_REF_RULES,
- convSubQueryPlan,
- relNode.getTraitSet)
-
-// 1. decorrelate
+val convSubQueryPlan = optimizeConvertSubQueries(relNode)
Review comment:
Yes, that would be the general solution to this problem. I'm still waiting
for prioritization to finally implement that.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16619331#comment-16619331
]
ASF GitHub Bot commented on FLINK-9713:
---
pnowojski commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r218414712
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/StreamTableEnvironment.scala
##
@@ -803,70 +803,37 @@ abstract class StreamTableEnvironment(
* @return The optimized [[RelNode]] tree
*/
private[flink] def optimize(relNode: RelNode, updatesAsRetraction: Boolean):
RelNode = {
-
-// 0. convert sub-queries before query decorrelation
-val convSubQueryPlan = runHepPlanner(
- HepMatchOrder.BOTTOM_UP, FlinkRuleSets.TABLE_SUBQUERY_RULES, relNode,
relNode.getTraitSet)
-
-// 0. convert table references
-val fullRelNode = runHepPlanner(
- HepMatchOrder.BOTTOM_UP,
- FlinkRuleSets.TABLE_REF_RULES,
- convSubQueryPlan,
- relNode.getTraitSet)
-
-// 1. decorrelate
+val convSubQueryPlan = optimizeConvertSubQueries(relNode)
+val temporalTableJoinPlan = optimizeConvertToTemporalJoin(convSubQueryPlan)
+val fullRelNode = optimizeConvertTableReferences(temporalTableJoinPlan)
val decorPlan = RelDecorrelator.decorrelateQuery(fullRelNode)
-
-// 2. convert time indicators
val convPlan = RelTimeIndicatorConverter.convert(decorPlan,
getRelBuilder.getRexBuilder)
Review comment:
Any suggestions maybe? I renamed it to `planWithMaterializedTimeAttributes`
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16619314#comment-16619314
]
ASF GitHub Bot commented on FLINK-9713:
---
pnowojski commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r218462800
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamTemporalJoinToCoProcessTranslator.scala
##
@@ -0,0 +1,198 @@
+/*
+ * 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.table.plan.nodes.datastream
+
+import org.apache.calcite.rel.core.{JoinInfo, JoinRelType}
+import org.apache.calcite.rex._
+import org.apache.flink.api.common.functions.{FlatJoinFunction, MapFunction}
+import org.apache.flink.api.common.typeinfo.TypeInformation
+import org.apache.flink.streaming.api.functions.co.CoProcessFunction
+import org.apache.flink.table.api.{StreamQueryConfig, TableConfig,
TableException, ValidationException}
+import org.apache.flink.table.calcite.FlinkTypeFactory._
+import org.apache.flink.table.codegen.{FunctionCodeGenerator,
GeneratedFunction}
+import org.apache.flink.table.plan.logical.rel.LogicalTemporalTableJoin._
+import org.apache.flink.table.plan.schema.RowSchema
+import org.apache.flink.table.plan.util.RexDefaultVisitor
+import org.apache.flink.table.runtime.join.TemporalJoin
+import org.apache.flink.table.runtime.types.{CRow, CRowTypeInfo}
+import org.apache.flink.types.Row
+import org.apache.flink.util.Preconditions.checkState
+
+class DataStreamTemporalJoinToCoProcessTranslator(
+textualRepresentation: String,
+config: TableConfig,
+returnType: TypeInformation[Row],
+leftSchema: RowSchema,
+rightSchema: RowSchema,
+joinInfo: JoinInfo,
+rexBuilder: RexBuilder)
+ extends DataStreamJoinToCoProcessTranslator(
+config,
+returnType,
+leftSchema,
+rightSchema,
+joinInfo,
+rexBuilder) {
+
+ var leftTimeAttribute: Option[RexNode] = None
+
+ var rightTimeAttribute: Option[RexNode] = None
+
+ var rightPrimaryKeyExpression: Option[RexNode] = None
+
+ override val nonEquiJoinPredicates: Option[RexNode] =
extractTemporalJoinCondition()
+
+ def extractTemporalJoinCondition(): Option[RexNode] = {
+checkState(
+ !joinInfo.isEqui,
+ "Missing %s in join condition",
+ TEMPORAL_JOIN_CONDITION)
+
+val nonEquiJoinRex: RexNode = joinInfo.getRemaining(rexBuilder)
+val temporalJoinConditionExtractor = new TemporalJoinConditionExtractor(
+ nonEquiJoinRex.toString)
+
+Some(temporalJoinConditionExtractor.apply(nonEquiJoinRex))
+ }
+
+ override protected def createCoProcessFunction(
+ joinType: JoinRelType,
+ queryConfig: StreamQueryConfig,
+ joinFunction: GeneratedFunction[FlatJoinFunction[Row, Row, Row], Row])
+: CoProcessFunction[CRow, CRow, CRow] = {
+
+checkState(
+ leftTimeAttribute.isDefined &&
+rightPrimaryKeyExpression.isDefined,
+ "Missing %s in join condition",
+ TEMPORAL_JOIN_CONDITION)
+
+if (rightTimeAttribute.isDefined) {
Review comment:
If we add an additional flag, we just duplicate the same logic, since we
would still need to handle this `Optional` field:
```
if (isRowtimeMode) {
throw new ValidationException(
s"Currently only proctime temporal joins are supported in
[${textualRepresentation}]")
}
checkState(!rightTimeAttribute.isDefined);
```
Alternative solution (IMO not worth the effort here) would be to split this
class and have one for processing time and one for rowtime. Then we could skip
this check altogether.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
>
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16619328#comment-16619328
]
ASF GitHub Bot commented on FLINK-9713:
---
pnowojski commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r218414712
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/StreamTableEnvironment.scala
##
@@ -803,70 +803,37 @@ abstract class StreamTableEnvironment(
* @return The optimized [[RelNode]] tree
*/
private[flink] def optimize(relNode: RelNode, updatesAsRetraction: Boolean):
RelNode = {
-
-// 0. convert sub-queries before query decorrelation
-val convSubQueryPlan = runHepPlanner(
- HepMatchOrder.BOTTOM_UP, FlinkRuleSets.TABLE_SUBQUERY_RULES, relNode,
relNode.getTraitSet)
-
-// 0. convert table references
-val fullRelNode = runHepPlanner(
- HepMatchOrder.BOTTOM_UP,
- FlinkRuleSets.TABLE_REF_RULES,
- convSubQueryPlan,
- relNode.getTraitSet)
-
-// 1. decorrelate
+val convSubQueryPlan = optimizeConvertSubQueries(relNode)
+val temporalTableJoinPlan = optimizeConvertToTemporalJoin(convSubQueryPlan)
+val fullRelNode = optimizeConvertTableReferences(temporalTableJoinPlan)
val decorPlan = RelDecorrelator.decorrelateQuery(fullRelNode)
-
-// 2. convert time indicators
val convPlan = RelTimeIndicatorConverter.convert(decorPlan,
getRelBuilder.getRexBuilder)
Review comment:
Any suggestions maybe?
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16619321#comment-16619321
]
ASF GitHub Bot commented on FLINK-9713:
---
pnowojski commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r218422026
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/logical/rel/LogicalTemporalTableJoin.scala
##
@@ -0,0 +1,173 @@
+/*
+ * 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.table.plan.logical.rel
+
+import java.util.Collections
+
+import org.apache.calcite.plan.{RelOptCluster, RelTraitSet}
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.core._
+import org.apache.calcite.rex.{RexBuilder, RexNode}
+import org.apache.calcite.sql.`type`.{OperandTypes, ReturnTypes}
+import org.apache.calcite.sql.{SqlFunction, SqlFunctionCategory, SqlKind}
+import org.apache.flink.util.Preconditions.checkArgument
+
+/**
+ * Represents a join between a table and
[[org.apache.flink.table.functions.TemporalTableFunction]]
+ *
+ * @param cluster
+ * @param traitSet
+ * @param left stream
+ * @param right table scan of underlying
+ * [[org.apache.flink.table.functions.TemporalTableFunction]]
+ * @param condition must contain
[[LogicalTemporalTableJoin#TEMPORAL_JOIN_CONDITION]] with
+ * correctly defined references to rightTimeAttribute,
+ * rightPrimaryKeyExpression and leftTimeAttribute. We can
not implement
+ * those references as separate fields, because of problems
with Calcite's
+ * optimization rules like projections push downs, column
+ * pruning/renaming/reordering, etc. Later
rightTimeAttribute,
+ * rightPrimaryKeyExpression and leftTimeAttribute will be
extracted from
+ * the condition.
+ */
+class LogicalTemporalTableJoin private(
+cluster: RelOptCluster,
+traitSet: RelTraitSet,
+left: RelNode,
+right: RelNode,
+condition: RexNode)
+ extends Join(
+cluster,
+traitSet,
+left,
+right,
+condition,
+Collections.emptySet().asInstanceOf[java.util.Set[CorrelationId]],
+JoinRelType.INNER) {
+
+ override def copy(
+ traitSet: RelTraitSet,
+ condition: RexNode,
+ left: RelNode,
+ right: RelNode,
+ joinType: JoinRelType,
+ semiJoinDone: Boolean): LogicalTemporalTableJoin = {
+checkArgument(joinType == this.getJoinType,
+ "Can not change join type".asInstanceOf[Object])
+checkArgument(semiJoinDone == this.isSemiJoinDone,
+ "Can not change semiJoinDone".asInstanceOf[Object])
+new LogicalTemporalTableJoin(
+ cluster,
+ traitSet,
+ left,
+ right,
+ condition)
+ }
+}
+
+object LogicalTemporalTableJoin {
+ /**
+* See [[LogicalTemporalTableJoin#condition]]
+*/
+ val TEMPORAL_JOIN_CONDITION = new SqlFunction(
+"__TEMPORAL_JOIN_CONDITION",
+SqlKind.OTHER_FUNCTION,
+ReturnTypes.BOOLEAN_NOT_NULL,
+null,
+OperandTypes.or(
+ OperandTypes.sequence(
+"'(LEFT_TIME_ATTRIBUTE, RIGHT_TIME_ATTRIBUTE, PRIMARY_KEY)'",
+OperandTypes.DATETIME,
+OperandTypes.DATETIME,
+OperandTypes.ANY),
+ OperandTypes.sequence(
+"'(LEFT_TIME_ATTRIBUTE, PRIMARY_KEY)'",
+OperandTypes.DATETIME,
+OperandTypes.ANY)),
+SqlFunctionCategory.SYSTEM)
+
+ def makeRowTimeTemporalJoinConditionCall(
+ rexBuilder: RexBuilder,
+ leftTimeAttribute: RexNode,
+ rightTimeAttribute: RexNode,
+ rightPrimaryKeyExpression: RexNode): RexNode = {
+rexBuilder.makeCall(
+ TEMPORAL_JOIN_CONDITION,
+ leftTimeAttribute,
+ rightTimeAttribute,
+ rightPrimaryKeyExpression)
+ }
+
+ def makeProcTimeTemporalJoinConditionCall(
+ rexBuilder: RexBuilder,
+ leftTimeAttribute: RexNode,
+ rightPrimaryKeyExpression: RexNode): RexNode = {
+rexBuilder.makeCall(
+
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16619320#comment-16619320
]
ASF GitHub Bot commented on FLINK-9713:
---
pnowojski commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r218414391
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/BatchTableEnvironment.scala
##
@@ -448,48 +448,13 @@ abstract class BatchTableEnvironment(
* @return The optimized [[RelNode]] tree
*/
private[flink] def optimize(relNode: RelNode): RelNode = {
-
-// 0. convert sub-queries before query decorrelation
-val convSubQueryPlan = runHepPlanner(
- HepMatchOrder.BOTTOM_UP, FlinkRuleSets.TABLE_SUBQUERY_RULES, relNode,
relNode.getTraitSet)
-
-// 0. convert table references
-val fullRelNode = runHepPlanner(
- HepMatchOrder.BOTTOM_UP,
- FlinkRuleSets.TABLE_REF_RULES,
- convSubQueryPlan,
- relNode.getTraitSet)
-
-// 1. decorrelate
+val convSubQueryPlan = optimizeConvertSubQueries(relNode)
Review comment:
Isn't the general solution to define `TableEnvironment` interface in java
and only expanding it in Scala? Moving those methods to static utility class
doesn't seems as a proper solution. I think it would deserve a separate ticket.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16619318#comment-16619318
]
ASF GitHub Bot commented on FLINK-9713:
---
pnowojski commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r218417820
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/calcite/RelTimeIndicatorConverter.scala
##
@@ -166,9 +172,14 @@ class RelTimeIndicatorConverter(rexBuilder: RexBuilder)
extends RelShuttle {
val right = join.getRight.accept(this)
LogicalJoin.create(left, right, join.getCondition, join.getVariablesSet,
join.getJoinType)
-
}
+ def visit(temporalJoin: LogicalTemporalTableJoin): RelNode = {
+val left = temporalJoin.getLeft.accept(this)
+val right = temporalJoin.getRight.accept(this)
+
+temporalJoin.copy(temporalJoin.getTraitSet, List(left, right))
Review comment:
Why? The implementation looks like it's working and logically the output has
still a valid processing time attribute.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16619324#comment-16619324
]
ASF GitHub Bot commented on FLINK-9713:
---
pnowojski commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r218442168
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/logical/LogicalCorrelateToTemporalTableJoinRule.scala
##
@@ -0,0 +1,208 @@
+/*
+ * 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.table.plan.rules.logical
+
+import org.apache.calcite.plan.RelOptRule.{any, none, operand, some}
+import org.apache.calcite.plan.{RelOptRule, RelOptRuleCall}
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.core.TableFunctionScan
+import org.apache.calcite.rel.logical.LogicalCorrelate
+import org.apache.calcite.rex._
+import org.apache.flink.table.api.{Table, Types, ValidationException}
+import org.apache.flink.table.calcite.FlinkTypeFactory.isProctimeIndicatorType
+import org.apache.flink.table.expressions._
+import org.apache.flink.table.functions.TemporalTableFunction
+import org.apache.flink.table.functions.utils.TableSqlFunction
+import org.apache.flink.table.plan.logical.rel.LogicalTemporalTableJoin
+import org.apache.flink.table.plan.util.RexDefaultVisitor
+
+class LogicalCorrelateToTemporalTableJoinRule
+ extends RelOptRule(
+operand(classOf[LogicalCorrelate],
+ some(
+operand(classOf[RelNode], any()),
+operand(classOf[TableFunctionScan], none(,
+"LogicalCorrelateToTemporalTableJoinRule") {
+
+ def extractNameFromTimeAttribute(timeAttribute: Expression): String = {
+timeAttribute match {
+ case RowtimeAttribute(expr: TemporalTableReference) =>
+expr.name
+ case ProctimeAttribute(expr: TemporalTableReference) =>
+expr.name
+ case TemporalTableReference(name, _) =>
+name
+ case ResolvedFieldReference(name, _)
+if timeAttribute.resultType == Types.LONG ||
+ timeAttribute.resultType == Types.SQL_TIMESTAMP =>
+name
+ case _ => throw new ValidationException(
+s"Invalid timeAttribute [${timeAttribute}] in TemporalTableFunction")
+}
+ }
+
+ override def onMatch(call: RelOptRuleCall): Unit = {
+val logicalCorrelate: LogicalCorrelate = call.rel(0)
+val leftNode: RelNode = call.rel(1)
+val rightTableFunctionScan: TableFunctionScan = call.rel(2)
+
+val cluster = logicalCorrelate.getCluster
+
+new GetTemporalTableFunctionCall(cluster.getRexBuilder, leftNode)
+ .visit(rightTableFunctionScan.getCall) match {
+ case Some(TemporalTableFunctionCall(rightTemporalTableFunction,
leftTimeAttribute)) =>
+val underlyingHistoryTable: Table =
rightTemporalTableFunction.getUnderlyingHistoryTable
+val relBuilder = this.relBuilderFactory.create(
+ cluster,
+ underlyingHistoryTable.relBuilder.getRelOptSchema)
+val rexBuilder = cluster.getRexBuilder
+
+val rightNode: RelNode =
underlyingHistoryTable.logicalPlan.toRelNode(relBuilder)
+
+val rightTimeIndicatorExpression = createRightExpression(
+ rexBuilder,
+ leftNode,
+ rightNode,
+
extractNameFromTimeAttribute(rightTemporalTableFunction.getTimeAttribute))
+
+val rightPrimaryKeyExpression = createRightExpression(
+ rexBuilder,
+ leftNode,
+ rightNode,
+ rightTemporalTableFunction.getPrimaryKey)
+
+relBuilder.push(
+ if
(isProctimeIndicatorType(rightTemporalTableFunction.getTimeAttribute.resultType))
{
+LogicalTemporalTableJoin.create(
+ rexBuilder,
+ cluster,
+ logicalCorrelate.getTraitSet,
+ leftNode,
+ rightNode,
+ leftTimeAttribute,
+ rightPrimaryKeyExpression)
+ }
+ else {
+LogicalTemporalTableJoin.create(
+ rexBuilder,
+ cluster,
+
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16619327#comment-16619327
]
ASF GitHub Bot commented on FLINK-9713:
---
pnowojski commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r218420897
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/calcite/RelTimeIndicatorConverter.scala
##
@@ -400,6 +411,10 @@ class RexTimeIndicatorMaterializer(
// materialize operands with time indicators
val materializedOperands = updatedCall.getOperator match {
+ case tableFunction: TableSqlFunction
Review comment:
Good catch. This must have turned into a dead code. Removed.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16619319#comment-16619319
]
ASF GitHub Bot commented on FLINK-9713:
---
pnowojski commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r218478087
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamTemporalJoinToCoProcessTranslator.scala
##
@@ -0,0 +1,198 @@
+/*
+ * 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.table.plan.nodes.datastream
+
+import org.apache.calcite.rel.core.{JoinInfo, JoinRelType}
+import org.apache.calcite.rex._
+import org.apache.flink.api.common.functions.{FlatJoinFunction, MapFunction}
+import org.apache.flink.api.common.typeinfo.TypeInformation
+import org.apache.flink.streaming.api.functions.co.CoProcessFunction
+import org.apache.flink.table.api.{StreamQueryConfig, TableConfig,
TableException, ValidationException}
+import org.apache.flink.table.calcite.FlinkTypeFactory._
+import org.apache.flink.table.codegen.{FunctionCodeGenerator,
GeneratedFunction}
+import org.apache.flink.table.plan.logical.rel.LogicalTemporalTableJoin._
+import org.apache.flink.table.plan.schema.RowSchema
+import org.apache.flink.table.plan.util.RexDefaultVisitor
+import org.apache.flink.table.runtime.join.TemporalJoin
+import org.apache.flink.table.runtime.types.{CRow, CRowTypeInfo}
+import org.apache.flink.types.Row
+import org.apache.flink.util.Preconditions.checkState
+
+class DataStreamTemporalJoinToCoProcessTranslator(
+textualRepresentation: String,
+config: TableConfig,
+returnType: TypeInformation[Row],
+leftSchema: RowSchema,
+rightSchema: RowSchema,
+joinInfo: JoinInfo,
+rexBuilder: RexBuilder)
+ extends DataStreamJoinToCoProcessTranslator(
+config,
+returnType,
+leftSchema,
+rightSchema,
+joinInfo,
+rexBuilder) {
+
+ var leftTimeAttribute: Option[RexNode] = None
+
+ var rightTimeAttribute: Option[RexNode] = None
+
+ var rightPrimaryKeyExpression: Option[RexNode] = None
+
+ override val nonEquiJoinPredicates: Option[RexNode] =
extractTemporalJoinCondition()
+
+ def extractTemporalJoinCondition(): Option[RexNode] = {
+checkState(
+ !joinInfo.isEqui,
+ "Missing %s in join condition",
+ TEMPORAL_JOIN_CONDITION)
+
+val nonEquiJoinRex: RexNode = joinInfo.getRemaining(rexBuilder)
+val temporalJoinConditionExtractor = new TemporalJoinConditionExtractor(
+ nonEquiJoinRex.toString)
+
+Some(temporalJoinConditionExtractor.apply(nonEquiJoinRex))
+ }
+
+ override protected def createCoProcessFunction(
+ joinType: JoinRelType,
+ queryConfig: StreamQueryConfig,
+ joinFunction: GeneratedFunction[FlatJoinFunction[Row, Row, Row], Row])
+: CoProcessFunction[CRow, CRow, CRow] = {
+
+checkState(
+ leftTimeAttribute.isDefined &&
+rightPrimaryKeyExpression.isDefined,
+ "Missing %s in join condition",
+ TEMPORAL_JOIN_CONDITION)
+
+if (rightTimeAttribute.isDefined) {
+ throw new ValidationException(
+s"Currently only proctime temporal joins are supported in
[${textualRepresentation}]")
+}
+
+joinType match {
+ case JoinRelType.INNER =>
+new TemporalJoin(
+ leftSchema.typeInfo,
+ rightSchema.typeInfo,
+ joinFunction.name,
+ joinFunction.code,
+ queryConfig)
+ case _ =>
+ throw new ValidationException(
+ s"Only ${JoinRelType.INNER} temporal join is supported in
[${textualRepresentation}]")
+}
+ }
+
+ private class TemporalJoinConditionExtractor(
+ nonEquiJoinCondition: String)
+extends RexShuttle {
+
+override def visitCall(call: RexCall): RexNode = {
+ if (call.getOperator != TEMPORAL_JOIN_CONDITION) {
+return super.visitCall(call)
+ }
+
+ if (leftTimeAttribute.isDefined
+|| rightPrimaryKeyExpression.isDefined
+|| rightTimeAttribute.isDefined) {
+throw new
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16619323#comment-16619323
]
ASF GitHub Bot commented on FLINK-9713:
---
pnowojski commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r218369289
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/TemporalTableFunction.scala
##
@@ -0,0 +1,75 @@
+/*
+ * 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.table.functions
+
+import java.sql.Timestamp
+
+import org.apache.flink.api.java.typeutils.RowTypeInfo
+import org.apache.flink.table.api.Table
+import org.apache.flink.table.expressions.Expression
+import org.apache.flink.types.Row
+
+/**
+ * Class representing temporal table function over some history table.
+ */
+class TemporalTableFunction private(
+@transient private val underlyingHistoryTable: Table,
+private val timeAttribute: Expression,
+private val primaryKey: String,
+private val resultType: RowTypeInfo)
+ extends TableFunction[Row] {
+
+ def eval(row: Timestamp): Unit = {
Review comment:
I don't think that's a good idea. That would brake the abstraction, and
would require to adjust/refactor more code, while this is the special case.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16619322#comment-16619322
]
ASF GitHub Bot commented on FLINK-9713:
---
pnowojski commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r218439267
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/logical/LogicalCorrelateToTemporalTableJoinRule.scala
##
@@ -0,0 +1,208 @@
+/*
+ * 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.table.plan.rules.logical
+
+import org.apache.calcite.plan.RelOptRule.{any, none, operand, some}
+import org.apache.calcite.plan.{RelOptRule, RelOptRuleCall}
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.core.TableFunctionScan
+import org.apache.calcite.rel.logical.LogicalCorrelate
+import org.apache.calcite.rex._
+import org.apache.flink.table.api.{Table, Types, ValidationException}
+import org.apache.flink.table.calcite.FlinkTypeFactory.isProctimeIndicatorType
+import org.apache.flink.table.expressions._
+import org.apache.flink.table.functions.TemporalTableFunction
+import org.apache.flink.table.functions.utils.TableSqlFunction
+import org.apache.flink.table.plan.logical.rel.LogicalTemporalTableJoin
+import org.apache.flink.table.plan.util.RexDefaultVisitor
+
+class LogicalCorrelateToTemporalTableJoinRule
+ extends RelOptRule(
+operand(classOf[LogicalCorrelate],
+ some(
+operand(classOf[RelNode], any()),
+operand(classOf[TableFunctionScan], none(,
+"LogicalCorrelateToTemporalTableJoinRule") {
+
+ def extractNameFromTimeAttribute(timeAttribute: Expression): String = {
+timeAttribute match {
+ case RowtimeAttribute(expr: TemporalTableReference) =>
+expr.name
+ case ProctimeAttribute(expr: TemporalTableReference) =>
+expr.name
+ case TemporalTableReference(name, _) =>
+name
+ case ResolvedFieldReference(name, _)
+if timeAttribute.resultType == Types.LONG ||
+ timeAttribute.resultType == Types.SQL_TIMESTAMP =>
+name
+ case _ => throw new ValidationException(
+s"Invalid timeAttribute [${timeAttribute}] in TemporalTableFunction")
+}
+ }
+
+ override def onMatch(call: RelOptRuleCall): Unit = {
+val logicalCorrelate: LogicalCorrelate = call.rel(0)
+val leftNode: RelNode = call.rel(1)
+val rightTableFunctionScan: TableFunctionScan = call.rel(2)
+
+val cluster = logicalCorrelate.getCluster
+
+new GetTemporalTableFunctionCall(cluster.getRexBuilder, leftNode)
+ .visit(rightTableFunctionScan.getCall) match {
+ case Some(TemporalTableFunctionCall(rightTemporalTableFunction,
leftTimeAttribute)) =>
+val underlyingHistoryTable: Table =
rightTemporalTableFunction.getUnderlyingHistoryTable
+val relBuilder = this.relBuilderFactory.create(
+ cluster,
+ underlyingHistoryTable.relBuilder.getRelOptSchema)
+val rexBuilder = cluster.getRexBuilder
+
+val rightNode: RelNode =
underlyingHistoryTable.logicalPlan.toRelNode(relBuilder)
+
+val rightTimeIndicatorExpression = createRightExpression(
+ rexBuilder,
+ leftNode,
+ rightNode,
+
extractNameFromTimeAttribute(rightTemporalTableFunction.getTimeAttribute))
+
+val rightPrimaryKeyExpression = createRightExpression(
+ rexBuilder,
+ leftNode,
+ rightNode,
+ rightTemporalTableFunction.getPrimaryKey)
+
+relBuilder.push(
+ if
(isProctimeIndicatorType(rightTemporalTableFunction.getTimeAttribute.resultType))
{
+LogicalTemporalTableJoin.create(
+ rexBuilder,
+ cluster,
+ logicalCorrelate.getTraitSet,
+ leftNode,
+ rightNode,
+ leftTimeAttribute,
+ rightPrimaryKeyExpression)
+ }
+ else {
+LogicalTemporalTableJoin.create(
+ rexBuilder,
+ cluster,
+
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16619316#comment-16619316
]
ASF GitHub Bot commented on FLINK-9713:
---
pnowojski commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r218479232
##
File path:
flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/table/validation/TemporalTableJoinValidationTest.scala
##
@@ -0,0 +1,124 @@
+/*
+ * 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.table.api.stream.table.validation
+
+import java.sql.Timestamp
+
+import org.apache.flink.api.scala._
+import org.apache.flink.table.api.{TableException, ValidationException}
+import org.apache.flink.table.api.scala._
+import org.apache.flink.table.utils._
+import org.junit.Test
+
+class TemporalTableJoinValidationTest extends TableTestBase {
+
+ val util: TableTestUtil = streamTestUtil()
+
+ val orders = util.addTable[(Long, String, Timestamp)](
+"Orders", 'o_amount, 'o_currency, 'o_rowtime.rowtime)
+
+ val ordersProctime = util.addTable[(Long, String)](
+"OrdersProctime", 'o_amount, 'o_currency, 'o_rowtime.proctime)
+
+ val ordersWithoutTimeAttribute = util.addTable[(Long, String, Timestamp)](
+"OrdersWithoutTimeAttribute", 'o_amount, 'o_currency, 'o_rowtime)
+
+ val ratesHistory = util.addTable[(String, Int, Timestamp)](
+"RatesHistory", 'currency, 'rate, 'rowtime.rowtime)
+
+ val ratesHistoryWithoutTimeAttribute = util.addTable[(String, Int,
Timestamp)](
+"ratesHistoryWithoutTimeAttribute", 'currency, 'rate, 'rowtime)
+
+ @Test
+ def testInvalidFieldReference(): Unit = {
+expectedException.expect(classOf[ValidationException])
+expectedException.expectMessage("Cannot resolve field [foobar]")
+
+ratesHistory.createTemporalTableFunction('rowtime.rowtime, 'foobar)
+ }
+
+ @Test
+ def testInvalidStringFieldReference(): Unit = {
+expectedException.expect(classOf[ValidationException])
+expectedException.expectMessage("Cannot resolve field [foobar]")
+
+ratesHistory.createTemporalTableFunction("rowtime", "foobar")
+ }
+
+ @Test
+ def testNonTimeIndicatorOnRightSide(): Unit = {
+expectedException.expect(classOf[ValidationException])
+expectedException.expectMessage(
+ "Non rowtime timeAttribute [TIMESTAMP(3)] used to create
TemporalTableFunction")
+
+val rates =
ratesHistoryWithoutTimeAttribute.createTemporalTableFunction('rowtime,
'currency)
Review comment:
But the previous version of this PR was doing it earlier on the API level.
How is it handled by windowed aggregations/windowed joins?
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16619325#comment-16619325
]
ASF GitHub Bot commented on FLINK-9713:
---
pnowojski commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r218460055
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamTemporalJoinToCoProcessTranslator.scala
##
@@ -0,0 +1,198 @@
+/*
+ * 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.table.plan.nodes.datastream
+
+import org.apache.calcite.rel.core.{JoinInfo, JoinRelType}
+import org.apache.calcite.rex._
+import org.apache.flink.api.common.functions.{FlatJoinFunction, MapFunction}
+import org.apache.flink.api.common.typeinfo.TypeInformation
+import org.apache.flink.streaming.api.functions.co.CoProcessFunction
+import org.apache.flink.table.api.{StreamQueryConfig, TableConfig,
TableException, ValidationException}
+import org.apache.flink.table.calcite.FlinkTypeFactory._
+import org.apache.flink.table.codegen.{FunctionCodeGenerator,
GeneratedFunction}
+import org.apache.flink.table.plan.logical.rel.LogicalTemporalTableJoin._
+import org.apache.flink.table.plan.schema.RowSchema
+import org.apache.flink.table.plan.util.RexDefaultVisitor
+import org.apache.flink.table.runtime.join.TemporalJoin
+import org.apache.flink.table.runtime.types.{CRow, CRowTypeInfo}
+import org.apache.flink.types.Row
+import org.apache.flink.util.Preconditions.checkState
+
+class DataStreamTemporalJoinToCoProcessTranslator(
+textualRepresentation: String,
+config: TableConfig,
+returnType: TypeInformation[Row],
+leftSchema: RowSchema,
+rightSchema: RowSchema,
+joinInfo: JoinInfo,
+rexBuilder: RexBuilder)
+ extends DataStreamJoinToCoProcessTranslator(
+config,
+returnType,
+leftSchema,
+rightSchema,
+joinInfo,
+rexBuilder) {
+
+ var leftTimeAttribute: Option[RexNode] = None
+
+ var rightTimeAttribute: Option[RexNode] = None
+
+ var rightPrimaryKeyExpression: Option[RexNode] = None
+
+ override val nonEquiJoinPredicates: Option[RexNode] =
extractTemporalJoinCondition()
+
+ def extractTemporalJoinCondition(): Option[RexNode] = {
+checkState(
+ !joinInfo.isEqui,
+ "Missing %s in join condition",
+ TEMPORAL_JOIN_CONDITION)
+
+val nonEquiJoinRex: RexNode = joinInfo.getRemaining(rexBuilder)
+val temporalJoinConditionExtractor = new TemporalJoinConditionExtractor(
Review comment:
I have refactored this class and moved out all of the construction code. It
makes it nicer (for example `leftTimeAttribute`, `rightTimeAttribute` and
`rightPrimaryKeyExpression` are now proper final fields
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16619329#comment-16619329
]
ASF GitHub Bot commented on FLINK-9713:
---
pnowojski commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r218423118
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/datastream/DataStreamTemporalTableJoinRule.scala
##
@@ -0,0 +1,78 @@
+/*
+ * 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.table.plan.rules.datastream
+
+import org.apache.calcite.plan.{RelOptRule, RelOptRuleCall, RelTraitSet}
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.convert.ConverterRule
+import org.apache.calcite.rel.core.JoinRelType
+import org.apache.flink.table.api.TableConfig
+import org.apache.flink.table.plan.nodes.FlinkConventions
+import org.apache.flink.table.plan.nodes.datastream.DataStreamTemporalTableJoin
+import org.apache.flink.table.plan.nodes.logical.FlinkLogicalTemporalTableJoin
+import org.apache.flink.table.plan.schema.RowSchema
+import org.apache.flink.table.runtime.join.WindowJoinUtil
+
+class DataStreamTemporalTableJoinRule
+ extends ConverterRule(
+classOf[FlinkLogicalTemporalTableJoin],
+FlinkConventions.LOGICAL,
+FlinkConventions.DATASTREAM,
+"DataStreamTemporalTableJoinRule") {
+
+ override def matches(call: RelOptRuleCall): Boolean = {
+val join: FlinkLogicalTemporalTableJoin = call.rel(0)
+val joinInfo = join.analyzeCondition
+
+val (windowBounds, remainingPreds) =
WindowJoinUtil.extractWindowBoundsFromPredicate(
Review comment:
ok :(
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16619326#comment-16619326
]
ASF GitHub Bot commented on FLINK-9713:
---
pnowojski commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r218374441
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/logical/operators.scala
##
@@ -678,6 +678,32 @@ case class WindowAggregate(
}
}
+case class TemporalTable(
Review comment:
`Constructor` is inconsistent with other case classes here.
This is behaviour is already quite clearly visible in this class
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16619317#comment-16619317
]
ASF GitHub Bot commented on FLINK-9713:
---
pnowojski commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r218486273
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/logical/LogicalCorrelateToTemporalTableJoinRule.scala
##
@@ -0,0 +1,208 @@
+/*
+ * 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.table.plan.rules.logical
+
+import org.apache.calcite.plan.RelOptRule.{any, none, operand, some}
+import org.apache.calcite.plan.{RelOptRule, RelOptRuleCall}
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.core.TableFunctionScan
+import org.apache.calcite.rel.logical.LogicalCorrelate
+import org.apache.calcite.rex._
+import org.apache.flink.table.api.{Table, Types, ValidationException}
+import org.apache.flink.table.calcite.FlinkTypeFactory.isProctimeIndicatorType
+import org.apache.flink.table.expressions._
+import org.apache.flink.table.functions.TemporalTableFunction
+import org.apache.flink.table.functions.utils.TableSqlFunction
+import org.apache.flink.table.plan.logical.rel.LogicalTemporalTableJoin
+import org.apache.flink.table.plan.util.RexDefaultVisitor
+
+class LogicalCorrelateToTemporalTableJoinRule
+ extends RelOptRule(
+operand(classOf[LogicalCorrelate],
+ some(
+operand(classOf[RelNode], any()),
+operand(classOf[TableFunctionScan], none(,
+"LogicalCorrelateToTemporalTableJoinRule") {
+
+ def extractNameFromTimeAttribute(timeAttribute: Expression): String = {
+timeAttribute match {
+ case RowtimeAttribute(expr: TemporalTableReference) =>
Review comment:
Dropped
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16619315#comment-16619315
]
ASF GitHub Bot commented on FLINK-9713:
---
pnowojski commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r218364079
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/expressions/fieldExpression.scala
##
@@ -139,6 +139,23 @@ case class WindowReference(name: String, tpe:
Option[TypeInformation[_]] = None)
override def toString: String = s"'$name"
}
+case class TemporalTableReference(name: String, typeInformation:
TypeInformation[_])
Review comment:
If I remember correctly, I introduced it in order to implement
`RowtimeAttribute#validateInput` and `RowtimeAttribute#resultType`.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16614726#comment-16614726
]
ASF GitHub Bot commented on FLINK-9713:
---
twalthr commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r217403556
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/logical/rel/LogicalTemporalTableJoin.scala
##
@@ -0,0 +1,173 @@
+/*
+ * 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.table.plan.logical.rel
+
+import java.util.Collections
+
+import org.apache.calcite.plan.{RelOptCluster, RelTraitSet}
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.core._
+import org.apache.calcite.rex.{RexBuilder, RexNode}
+import org.apache.calcite.sql.`type`.{OperandTypes, ReturnTypes}
+import org.apache.calcite.sql.{SqlFunction, SqlFunctionCategory, SqlKind}
+import org.apache.flink.util.Preconditions.checkArgument
+
+/**
+ * Represents a join between a table and
[[org.apache.flink.table.functions.TemporalTableFunction]]
+ *
+ * @param cluster
+ * @param traitSet
+ * @param left stream
+ * @param right table scan of underlying
+ * [[org.apache.flink.table.functions.TemporalTableFunction]]
+ * @param condition must contain
[[LogicalTemporalTableJoin#TEMPORAL_JOIN_CONDITION]] with
+ * correctly defined references to rightTimeAttribute,
+ * rightPrimaryKeyExpression and leftTimeAttribute. We can
not implement
+ * those references as separate fields, because of problems
with Calcite's
+ * optimization rules like projections push downs, column
+ * pruning/renaming/reordering, etc. Later
rightTimeAttribute,
+ * rightPrimaryKeyExpression and leftTimeAttribute will be
extracted from
+ * the condition.
+ */
+class LogicalTemporalTableJoin private(
+cluster: RelOptCluster,
+traitSet: RelTraitSet,
+left: RelNode,
+right: RelNode,
+condition: RexNode)
+ extends Join(
+cluster,
+traitSet,
+left,
+right,
+condition,
+Collections.emptySet().asInstanceOf[java.util.Set[CorrelationId]],
+JoinRelType.INNER) {
+
+ override def copy(
+ traitSet: RelTraitSet,
+ condition: RexNode,
+ left: RelNode,
+ right: RelNode,
+ joinType: JoinRelType,
+ semiJoinDone: Boolean): LogicalTemporalTableJoin = {
+checkArgument(joinType == this.getJoinType,
+ "Can not change join type".asInstanceOf[Object])
+checkArgument(semiJoinDone == this.isSemiJoinDone,
+ "Can not change semiJoinDone".asInstanceOf[Object])
+new LogicalTemporalTableJoin(
+ cluster,
+ traitSet,
+ left,
+ right,
+ condition)
+ }
+}
+
+object LogicalTemporalTableJoin {
+ /**
+* See [[LogicalTemporalTableJoin#condition]]
+*/
+ val TEMPORAL_JOIN_CONDITION = new SqlFunction(
+"__TEMPORAL_JOIN_CONDITION",
+SqlKind.OTHER_FUNCTION,
+ReturnTypes.BOOLEAN_NOT_NULL,
+null,
+OperandTypes.or(
+ OperandTypes.sequence(
+"'(LEFT_TIME_ATTRIBUTE, RIGHT_TIME_ATTRIBUTE, PRIMARY_KEY)'",
+OperandTypes.DATETIME,
+OperandTypes.DATETIME,
+OperandTypes.ANY),
+ OperandTypes.sequence(
+"'(LEFT_TIME_ATTRIBUTE, PRIMARY_KEY)'",
+OperandTypes.DATETIME,
+OperandTypes.ANY)),
+SqlFunctionCategory.SYSTEM)
+
+ def makeRowTimeTemporalJoinConditionCall(
+ rexBuilder: RexBuilder,
+ leftTimeAttribute: RexNode,
+ rightTimeAttribute: RexNode,
+ rightPrimaryKeyExpression: RexNode): RexNode = {
+rexBuilder.makeCall(
+ TEMPORAL_JOIN_CONDITION,
+ leftTimeAttribute,
+ rightTimeAttribute,
+ rightPrimaryKeyExpression)
+ }
+
+ def makeProcTimeTemporalJoinConditionCall(
+ rexBuilder: RexBuilder,
+ leftTimeAttribute: RexNode,
+ rightPrimaryKeyExpression: RexNode): RexNode = {
+rexBuilder.makeCall(
+
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16614759#comment-16614759
]
ASF GitHub Bot commented on FLINK-9713:
---
twalthr commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r217686794
##
File path:
flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/table/validation/TemporalTableJoinValidationTest.scala
##
@@ -0,0 +1,124 @@
+/*
+ * 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.table.api.stream.table.validation
+
+import java.sql.Timestamp
+
+import org.apache.flink.api.scala._
+import org.apache.flink.table.api.{TableException, ValidationException}
+import org.apache.flink.table.api.scala._
+import org.apache.flink.table.utils._
+import org.junit.Test
+
+class TemporalTableJoinValidationTest extends TableTestBase {
+
+ val util: TableTestUtil = streamTestUtil()
+
+ val orders = util.addTable[(Long, String, Timestamp)](
+"Orders", 'o_amount, 'o_currency, 'o_rowtime.rowtime)
+
+ val ordersProctime = util.addTable[(Long, String)](
+"OrdersProctime", 'o_amount, 'o_currency, 'o_rowtime.proctime)
+
+ val ordersWithoutTimeAttribute = util.addTable[(Long, String, Timestamp)](
+"OrdersWithoutTimeAttribute", 'o_amount, 'o_currency, 'o_rowtime)
+
+ val ratesHistory = util.addTable[(String, Int, Timestamp)](
+"RatesHistory", 'currency, 'rate, 'rowtime.rowtime)
+
+ val ratesHistoryWithoutTimeAttribute = util.addTable[(String, Int,
Timestamp)](
+"ratesHistoryWithoutTimeAttribute", 'currency, 'rate, 'rowtime)
+
+ @Test
+ def testInvalidFieldReference(): Unit = {
+expectedException.expect(classOf[ValidationException])
+expectedException.expectMessage("Cannot resolve field [foobar]")
+
+ratesHistory.createTemporalTableFunction('rowtime.rowtime, 'foobar)
+ }
+
+ @Test
+ def testInvalidStringFieldReference(): Unit = {
+expectedException.expect(classOf[ValidationException])
+expectedException.expectMessage("Cannot resolve field [foobar]")
+
+ratesHistory.createTemporalTableFunction("rowtime", "foobar")
+ }
+
+ @Test
+ def testNonTimeIndicatorOnRightSide(): Unit = {
+expectedException.expect(classOf[ValidationException])
+expectedException.expectMessage(
+ "Non rowtime timeAttribute [TIMESTAMP(3)] used to create
TemporalTableFunction")
+
+val rates =
ratesHistoryWithoutTimeAttribute.createTemporalTableFunction('rowtime,
'currency)
Review comment:
This exception occurs very late during optimization. I'm wondering if we
should perform the basic check a bit earlier, even though this check could have
false positives.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16614753#comment-16614753
]
ASF GitHub Bot commented on FLINK-9713:
---
twalthr commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r217364300
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/BatchTableEnvironment.scala
##
@@ -448,48 +448,13 @@ abstract class BatchTableEnvironment(
* @return The optimized [[RelNode]] tree
*/
private[flink] def optimize(relNode: RelNode): RelNode = {
-
-// 0. convert sub-queries before query decorrelation
-val convSubQueryPlan = runHepPlanner(
- HepMatchOrder.BOTTOM_UP, FlinkRuleSets.TABLE_SUBQUERY_RULES, relNode,
relNode.getTraitSet)
-
-// 0. convert table references
-val fullRelNode = runHepPlanner(
- HepMatchOrder.BOTTOM_UP,
- FlinkRuleSets.TABLE_REF_RULES,
- convSubQueryPlan,
- relNode.getTraitSet)
-
-// 1. decorrelate
+val convSubQueryPlan = optimizeConvertSubQueries(relNode)
+val temporalTableJoinPlan = optimizeConvertToTemporalJoin(convSubQueryPlan)
+val fullRelNode = optimizeConvertTableReferences(temporalTableJoinPlan)
Review comment:
nit: `fullRelNode` -> `fullPlan`
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16614732#comment-16614732
]
ASF GitHub Bot commented on FLINK-9713:
---
twalthr commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r217639947
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/logical/LogicalCorrelateToTemporalTableJoinRule.scala
##
@@ -0,0 +1,208 @@
+/*
+ * 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.table.plan.rules.logical
+
+import org.apache.calcite.plan.RelOptRule.{any, none, operand, some}
+import org.apache.calcite.plan.{RelOptRule, RelOptRuleCall}
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.core.TableFunctionScan
+import org.apache.calcite.rel.logical.LogicalCorrelate
+import org.apache.calcite.rex._
+import org.apache.flink.table.api.{Table, Types, ValidationException}
+import org.apache.flink.table.calcite.FlinkTypeFactory.isProctimeIndicatorType
+import org.apache.flink.table.expressions._
+import org.apache.flink.table.functions.TemporalTableFunction
+import org.apache.flink.table.functions.utils.TableSqlFunction
+import org.apache.flink.table.plan.logical.rel.LogicalTemporalTableJoin
+import org.apache.flink.table.plan.util.RexDefaultVisitor
+
+class LogicalCorrelateToTemporalTableJoinRule
+ extends RelOptRule(
+operand(classOf[LogicalCorrelate],
+ some(
+operand(classOf[RelNode], any()),
+operand(classOf[TableFunctionScan], none(,
+"LogicalCorrelateToTemporalTableJoinRule") {
+
+ def extractNameFromTimeAttribute(timeAttribute: Expression): String = {
+timeAttribute match {
+ case RowtimeAttribute(expr: TemporalTableReference) =>
+expr.name
+ case ProctimeAttribute(expr: TemporalTableReference) =>
+expr.name
+ case TemporalTableReference(name, _) =>
+name
+ case ResolvedFieldReference(name, _)
+if timeAttribute.resultType == Types.LONG ||
+ timeAttribute.resultType == Types.SQL_TIMESTAMP =>
+name
+ case _ => throw new ValidationException(
+s"Invalid timeAttribute [${timeAttribute}] in TemporalTableFunction")
+}
+ }
+
+ override def onMatch(call: RelOptRuleCall): Unit = {
+val logicalCorrelate: LogicalCorrelate = call.rel(0)
+val leftNode: RelNode = call.rel(1)
+val rightTableFunctionScan: TableFunctionScan = call.rel(2)
+
+val cluster = logicalCorrelate.getCluster
+
+new GetTemporalTableFunctionCall(cluster.getRexBuilder, leftNode)
+ .visit(rightTableFunctionScan.getCall) match {
+ case Some(TemporalTableFunctionCall(rightTemporalTableFunction,
leftTimeAttribute)) =>
+val underlyingHistoryTable: Table =
rightTemporalTableFunction.getUnderlyingHistoryTable
+val relBuilder = this.relBuilderFactory.create(
+ cluster,
+ underlyingHistoryTable.relBuilder.getRelOptSchema)
+val rexBuilder = cluster.getRexBuilder
+
+val rightNode: RelNode =
underlyingHistoryTable.logicalPlan.toRelNode(relBuilder)
+
+val rightTimeIndicatorExpression = createRightExpression(
+ rexBuilder,
+ leftNode,
+ rightNode,
+
extractNameFromTimeAttribute(rightTemporalTableFunction.getTimeAttribute))
+
+val rightPrimaryKeyExpression = createRightExpression(
+ rexBuilder,
+ leftNode,
+ rightNode,
+ rightTemporalTableFunction.getPrimaryKey)
+
+relBuilder.push(
+ if
(isProctimeIndicatorType(rightTemporalTableFunction.getTimeAttribute.resultType))
{
+LogicalTemporalTableJoin.create(
+ rexBuilder,
+ cluster,
+ logicalCorrelate.getTraitSet,
+ leftNode,
+ rightNode,
+ leftTimeAttribute,
+ rightPrimaryKeyExpression)
+ }
+ else {
+LogicalTemporalTableJoin.create(
+ rexBuilder,
+ cluster,
+
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16614720#comment-16614720
]
ASF GitHub Bot commented on FLINK-9713:
---
twalthr commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r217369978
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/table.scala
##
@@ -155,6 +156,50 @@ class Table(
select(withResolvedAggFunctionCall: _*)
}
+ /**
+* Creates [[TemporalTableFunction]] backed up by this table as a history
table.
+*
+* @param timeIndicator field for the [[TemporalTableFunction]]. Must
points to a time indicator
+* @param primaryKeyfield for the [[TemporalTableFunction]].
+* @return [[TemporalTableFunction]]
Review comment:
Add documentation what users can do with this return value.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16614721#comment-16614721
]
ASF GitHub Bot commented on FLINK-9713:
---
twalthr commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r217414676
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamTemporalJoinToCoProcessTranslator.scala
##
@@ -0,0 +1,198 @@
+/*
+ * 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.table.plan.nodes.datastream
+
+import org.apache.calcite.rel.core.{JoinInfo, JoinRelType}
+import org.apache.calcite.rex._
+import org.apache.flink.api.common.functions.{FlatJoinFunction, MapFunction}
+import org.apache.flink.api.common.typeinfo.TypeInformation
+import org.apache.flink.streaming.api.functions.co.CoProcessFunction
+import org.apache.flink.table.api.{StreamQueryConfig, TableConfig,
TableException, ValidationException}
+import org.apache.flink.table.calcite.FlinkTypeFactory._
+import org.apache.flink.table.codegen.{FunctionCodeGenerator,
GeneratedFunction}
+import org.apache.flink.table.plan.logical.rel.LogicalTemporalTableJoin._
+import org.apache.flink.table.plan.schema.RowSchema
+import org.apache.flink.table.plan.util.RexDefaultVisitor
+import org.apache.flink.table.runtime.join.TemporalJoin
+import org.apache.flink.table.runtime.types.{CRow, CRowTypeInfo}
+import org.apache.flink.types.Row
+import org.apache.flink.util.Preconditions.checkState
+
+class DataStreamTemporalJoinToCoProcessTranslator(
+textualRepresentation: String,
+config: TableConfig,
+returnType: TypeInformation[Row],
+leftSchema: RowSchema,
+rightSchema: RowSchema,
+joinInfo: JoinInfo,
+rexBuilder: RexBuilder)
+ extends DataStreamJoinToCoProcessTranslator(
+config,
+returnType,
+leftSchema,
+rightSchema,
+joinInfo,
+rexBuilder) {
+
+ var leftTimeAttribute: Option[RexNode] = None
+
+ var rightTimeAttribute: Option[RexNode] = None
+
+ var rightPrimaryKeyExpression: Option[RexNode] = None
+
+ override val nonEquiJoinPredicates: Option[RexNode] =
extractTemporalJoinCondition()
+
+ def extractTemporalJoinCondition(): Option[RexNode] = {
+checkState(
+ !joinInfo.isEqui,
+ "Missing %s in join condition",
+ TEMPORAL_JOIN_CONDITION)
+
+val nonEquiJoinRex: RexNode = joinInfo.getRemaining(rexBuilder)
+val temporalJoinConditionExtractor = new TemporalJoinConditionExtractor(
+ nonEquiJoinRex.toString)
+
+Some(temporalJoinConditionExtractor.apply(nonEquiJoinRex))
+ }
+
+ override protected def createCoProcessFunction(
+ joinType: JoinRelType,
+ queryConfig: StreamQueryConfig,
+ joinFunction: GeneratedFunction[FlatJoinFunction[Row, Row, Row], Row])
+: CoProcessFunction[CRow, CRow, CRow] = {
+
+checkState(
+ leftTimeAttribute.isDefined &&
+rightPrimaryKeyExpression.isDefined,
+ "Missing %s in join condition",
+ TEMPORAL_JOIN_CONDITION)
+
+if (rightTimeAttribute.isDefined) {
+ throw new ValidationException(
+s"Currently only proctime temporal joins are supported in
[${textualRepresentation}]")
+}
+
+joinType match {
+ case JoinRelType.INNER =>
+new TemporalJoin(
+ leftSchema.typeInfo,
+ rightSchema.typeInfo,
+ joinFunction.name,
+ joinFunction.code,
+ queryConfig)
+ case _ =>
+ throw new ValidationException(
+ s"Only ${JoinRelType.INNER} temporal join is supported in
[${textualRepresentation}]")
+}
+ }
+
+ private class TemporalJoinConditionExtractor(
+ nonEquiJoinCondition: String)
+extends RexShuttle {
+
+override def visitCall(call: RexCall): RexNode = {
+ if (call.getOperator != TEMPORAL_JOIN_CONDITION) {
+return super.visitCall(call)
+ }
+
+ if (leftTimeAttribute.isDefined
+|| rightPrimaryKeyExpression.isDefined
+|| rightTimeAttribute.isDefined) {
+throw new
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16614760#comment-16614760
]
ASF GitHub Bot commented on FLINK-9713:
---
twalthr commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r217685337
##
File path:
flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/table/TemporalTableJoinTest.scala
##
@@ -0,0 +1,276 @@
+/*
+ * 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.table.api.stream.table
+
+import java.sql.Timestamp
+
+import org.apache.flink.api.scala._
+import org.apache.flink.table.api.{TableSchema, ValidationException}
+import org.apache.flink.table.api.scala._
+import org.apache.flink.table.api.stream.table.TemporalTableJoinTest._
+import org.apache.flink.table.expressions.TemporalTableReference
+import org.apache.flink.table.functions.TemporalTableFunction
+import org.apache.flink.table.plan.logical.rel.LogicalTemporalTableJoin._
+import org.apache.flink.table.utils.TableTestUtil._
+import org.apache.flink.table.utils._
+import org.hamcrest.Matchers.startsWith
+import org.junit.Assert.{assertArrayEquals, assertEquals, assertTrue}
+import org.junit.Test
+
+class TemporalTableJoinTest extends TableTestBase {
+
+ val util: TableTestUtil = streamTestUtil()
+
+ val orders = util.addTable[(Long, String, Timestamp)](
+"Orders", 'o_amount, 'o_currency, 'o_rowtime.rowtime)
+
+ val ratesHistory = util.addTable[(String, Int, Timestamp)](
+"RatesHistory", 'currency, 'rate, 'rowtime.rowtime)
+
+ val rates = util.addFunction(
+"Rates",
+ratesHistory.createTemporalTableFunction('rowtime.rowtime, 'currency))
+
+ val proctimeOrders = util.addTable[(Long, String)](
+"ProctimeOrders", 'o_amount, 'o_currency, 'o_proctime.proctime)
+
+ val proctimeRatesHistory = util.addTable[(String, Int)](
+"ProctimeRatesHistory", 'currency, 'rate, 'proctime.proctime)
+
+ val proctimeRates =
proctimeRatesHistory.createTemporalTableFunction('proctime, 'currency)
+
+ @Test
+ def testSimpleJoin(): Unit = {
+val result = orders
+ .join(rates('o_rowtime), "currency = o_currency")
+ .select("o_amount * rate").as("rate")
+
+util.verifyTable(result, getExpectedSimpleJoinPlan())
+ }
+
+ @Test
+ def testSimpleProctimeJoin(): Unit = {
+val result = proctimeOrders
+ .join(proctimeRates('o_proctime), "currency = o_currency")
+ .select("o_amount * rate").as("rate")
+
+util.verifyTable(result, getExpectedSimpleProctimeJoinPlan())
+ }
+
+ /**
+* Test versioned joins with more complicated query.
+* Important thing here is that we have complex OR join condition
+* and there are some columns that are not being used (are being pruned).
+*/
+ @Test
+ def testComplexJoin(): Unit = {
+val util = streamTestUtil()
+val thirdTable = util.addTable[(String, Int)]("ThirdTable", 't3_comment,
't3_secondary_key)
+val orders = util.addTable[(Timestamp, String, Long, String, Int)](
+ "Orders", 'o_rowtime.rowtime, 'o_comment, 'o_amount, 'o_currency,
'o_secondary_key)
+
+val ratesHistory = util.addTable[(Timestamp, String, String, Int, Int)](
+ "RatesHistory", 'rowtime.rowtime, 'comment, 'currency, 'rate,
'secondary_key)
+val rates = ratesHistory.createTemporalTableFunction('rowtime.rowtime,
'currency)
Review comment:
remove `.rowtime`
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16614749#comment-16614749
]
ASF GitHub Bot commented on FLINK-9713:
---
twalthr commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r217641707
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/logical/LogicalCorrelateToTemporalTableJoinRule.scala
##
@@ -0,0 +1,208 @@
+/*
+ * 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.table.plan.rules.logical
+
+import org.apache.calcite.plan.RelOptRule.{any, none, operand, some}
+import org.apache.calcite.plan.{RelOptRule, RelOptRuleCall}
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.core.TableFunctionScan
+import org.apache.calcite.rel.logical.LogicalCorrelate
+import org.apache.calcite.rex._
+import org.apache.flink.table.api.{Table, Types, ValidationException}
+import org.apache.flink.table.calcite.FlinkTypeFactory.isProctimeIndicatorType
+import org.apache.flink.table.expressions._
+import org.apache.flink.table.functions.TemporalTableFunction
+import org.apache.flink.table.functions.utils.TableSqlFunction
+import org.apache.flink.table.plan.logical.rel.LogicalTemporalTableJoin
+import org.apache.flink.table.plan.util.RexDefaultVisitor
+
+class LogicalCorrelateToTemporalTableJoinRule
+ extends RelOptRule(
+operand(classOf[LogicalCorrelate],
+ some(
+operand(classOf[RelNode], any()),
+operand(classOf[TableFunctionScan], none(,
+"LogicalCorrelateToTemporalTableJoinRule") {
+
+ def extractNameFromTimeAttribute(timeAttribute: Expression): String = {
+timeAttribute match {
+ case RowtimeAttribute(expr: TemporalTableReference) =>
+expr.name
+ case ProctimeAttribute(expr: TemporalTableReference) =>
+expr.name
+ case TemporalTableReference(name, _) =>
+name
+ case ResolvedFieldReference(name, _)
+if timeAttribute.resultType == Types.LONG ||
+ timeAttribute.resultType == Types.SQL_TIMESTAMP =>
+name
+ case _ => throw new ValidationException(
+s"Invalid timeAttribute [${timeAttribute}] in TemporalTableFunction")
+}
+ }
+
+ override def onMatch(call: RelOptRuleCall): Unit = {
+val logicalCorrelate: LogicalCorrelate = call.rel(0)
+val leftNode: RelNode = call.rel(1)
+val rightTableFunctionScan: TableFunctionScan = call.rel(2)
+
+val cluster = logicalCorrelate.getCluster
+
+new GetTemporalTableFunctionCall(cluster.getRexBuilder, leftNode)
+ .visit(rightTableFunctionScan.getCall) match {
+ case Some(TemporalTableFunctionCall(rightTemporalTableFunction,
leftTimeAttribute)) =>
+val underlyingHistoryTable: Table =
rightTemporalTableFunction.getUnderlyingHistoryTable
+val relBuilder = this.relBuilderFactory.create(
+ cluster,
+ underlyingHistoryTable.relBuilder.getRelOptSchema)
+val rexBuilder = cluster.getRexBuilder
+
+val rightNode: RelNode =
underlyingHistoryTable.logicalPlan.toRelNode(relBuilder)
+
+val rightTimeIndicatorExpression = createRightExpression(
+ rexBuilder,
+ leftNode,
+ rightNode,
+
extractNameFromTimeAttribute(rightTemporalTableFunction.getTimeAttribute))
+
+val rightPrimaryKeyExpression = createRightExpression(
+ rexBuilder,
+ leftNode,
+ rightNode,
+ rightTemporalTableFunction.getPrimaryKey)
+
+relBuilder.push(
+ if
(isProctimeIndicatorType(rightTemporalTableFunction.getTimeAttribute.resultType))
{
+LogicalTemporalTableJoin.create(
+ rexBuilder,
+ cluster,
+ logicalCorrelate.getTraitSet,
+ leftNode,
+ rightNode,
+ leftTimeAttribute,
+ rightPrimaryKeyExpression)
+ }
+ else {
+LogicalTemporalTableJoin.create(
+ rexBuilder,
+ cluster,
+
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16614748#comment-16614748
]
ASF GitHub Bot commented on FLINK-9713:
---
twalthr commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r217364622
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/StreamTableEnvironment.scala
##
@@ -803,70 +803,37 @@ abstract class StreamTableEnvironment(
* @return The optimized [[RelNode]] tree
*/
private[flink] def optimize(relNode: RelNode, updatesAsRetraction: Boolean):
RelNode = {
-
-// 0. convert sub-queries before query decorrelation
-val convSubQueryPlan = runHepPlanner(
- HepMatchOrder.BOTTOM_UP, FlinkRuleSets.TABLE_SUBQUERY_RULES, relNode,
relNode.getTraitSet)
-
-// 0. convert table references
-val fullRelNode = runHepPlanner(
- HepMatchOrder.BOTTOM_UP,
- FlinkRuleSets.TABLE_REF_RULES,
- convSubQueryPlan,
- relNode.getTraitSet)
-
-// 1. decorrelate
+val convSubQueryPlan = optimizeConvertSubQueries(relNode)
+val temporalTableJoinPlan = optimizeConvertToTemporalJoin(convSubQueryPlan)
+val fullRelNode = optimizeConvertTableReferences(temporalTableJoinPlan)
val decorPlan = RelDecorrelator.decorrelateQuery(fullRelNode)
-
-// 2. convert time indicators
val convPlan = RelTimeIndicatorConverter.convert(decorPlan,
getRelBuilder.getRexBuilder)
Review comment:
Give a more meaningful name to this variable.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16614761#comment-16614761
]
ASF GitHub Bot commented on FLINK-9713:
---
twalthr commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r217630735
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/datastream/DataStreamTemporalTableJoinRule.scala
##
@@ -0,0 +1,78 @@
+/*
+ * 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.table.plan.rules.datastream
+
+import org.apache.calcite.plan.{RelOptRule, RelOptRuleCall, RelTraitSet}
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.convert.ConverterRule
+import org.apache.calcite.rel.core.JoinRelType
+import org.apache.flink.table.api.TableConfig
+import org.apache.flink.table.plan.nodes.FlinkConventions
+import org.apache.flink.table.plan.nodes.datastream.DataStreamTemporalTableJoin
+import org.apache.flink.table.plan.nodes.logical.FlinkLogicalTemporalTableJoin
+import org.apache.flink.table.plan.schema.RowSchema
+import org.apache.flink.table.runtime.join.WindowJoinUtil
+
+class DataStreamTemporalTableJoinRule
+ extends ConverterRule(
+classOf[FlinkLogicalTemporalTableJoin],
+FlinkConventions.LOGICAL,
+FlinkConventions.DATASTREAM,
+"DataStreamTemporalTableJoinRule") {
+
+ override def matches(call: RelOptRuleCall): Boolean = {
+val join: FlinkLogicalTemporalTableJoin = call.rel(0)
+val joinInfo = join.analyzeCondition
+
+val (windowBounds, remainingPreds) =
WindowJoinUtil.extractWindowBoundsFromPredicate(
Review comment:
side note for the future: this check is very expensive since the method
performs code compilation. If we find another way, that would be great :/
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16614724#comment-16614724
]
ASF GitHub Bot commented on FLINK-9713:
---
twalthr commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r217425419
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/logical/FlinkAbstractLogicalJoinWithCost.scala
##
@@ -0,0 +1,59 @@
+/*
+ * 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.table.plan.nodes.logical
+
+import org.apache.calcite.plan._
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.core._
+import org.apache.calcite.rel.metadata.RelMetadataQuery
+import org.apache.calcite.rex.RexNode
+
+import scala.collection.JavaConverters._
+
+abstract class FlinkAbstractLogicalJoinWithCost(
Review comment:
nit: `FlinkAbstractLogicalJoinWithCost` -> `FlinkLogicalJoinBase`. Because
actually all relnodes have cost.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16614728#comment-16614728
]
ASF GitHub Bot commented on FLINK-9713:
---
twalthr commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r217691162
##
File path:
flink-libraries/flink-table/src/test/scala/org/apache/flink/table/runtime/stream/sql/TemporalJoinITCase.scala
##
@@ -0,0 +1,135 @@
+/*
+ * 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.table.runtime.stream.sql
+
+import java.sql.Timestamp
+
+import org.apache.flink.api.scala._
+import org.apache.flink.streaming.api.TimeCharacteristic
+import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
+import org.apache.flink.table.api.{TableEnvironment, TableException}
+import org.apache.flink.table.api.scala._
+import org.apache.flink.table.runtime.utils.{StreamITCase,
StreamingWithStateTestBase}
+import org.apache.flink.types.Row
+import org.junit._
+import org.junit.rules.ExpectedException
+
+import scala.collection.mutable
+
+class TemporalJoinITCase extends StreamingWithStateTestBase {
+ val expectedException = ExpectedException.none()
+
+ @Rule
+ def thrown = expectedException
+
+ @Test
+ def testProcessTimeInnerJoin(): Unit = {
Review comment:
We should mention in a comment that this test is actually not validating the
actual results.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16614747#comment-16614747
]
ASF GitHub Bot commented on FLINK-9713:
---
twalthr commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r217373119
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/logical/operators.scala
##
@@ -678,6 +678,32 @@ case class WindowAggregate(
}
}
+case class TemporalTable(
Review comment:
Call it `TemporalTableConstructor`? Add docs to this class to remind us that
we need it only to resolve time indicators.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16614758#comment-16614758
]
ASF GitHub Bot commented on FLINK-9713:
---
twalthr commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r217687627
##
File path:
flink-libraries/flink-table/src/test/scala/org/apache/flink/table/runtime/harness/TemporalJoinHarnessTest.scala
##
@@ -0,0 +1,458 @@
+/*
+ * 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.table.runtime.harness
+
+import java.util.concurrent.ConcurrentLinkedQueue
+
+import org.apache.calcite.rel.core.{JoinInfo, JoinRelType}
+import org.apache.calcite.rex.{RexBuilder, RexNode}
+import org.apache.calcite.sql.fun.SqlStdOperatorTable
+import org.apache.calcite.util.ImmutableIntList
+import org.apache.flink.api.common.time.Time
+import org.apache.flink.api.common.typeinfo.{BasicTypeInfo, TypeInformation}
+import org.apache.flink.api.java.functions.KeySelector
+import org.apache.flink.api.java.typeutils.RowTypeInfo
+import org.apache.flink.streaming.api.functions.co.CoProcessFunction
+import org.apache.flink.streaming.api.operators.co.KeyedCoProcessOperator
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord
+import org.apache.flink.streaming.util.KeyedTwoInputStreamOperatorTestHarness
+import org.apache.flink.table.api.{TableConfig, Types, ValidationException}
+import org.apache.flink.table.calcite.{FlinkTypeFactory, FlinkTypeSystem}
+import org.apache.flink.table.plan.logical.rel.LogicalTemporalTableJoin
+import
org.apache.flink.table.plan.logical.rel.LogicalTemporalTableJoin.TEMPORAL_JOIN_CONDITION
+import
org.apache.flink.table.plan.nodes.datastream.DataStreamTemporalJoinToCoProcessTranslator
+import org.apache.flink.table.plan.schema.RowSchema
+import org.apache.flink.table.runtime.CRowKeySelector
+import
org.apache.flink.table.runtime.harness.HarnessTestBase.{RowResultSortComparator,
TestStreamQueryConfig}
+import org.apache.flink.table.runtime.types.CRow
+import org.apache.flink.table.typeutils.TimeIndicatorTypeInfo
+import org.hamcrest.Matchers.startsWith
+import org.junit.rules.ExpectedException
+import org.junit.{Rule, Test}
+
+class TemporalJoinHarnessTest extends HarnessTestBase {
+ // used for accurate exception information checking.
Review comment:
Move this to the test base as we do it in `TableTestBase`.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16614735#comment-16614735
]
ASF GitHub Bot commented on FLINK-9713:
---
twalthr commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r217405733
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamJoinToCoProcessTranslator.scala
##
@@ -0,0 +1,149 @@
+/*
+ * 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.table.plan.nodes.datastream
+
+import org.apache.calcite.rel.core.{JoinInfo, JoinRelType}
+import org.apache.calcite.rex.{RexBuilder, RexNode}
+import org.apache.flink.api.common.functions.FlatJoinFunction
+import org.apache.flink.api.common.typeinfo.TypeInformation
+import org.apache.flink.streaming.api.functions.co.CoProcessFunction
+import org.apache.flink.table.api.{StreamQueryConfig, TableConfig}
+import org.apache.flink.table.codegen.{FunctionCodeGenerator,
GeneratedFunction}
+import org.apache.flink.table.plan.schema.RowSchema
+import org.apache.flink.table.runtime.CRowKeySelector
+import org.apache.flink.table.runtime.join._
+import org.apache.flink.table.runtime.types.{CRow, CRowTypeInfo}
Review comment:
Unused import.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16614755#comment-16614755
]
ASF GitHub Bot commented on FLINK-9713:
---
twalthr commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r217373861
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/table.scala
##
@@ -155,6 +156,50 @@ class Table(
select(withResolvedAggFunctionCall: _*)
}
+ /**
+* Creates [[TemporalTableFunction]] backed up by this table as a history
table.
+*
+* @param timeIndicator field for the [[TemporalTableFunction]]. Must
points to a time indicator
+* @param primaryKeyfield for the [[TemporalTableFunction]].
+* @return [[TemporalTableFunction]]
+*/
+ def createTemporalTableFunction(
+ timeIndicator: String,
+ primaryKey: String): TemporalTableFunction = {
+createTemporalTableFunction(
+ ExpressionParser.parseExpression(timeIndicator),
+ ExpressionParser.parseExpression(primaryKey))
+ }
+
+ /**
+* Creates [[TemporalTableFunction]] backed up by this table as a history
table.
+*
+* @param timeIndicator field for the [[TemporalTableFunction]]. Must
points to a time indicator
+* @param primaryKeyfield for the [[TemporalTableFunction]].
+* @return [[TemporalTableFunction]]
+*/
+ def createTemporalTableFunction(
+ timeIndicator: Expression,
+ primaryKey: Expression): TemporalTableFunction = {
+val temporalTable = new TemporalTable(timeIndicator, primaryKey,
logicalPlan)
+ .validate(tableEnv)
+ .asInstanceOf[TemporalTable]
+
+TemporalTableFunction.create(
+ this,
+ temporalTable.timeIndicator,
+ validatePrimaryKeyExpression(temporalTable.primaryKey))
+ }
+
+ private def validatePrimaryKeyExpression(expression: Expression): String = {
+expression match {
+ case fieldReference: ResolvedFieldReference =>
+fieldReference.name
+ case _ => throw new ValidationException(
+s"Unsupported expression [${expression}] as primary key")
Review comment:
Add description what is supported as primary key. E.g. "Only single,
top-level field references are supported."
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16614750#comment-16614750
]
ASF GitHub Bot commented on FLINK-9713:
---
twalthr commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r217377012
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/calcite/RelTimeIndicatorConverter.scala
##
@@ -18,6 +18,8 @@
package org.apache.flink.table.calcite
+import java.util.Collections
Review comment:
Remove unused import.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16614725#comment-16614725
]
ASF GitHub Bot commented on FLINK-9713:
---
twalthr commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r217397897
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/logical/rel/LogicalTemporalTableJoin.scala
##
@@ -0,0 +1,173 @@
+/*
+ * 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.table.plan.logical.rel
+
+import java.util.Collections
+
+import org.apache.calcite.plan.{RelOptCluster, RelTraitSet}
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.core._
+import org.apache.calcite.rex.{RexBuilder, RexNode}
+import org.apache.calcite.sql.`type`.{OperandTypes, ReturnTypes}
+import org.apache.calcite.sql.{SqlFunction, SqlFunctionCategory, SqlKind}
+import org.apache.flink.util.Preconditions.checkArgument
+
+/**
+ * Represents a join between a table and
[[org.apache.flink.table.functions.TemporalTableFunction]]
+ *
+ * @param cluster
+ * @param traitSet
+ * @param left stream
Review comment:
Logically this is not a stream, because a logical node applies to both batch
and streaming.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16614739#comment-16614739
]
ASF GitHub Bot commented on FLINK-9713:
---
twalthr commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r217412201
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamTemporalJoinToCoProcessTranslator.scala
##
@@ -0,0 +1,198 @@
+/*
+ * 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.table.plan.nodes.datastream
+
+import org.apache.calcite.rel.core.{JoinInfo, JoinRelType}
+import org.apache.calcite.rex._
+import org.apache.flink.api.common.functions.{FlatJoinFunction, MapFunction}
+import org.apache.flink.api.common.typeinfo.TypeInformation
+import org.apache.flink.streaming.api.functions.co.CoProcessFunction
+import org.apache.flink.table.api.{StreamQueryConfig, TableConfig,
TableException, ValidationException}
+import org.apache.flink.table.calcite.FlinkTypeFactory._
+import org.apache.flink.table.codegen.{FunctionCodeGenerator,
GeneratedFunction}
+import org.apache.flink.table.plan.logical.rel.LogicalTemporalTableJoin._
+import org.apache.flink.table.plan.schema.RowSchema
+import org.apache.flink.table.plan.util.RexDefaultVisitor
+import org.apache.flink.table.runtime.join.TemporalJoin
+import org.apache.flink.table.runtime.types.{CRow, CRowTypeInfo}
+import org.apache.flink.types.Row
+import org.apache.flink.util.Preconditions.checkState
+
+class DataStreamTemporalJoinToCoProcessTranslator(
+textualRepresentation: String,
+config: TableConfig,
+returnType: TypeInformation[Row],
+leftSchema: RowSchema,
+rightSchema: RowSchema,
+joinInfo: JoinInfo,
+rexBuilder: RexBuilder)
+ extends DataStreamJoinToCoProcessTranslator(
+config,
+returnType,
+leftSchema,
+rightSchema,
+joinInfo,
+rexBuilder) {
+
+ var leftTimeAttribute: Option[RexNode] = None
+
+ var rightTimeAttribute: Option[RexNode] = None
+
+ var rightPrimaryKeyExpression: Option[RexNode] = None
+
+ override val nonEquiJoinPredicates: Option[RexNode] =
extractTemporalJoinCondition()
+
+ def extractTemporalJoinCondition(): Option[RexNode] = {
+checkState(
+ !joinInfo.isEqui,
+ "Missing %s in join condition",
+ TEMPORAL_JOIN_CONDITION)
+
+val nonEquiJoinRex: RexNode = joinInfo.getRemaining(rexBuilder)
+val temporalJoinConditionExtractor = new TemporalJoinConditionExtractor(
+ nonEquiJoinRex.toString)
+
+Some(temporalJoinConditionExtractor.apply(nonEquiJoinRex))
+ }
+
+ override protected def createCoProcessFunction(
+ joinType: JoinRelType,
+ queryConfig: StreamQueryConfig,
+ joinFunction: GeneratedFunction[FlatJoinFunction[Row, Row, Row], Row])
+: CoProcessFunction[CRow, CRow, CRow] = {
+
+checkState(
+ leftTimeAttribute.isDefined &&
+rightPrimaryKeyExpression.isDefined,
+ "Missing %s in join condition",
+ TEMPORAL_JOIN_CONDITION)
+
+if (rightTimeAttribute.isDefined) {
Review comment:
Shouldn't we have a flag instead of checking if the right time attribute is
defined?
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16614734#comment-16614734
]
ASF GitHub Bot commented on FLINK-9713:
---
twalthr commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r217367243
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/table.scala
##
@@ -155,6 +156,50 @@ class Table(
select(withResolvedAggFunctionCall: _*)
}
+ /**
+* Creates [[TemporalTableFunction]] backed up by this table as a history
table.
+*
+* @param timeIndicator field for the [[TemporalTableFunction]]. Must
points to a time indicator
Review comment:
As mentioned before, please add more explanation and documentation to this
method. Maybe a little example? Otherwise people won't understand/won't use
this feature. Time indicator is an internal name, use `time attribute`.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16614738#comment-16614738
]
ASF GitHub Bot commented on FLINK-9713:
---
twalthr commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r217685225
##
File path:
flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/table/TemporalTableJoinTest.scala
##
@@ -0,0 +1,276 @@
+/*
+ * 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.table.api.stream.table
+
+import java.sql.Timestamp
+
+import org.apache.flink.api.scala._
+import org.apache.flink.table.api.{TableSchema, ValidationException}
+import org.apache.flink.table.api.scala._
+import org.apache.flink.table.api.stream.table.TemporalTableJoinTest._
+import org.apache.flink.table.expressions.TemporalTableReference
+import org.apache.flink.table.functions.TemporalTableFunction
+import org.apache.flink.table.plan.logical.rel.LogicalTemporalTableJoin._
+import org.apache.flink.table.utils.TableTestUtil._
+import org.apache.flink.table.utils._
+import org.hamcrest.Matchers.startsWith
+import org.junit.Assert.{assertArrayEquals, assertEquals, assertTrue}
+import org.junit.Test
+
+class TemporalTableJoinTest extends TableTestBase {
+
+ val util: TableTestUtil = streamTestUtil()
+
+ val orders = util.addTable[(Long, String, Timestamp)](
+"Orders", 'o_amount, 'o_currency, 'o_rowtime.rowtime)
+
+ val ratesHistory = util.addTable[(String, Int, Timestamp)](
+"RatesHistory", 'currency, 'rate, 'rowtime.rowtime)
+
+ val rates = util.addFunction(
+"Rates",
+ratesHistory.createTemporalTableFunction('rowtime.rowtime, 'currency))
Review comment:
Remove `.rowtime` here as well.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16614741#comment-16614741
]
ASF GitHub Bot commented on FLINK-9713:
---
twalthr commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r217421138
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamTemporalJoinToCoProcessTranslator.scala
##
@@ -0,0 +1,198 @@
+/*
+ * 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.table.plan.nodes.datastream
+
+import org.apache.calcite.rel.core.{JoinInfo, JoinRelType}
+import org.apache.calcite.rex._
+import org.apache.flink.api.common.functions.{FlatJoinFunction, MapFunction}
+import org.apache.flink.api.common.typeinfo.TypeInformation
+import org.apache.flink.streaming.api.functions.co.CoProcessFunction
+import org.apache.flink.table.api.{StreamQueryConfig, TableConfig,
TableException, ValidationException}
+import org.apache.flink.table.calcite.FlinkTypeFactory._
+import org.apache.flink.table.codegen.{FunctionCodeGenerator,
GeneratedFunction}
+import org.apache.flink.table.plan.logical.rel.LogicalTemporalTableJoin._
+import org.apache.flink.table.plan.schema.RowSchema
+import org.apache.flink.table.plan.util.RexDefaultVisitor
+import org.apache.flink.table.runtime.join.TemporalJoin
+import org.apache.flink.table.runtime.types.{CRow, CRowTypeInfo}
+import org.apache.flink.types.Row
+import org.apache.flink.util.Preconditions.checkState
+
+class DataStreamTemporalJoinToCoProcessTranslator(
+textualRepresentation: String,
+config: TableConfig,
+returnType: TypeInformation[Row],
+leftSchema: RowSchema,
+rightSchema: RowSchema,
+joinInfo: JoinInfo,
+rexBuilder: RexBuilder)
+ extends DataStreamJoinToCoProcessTranslator(
+config,
+returnType,
+leftSchema,
+rightSchema,
+joinInfo,
+rexBuilder) {
+
+ var leftTimeAttribute: Option[RexNode] = None
+
+ var rightTimeAttribute: Option[RexNode] = None
+
+ var rightPrimaryKeyExpression: Option[RexNode] = None
+
+ override val nonEquiJoinPredicates: Option[RexNode] =
extractTemporalJoinCondition()
+
+ def extractTemporalJoinCondition(): Option[RexNode] = {
+checkState(
+ !joinInfo.isEqui,
+ "Missing %s in join condition",
+ TEMPORAL_JOIN_CONDITION)
+
+val nonEquiJoinRex: RexNode = joinInfo.getRemaining(rexBuilder)
+val temporalJoinConditionExtractor = new TemporalJoinConditionExtractor(
Review comment:
We should add a comment that calling this extractor has side effects on
setting the optional attributes. This is not obvious at the first glance. Or we
move these variable into the extractor class as we do it in `PrimaryKeyVisitor`.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16614744#comment-16614744
]
ASF GitHub Bot commented on FLINK-9713:
---
twalthr commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r217641328
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/logical/LogicalCorrelateToTemporalTableJoinRule.scala
##
@@ -0,0 +1,208 @@
+/*
+ * 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.table.plan.rules.logical
+
+import org.apache.calcite.plan.RelOptRule.{any, none, operand, some}
+import org.apache.calcite.plan.{RelOptRule, RelOptRuleCall}
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.core.TableFunctionScan
+import org.apache.calcite.rel.logical.LogicalCorrelate
+import org.apache.calcite.rex._
+import org.apache.flink.table.api.{Table, Types, ValidationException}
+import org.apache.flink.table.calcite.FlinkTypeFactory.isProctimeIndicatorType
+import org.apache.flink.table.expressions._
+import org.apache.flink.table.functions.TemporalTableFunction
+import org.apache.flink.table.functions.utils.TableSqlFunction
+import org.apache.flink.table.plan.logical.rel.LogicalTemporalTableJoin
+import org.apache.flink.table.plan.util.RexDefaultVisitor
+
+class LogicalCorrelateToTemporalTableJoinRule
+ extends RelOptRule(
+operand(classOf[LogicalCorrelate],
+ some(
+operand(classOf[RelNode], any()),
+operand(classOf[TableFunctionScan], none(,
+"LogicalCorrelateToTemporalTableJoinRule") {
+
+ def extractNameFromTimeAttribute(timeAttribute: Expression): String = {
+timeAttribute match {
+ case RowtimeAttribute(expr: TemporalTableReference) =>
+expr.name
+ case ProctimeAttribute(expr: TemporalTableReference) =>
+expr.name
+ case TemporalTableReference(name, _) =>
+name
+ case ResolvedFieldReference(name, _)
+if timeAttribute.resultType == Types.LONG ||
+ timeAttribute.resultType == Types.SQL_TIMESTAMP =>
+name
+ case _ => throw new ValidationException(
+s"Invalid timeAttribute [${timeAttribute}] in TemporalTableFunction")
+}
+ }
+
+ override def onMatch(call: RelOptRuleCall): Unit = {
+val logicalCorrelate: LogicalCorrelate = call.rel(0)
+val leftNode: RelNode = call.rel(1)
+val rightTableFunctionScan: TableFunctionScan = call.rel(2)
+
+val cluster = logicalCorrelate.getCluster
+
+new GetTemporalTableFunctionCall(cluster.getRexBuilder, leftNode)
+ .visit(rightTableFunctionScan.getCall) match {
+ case Some(TemporalTableFunctionCall(rightTemporalTableFunction,
leftTimeAttribute)) =>
+val underlyingHistoryTable: Table =
rightTemporalTableFunction.getUnderlyingHistoryTable
+val relBuilder = this.relBuilderFactory.create(
+ cluster,
+ underlyingHistoryTable.relBuilder.getRelOptSchema)
+val rexBuilder = cluster.getRexBuilder
+
+val rightNode: RelNode =
underlyingHistoryTable.logicalPlan.toRelNode(relBuilder)
+
+val rightTimeIndicatorExpression = createRightExpression(
+ rexBuilder,
+ leftNode,
+ rightNode,
+
extractNameFromTimeAttribute(rightTemporalTableFunction.getTimeAttribute))
+
+val rightPrimaryKeyExpression = createRightExpression(
+ rexBuilder,
+ leftNode,
+ rightNode,
+ rightTemporalTableFunction.getPrimaryKey)
+
+relBuilder.push(
+ if
(isProctimeIndicatorType(rightTemporalTableFunction.getTimeAttribute.resultType))
{
+LogicalTemporalTableJoin.create(
+ rexBuilder,
+ cluster,
+ logicalCorrelate.getTraitSet,
+ leftNode,
+ rightNode,
+ leftTimeAttribute,
+ rightPrimaryKeyExpression)
+ }
+ else {
+LogicalTemporalTableJoin.create(
+ rexBuilder,
+ cluster,
+
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16614756#comment-16614756
]
ASF GitHub Bot commented on FLINK-9713:
---
twalthr commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r217688365
##
File path:
flink-libraries/flink-table/src/test/scala/org/apache/flink/table/runtime/harness/TemporalJoinHarnessTest.scala
##
@@ -0,0 +1,458 @@
+/*
+ * 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.table.runtime.harness
+
+import java.util.concurrent.ConcurrentLinkedQueue
+
+import org.apache.calcite.rel.core.{JoinInfo, JoinRelType}
+import org.apache.calcite.rex.{RexBuilder, RexNode}
+import org.apache.calcite.sql.fun.SqlStdOperatorTable
+import org.apache.calcite.util.ImmutableIntList
+import org.apache.flink.api.common.time.Time
+import org.apache.flink.api.common.typeinfo.{BasicTypeInfo, TypeInformation}
+import org.apache.flink.api.java.functions.KeySelector
+import org.apache.flink.api.java.typeutils.RowTypeInfo
+import org.apache.flink.streaming.api.functions.co.CoProcessFunction
+import org.apache.flink.streaming.api.operators.co.KeyedCoProcessOperator
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord
+import org.apache.flink.streaming.util.KeyedTwoInputStreamOperatorTestHarness
+import org.apache.flink.table.api.{TableConfig, Types, ValidationException}
+import org.apache.flink.table.calcite.{FlinkTypeFactory, FlinkTypeSystem}
+import org.apache.flink.table.plan.logical.rel.LogicalTemporalTableJoin
+import
org.apache.flink.table.plan.logical.rel.LogicalTemporalTableJoin.TEMPORAL_JOIN_CONDITION
+import
org.apache.flink.table.plan.nodes.datastream.DataStreamTemporalJoinToCoProcessTranslator
+import org.apache.flink.table.plan.schema.RowSchema
+import org.apache.flink.table.runtime.CRowKeySelector
+import
org.apache.flink.table.runtime.harness.HarnessTestBase.{RowResultSortComparator,
TestStreamQueryConfig}
+import org.apache.flink.table.runtime.types.CRow
+import org.apache.flink.table.typeutils.TimeIndicatorTypeInfo
+import org.hamcrest.Matchers.startsWith
+import org.junit.rules.ExpectedException
+import org.junit.{Rule, Test}
+
+class TemporalJoinHarnessTest extends HarnessTestBase {
+ // used for accurate exception information checking.
+ val expectedException = ExpectedException.none()
+
+ @Rule
+ def thrown = expectedException
+
+ private val typeFactory = new FlinkTypeFactory(new FlinkTypeSystem)
+
+ private val tableConfig = new TableConfig
+
+ private val queryConfig =
+new TestStreamQueryConfig(Time.milliseconds(2), Time.milliseconds(4))
+
+ private val ORDERS_KEY = "o_currency"
+
+ private val ORDERS_PROCTIME = "o_proctime"
+
+ private val RATES_KEY = "r_currency"
+
+ private val ordersRowtimeType = new RowTypeInfo(
+Array[TypeInformation[_]](
+ Types.LONG,
+ Types.STRING,
+ TimeIndicatorTypeInfo.ROWTIME_INDICATOR),
+Array("o_amount", ORDERS_KEY, "o_rowtime"))
+
+ private val ordersProctimeType = new RowTypeInfo(
+Array[TypeInformation[_]](
+ Types.LONG,
+ Types.STRING,
+ TimeIndicatorTypeInfo.PROCTIME_INDICATOR),
+Array("o_amount", ORDERS_KEY, ORDERS_PROCTIME))
+
+ private val ratesRowtimeType = new RowTypeInfo(
+Array[TypeInformation[_]](
+ Types.STRING,
+ Types.LONG,
+ TimeIndicatorTypeInfo.ROWTIME_INDICATOR),
+Array(RATES_KEY, "r_rate", "r_rowtime"))
+
+ private val ratesProctimeType = new RowTypeInfo(
+Array[TypeInformation[_]](
+ Types.STRING,
+ Types.LONG,
+ TimeIndicatorTypeInfo.PROCTIME_INDICATOR),
+Array(RATES_KEY, "r_rate", "r_proctime"))
+
+ private val joinRowtimeType = new RowTypeInfo(
+ordersRowtimeType.getFieldTypes ++ ratesRowtimeType.getFieldTypes,
+ordersRowtimeType.getFieldNames ++ ratesRowtimeType.getFieldNames)
+
+ private val rexBuilder = new RexBuilder(typeFactory)
+
+ @Test
+ def testProctime() {
+val testHarness = createTestHarness(new
OrdersRatesProctimeTemporalJoinInfo)
+
+
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16614757#comment-16614757
]
ASF GitHub Bot commented on FLINK-9713:
---
twalthr commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r217645653
##
File path:
flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/sql/TemporalTableJoinTest.scala
##
@@ -0,0 +1,130 @@
+/*
+ * 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.table.api.stream.sql
+
+import java.sql.Timestamp
+
+import org.apache.flink.api.scala._
+import org.apache.flink.table.api.TableException
+import org.apache.flink.table.api.scala._
+import org.apache.flink.table.api.stream.table.TemporalTableJoinTest._
+import org.apache.flink.table.utils._
+import org.hamcrest.Matchers.startsWith
+import org.junit.Test
+
+class TemporalTableJoinTest extends TableTestBase {
+
+ val util: TableTestUtil = streamTestUtil()
+
+ val orders = util.addTable[(Long, String, Timestamp)](
+"Orders", 'o_amount, 'o_currency, 'o_rowtime.rowtime)
+
+ val ratesHistory = util.addTable[(String, Int, Timestamp)](
+"RatesHistory", 'currency, 'rate, 'rowtime.rowtime)
+
+ val rates = util.addFunction(
+"Rates",
+ratesHistory.createTemporalTableFunction('rowtime.rowtime, 'currency))
Review comment:
Remove `.rowtime` here. This should also not be allowed. We should check why
no exception is thrown.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16614751#comment-16614751
]
ASF GitHub Bot commented on FLINK-9713:
---
twalthr commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r217364412
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/StreamTableEnvironment.scala
##
@@ -803,70 +803,37 @@ abstract class StreamTableEnvironment(
* @return The optimized [[RelNode]] tree
*/
private[flink] def optimize(relNode: RelNode, updatesAsRetraction: Boolean):
RelNode = {
-
-// 0. convert sub-queries before query decorrelation
-val convSubQueryPlan = runHepPlanner(
- HepMatchOrder.BOTTOM_UP, FlinkRuleSets.TABLE_SUBQUERY_RULES, relNode,
relNode.getTraitSet)
-
-// 0. convert table references
-val fullRelNode = runHepPlanner(
- HepMatchOrder.BOTTOM_UP,
- FlinkRuleSets.TABLE_REF_RULES,
- convSubQueryPlan,
- relNode.getTraitSet)
-
-// 1. decorrelate
+val convSubQueryPlan = optimizeConvertSubQueries(relNode)
+val temporalTableJoinPlan = optimizeConvertToTemporalJoin(convSubQueryPlan)
+val fullRelNode = optimizeConvertTableReferences(temporalTableJoinPlan)
Review comment:
nit: `fullRelNode` -> `fullPlan`
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16614754#comment-16614754
]
ASF GitHub Bot commented on FLINK-9713:
---
twalthr commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r217388272
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/calcite/RelTimeIndicatorConverter.scala
##
@@ -400,6 +411,10 @@ class RexTimeIndicatorMaterializer(
// materialize operands with time indicators
val materializedOperands = updatedCall.getOperator match {
+ case tableFunction: TableSqlFunction
Review comment:
Is this change still necessary? No test is failing if I remove it.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16614752#comment-16614752
]
ASF GitHub Bot commented on FLINK-9713:
---
twalthr commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r217639462
##
File path:
flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/table/TemporalTableJoinTest.scala
##
@@ -0,0 +1,276 @@
+/*
+ * 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.table.api.stream.table
+
+import java.sql.Timestamp
+
+import org.apache.flink.api.scala._
+import org.apache.flink.table.api.{TableSchema, ValidationException}
+import org.apache.flink.table.api.scala._
+import org.apache.flink.table.api.stream.table.TemporalTableJoinTest._
+import org.apache.flink.table.expressions.TemporalTableReference
+import org.apache.flink.table.functions.TemporalTableFunction
+import org.apache.flink.table.plan.logical.rel.LogicalTemporalTableJoin._
+import org.apache.flink.table.utils.TableTestUtil._
+import org.apache.flink.table.utils._
+import org.hamcrest.Matchers.startsWith
+import org.junit.Assert.{assertArrayEquals, assertEquals, assertTrue}
+import org.junit.Test
+
+class TemporalTableJoinTest extends TableTestBase {
+
+ val util: TableTestUtil = streamTestUtil()
+
+ val orders = util.addTable[(Long, String, Timestamp)](
+"Orders", 'o_amount, 'o_currency, 'o_rowtime.rowtime)
+
+ val ratesHistory = util.addTable[(String, Int, Timestamp)](
+"RatesHistory", 'currency, 'rate, 'rowtime.rowtime)
+
+ val rates = util.addFunction(
+"Rates",
+ratesHistory.createTemporalTableFunction('rowtime.rowtime, 'currency))
+
+ val proctimeOrders = util.addTable[(Long, String)](
+"ProctimeOrders", 'o_amount, 'o_currency, 'o_proctime.proctime)
+
+ val proctimeRatesHistory = util.addTable[(String, Int)](
+"ProctimeRatesHistory", 'currency, 'rate, 'proctime.proctime)
+
+ val proctimeRates =
proctimeRatesHistory.createTemporalTableFunction('proctime, 'currency)
+
+ @Test
+ def testSimpleJoin(): Unit = {
+val result = orders
+ .join(rates('o_rowtime), "currency = o_currency")
+ .select("o_amount * rate").as("rate")
+
+util.verifyTable(result, getExpectedSimpleJoinPlan())
+ }
+
+ @Test
+ def testSimpleProctimeJoin(): Unit = {
+val result = proctimeOrders
+ .join(proctimeRates('o_proctime), "currency = o_currency")
+ .select("o_amount * rate").as("rate")
+
+util.verifyTable(result, getExpectedSimpleProctimeJoinPlan())
+ }
+
+ /**
+* Test versioned joins with more complicated query.
+* Important thing here is that we have complex OR join condition
+* and there are some columns that are not being used (are being pruned).
+*/
+ @Test
+ def testComplexJoin(): Unit = {
+val util = streamTestUtil()
+val thirdTable = util.addTable[(String, Int)]("ThirdTable", 't3_comment,
't3_secondary_key)
+val orders = util.addTable[(Timestamp, String, Long, String, Int)](
+ "Orders", 'o_rowtime.rowtime, 'o_comment, 'o_amount, 'o_currency,
'o_secondary_key)
+
+val ratesHistory = util.addTable[(Timestamp, String, String, Int, Int)](
+ "RatesHistory", 'rowtime.rowtime, 'comment, 'currency, 'rate,
'secondary_key)
+val rates = ratesHistory.createTemporalTableFunction('rowtime.rowtime,
'currency)
+util.addFunction("Rates", rates)
+
+val result = orders
+ .join(rates('o_rowtime))
+ .filter('currency === 'o_currency || 'secondary_key === 'o_secondary_key)
+ .select('o_amount * 'rate, 'secondary_key).as('rate, 'secondary_key)
+ .join(thirdTable, 't3_secondary_key === 'secondary_key)
+
+util.verifyTable(result, getExpectedComplexJoinPlan())
+ }
+
+ @Test
+ def testTemporalTableFunctionOnTopOfQuery(): Unit = {
+val filteredRatesHistory = ratesHistory
+ .filter('rate > 100)
+ .select('currency, 'rate * 2, 'rowtime)
+ .as('currency, 'rate, 'rowtime)
+
+val filteredRates = util.addFunction(
+ "FilteredRates",
+
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16614746#comment-16614746
]
ASF GitHub Bot commented on FLINK-9713:
---
twalthr commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r217393776
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/calcite/RelTimeIndicatorConverter.scala
##
@@ -400,6 +411,10 @@ class RexTimeIndicatorMaterializer(
// materialize operands with time indicators
val materializedOperands = updatedCall.getOperator match {
+ case tableFunction: TableSqlFunction
Review comment:
Additionally, for the future event-time case, add a dedicated test case to
`TimeAttributesITCase`.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16614745#comment-16614745
]
ASF GitHub Bot commented on FLINK-9713:
---
twalthr commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r217370821
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/table.scala
##
@@ -155,6 +156,50 @@ class Table(
select(withResolvedAggFunctionCall: _*)
}
+ /**
+* Creates [[TemporalTableFunction]] backed up by this table as a history
table.
+*
+* @param timeIndicator field for the [[TemporalTableFunction]]. Must
points to a time indicator
+* @param primaryKeyfield for the [[TemporalTableFunction]].
+* @return [[TemporalTableFunction]]
+*/
+ def createTemporalTableFunction(
+ timeIndicator: String,
+ primaryKey: String): TemporalTableFunction = {
+createTemporalTableFunction(
+ ExpressionParser.parseExpression(timeIndicator),
+ ExpressionParser.parseExpression(primaryKey))
+ }
+
+ /**
+* Creates [[TemporalTableFunction]] backed up by this table as a history
table.
+*
+* @param timeIndicator field for the [[TemporalTableFunction]]. Must
points to a time indicator
+* @param primaryKeyfield for the [[TemporalTableFunction]].
+* @return [[TemporalTableFunction]]
+*/
+ def createTemporalTableFunction(
+ timeIndicator: Expression,
+ primaryKey: Expression): TemporalTableFunction = {
+val temporalTable = new TemporalTable(timeIndicator, primaryKey,
logicalPlan)
Review comment:
remove `new` as it is a case class
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16614731#comment-16614731
]
ASF GitHub Bot commented on FLINK-9713:
---
twalthr commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r217397054
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/table.scala
##
@@ -155,6 +156,50 @@ class Table(
select(withResolvedAggFunctionCall: _*)
}
+ /**
+* Creates [[TemporalTableFunction]] backed up by this table as a history
table.
+*
+* @param timeIndicator field for the [[TemporalTableFunction]]. Must
points to a time indicator
Review comment:
We should also add a comment that batch is not supported.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16614740#comment-16614740
]
ASF GitHub Bot commented on FLINK-9713:
---
twalthr commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r217370389
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/table.scala
##
@@ -155,6 +156,50 @@ class Table(
select(withResolvedAggFunctionCall: _*)
}
+ /**
+* Creates [[TemporalTableFunction]] backed up by this table as a history
table.
+*
+* @param timeIndicator field for the [[TemporalTableFunction]]. Must
points to a time indicator
+* @param primaryKeyfield for the [[TemporalTableFunction]].
+* @return [[TemporalTableFunction]]
+*/
+ def createTemporalTableFunction(
+ timeIndicator: String,
+ primaryKey: String): TemporalTableFunction = {
+createTemporalTableFunction(
+ ExpressionParser.parseExpression(timeIndicator),
+ ExpressionParser.parseExpression(primaryKey))
+ }
+
+ /**
+* Creates [[TemporalTableFunction]] backed up by this table as a history
table.
+*
+* @param timeIndicator field for the [[TemporalTableFunction]]. Must
points to a time indicator
+* @param primaryKeyfield for the [[TemporalTableFunction]].
+* @return [[TemporalTableFunction]]
+*/
+ def createTemporalTableFunction(
Review comment:
Some comment as above.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16614727#comment-16614727
]
ASF GitHub Bot commented on FLINK-9713:
---
twalthr commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r217638668
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/expressions/fieldExpression.scala
##
@@ -139,6 +139,23 @@ case class WindowReference(name: String, tpe:
Option[TypeInformation[_]] = None)
override def toString: String = s"'$name"
}
+case class TemporalTableReference(name: String, typeInformation:
TypeInformation[_])
Review comment:
Why do we need this class? Window references and table references are
necessary to call `window.rowtime` (extract the rowtime out of a window) or
`table.in(1, 2, 3)` but when do we need to references the temporal table?
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16614730#comment-16614730
]
ASF GitHub Bot commented on FLINK-9713:
---
twalthr commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r217412622
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamTemporalJoinToCoProcessTranslator.scala
##
@@ -0,0 +1,198 @@
+/*
+ * 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.table.plan.nodes.datastream
+
+import org.apache.calcite.rel.core.{JoinInfo, JoinRelType}
+import org.apache.calcite.rex._
+import org.apache.flink.api.common.functions.{FlatJoinFunction, MapFunction}
+import org.apache.flink.api.common.typeinfo.TypeInformation
+import org.apache.flink.streaming.api.functions.co.CoProcessFunction
+import org.apache.flink.table.api.{StreamQueryConfig, TableConfig,
TableException, ValidationException}
+import org.apache.flink.table.calcite.FlinkTypeFactory._
+import org.apache.flink.table.codegen.{FunctionCodeGenerator,
GeneratedFunction}
+import org.apache.flink.table.plan.logical.rel.LogicalTemporalTableJoin._
+import org.apache.flink.table.plan.schema.RowSchema
+import org.apache.flink.table.plan.util.RexDefaultVisitor
+import org.apache.flink.table.runtime.join.TemporalJoin
+import org.apache.flink.table.runtime.types.{CRow, CRowTypeInfo}
+import org.apache.flink.types.Row
+import org.apache.flink.util.Preconditions.checkState
+
+class DataStreamTemporalJoinToCoProcessTranslator(
+textualRepresentation: String,
+config: TableConfig,
+returnType: TypeInformation[Row],
+leftSchema: RowSchema,
+rightSchema: RowSchema,
+joinInfo: JoinInfo,
+rexBuilder: RexBuilder)
+ extends DataStreamJoinToCoProcessTranslator(
+config,
+returnType,
+leftSchema,
+rightSchema,
+joinInfo,
+rexBuilder) {
+
+ var leftTimeAttribute: Option[RexNode] = None
+
+ var rightTimeAttribute: Option[RexNode] = None
+
+ var rightPrimaryKeyExpression: Option[RexNode] = None
+
+ override val nonEquiJoinPredicates: Option[RexNode] =
extractTemporalJoinCondition()
+
+ def extractTemporalJoinCondition(): Option[RexNode] = {
+checkState(
+ !joinInfo.isEqui,
+ "Missing %s in join condition",
+ TEMPORAL_JOIN_CONDITION)
+
+val nonEquiJoinRex: RexNode = joinInfo.getRemaining(rexBuilder)
+val temporalJoinConditionExtractor = new TemporalJoinConditionExtractor(
+ nonEquiJoinRex.toString)
+
+Some(temporalJoinConditionExtractor.apply(nonEquiJoinRex))
+ }
+
+ override protected def createCoProcessFunction(
+ joinType: JoinRelType,
+ queryConfig: StreamQueryConfig,
+ joinFunction: GeneratedFunction[FlatJoinFunction[Row, Row, Row], Row])
+: CoProcessFunction[CRow, CRow, CRow] = {
+
+checkState(
+ leftTimeAttribute.isDefined &&
+rightPrimaryKeyExpression.isDefined,
+ "Missing %s in join condition",
+ TEMPORAL_JOIN_CONDITION)
+
+if (rightTimeAttribute.isDefined) {
+ throw new ValidationException(
+s"Currently only proctime temporal joins are supported in
[${textualRepresentation}]")
+}
+
+joinType match {
+ case JoinRelType.INNER =>
+new TemporalJoin(
+ leftSchema.typeInfo,
+ rightSchema.typeInfo,
+ joinFunction.name,
+ joinFunction.code,
+ queryConfig)
+ case _ =>
+ throw new ValidationException(
+ s"Only ${JoinRelType.INNER} temporal join is supported in
[${textualRepresentation}]")
+}
+ }
+
+ private class TemporalJoinConditionExtractor(
+ nonEquiJoinCondition: String)
+extends RexShuttle {
+
+override def visitCall(call: RexCall): RexNode = {
+ if (call.getOperator != TEMPORAL_JOIN_CONDITION) {
+return super.visitCall(call)
+ }
+
+ if (leftTimeAttribute.isDefined
+|| rightPrimaryKeyExpression.isDefined
+|| rightTimeAttribute.isDefined) {
+throw new
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16614742#comment-16614742
]
ASF GitHub Bot commented on FLINK-9713:
---
twalthr commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r217395231
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/TemporalTableFunction.scala
##
@@ -0,0 +1,75 @@
+/*
+ * 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.table.functions
+
+import java.sql.Timestamp
+
+import org.apache.flink.api.java.typeutils.RowTypeInfo
+import org.apache.flink.table.api.Table
+import org.apache.flink.table.expressions.Expression
+import org.apache.flink.types.Row
+
+/**
+ * Class representing temporal table function over some history table.
+ */
+class TemporalTableFunction private(
+@transient private val underlyingHistoryTable: Table,
+private val timeAttribute: Expression,
+private val primaryKey: String,
+private val resultType: RowTypeInfo)
+ extends TableFunction[Row] {
+
+ def eval(row: Timestamp): Unit = {
Review comment:
I would add a special case to the table function checking logic. Otherwise
this method is visible to users, after creating a table function.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16614722#comment-16614722
]
ASF GitHub Bot commented on FLINK-9713:
---
twalthr commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r217686456
##
File path:
flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/table/validation/TemporalTableJoinValidationTest.scala
##
@@ -0,0 +1,124 @@
+/*
+ * 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.table.api.stream.table.validation
+
+import java.sql.Timestamp
+
+import org.apache.flink.api.scala._
+import org.apache.flink.table.api.{TableException, ValidationException}
+import org.apache.flink.table.api.scala._
+import org.apache.flink.table.utils._
+import org.junit.Test
+
+class TemporalTableJoinValidationTest extends TableTestBase {
+
+ val util: TableTestUtil = streamTestUtil()
+
+ val orders = util.addTable[(Long, String, Timestamp)](
+"Orders", 'o_amount, 'o_currency, 'o_rowtime.rowtime)
+
+ val ordersProctime = util.addTable[(Long, String)](
+"OrdersProctime", 'o_amount, 'o_currency, 'o_rowtime.proctime)
+
+ val ordersWithoutTimeAttribute = util.addTable[(Long, String, Timestamp)](
+"OrdersWithoutTimeAttribute", 'o_amount, 'o_currency, 'o_rowtime)
+
+ val ratesHistory = util.addTable[(String, Int, Timestamp)](
+"RatesHistory", 'currency, 'rate, 'rowtime.rowtime)
+
+ val ratesHistoryWithoutTimeAttribute = util.addTable[(String, Int,
Timestamp)](
+"ratesHistoryWithoutTimeAttribute", 'currency, 'rate, 'rowtime)
+
+ @Test
+ def testInvalidFieldReference(): Unit = {
+expectedException.expect(classOf[ValidationException])
+expectedException.expectMessage("Cannot resolve field [foobar]")
+
+ratesHistory.createTemporalTableFunction('rowtime.rowtime, 'foobar)
Review comment:
remove `.rowtime`
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16614737#comment-16614737
]
ASF GitHub Bot commented on FLINK-9713:
---
twalthr commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r217690801
##
File path:
flink-libraries/flink-table/src/test/scala/org/apache/flink/table/runtime/stream/sql/TemporalJoinITCase.scala
##
@@ -0,0 +1,135 @@
+/*
+ * 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.table.runtime.stream.sql
+
+import java.sql.Timestamp
+
+import org.apache.flink.api.scala._
+import org.apache.flink.streaming.api.TimeCharacteristic
+import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
+import org.apache.flink.table.api.{TableEnvironment, TableException}
+import org.apache.flink.table.api.scala._
+import org.apache.flink.table.runtime.utils.{StreamITCase,
StreamingWithStateTestBase}
+import org.apache.flink.types.Row
+import org.junit._
+import org.junit.rules.ExpectedException
+
+import scala.collection.mutable
+
+class TemporalJoinITCase extends StreamingWithStateTestBase {
+ val expectedException = ExpectedException.none()
Review comment:
Move this to the test base too. All test bases should be able to deal with
exceptions.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16614723#comment-16614723
]
ASF GitHub Bot commented on FLINK-9713:
---
twalthr commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r217382202
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/calcite/RelTimeIndicatorConverter.scala
##
@@ -166,9 +172,14 @@ class RelTimeIndicatorConverter(rexBuilder: RexBuilder)
extends RelShuttle {
val right = join.getRight.accept(this)
LogicalJoin.create(left, right, join.getCondition, join.getVariablesSet,
join.getJoinType)
-
}
+ def visit(temporalJoin: LogicalTemporalTableJoin): RelNode = {
+val left = temporalJoin.getLeft.accept(this)
+val right = temporalJoin.getRight.accept(this)
+
+temporalJoin.copy(temporalJoin.getTraitSet, List(left, right))
Review comment:
I guess we need more logic here. At first I also thought that we don't need
special treatment of time attributes here. However, if we have a
processing-time temporal join we need to materialize all rowtime attributes of
both inputs. So this method needs to look at the time attributes defined by the
join condition. What do you think?
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16614743#comment-16614743
]
ASF GitHub Bot commented on FLINK-9713:
---
twalthr commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r217365862
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/BatchTableEnvironment.scala
##
@@ -448,48 +448,13 @@ abstract class BatchTableEnvironment(
* @return The optimized [[RelNode]] tree
*/
private[flink] def optimize(relNode: RelNode): RelNode = {
-
-// 0. convert sub-queries before query decorrelation
-val convSubQueryPlan = runHepPlanner(
- HepMatchOrder.BOTTOM_UP, FlinkRuleSets.TABLE_SUBQUERY_RULES, relNode,
relNode.getTraitSet)
-
-// 0. convert table references
-val fullRelNode = runHepPlanner(
- HepMatchOrder.BOTTOM_UP,
- FlinkRuleSets.TABLE_REF_RULES,
- convSubQueryPlan,
- relNode.getTraitSet)
-
-// 1. decorrelate
+val convSubQueryPlan = optimizeConvertSubQueries(relNode)
Review comment:
This again adds a lot of public methods to the table environments in Java.
Can we move the entire optimize method and sub-methods to a
`TableEnvironmentUtils` class?
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16614736#comment-16614736
]
ASF GitHub Bot commented on FLINK-9713:
---
twalthr commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r217401379
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/logical/rel/LogicalTemporalTableJoin.scala
##
@@ -0,0 +1,173 @@
+/*
+ * 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.table.plan.logical.rel
+
+import java.util.Collections
+
+import org.apache.calcite.plan.{RelOptCluster, RelTraitSet}
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.core._
+import org.apache.calcite.rex.{RexBuilder, RexNode}
+import org.apache.calcite.sql.`type`.{OperandTypes, ReturnTypes}
+import org.apache.calcite.sql.{SqlFunction, SqlFunctionCategory, SqlKind}
+import org.apache.flink.util.Preconditions.checkArgument
+
+/**
+ * Represents a join between a table and
[[org.apache.flink.table.functions.TemporalTableFunction]]
+ *
+ * @param cluster
+ * @param traitSet
+ * @param left stream
+ * @param right table scan of underlying
+ * [[org.apache.flink.table.functions.TemporalTableFunction]]
+ * @param condition must contain
[[LogicalTemporalTableJoin#TEMPORAL_JOIN_CONDITION]] with
+ * correctly defined references to rightTimeAttribute,
+ * rightPrimaryKeyExpression and leftTimeAttribute. We can
not implement
+ * those references as separate fields, because of problems
with Calcite's
+ * optimization rules like projections push downs, column
+ * pruning/renaming/reordering, etc. Later
rightTimeAttribute,
+ * rightPrimaryKeyExpression and leftTimeAttribute will be
extracted from
+ * the condition.
+ */
+class LogicalTemporalTableJoin private(
+cluster: RelOptCluster,
+traitSet: RelTraitSet,
+left: RelNode,
+right: RelNode,
+condition: RexNode)
+ extends Join(
+cluster,
+traitSet,
+left,
+right,
+condition,
+Collections.emptySet().asInstanceOf[java.util.Set[CorrelationId]],
+JoinRelType.INNER) {
+
+ override def copy(
+ traitSet: RelTraitSet,
+ condition: RexNode,
+ left: RelNode,
+ right: RelNode,
+ joinType: JoinRelType,
+ semiJoinDone: Boolean): LogicalTemporalTableJoin = {
+checkArgument(joinType == this.getJoinType,
+ "Can not change join type".asInstanceOf[Object])
+checkArgument(semiJoinDone == this.isSemiJoinDone,
+ "Can not change semiJoinDone".asInstanceOf[Object])
+new LogicalTemporalTableJoin(
+ cluster,
+ traitSet,
+ left,
+ right,
+ condition)
+ }
+}
+
+object LogicalTemporalTableJoin {
+ /**
+* See [[LogicalTemporalTableJoin#condition]]
+*/
+ val TEMPORAL_JOIN_CONDITION = new SqlFunction(
+"__TEMPORAL_JOIN_CONDITION",
+SqlKind.OTHER_FUNCTION,
+ReturnTypes.BOOLEAN_NOT_NULL,
+null,
+OperandTypes.or(
+ OperandTypes.sequence(
+"'(LEFT_TIME_ATTRIBUTE, RIGHT_TIME_ATTRIBUTE, PRIMARY_KEY)'",
+OperandTypes.DATETIME,
+OperandTypes.DATETIME,
+OperandTypes.ANY),
+ OperandTypes.sequence(
+"'(LEFT_TIME_ATTRIBUTE, PRIMARY_KEY)'",
+OperandTypes.DATETIME,
+OperandTypes.ANY)),
+SqlFunctionCategory.SYSTEM)
+
+ def makeRowTimeTemporalJoinConditionCall(
+ rexBuilder: RexBuilder,
+ leftTimeAttribute: RexNode,
+ rightTimeAttribute: RexNode,
+ rightPrimaryKeyExpression: RexNode): RexNode = {
+rexBuilder.makeCall(
+ TEMPORAL_JOIN_CONDITION,
+ leftTimeAttribute,
+ rightTimeAttribute,
+ rightPrimaryKeyExpression)
+ }
+
+ def makeProcTimeTemporalJoinConditionCall(
+ rexBuilder: RexBuilder,
+ leftTimeAttribute: RexNode,
+ rightPrimaryKeyExpression: RexNode): RexNode = {
+rexBuilder.makeCall(
+
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16614733#comment-16614733
]
ASF GitHub Bot commented on FLINK-9713:
---
twalthr commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r217633369
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/logical/LogicalCorrelateToTemporalTableJoinRule.scala
##
@@ -0,0 +1,208 @@
+/*
+ * 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.table.plan.rules.logical
+
+import org.apache.calcite.plan.RelOptRule.{any, none, operand, some}
+import org.apache.calcite.plan.{RelOptRule, RelOptRuleCall}
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.core.TableFunctionScan
+import org.apache.calcite.rel.logical.LogicalCorrelate
+import org.apache.calcite.rex._
+import org.apache.flink.table.api.{Table, Types, ValidationException}
+import org.apache.flink.table.calcite.FlinkTypeFactory.isProctimeIndicatorType
+import org.apache.flink.table.expressions._
+import org.apache.flink.table.functions.TemporalTableFunction
+import org.apache.flink.table.functions.utils.TableSqlFunction
+import org.apache.flink.table.plan.logical.rel.LogicalTemporalTableJoin
+import org.apache.flink.table.plan.util.RexDefaultVisitor
+
+class LogicalCorrelateToTemporalTableJoinRule
+ extends RelOptRule(
+operand(classOf[LogicalCorrelate],
+ some(
+operand(classOf[RelNode], any()),
+operand(classOf[TableFunctionScan], none(,
+"LogicalCorrelateToTemporalTableJoinRule") {
+
+ def extractNameFromTimeAttribute(timeAttribute: Expression): String = {
+timeAttribute match {
+ case RowtimeAttribute(expr: TemporalTableReference) =>
Review comment:
Why do we need a distinction here? Shouldn't a field `timeAttribute:
Expression` just contain what it says? A time attribute which is resolved field
reference with a special data type.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16614729#comment-16614729
]
ASF GitHub Bot commented on FLINK-9713:
---
twalthr commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r217366797
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/TableEnvironment.scala
##
@@ -234,6 +235,54 @@ abstract class TableEnvironment(val config: TableConfig) {
*/
protected def getBuiltInPhysicalOptRuleSet: RuleSet
+ protected def optimizeConvertSubQueries(relNode: RelNode): RelNode = {
+runHepPlanner(
Review comment:
nit: new line for every parameter to be consistent with
`optimizeConvertTableReferences`.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16591638#comment-16591638
]
ASF GitHub Bot commented on FLINK-9713:
---
pnowojski commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r208833159
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/logical/LogicalCorrelateToVersionedJoinRule.scala
##
@@ -0,0 +1,243 @@
+/*
+ * 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.table.plan.rules.logical
+
+import java.util
+import java.util.Collections
+
+import org.apache.calcite.plan.RelOptRule.{any, none, operand, some}
+import org.apache.calcite.plan.{RelOptRule, RelOptRuleCall}
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.core.TableFunctionScan
+import org.apache.calcite.rel.logical.LogicalCorrelate
+import org.apache.calcite.rex._
+import org.apache.calcite.sql.fun.SqlCastFunction
+import org.apache.flink.table.api.{Table, ValidationException}
+import org.apache.flink.table.functions.TableVersionFunction
+import org.apache.flink.table.functions.sql.ProctimeSqlFunction
+import org.apache.flink.table.functions.utils.TableSqlFunction
+import org.apache.flink.table.plan.logical.rel.LogicalVersionedJoin
+import org.apache.flink.table.plan.schema.TimeIndicatorRelDataType
+import org.apache.flink.table.plan.util.RexDefaultVisitor
+import org.apache.flink.table.typeutils.TypeStringUtils
+import org.apache.flink.util.Preconditions.checkState
+
+class LogicalCorrelateToVersionedJoinRule
+ extends RelOptRule(
+operand(classOf[LogicalCorrelate],
+ some(
+operand(classOf[RelNode], any()),
+operand(classOf[TableFunctionScan], none(,
+"LogicalCorrelateToVersionedJoinRule") {
+
+ override def onMatch(call: RelOptRuleCall): Unit = {
+val logicalCorrelate: LogicalCorrelate = call.rel(0)
+val leftNode: RelNode = call.rel(1)
+val rightTableFunctionScan: TableFunctionScan = call.rel(2)
+
+val cluster = logicalCorrelate.getCluster
+
+new GetTableVersionFunctionCall(cluster.getRexBuilder, leftNode)
+ .visit(rightTableFunctionScan.getCall) match {
+ case Some(rightVersionedTableCall) =>
+val underlyingTableHistory: Table =
rightVersionedTableCall.tableVersionFunction.table
+val relBuilder = this.relBuilderFactory.create(
+ cluster,
+ underlyingTableHistory.relBuilder.getRelOptSchema)
+val rexBuilder = cluster.getRexBuilder
+
+val rightNode: RelNode =
underlyingTableHistory.logicalPlan.toRelNode(relBuilder)
+
+val rightVersionExpression = createRightExpression(
+ rexBuilder,
+ leftNode,
+ rightNode,
+ rightVersionedTableCall.tableVersionFunction.versionField)
+
+val rightPrimaryKeyExpression = createRightExpression(
+ rexBuilder,
+ leftNode,
+ rightNode,
+ rightVersionedTableCall.tableVersionFunction.primaryKey)
+
+relBuilder.push(
+ if (rightVersionedTableCall.tableVersionFunction.isProctime) {
+LogicalVersionedJoin.create(
+ rexBuilder,
+ cluster,
+ logicalCorrelate.getTraitSet,
+ leftNode,
+ rightNode,
+ rightVersionedTableCall.versionExpression,
+ rightPrimaryKeyExpression)
+ }
+ else {
+LogicalVersionedJoin.create(
+ rexBuilder,
+ cluster,
+ logicalCorrelate.getTraitSet,
+ leftNode,
+ rightNode,
+ rightVersionedTableCall.versionExpression,
+ rightVersionExpression,
+ rightPrimaryKeyExpression)
+ })
+call.transformTo(relBuilder.build())
+ case None =>
+}
+ }
+
+ private def createRightExpression(
+ rexBuilder: RexBuilder,
+ leftNode: RelNode,
+ rightNode: RelNode,
+ field: String): RexNode = {
+val rightReferencesOffset =
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16591639#comment-16591639
]
ASF GitHub Bot commented on FLINK-9713:
---
pnowojski commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r212625803
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/join/VersionedJoin.scala
##
@@ -0,0 +1,93 @@
+/*
+ * 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.table.runtime.join
+
+import org.apache.flink.api.common.functions.FlatJoinFunction
+import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
+import org.apache.flink.api.common.typeinfo.TypeInformation
+import org.apache.flink.configuration.Configuration
+import org.apache.flink.streaming.api.functions.co.CoProcessFunction
+import org.apache.flink.table.api.StreamQueryConfig
+import org.apache.flink.table.codegen.Compiler
+import org.apache.flink.table.runtime.CRowWrappingCollector
+import org.apache.flink.table.runtime.types.CRow
+import org.apache.flink.table.typeutils.TypeCheckUtils._
+import org.apache.flink.table.util.Logging
+import org.apache.flink.types.Row
+import org.apache.flink.util.Collector
+
+class VersionedJoin(
+leftType: TypeInformation[Row],
+rightType: TypeInformation[Row],
+genJoinFuncName: String,
+genJoinFuncCode: String,
+queryConfig: StreamQueryConfig)
+ extends CoProcessFunction[CRow, CRow, CRow]
+ with Compiler[FlatJoinFunction[Row, Row, Row]]
+ with Logging {
+
+ validateEqualsHashCode("join", leftType)
+ validateEqualsHashCode("join", rightType)
+
+ protected var rightState: ValueState[Row] = _
+ protected var cRowWrapper: CRowWrappingCollector = _
+
+ protected var joinFunction: FlatJoinFunction[Row, Row, Row] = _
+
+ override def open(parameters: Configuration): Unit = {
+val clazz = compile(
+ getRuntimeContext.getUserCodeClassLoader,
+ genJoinFuncName,
+ genJoinFuncCode)
+
+joinFunction = clazz.newInstance()
+
+val rightStateDescriptor = new ValueStateDescriptor[Row]("right",
rightType)
Review comment:
Could we do it as a follow up task to minimize PR size?
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16590035#comment-16590035
]
ASF GitHub Bot commented on FLINK-9713:
---
pnowojski commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r212257550
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/TableVersionFunction.scala
##
@@ -0,0 +1,70 @@
+/*
+ * 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.table.functions
+
+import java.sql.Timestamp
+
+import org.apache.flink.api.common.typeinfo.TypeInformation
+import org.apache.flink.api.java.typeutils.RowTypeInfo
+import org.apache.flink.table.api.Table
+import org.apache.flink.types.Row
+
+/**
+ * Class represnting table version functions over some history table.
+ */
+class TableVersionFunction private(
+@transient private val _table: Table,
+private[flink] val versionField: String,
+private[flink] val primaryKey: String,
+private[flink] val resultType: RowTypeInfo,
+private[flink] val isProctime: Boolean)
+ extends TableFunction[Row] {
+
+ def eval(row: Timestamp): Unit = {
Review comment:
Without it I'm getting:
```
org.apache.flink.table.api.ValidationException: Function class
'org.apache.flink.table.functions.TemporalTableFunction' does not implement at
least one method named 'eval' which is public, not abstract and (in case of
table functions) not static.
```
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16574654#comment-16574654
]
ASF GitHub Bot commented on FLINK-9713:
---
twalthr commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r208876405
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/logical/LogicalCorrelateToVersionedJoinRule.scala
##
@@ -0,0 +1,243 @@
+/*
+ * 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.table.plan.rules.logical
+
+import java.util
+import java.util.Collections
+
+import org.apache.calcite.plan.RelOptRule.{any, none, operand, some}
+import org.apache.calcite.plan.{RelOptRule, RelOptRuleCall}
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.core.TableFunctionScan
+import org.apache.calcite.rel.logical.LogicalCorrelate
+import org.apache.calcite.rex._
+import org.apache.calcite.sql.fun.SqlCastFunction
+import org.apache.flink.table.api.{Table, ValidationException}
+import org.apache.flink.table.functions.TableVersionFunction
+import org.apache.flink.table.functions.sql.ProctimeSqlFunction
+import org.apache.flink.table.functions.utils.TableSqlFunction
+import org.apache.flink.table.plan.logical.rel.LogicalVersionedJoin
+import org.apache.flink.table.plan.schema.TimeIndicatorRelDataType
+import org.apache.flink.table.plan.util.RexDefaultVisitor
+import org.apache.flink.table.typeutils.TypeStringUtils
+import org.apache.flink.util.Preconditions.checkState
+
+class LogicalCorrelateToVersionedJoinRule
+ extends RelOptRule(
+operand(classOf[LogicalCorrelate],
+ some(
+operand(classOf[RelNode], any()),
+operand(classOf[TableFunctionScan], none(,
+"LogicalCorrelateToVersionedJoinRule") {
+
+ override def onMatch(call: RelOptRuleCall): Unit = {
+val logicalCorrelate: LogicalCorrelate = call.rel(0)
+val leftNode: RelNode = call.rel(1)
+val rightTableFunctionScan: TableFunctionScan = call.rel(2)
+
+val cluster = logicalCorrelate.getCluster
+
+new GetTableVersionFunctionCall(cluster.getRexBuilder, leftNode)
+ .visit(rightTableFunctionScan.getCall) match {
+ case Some(rightVersionedTableCall) =>
+val underlyingTableHistory: Table =
rightVersionedTableCall.tableVersionFunction.table
+val relBuilder = this.relBuilderFactory.create(
+ cluster,
+ underlyingTableHistory.relBuilder.getRelOptSchema)
+val rexBuilder = cluster.getRexBuilder
+
+val rightNode: RelNode =
underlyingTableHistory.logicalPlan.toRelNode(relBuilder)
Review comment:
Creating a logical plan and inserting it during a phase of the optimization
is dangerous. You would skip steps that need to be applied before. In our case
these steps are shown in
`org.apache.flink.table.api.StreamTableEnvironment#optimize`:
- convert sub-queries before query decorrelation
- convert table references
- decorrelate
- convert time indicators
It might be better to execute this rule in the "convert table references"
step.
Btw in this stage time indicators have not been materialized. So you can
also remove the `CAST` special case. Otherwise I would also recommend changes
in `org.apache.flink.table.calcite.RexTimeIndicatorMaterializer#visitCall`.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16574443#comment-16574443
]
ASF GitHub Bot commented on FLINK-9713:
---
pnowojski commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r208835295
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamJoin.scala
##
@@ -103,27 +97,42 @@ class DataStreamJoin(
tableEnv: StreamTableEnvironment,
queryConfig: StreamQueryConfig): DataStream[CRow] = {
-val config = tableEnv.getConfig
-val returnType = schema.typeInfo
-val keyPairs = joinInfo.pairs().toList
+validateKeyTypes()
-// get the equality keys
-val leftKeys = ArrayBuffer.empty[Int]
-val rightKeys = ArrayBuffer.empty[Int]
+val leftDataStream =
Review comment:
I would be against that. I'm depending on this change in this PR, so it
defines strict order of those two commits. Either I wouldn't publish the actual
versioned joins PR after this commit is merged, or the versioned joins PR would
look exactly the same as it looks now. Splitting into two PRs adds a lot of
overhead (in case of rebases or applying review changes you need to keep
updating multiple PRs) and room for a lot of problems with PRs going out of
sync. During reviewing it's also annoying, because you see the same code twice
and reviewers (same or different ones) are often (almost always?) making half
of the comments in one PR and half in the other.
Besides, this is already in a separate commit so it can be reviewed
separately. Maybe instead of reviewing PR all at once try reviewing it commit
by commit?
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16574434#comment-16574434
]
ASF GitHub Bot commented on FLINK-9713:
---
pnowojski commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r208833101
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/logical/LogicalCorrelateToVersionedJoinRule.scala
##
@@ -0,0 +1,243 @@
+/*
+ * 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.table.plan.rules.logical
+
+import java.util
+import java.util.Collections
+
+import org.apache.calcite.plan.RelOptRule.{any, none, operand, some}
+import org.apache.calcite.plan.{RelOptRule, RelOptRuleCall}
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.core.TableFunctionScan
+import org.apache.calcite.rel.logical.LogicalCorrelate
+import org.apache.calcite.rex._
+import org.apache.calcite.sql.fun.SqlCastFunction
+import org.apache.flink.table.api.{Table, ValidationException}
+import org.apache.flink.table.functions.TableVersionFunction
+import org.apache.flink.table.functions.sql.ProctimeSqlFunction
+import org.apache.flink.table.functions.utils.TableSqlFunction
+import org.apache.flink.table.plan.logical.rel.LogicalVersionedJoin
+import org.apache.flink.table.plan.schema.TimeIndicatorRelDataType
+import org.apache.flink.table.plan.util.RexDefaultVisitor
+import org.apache.flink.table.typeutils.TypeStringUtils
+import org.apache.flink.util.Preconditions.checkState
+
+class LogicalCorrelateToVersionedJoinRule
+ extends RelOptRule(
+operand(classOf[LogicalCorrelate],
+ some(
+operand(classOf[RelNode], any()),
+operand(classOf[TableFunctionScan], none(,
+"LogicalCorrelateToVersionedJoinRule") {
+
+ override def onMatch(call: RelOptRuleCall): Unit = {
+val logicalCorrelate: LogicalCorrelate = call.rel(0)
+val leftNode: RelNode = call.rel(1)
+val rightTableFunctionScan: TableFunctionScan = call.rel(2)
+
+val cluster = logicalCorrelate.getCluster
+
+new GetTableVersionFunctionCall(cluster.getRexBuilder, leftNode)
+ .visit(rightTableFunctionScan.getCall) match {
+ case Some(rightVersionedTableCall) =>
+val underlyingTableHistory: Table =
rightVersionedTableCall.tableVersionFunction.table
+val relBuilder = this.relBuilderFactory.create(
+ cluster,
+ underlyingTableHistory.relBuilder.getRelOptSchema)
+val rexBuilder = cluster.getRexBuilder
+
+val rightNode: RelNode =
underlyingTableHistory.logicalPlan.toRelNode(relBuilder)
+
+val rightVersionExpression = createRightExpression(
+ rexBuilder,
+ leftNode,
+ rightNode,
+ rightVersionedTableCall.tableVersionFunction.versionField)
+
+val rightPrimaryKeyExpression = createRightExpression(
+ rexBuilder,
+ leftNode,
+ rightNode,
+ rightVersionedTableCall.tableVersionFunction.primaryKey)
+
+relBuilder.push(
+ if (rightVersionedTableCall.tableVersionFunction.isProctime) {
+LogicalVersionedJoin.create(
+ rexBuilder,
+ cluster,
+ logicalCorrelate.getTraitSet,
+ leftNode,
+ rightNode,
+ rightVersionedTableCall.versionExpression,
+ rightPrimaryKeyExpression)
+ }
+ else {
+LogicalVersionedJoin.create(
+ rexBuilder,
+ cluster,
+ logicalCorrelate.getTraitSet,
+ leftNode,
+ rightNode,
+ rightVersionedTableCall.versionExpression,
+ rightVersionExpression,
+ rightPrimaryKeyExpression)
+ })
+call.transformTo(relBuilder.build())
+ case None =>
+}
+ }
+
+ private def createRightExpression(
+ rexBuilder: RexBuilder,
+ leftNode: RelNode,
+ rightNode: RelNode,
+ field: String): RexNode = {
+val rightReferencesOffset =
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16574440#comment-16574440
]
ASF GitHub Bot commented on FLINK-9713:
---
pnowojski commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r208835846
##
File path:
flink-libraries/flink-table/src/test/scala/org/apache/flink/table/runtime/harness/JoinHarnessTest.scala
##
@@ -167,28 +167,28 @@ class JoinHarnessTest extends HarnessTestBase {
testHarness.setProcessingTime(1)
testHarness.processElement1(new StreamRecord(
- CRow(Row.of(1L: JLong, "1a1"), change = true), 1))
+ CRow(1L: JLong, "1a1"), 1))
Review comment:
Again, I'm depending on this change in the following commit. What harm does
it do here?
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16574437#comment-16574437
]
ASF GitHub Bot commented on FLINK-9713:
---
pnowojski commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r208830206
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/calcite/FlinkRelBuilder.scala
##
@@ -46,6 +46,8 @@ class FlinkRelBuilder(
relOptCluster,
relOptSchema) {
+ def getRelOptSchema: RelOptSchema = relOptSchema
Review comment:
Yes, check for `getRelOptSchema` usage in this commit
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16574438#comment-16574438
]
ASF GitHub Bot commented on FLINK-9713:
---
pnowojski commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r208829502
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/table.scala
##
@@ -155,6 +158,90 @@ class Table(
select(withResolvedAggFunctionCall: _*)
}
+ /**
+* Creates table version function backed up by this table as a history
table.
+*
+* @param version field for the [[TableVersionFunction]]. Must points to a
time indicator
+* @param primaryKey field for the [[TableVersionFunction]].
+* @return [[TableVersionFunction]]
+*/
+ def createTableVersionFunction(
+ version: String,
+ primaryKey: String): TableVersionFunction = {
Review comment:
In first version me and @fhueske didn't want to focus on this and we agreed
to do this as a follow up task.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16574435#comment-16574435
]
ASF GitHub Bot commented on FLINK-9713:
---
pnowojski commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r208833159
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/logical/LogicalCorrelateToVersionedJoinRule.scala
##
@@ -0,0 +1,243 @@
+/*
+ * 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.table.plan.rules.logical
+
+import java.util
+import java.util.Collections
+
+import org.apache.calcite.plan.RelOptRule.{any, none, operand, some}
+import org.apache.calcite.plan.{RelOptRule, RelOptRuleCall}
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.core.TableFunctionScan
+import org.apache.calcite.rel.logical.LogicalCorrelate
+import org.apache.calcite.rex._
+import org.apache.calcite.sql.fun.SqlCastFunction
+import org.apache.flink.table.api.{Table, ValidationException}
+import org.apache.flink.table.functions.TableVersionFunction
+import org.apache.flink.table.functions.sql.ProctimeSqlFunction
+import org.apache.flink.table.functions.utils.TableSqlFunction
+import org.apache.flink.table.plan.logical.rel.LogicalVersionedJoin
+import org.apache.flink.table.plan.schema.TimeIndicatorRelDataType
+import org.apache.flink.table.plan.util.RexDefaultVisitor
+import org.apache.flink.table.typeutils.TypeStringUtils
+import org.apache.flink.util.Preconditions.checkState
+
+class LogicalCorrelateToVersionedJoinRule
+ extends RelOptRule(
+operand(classOf[LogicalCorrelate],
+ some(
+operand(classOf[RelNode], any()),
+operand(classOf[TableFunctionScan], none(,
+"LogicalCorrelateToVersionedJoinRule") {
+
+ override def onMatch(call: RelOptRuleCall): Unit = {
+val logicalCorrelate: LogicalCorrelate = call.rel(0)
+val leftNode: RelNode = call.rel(1)
+val rightTableFunctionScan: TableFunctionScan = call.rel(2)
+
+val cluster = logicalCorrelate.getCluster
+
+new GetTableVersionFunctionCall(cluster.getRexBuilder, leftNode)
+ .visit(rightTableFunctionScan.getCall) match {
+ case Some(rightVersionedTableCall) =>
+val underlyingTableHistory: Table =
rightVersionedTableCall.tableVersionFunction.table
+val relBuilder = this.relBuilderFactory.create(
+ cluster,
+ underlyingTableHistory.relBuilder.getRelOptSchema)
+val rexBuilder = cluster.getRexBuilder
+
+val rightNode: RelNode =
underlyingTableHistory.logicalPlan.toRelNode(relBuilder)
+
+val rightVersionExpression = createRightExpression(
+ rexBuilder,
+ leftNode,
+ rightNode,
+ rightVersionedTableCall.tableVersionFunction.versionField)
+
+val rightPrimaryKeyExpression = createRightExpression(
+ rexBuilder,
+ leftNode,
+ rightNode,
+ rightVersionedTableCall.tableVersionFunction.primaryKey)
+
+relBuilder.push(
+ if (rightVersionedTableCall.tableVersionFunction.isProctime) {
+LogicalVersionedJoin.create(
+ rexBuilder,
+ cluster,
+ logicalCorrelate.getTraitSet,
+ leftNode,
+ rightNode,
+ rightVersionedTableCall.versionExpression,
+ rightPrimaryKeyExpression)
+ }
+ else {
+LogicalVersionedJoin.create(
+ rexBuilder,
+ cluster,
+ logicalCorrelate.getTraitSet,
+ leftNode,
+ rightNode,
+ rightVersionedTableCall.versionExpression,
+ rightVersionExpression,
+ rightPrimaryKeyExpression)
+ })
+call.transformTo(relBuilder.build())
+ case None =>
+}
+ }
+
+ private def createRightExpression(
+ rexBuilder: RexBuilder,
+ leftNode: RelNode,
+ rightNode: RelNode,
+ field: String): RexNode = {
+val rightReferencesOffset =
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16574439#comment-16574439
]
ASF GitHub Bot commented on FLINK-9713:
---
pnowojski commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r208831842
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/logical/rel/LogicalVersionedJoin.scala
##
@@ -0,0 +1,172 @@
+/*
+ * 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.table.plan.logical.rel
+
+import java.util.Collections
+
+import org.apache.calcite.plan.{RelOptCluster, RelTraitSet}
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.core._
+import org.apache.calcite.rex.{RexBuilder, RexNode}
+import org.apache.calcite.sql.`type`.{OperandTypes, ReturnTypes}
+import org.apache.calcite.sql.{SqlFunction, SqlFunctionCategory, SqlKind}
+import org.apache.flink.util.Preconditions.checkArgument
+
+/**
+ * Represents a join between a table and
[[org.apache.flink.table.functions.TableVersionFunction]]
+ *
+ * @param cluster
+ * @param traitSet
+ * @param left stream
+ * @param right table scan of underlying
[[org.apache.flink.table.functions.TableVersionFunction]]
+ * @param condition must contain
[[LogicalVersionedJoin#VERSIONING_JOIN_CONDITION()]] with
+ * correctly defined references to rightVersioneExpression,
+ * rightPrimaryKeyExpression and leftVersionExpression. We
can not implement
+ * those references as separate fields, because of problems
with Calcite's
+ * optimization rules like projections push downs, column
+ * pruning/renaming/reordering, etc. Later
rightVersioneExpression,
+ * rightPrimaryKeyExpression and leftVersionExpression will
be extracted from
+ * the condition.
+ */
+class LogicalVersionedJoin private (
+cluster: RelOptCluster,
+traitSet: RelTraitSet,
+left: RelNode,
+right: RelNode,
+condition: RexNode)
+ extends Join(
+cluster,
+traitSet,
+left,
+right,
+condition,
+Collections.emptySet().asInstanceOf[java.util.Set[CorrelationId]],
+JoinRelType.INNER) {
+
+ override def copy(
+ traitSet: RelTraitSet,
+ condition: RexNode,
+ left: RelNode,
+ right: RelNode,
+ joinType: JoinRelType,
+ semiJoinDone: Boolean): LogicalVersionedJoin = {
+checkArgument(joinType == this.getJoinType,
+ "Can not change join type".asInstanceOf[Object])
+checkArgument(semiJoinDone == this.isSemiJoinDone,
+ "Can not change semiJoinDone".asInstanceOf[Object])
+new LogicalVersionedJoin(
+ cluster,
+ traitSet,
+ left,
+ right,
+ condition)
+ }
+}
+
+object LogicalVersionedJoin {
+ /**
+* See [[LogicalVersionedJoin#condition]]
+*/
+ val VERSIONING_JOIN_CONDITION = new SqlFunction(
Review comment:
We want and need to use Calcite's optimisations for joins (like pushing down
predicate, column pruning, predicate pushdown etc). To enable that,
`LogicalVersionedJoin` extends calcite's `Join`. However there is no way to
expand `Join`'s semantic, that beside actual join condition we need some extra
fields to handle versioning. The only way to do this, is to expose
`leftVersionExpression`, `rightVersionExpression` and
`rightPrimaryKeyExpression` via this `VERSIONING_JOIN_CONDITION`. Otherwise, if
we kept those expressions on some hidden state, they could be
pruned/renamed/reordered/... .
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL:
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16574436#comment-16574436
]
ASF GitHub Bot commented on FLINK-9713:
---
pnowojski commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r208835295
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamJoin.scala
##
@@ -103,27 +97,42 @@ class DataStreamJoin(
tableEnv: StreamTableEnvironment,
queryConfig: StreamQueryConfig): DataStream[CRow] = {
-val config = tableEnv.getConfig
-val returnType = schema.typeInfo
-val keyPairs = joinInfo.pairs().toList
+validateKeyTypes()
-// get the equality keys
-val leftKeys = ArrayBuffer.empty[Int]
-val rightKeys = ArrayBuffer.empty[Int]
+val leftDataStream =
Review comment:
I would be against that. I'm depending on this change in this PR, so it
defines strict order of those two commits. Either I wouldn't publish the actual
versioned joins PR after this commit is merged, or the versioned joins PR would
look exactly the same as it looks now. Splitting into two PRs adds a lot of
overhead and room for a lot of problems with PRs going out of sync. During
reviewing it's also annoying, because you see the same code twice and reviewers
(same or different ones) are often (almost always?) making half of the comments
in one PR and half in the other.
Besides, this is already in a separate commit so it can be reviewed
separately. Maybe instead of reviewing PR all at once try reviewing it commit
by commit?
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16573054#comment-16573054
]
ASF GitHub Bot commented on FLINK-9713:
---
twalthr commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r208512822
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/TableVersionFunction.scala
##
@@ -0,0 +1,70 @@
+/*
+ * 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.table.functions
+
+import java.sql.Timestamp
+
+import org.apache.flink.api.common.typeinfo.TypeInformation
+import org.apache.flink.api.java.typeutils.RowTypeInfo
+import org.apache.flink.table.api.Table
+import org.apache.flink.types.Row
+
+/**
+ * Class represnting table version functions over some history table.
+ */
+class TableVersionFunction private(
+@transient private val _table: Table,
Review comment:
`_` has so many meanings in Scala. We should not use it in variable names.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16573064#comment-16573064
]
ASF GitHub Bot commented on FLINK-9713:
---
twalthr commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r208529690
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamJoin.scala
##
@@ -103,27 +97,42 @@ class DataStreamJoin(
tableEnv: StreamTableEnvironment,
queryConfig: StreamQueryConfig): DataStream[CRow] = {
-val config = tableEnv.getConfig
-val returnType = schema.typeInfo
-val keyPairs = joinInfo.pairs().toList
+validateKeyTypes()
-// get the equality keys
-val leftKeys = ArrayBuffer.empty[Int]
-val rightKeys = ArrayBuffer.empty[Int]
+val leftDataStream =
Review comment:
We could also move this refactoring into a separate PR as subtask of
time-versioned joins. It would reduce the amount of changes in this PR.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16573055#comment-16573055
]
ASF GitHub Bot commented on FLINK-9713:
---
twalthr commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r208513757
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/TableVersionFunction.scala
##
@@ -0,0 +1,70 @@
+/*
+ * 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.table.functions
+
+import java.sql.Timestamp
+
+import org.apache.flink.api.common.typeinfo.TypeInformation
+import org.apache.flink.api.java.typeutils.RowTypeInfo
+import org.apache.flink.table.api.Table
+import org.apache.flink.types.Row
+
+/**
+ * Class represnting table version functions over some history table.
+ */
+class TableVersionFunction private(
+@transient private val _table: Table,
+private[flink] val versionField: String,
+private[flink] val primaryKey: String,
+private[flink] val resultType: RowTypeInfo,
+private[flink] val isProctime: Boolean)
+ extends TableFunction[Row] {
+
+ def eval(row: Timestamp): Unit = {
+throw new IllegalStateException("This should never be called")
+ }
+
+ override def getResultType: RowTypeInfo = {
+resultType
+ }
+
+ private[flink] def table: Table = {
+if (_table == null) {
+ throw new IllegalStateException("Accessing table field after
planing/serialization")
+}
+_table
+ }
+}
+
+object TableVersionFunction {
+ def create(
+ table: Table,
+ versionField: String,
+ primaryKey: String,
+ isProctime: Boolean): TableVersionFunction = {
+return new TableVersionFunction(
Review comment:
Please enable more warnings in your IDE. I get multiple warnings in your
changes. Here for example the unnecessary `return`.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16573057#comment-16573057
]
ASF GitHub Bot commented on FLINK-9713:
---
twalthr commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r208541703
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/join/VersionedJoin.scala
##
@@ -0,0 +1,93 @@
+/*
+ * 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.table.runtime.join
+
+import org.apache.flink.api.common.functions.FlatJoinFunction
+import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
+import org.apache.flink.api.common.typeinfo.TypeInformation
+import org.apache.flink.configuration.Configuration
+import org.apache.flink.streaming.api.functions.co.CoProcessFunction
+import org.apache.flink.table.api.StreamQueryConfig
+import org.apache.flink.table.codegen.Compiler
+import org.apache.flink.table.runtime.CRowWrappingCollector
+import org.apache.flink.table.runtime.types.CRow
+import org.apache.flink.table.typeutils.TypeCheckUtils._
+import org.apache.flink.table.util.Logging
+import org.apache.flink.types.Row
+import org.apache.flink.util.Collector
+
+class VersionedJoin(
+leftType: TypeInformation[Row],
+rightType: TypeInformation[Row],
+genJoinFuncName: String,
+genJoinFuncCode: String,
+queryConfig: StreamQueryConfig)
+ extends CoProcessFunction[CRow, CRow, CRow]
+ with Compiler[FlatJoinFunction[Row, Row, Row]]
+ with Logging {
+
+ validateEqualsHashCode("join", leftType)
+ validateEqualsHashCode("join", rightType)
+
+ protected var rightState: ValueState[Row] = _
+ protected var cRowWrapper: CRowWrappingCollector = _
+
+ protected var joinFunction: FlatJoinFunction[Row, Row, Row] = _
+
+ override def open(parameters: Configuration): Unit = {
+val clazz = compile(
+ getRuntimeContext.getUserCodeClassLoader,
+ genJoinFuncName,
+ genJoinFuncCode)
+
+joinFunction = clazz.newInstance()
+
+val rightStateDescriptor = new ValueStateDescriptor[Row]("right",
rightType)
Review comment:
Implement state clean-up depending on min/max retention time.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16573067#comment-16573067
]
ASF GitHub Bot commented on FLINK-9713:
---
twalthr commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r208519798
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/TableVersionFunction.scala
##
@@ -0,0 +1,70 @@
+/*
+ * 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.table.functions
+
+import java.sql.Timestamp
+
+import org.apache.flink.api.common.typeinfo.TypeInformation
+import org.apache.flink.api.java.typeutils.RowTypeInfo
+import org.apache.flink.table.api.Table
+import org.apache.flink.types.Row
+
+/**
+ * Class represnting table version functions over some history table.
+ */
+class TableVersionFunction private(
+@transient private val _table: Table,
+private[flink] val versionField: String,
+private[flink] val primaryKey: String,
Review comment:
These attributes are public in Java anyway. So we can have proper getters
for all attributes instead.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16573056#comment-16573056
]
ASF GitHub Bot commented on FLINK-9713:
---
twalthr commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r208518496
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/TableVersionFunction.scala
##
@@ -0,0 +1,70 @@
+/*
+ * 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.table.functions
+
+import java.sql.Timestamp
+
+import org.apache.flink.api.common.typeinfo.TypeInformation
+import org.apache.flink.api.java.typeutils.RowTypeInfo
+import org.apache.flink.table.api.Table
+import org.apache.flink.types.Row
+
+/**
+ * Class represnting table version functions over some history table.
Review comment:
typo and more documentation is needed also here. This is an API class.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-9713) Support versioned joins in planning phase
[
https://issues.apache.org/jira/browse/FLINK-9713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16573052#comment-16573052
]
ASF GitHub Bot commented on FLINK-9713:
---
twalthr commented on a change in pull request #6299: [FLINK-9713][table][sql]
Support processing time versioned joins
URL: https://github.com/apache/flink/pull/6299#discussion_r208508180
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/calcite/FlinkRelBuilder.scala
##
@@ -46,6 +46,8 @@ class FlinkRelBuilder(
relOptCluster,
relOptSchema) {
+ def getRelOptSchema: RelOptSchema = relOptSchema
Review comment:
Do we need this change?
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support versioned joins in planning phase
> -
>
> Key: FLINK-9713
> URL: https://issues.apache.org/jira/browse/FLINK-9713
> Project: Flink
> Issue Type: Sub-task
> Components: Table API SQL
>Affects Versions: 1.5.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Major
> Labels: pull-request-available
>
> Queries like:
> {code:java}
> SELECT
> o.amount * r.rate
> FROM
> Orders AS o,
> LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE o.currency = r.currency{code}
> should evaluate to valid plan with versioned joins plan node.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
