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_r269490282
##########
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,
+ operationTreeBuilder.minus(logicalPlan, rightImpl.logicalPlan, all =
true))
}
override def union(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 unioned.")
}
- new TableImpl(
- tableEnv, Union(logicalPlan, right.asInstanceOf[TableImpl].logicalPlan,
all = false)
- .validate(tableEnv))
+
+ new TableImpl(tableEnv,
+ operationTreeBuilder.union(logicalPlan, rightImpl.logicalPlan, all =
false))
}
override def unionAll(right: Table): Table = {
// check that right table belongs to the same TableEnvironment
+ val rightImpl = right.asInstanceOf[TableImpl]
if (right.asInstanceOf[TableImpl].tableEnv != this.tableEnv) {
throw new ValidationException("Only tables from the same
TableEnvironment can be unioned.")
}
- new TableImpl(
- tableEnv, Union(logicalPlan, right.asInstanceOf[TableImpl].logicalPlan,
all = true)
- .validate(tableEnv))
+
+ new TableImpl(tableEnv,
+ operationTreeBuilder.union(logicalPlan, rightImpl.logicalPlan, all =
true))
}
override def intersect(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 intersected.")
}
- new TableImpl(
- tableEnv, Intersect(logicalPlan,
right.asInstanceOf[TableImpl].logicalPlan, all = false)
- .validate(tableEnv))
+
+ new TableImpl(tableEnv,
+ operationTreeBuilder.intersect(logicalPlan, rightImpl.logicalPlan, all =
false))
}
override def intersectAll(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 intersected.")
}
- new TableImpl(
- tableEnv, Intersect(logicalPlan,
right.asInstanceOf[TableImpl].logicalPlan, all = true)
- .validate(tableEnv))
+
+ new TableImpl(tableEnv,
+ operationTreeBuilder.intersect(logicalPlan, rightImpl.logicalPlan, all =
true))
}
override def orderBy(fields: String): Table = {
orderBy(ExpressionParser.parseExpressionList(fields).asScala: _*)
}
override def orderBy(fields: Expression*): Table = {
- orderByInternal(fields.map(expressionBridge.bridge))
+ orderByInternal(fields)
}
- private def orderByInternal(fields: Seq[PlannerExpression]): Table = {
- val order: Seq[Ordering] = fields.map {
- case o: Ordering => o
- case e => Asc(e)
- }
- new TableImpl(tableEnv, Sort(order, logicalPlan).validate(tableEnv))
+ private def orderByInternal(fields: Seq[Expression]): Table = {
+ new TableImpl(tableEnv,
+ operationTreeBuilder.sort(fields.map(_.accept(callResolver)).asJava,
logicalPlan))
}
override def offset(offset: Int): Table = {
- new TableImpl(tableEnv, Limit(offset, -1, logicalPlan).validate(tableEnv))
+ new TableImpl(tableEnv,
+ operationTreeBuilder.limit(Optional.of(offset), Optional.empty(),
logicalPlan))
Review comment:
Should we add `.limitWithOffset`, `.limitWithFetch`, `.limit()` instead of
optionals?
----------------------------------------------------------------
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
