[FLINK-22099][table-planner-blink] Fix bug about throwing ArrayIndexOutOfBoundsException when window join deals with semi/anti queries

This closes #15477

Author: beyond1920

flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamPhysicalWindowJoin.scala

@@ -97,7 +97,7 @@ class StreamPhysicalWindowJoin(
       .item("joinType", joinSpec.getJoinType)
       .item("where",
         getExpressionString(
-          remainingCondition, getRowType.getFieldNames.toList, None, preferExpressionFormat(pw)))
+          remainingCondition, inputRowType.getFieldNames.toList, None, preferExpressionFormat(pw)))
       .item("select", getRowType.getFieldNames.mkString(", "))
   }
 

flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/utils/WindowJoinUtil.scala

@@ -96,7 +96,8 @@ object WindowJoinUtil {
     val joinInfo = join.analyzeCondition()
     val (remainLeftKeys, remainRightKeys, remainCondition) = if (
       windowStartEqualityLeftKeys.nonEmpty || windowEndEqualityLeftKeys.nonEmpty) {
-      val joinFieldsType = join.getRowType.getFieldList
+      val leftChildFieldsType = join.getLeft.getRowType.getFieldList
+      val rightChildFieldsType = join.getRight.getRowType.getFieldList
       val leftFieldCnt = join.getLeft.getRowType.getFieldCount
       val rexBuilder = join.getCluster.getRexBuilder
       val remainEquals = mutable.ArrayBuffer[RexNode]()
@@ -106,10 +107,10 @@ object WindowJoinUtil {
       joinInfo.pairs().foreach { p =>
         if (!windowStartEqualityLeftKeys.contains(p.source) &&
           !windowEndEqualityLeftKeys.contains(p.source)) {
-          val leftFieldType = joinFieldsType.get(p.source).getType
+          val leftFieldType = leftChildFieldsType.get(p.source).getType
           val leftInputRef = new RexInputRef(p.source, leftFieldType)
+          val rightFieldType = rightChildFieldsType.get(p.target).getType
           val rightIndex = leftFieldCnt + p.target
-          val rightFieldType = joinFieldsType.get(rightIndex).getType
           val rightInputRef = new RexInputRef(rightIndex, rightFieldType)
           val remainEqual = rexBuilder.makeCall(
             SqlStdOperatorTable.EQUALS,