Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/4471#discussion_r142779879
--- Diff:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/datastream/DataStreamJoinRule.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.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.flink.table.api.TableConfig
+import org.apache.flink.table.calcite.FlinkTypeFactory
+import org.apache.flink.table.plan.nodes.FlinkConventions
+import org.apache.flink.table.plan.nodes.datastream.DataStreamJoin
+import org.apache.flink.table.plan.nodes.logical.FlinkLogicalJoin
+import org.apache.flink.table.plan.schema.RowSchema
+import org.apache.flink.table.runtime.join.WindowJoinUtil
+import scala.collection.JavaConverters._
+
+class DataStreamJoinRule
+ extends ConverterRule(
+ classOf[FlinkLogicalJoin],
+ FlinkConventions.LOGICAL,
+ FlinkConventions.DATASTREAM,
+ "DataStreamJoinRule") {
+
+ override def matches(call: RelOptRuleCall): Boolean = {
+ val join: FlinkLogicalJoin = call.rel(0).asInstanceOf[FlinkLogicalJoin]
+ val joinInfo = join.analyzeCondition
+
+ val (windowBounds, remainingPreds) =
WindowJoinUtil.extractWindowBoundsFromPredicate(
+ joinInfo.getRemaining(join.getCluster.getRexBuilder),
+ join.getLeft.getRowType.getFieldCount,
+ join.getRowType,
+ join.getCluster.getRexBuilder,
+ TableConfig.DEFAULT)
+
+ // remaining predicate must not access time attributes
+ val remainingPredsAccessTime = remainingPreds.isDefined &&
+ WindowJoinUtil.accessesTimeAttribute(remainingPreds.get,
join.getRowType)
+
+ // Check that no event-time attributes are in the input.
+ val rowTimeAttrInOutput = join.getRowType.getFieldList.asScala
+ .exists(f => FlinkTypeFactory.isRowtimeIndicatorType(f.getType))
+
+ if (!windowBounds.isDefined && !remainingPredsAccessTime &&
!rowTimeAttrInOutput) {
--- End diff --
@shaoxuan-wang, I thought about this issue again and think you are right.
It would be quite difficult for users to get the queries right and also
difficult to properly document the restrictions.
IMO, it would be good to evolve the relational APIs such that most
operators can be executed on time indicator attributes (event or proc time) or
not. In case of time indicator attributes, we can generate more efficient plans
with built-in state clean-up. A generic stream-stream join such as the one
proposed in the PR would be a first step in this direction.
As you said before, a major challenge with this approach would be to help
users configuring state cleanup timers. I would propose to extend the EXPLAIN
information with state size estimates. This would help users users to correctly
set the query configuration.
I will go over my comments for this PR again and adapt them where necessary.
Thanks, Fabian
---
