twalthr commented on a change in pull request #8018: [FLINK-11884][table]
Separate creation and validation of LogicalNodes from TableImpl
URL: https://github.com/apache/flink/pull/8018#discussion_r269488687
##########
File path:
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/tableImpl.scala
##########
@@ -313,135 +277,124 @@ class TableImpl(
override def leftOuterJoinLateral(
tableFunctionCall: Expression, joinPredicate: Expression): Table = {
joinLateralInternal(
- expressionBridge.bridge(tableFunctionCall),
- Some(expressionBridge.bridge(joinPredicate)),
+ tableFunctionCall,
+ Some(joinPredicate),
JoinType.LEFT_OUTER)
}
private def joinLateralInternal(
- callExpr: PlannerExpression,
- joinPredicate: Option[PlannerExpression],
+ callExpr: Expression,
+ joinPredicate: Option[Expression],
joinType: JoinType): Table = {
// check join type
if (joinType != JoinType.INNER && joinType != JoinType.LEFT_OUTER) {
throw new ValidationException(
"Table functions are currently only supported for inner and left outer
lateral joins.")
}
-
- val logicalCall = UserDefinedFunctionUtils.createLogicalFunctionCall(
- callExpr,
- logicalPlan)
- val validatedLogicalCall = logicalCall.validate(tableEnv)
-
new TableImpl(
tableEnv,
- Join(
+ operationTreeBuilder.joinLateral(
logicalPlan,
- validatedLogicalCall,
+ callExpr.accept(callResolver),
joinType,
- joinPredicate,
- correlated = true
- ).validate(tableEnv))
+ toJava(joinPredicate.map(_.accept(callResolver))
+ )))
}
override def minus(right: Table): Table = {
// check that right table belongs to the same TableEnvironment
- if (right.asInstanceOf[TableImpl].tableEnv != this.tableEnv) {
+ val rightImpl = right.asInstanceOf[TableImpl]
+ if (rightImpl.tableEnv != this.tableEnv) {
throw new ValidationException("Only tables from the same
TableEnvironment can be " +
"subtracted.")
}
- new TableImpl(
- tableEnv, Minus(logicalPlan, right.asInstanceOf[TableImpl].logicalPlan,
all = false)
- .validate(tableEnv))
+ new TableImpl(tableEnv,
+ operationTreeBuilder.minus(logicalPlan, rightImpl.logicalPlan, all =
false))
}
override def minusAll(right: Table): Table = {
// check that right table belongs to the same TableEnvironment
- if (right.asInstanceOf[TableImpl].tableEnv != this.tableEnv) {
+ val rightImpl = right.asInstanceOf[TableImpl]
+ if (rightImpl.tableEnv != this.tableEnv) {
throw new ValidationException("Only tables from the same
TableEnvironment can be " +
"subtracted.")
}
- new TableImpl(
- tableEnv, Minus(logicalPlan, right.asInstanceOf[TableImpl].logicalPlan,
all = true)
- .validate(tableEnv))
+
+ new TableImpl(tableEnv,
Review comment:
nit: we could introduce a helper method `wrap(Operation)` instead of copying
`new TableImpl(tableEnv,` all the time.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
