jnh5y commented on code in PR #24253:
URL: https://github.com/apache/flink/pull/24253#discussion_r1481681792
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/LogicalCorrelateToJoinFromTemporalTableRule.scala:
##########
@@ -237,57 +245,32 @@ abstract class
LogicalCorrelateToJoinFromGeneralTemporalTableRule(
val snapshot = getLogicalSnapshot(call)
val leftRowType = leftInput.getRowType
- val joinCondition = filterCondition.accept(new RexShuttle() {
- // change correlate variable expression to normal RexInputRef (which is
from left side)
- override def visitFieldAccess(fieldAccess: RexFieldAccess): RexNode = {
- fieldAccess.getReferenceExpr match {
- case corVar: RexCorrelVariable =>
- require(correlate.getCorrelationId.equals(corVar.id))
- val index = leftRowType.getFieldList.indexOf(fieldAccess.getField)
- RexInputRef.of(index, leftRowType)
- case _ => super.visitFieldAccess(fieldAccess)
- }
- }
-
- // update the field index from right side
- override def visitInputRef(inputRef: RexInputRef): RexNode = {
- val rightIndex = leftRowType.getFieldCount + inputRef.getIndex
- new RexInputRef(rightIndex, inputRef.getType)
- }
- })
+ val joinCondition = decorrelate(filterCondition, leftRowType,
correlate.getCorrelationId)
validateSnapshotInCorrelate(snapshot, correlate)
val rexBuilder = correlate.getCluster.getRexBuilder
- val (leftJoinKey, rightJoinKey) = {
- val relBuilder = call.builder()
- relBuilder.push(leftInput)
- relBuilder.push(snapshot)
- val rewriteJoin = relBuilder.join(correlate.getJoinType,
joinCondition).build()
- val joinInfo = rewriteJoin.asInstanceOf[LogicalJoin].analyzeCondition()
- val leftJoinKey = joinInfo.leftKeys.map(i =>
rexBuilder.makeInputRef(leftInput, i))
- val leftFieldCnt = leftInput.getRowType.getFieldCount
- val rightJoinKey = joinInfo.rightKeys.map(
- i => {
- val leftKeyType = snapshot.getRowType.getFieldList.get(i).getType
- rexBuilder.makeInputRef(leftKeyType, leftFieldCnt + i)
- })
- if (leftJoinKey.length == 0 || rightJoinKey.length == 0) {
- throw new ValidationException(
- "Currently the join key in Temporal Table Join " +
- "can not be empty.")
- }
- (leftJoinKey, rightJoinKey)
+ val relBuilder = call.builder()
+ relBuilder.push(leftInput)
+ relBuilder.push(snapshot)
+ val nonPushedJoin =
+ relBuilder.join(correlate.getJoinType,
joinCondition).build().asInstanceOf[LogicalJoin]
+ val rewriteJoin = RelOptUtil.pushDownJoinConditions(nonPushedJoin,
relBuilder)
Review Comment:
In what situations can the join conditions not be pushed down?
We need some validation for those situations?
--
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.
To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
