You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2021/01/20 02:12:57 UTC
[GitHub] [flink] wenlong88 opened a new pull request #14699: [FLINK-21011][table-planner-blink] Separate implementation of StreamExecIntervalJoin
wenlong88 opened a new pull request #14699:
URL: https://github.com/apache/flink/pull/14699
## What is the purpose of the change
Introduce StreamPhysicalIntervalJoin and make StreamExecIntervalJoin only extends from ExecNode
## Brief change log
1. introduce StreamPhysicalIntervalJoin and make StreamExecIntervalJoin only extends from ExecNode
2. introduce IntervalJoinFunction to wrapper non-equil JoinCondition and null join key filter in join spec, so that we do not need to compare join keys in runtime because the join keys is already compared by the group key in runtime.
## Verifying this change
This change is a refactor work, the change is already covered by existing tests.
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): (no)
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (no)
- The serializers: (no)
- The runtime per-record code paths (performance sensitive): (no)
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn/Mesos, ZooKeeper: (no)
- The S3 file system connector: (no)
## Documentation
- Does this pull request introduce a new feature? (no)
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #14699: [FLINK-21011][table-planner-blink] Separate implementation of StreamExecIntervalJoin
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #14699:
URL: https://github.com/apache/flink/pull/14699#issuecomment-763287840
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "ddd72a9067d9c7b0fed65b26947d00f530d53310",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=12258",
"triggerID" : "ddd72a9067d9c7b0fed65b26947d00f530d53310",
"triggerType" : "PUSH"
}, {
"hash" : "23ac68d1604e6f5e38af38caa73d11a8dc76ceb9",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=12268",
"triggerID" : "23ac68d1604e6f5e38af38caa73d11a8dc76ceb9",
"triggerType" : "PUSH"
}, {
"hash" : "e9d2494f67bf130a6a0da0c9a1433195b79a7c45",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=12287",
"triggerID" : "e9d2494f67bf130a6a0da0c9a1433195b79a7c45",
"triggerType" : "PUSH"
}, {
"hash" : "c463989c4c6afdc6a85485822d1350f6ab771a30",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "c463989c4c6afdc6a85485822d1350f6ab771a30",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* e9d2494f67bf130a6a0da0c9a1433195b79a7c45 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=12287)
* c463989c4c6afdc6a85485822d1350f6ab771a30 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #14699: [FLINK-21011][table-planner-blink] Separate implementation of StreamExecIntervalJoin
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #14699:
URL: https://github.com/apache/flink/pull/14699#issuecomment-763287840
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "ddd72a9067d9c7b0fed65b26947d00f530d53310",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=12258",
"triggerID" : "ddd72a9067d9c7b0fed65b26947d00f530d53310",
"triggerType" : "PUSH"
}, {
"hash" : "23ac68d1604e6f5e38af38caa73d11a8dc76ceb9",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "23ac68d1604e6f5e38af38caa73d11a8dc76ceb9",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* ddd72a9067d9c7b0fed65b26947d00f530d53310 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=12258)
* 23ac68d1604e6f5e38af38caa73d11a8dc76ceb9 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #14699: [FLINK-21011][table-planner-blink] Separate implementation of StreamExecIntervalJoin
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #14699:
URL: https://github.com/apache/flink/pull/14699#issuecomment-763287840
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "ddd72a9067d9c7b0fed65b26947d00f530d53310",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=12258",
"triggerID" : "ddd72a9067d9c7b0fed65b26947d00f530d53310",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* ddd72a9067d9c7b0fed65b26947d00f530d53310 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=12258)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] wenlong88 commented on a change in pull request #14699: [FLINK-21011][table-planner-blink] Separate implementation of StreamExecIntervalJoin
Posted by GitBox <gi...@apache.org>.
wenlong88 commented on a change in pull request #14699:
URL: https://github.com/apache/flink/pull/14699#discussion_r561513490
##########
File path: flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamPhysicalIntervalJoin.scala
##########
@@ -0,0 +1,94 @@
+/*
+ * 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.planner.plan.nodes.physical.stream
+
+import org.apache.flink.table.api.TableException
+import org.apache.flink.table.planner.calcite.FlinkTypeFactory
+import org.apache.flink.table.planner.plan.nodes.common.CommonPhysicalJoin
+import org.apache.flink.table.planner.plan.nodes.exec.{ExecEdge, ExecNode}
+import org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecIntervalJoin
+import org.apache.flink.table.planner.plan.utils.IntervalJoinUtil.WindowBounds
+import org.apache.flink.table.planner.plan.utils.PythonUtil.containsPythonCall
+import org.apache.flink.table.planner.plan.utils.RelExplainUtil.preferExpressionFormat
+
+import org.apache.calcite.plan._
+import org.apache.calcite.rel.core.{Join, JoinRelType}
+import org.apache.calcite.rel.{RelNode, RelWriter}
+import org.apache.calcite.rex.RexNode
+
+import scala.collection.JavaConversions._
+
+/**
+ * Stream physical RelNode for a time interval stream join.
+ */
+class StreamPhysicalIntervalJoin(
+ cluster: RelOptCluster,
+ traitSet: RelTraitSet,
+ leftRel: RelNode,
+ rightRel: RelNode,
+ joinType: JoinRelType,
+ condition: RexNode,
Review comment:
I have added back the originalCondition in the physical node.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #14699: [FLINK-21011][table-planner-blink] Separate implementation of StreamExecIntervalJoin
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #14699:
URL: https://github.com/apache/flink/pull/14699#issuecomment-763287840
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "ddd72a9067d9c7b0fed65b26947d00f530d53310",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=12258",
"triggerID" : "ddd72a9067d9c7b0fed65b26947d00f530d53310",
"triggerType" : "PUSH"
}, {
"hash" : "23ac68d1604e6f5e38af38caa73d11a8dc76ceb9",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=12268",
"triggerID" : "23ac68d1604e6f5e38af38caa73d11a8dc76ceb9",
"triggerType" : "PUSH"
}, {
"hash" : "e9d2494f67bf130a6a0da0c9a1433195b79a7c45",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=12287",
"triggerID" : "e9d2494f67bf130a6a0da0c9a1433195b79a7c45",
"triggerType" : "PUSH"
}, {
"hash" : "c463989c4c6afdc6a85485822d1350f6ab771a30",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=12304",
"triggerID" : "c463989c4c6afdc6a85485822d1350f6ab771a30",
"triggerType" : "PUSH"
}, {
"hash" : "316311b1eb3e13ae6ececc71d75c02b95403214d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=12314",
"triggerID" : "316311b1eb3e13ae6ececc71d75c02b95403214d",
"triggerType" : "PUSH"
}, {
"hash" : "5e4cf66cd16642cec55f3ed47c413f4dbfb856fe",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=12360",
"triggerID" : "5e4cf66cd16642cec55f3ed47c413f4dbfb856fe",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 5e4cf66cd16642cec55f3ed47c413f4dbfb856fe Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=12360)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] godfreyhe commented on a change in pull request #14699: [FLINK-21011][table-planner-blink] Separate implementation of StreamExecIntervalJoin
Posted by GitBox <gi...@apache.org>.
godfreyhe commented on a change in pull request #14699:
URL: https://github.com/apache/flink/pull/14699#discussion_r562339871
##########
File path: flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecIntervalJoin.java
##########
@@ -0,0 +1,357 @@
+/*
+ * 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.planner.plan.nodes.exec.stream;
+
+import org.apache.flink.api.common.functions.FlatMapFunction;
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.dag.Transformation;
+import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
+import org.apache.flink.streaming.api.operators.StreamFlatMap;
+import org.apache.flink.streaming.api.operators.StreamMap;
+import org.apache.flink.streaming.api.operators.co.KeyedCoProcessOperator;
+import org.apache.flink.streaming.api.transformations.OneInputTransformation;
+import org.apache.flink.streaming.api.transformations.TwoInputTransformation;
+import org.apache.flink.streaming.api.transformations.UnionTransformation;
+import org.apache.flink.table.api.TableException;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.planner.delegation.PlannerBase;
+import org.apache.flink.table.planner.plan.nodes.exec.ExecEdge;
+import org.apache.flink.table.planner.plan.nodes.exec.ExecNode;
+import org.apache.flink.table.planner.plan.nodes.exec.ExecNodeBase;
+import org.apache.flink.table.planner.plan.nodes.exec.utils.IntervalJoinSpec;
+import org.apache.flink.table.planner.plan.nodes.exec.utils.JoinSpec;
+import org.apache.flink.table.planner.plan.utils.JoinUtil;
+import org.apache.flink.table.planner.plan.utils.KeySelectorUtil;
+import org.apache.flink.table.runtime.generated.GeneratedJoinCondition;
+import org.apache.flink.table.runtime.keyselector.RowDataKeySelector;
+import org.apache.flink.table.runtime.operators.join.KeyedCoProcessOperatorWithWatermarkDelay;
+import org.apache.flink.table.runtime.operators.join.OuterJoinPaddingUtil;
+import org.apache.flink.table.runtime.operators.join.interval.IntervalJoinFunction;
+import org.apache.flink.table.runtime.operators.join.interval.ProcTimeIntervalJoin;
+import org.apache.flink.table.runtime.operators.join.interval.RowTimeIntervalJoin;
+import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.util.Collector;
+
+import org.apache.flink.shaded.guava18.com.google.common.collect.Lists;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/** {@link StreamExecNode} for a time interval stream join. */
+public class StreamExecIntervalJoin extends ExecNodeBase<RowData>
+ implements StreamExecNode<RowData> {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(StreamExecIntervalJoin.class);
+
+ private final IntervalJoinSpec intervalJoinSpec;
Review comment:
nit: This field is useless as class field ?
##########
File path: flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamPhysicalIntervalJoin.scala
##########
@@ -0,0 +1,95 @@
+/*
+ * 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.planner.plan.nodes.physical.stream
+
+import org.apache.flink.table.api.TableException
+import org.apache.flink.table.planner.calcite.FlinkTypeFactory
+import org.apache.flink.table.planner.plan.nodes.common.CommonPhysicalJoin
+import org.apache.flink.table.planner.plan.nodes.exec.{ExecEdge, ExecNode}
+import org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecIntervalJoin
+import org.apache.flink.table.planner.plan.nodes.exec.utils.IntervalJoinSpec
+import org.apache.flink.table.planner.plan.nodes.exec.utils.IntervalJoinSpec.WindowBounds
+import org.apache.flink.table.planner.plan.utils.PythonUtil.containsPythonCall
+import org.apache.flink.table.planner.plan.utils.RelExplainUtil.preferExpressionFormat
+import org.apache.calcite.plan._
+import org.apache.calcite.rel.core.{Join, JoinRelType}
+import org.apache.calcite.rel.{BiRel, RelNode, RelWriter}
+import org.apache.calcite.rex.RexNode
+
+import scala.collection.JavaConversions._
+
+/**
+ * Stream physical RelNode for a time interval stream join.
+ */
+class StreamPhysicalIntervalJoin(
+ cluster: RelOptCluster,
+ traitSet: RelTraitSet,
+ leftRel: RelNode,
+ rightRel: RelNode,
+ joinType: JoinRelType,
+ val originalCondition: RexNode,
+ remainingCondition: RexNode,
+ windowBounds: WindowBounds)
Review comment:
add some comments to explain `remainingCondition`, because in general, remainingCondition means the non-equal join condition part
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] wenlong88 commented on a change in pull request #14699: [FLINK-21011][table-planner-blink] Separate implementation of StreamExecIntervalJoin
Posted by GitBox <gi...@apache.org>.
wenlong88 commented on a change in pull request #14699:
URL: https://github.com/apache/flink/pull/14699#discussion_r561514662
##########
File path: flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdUniqueKeys.scala
##########
@@ -470,11 +470,11 @@ class FlinkRelMdUniqueKeys private extends MetadataHandler[BuiltInMetadata.Uniqu
}
def getUniqueKeys(
- rel: StreamExecIntervalJoin,
+ rel: StreamPhysicalIntervalJoin,
mq: RelMetadataQuery,
ignoreNulls: Boolean): JSet[ImmutableBitSet] = {
- val joinInfo = JoinInfo.of(rel.getLeft, rel.getRight, rel.joinCondition)
- getJoinUniqueKeys(joinInfo, rel.joinType, rel.getLeft, rel.getRight, mq, ignoreNulls)
+ val joinInfo = JoinInfo.of(rel.getLeft, rel.getRight, rel.getCondition)
Review comment:
we can not use analyzeCondition here, because join condition in interval join doesn't contain time range part
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] godfreyhe commented on a change in pull request #14699: [FLINK-21011][table-planner-blink] Separate implementation of StreamExecIntervalJoin
Posted by GitBox <gi...@apache.org>.
godfreyhe commented on a change in pull request #14699:
URL: https://github.com/apache/flink/pull/14699#discussion_r560693866
##########
File path: flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/operators/join/interval/IntervalJoinFunction.java
##########
@@ -0,0 +1,75 @@
+package org.apache.flink.table.runtime.operators.join.interval;
+
+import org.apache.flink.api.common.functions.RichFlatJoinFunction;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.binary.BinaryRowData;
+import org.apache.flink.table.data.utils.JoinedRowData;
+import org.apache.flink.table.runtime.generated.GeneratedJoinCondition;
+import org.apache.flink.table.runtime.generated.JoinCondition;
+import org.apache.flink.table.runtime.operators.join.NullAwareJoinHelper;
+import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
+import org.apache.flink.util.Collector;
+
+/**
+ * {@link IntervalJoinFunction} is a {@link RichFlatJoinFunction} wrappers a {@link
+ * GeneratedJoinCondition}.
+ */
+public class IntervalJoinFunction extends RichFlatJoinFunction<RowData, RowData, RowData>
Review comment:
@beyond1920 could you help review this change ? thanks
##########
File path: flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecIntervalJoin.java
##########
@@ -0,0 +1,357 @@
+/*
+ * 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.planner.plan.nodes.exec.stream;
+
+import org.apache.flink.api.common.functions.FlatMapFunction;
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.dag.Transformation;
+import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
+import org.apache.flink.streaming.api.operators.StreamFlatMap;
+import org.apache.flink.streaming.api.operators.StreamMap;
+import org.apache.flink.streaming.api.operators.co.KeyedCoProcessOperator;
+import org.apache.flink.streaming.api.transformations.OneInputTransformation;
+import org.apache.flink.streaming.api.transformations.TwoInputTransformation;
+import org.apache.flink.streaming.api.transformations.UnionTransformation;
+import org.apache.flink.table.api.TableException;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.planner.delegation.PlannerBase;
+import org.apache.flink.table.planner.plan.nodes.exec.ExecEdge;
+import org.apache.flink.table.planner.plan.nodes.exec.ExecNode;
+import org.apache.flink.table.planner.plan.nodes.exec.ExecNodeBase;
+import org.apache.flink.table.planner.plan.nodes.exec.utils.IntervalJoinSpec;
+import org.apache.flink.table.planner.plan.nodes.exec.utils.JoinSpec;
+import org.apache.flink.table.planner.plan.utils.JoinUtil;
+import org.apache.flink.table.planner.plan.utils.KeySelectorUtil;
+import org.apache.flink.table.runtime.generated.GeneratedJoinCondition;
+import org.apache.flink.table.runtime.keyselector.RowDataKeySelector;
+import org.apache.flink.table.runtime.operators.join.KeyedCoProcessOperatorWithWatermarkDelay;
+import org.apache.flink.table.runtime.operators.join.OuterJoinPaddingUtil;
+import org.apache.flink.table.runtime.operators.join.interval.IntervalJoinFunction;
+import org.apache.flink.table.runtime.operators.join.interval.ProcTimeIntervalJoin;
+import org.apache.flink.table.runtime.operators.join.interval.RowTimeIntervalJoin;
+import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.util.Collector;
+
+import org.apache.flink.shaded.guava18.com.google.common.collect.Lists;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/** {@link StreamExecNode} for a time interval stream join. */
+public class StreamExecIntervalJoin extends ExecNodeBase<RowData>
+ implements StreamExecNode<RowData> {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(StreamExecIntervalJoin.class);
+
+ private final IntervalJoinSpec intervalJoinSpec;
Review comment:
nit: This field is useless as class field ?
##########
File path: flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamPhysicalIntervalJoin.scala
##########
@@ -0,0 +1,95 @@
+/*
+ * 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.planner.plan.nodes.physical.stream
+
+import org.apache.flink.table.api.TableException
+import org.apache.flink.table.planner.calcite.FlinkTypeFactory
+import org.apache.flink.table.planner.plan.nodes.common.CommonPhysicalJoin
+import org.apache.flink.table.planner.plan.nodes.exec.{ExecEdge, ExecNode}
+import org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecIntervalJoin
+import org.apache.flink.table.planner.plan.nodes.exec.utils.IntervalJoinSpec
+import org.apache.flink.table.planner.plan.nodes.exec.utils.IntervalJoinSpec.WindowBounds
+import org.apache.flink.table.planner.plan.utils.PythonUtil.containsPythonCall
+import org.apache.flink.table.planner.plan.utils.RelExplainUtil.preferExpressionFormat
+import org.apache.calcite.plan._
+import org.apache.calcite.rel.core.{Join, JoinRelType}
+import org.apache.calcite.rel.{BiRel, RelNode, RelWriter}
+import org.apache.calcite.rex.RexNode
+
+import scala.collection.JavaConversions._
+
+/**
+ * Stream physical RelNode for a time interval stream join.
+ */
+class StreamPhysicalIntervalJoin(
+ cluster: RelOptCluster,
+ traitSet: RelTraitSet,
+ leftRel: RelNode,
+ rightRel: RelNode,
+ joinType: JoinRelType,
+ val originalCondition: RexNode,
+ remainingCondition: RexNode,
+ windowBounds: WindowBounds)
Review comment:
add some comments to explain `remainingCondition`, because in general, remainingCondition means the non-equal join condition part
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #14699: [FLINK-21011][table-planner-blink] Separate implementation of StreamExecIntervalJoin
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #14699:
URL: https://github.com/apache/flink/pull/14699#issuecomment-763287840
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "ddd72a9067d9c7b0fed65b26947d00f530d53310",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=12258",
"triggerID" : "ddd72a9067d9c7b0fed65b26947d00f530d53310",
"triggerType" : "PUSH"
}, {
"hash" : "23ac68d1604e6f5e38af38caa73d11a8dc76ceb9",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=12268",
"triggerID" : "23ac68d1604e6f5e38af38caa73d11a8dc76ceb9",
"triggerType" : "PUSH"
}, {
"hash" : "e9d2494f67bf130a6a0da0c9a1433195b79a7c45",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=12287",
"triggerID" : "e9d2494f67bf130a6a0da0c9a1433195b79a7c45",
"triggerType" : "PUSH"
}, {
"hash" : "c463989c4c6afdc6a85485822d1350f6ab771a30",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=12304",
"triggerID" : "c463989c4c6afdc6a85485822d1350f6ab771a30",
"triggerType" : "PUSH"
}, {
"hash" : "316311b1eb3e13ae6ececc71d75c02b95403214d",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=12314",
"triggerID" : "316311b1eb3e13ae6ececc71d75c02b95403214d",
"triggerType" : "PUSH"
}, {
"hash" : "5e4cf66cd16642cec55f3ed47c413f4dbfb856fe",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "5e4cf66cd16642cec55f3ed47c413f4dbfb856fe",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 316311b1eb3e13ae6ececc71d75c02b95403214d Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=12314)
* 5e4cf66cd16642cec55f3ed47c413f4dbfb856fe UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #14699: [FLINK-21011][table-planner-blink] Separate implementation of StreamExecIntervalJoin
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #14699:
URL: https://github.com/apache/flink/pull/14699#issuecomment-763287840
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "ddd72a9067d9c7b0fed65b26947d00f530d53310",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=12258",
"triggerID" : "ddd72a9067d9c7b0fed65b26947d00f530d53310",
"triggerType" : "PUSH"
}, {
"hash" : "23ac68d1604e6f5e38af38caa73d11a8dc76ceb9",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=12268",
"triggerID" : "23ac68d1604e6f5e38af38caa73d11a8dc76ceb9",
"triggerType" : "PUSH"
}, {
"hash" : "e9d2494f67bf130a6a0da0c9a1433195b79a7c45",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=12287",
"triggerID" : "e9d2494f67bf130a6a0da0c9a1433195b79a7c45",
"triggerType" : "PUSH"
}, {
"hash" : "c463989c4c6afdc6a85485822d1350f6ab771a30",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=12304",
"triggerID" : "c463989c4c6afdc6a85485822d1350f6ab771a30",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* e9d2494f67bf130a6a0da0c9a1433195b79a7c45 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=12287)
* c463989c4c6afdc6a85485822d1350f6ab771a30 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=12304)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #14699: [FLINK-21011][table-planner-blink] Separate implementation of StreamExecIntervalJoin
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #14699:
URL: https://github.com/apache/flink/pull/14699#issuecomment-763287840
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "ddd72a9067d9c7b0fed65b26947d00f530d53310",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=12258",
"triggerID" : "ddd72a9067d9c7b0fed65b26947d00f530d53310",
"triggerType" : "PUSH"
}, {
"hash" : "23ac68d1604e6f5e38af38caa73d11a8dc76ceb9",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=12268",
"triggerID" : "23ac68d1604e6f5e38af38caa73d11a8dc76ceb9",
"triggerType" : "PUSH"
}, {
"hash" : "e9d2494f67bf130a6a0da0c9a1433195b79a7c45",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=12287",
"triggerID" : "e9d2494f67bf130a6a0da0c9a1433195b79a7c45",
"triggerType" : "PUSH"
}, {
"hash" : "c463989c4c6afdc6a85485822d1350f6ab771a30",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=12304",
"triggerID" : "c463989c4c6afdc6a85485822d1350f6ab771a30",
"triggerType" : "PUSH"
}, {
"hash" : "316311b1eb3e13ae6ececc71d75c02b95403214d",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=12314",
"triggerID" : "316311b1eb3e13ae6ececc71d75c02b95403214d",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* c463989c4c6afdc6a85485822d1350f6ab771a30 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=12304)
* 316311b1eb3e13ae6ececc71d75c02b95403214d Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=12314)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #14699: [FLINK-21011][table-planner-blink] Separate implementation of StreamExecIntervalJoin
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #14699:
URL: https://github.com/apache/flink/pull/14699#issuecomment-763287840
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "ddd72a9067d9c7b0fed65b26947d00f530d53310",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=12258",
"triggerID" : "ddd72a9067d9c7b0fed65b26947d00f530d53310",
"triggerType" : "PUSH"
}, {
"hash" : "23ac68d1604e6f5e38af38caa73d11a8dc76ceb9",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=12268",
"triggerID" : "23ac68d1604e6f5e38af38caa73d11a8dc76ceb9",
"triggerType" : "PUSH"
}, {
"hash" : "e9d2494f67bf130a6a0da0c9a1433195b79a7c45",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=12287",
"triggerID" : "e9d2494f67bf130a6a0da0c9a1433195b79a7c45",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* e9d2494f67bf130a6a0da0c9a1433195b79a7c45 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=12287)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] wenlong88 commented on a change in pull request #14699: [FLINK-21011][table-planner-blink] Separate implementation of StreamExecIntervalJoin
Posted by GitBox <gi...@apache.org>.
wenlong88 commented on a change in pull request #14699:
URL: https://github.com/apache/flink/pull/14699#discussion_r561513490
##########
File path: flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamPhysicalIntervalJoin.scala
##########
@@ -0,0 +1,94 @@
+/*
+ * 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.planner.plan.nodes.physical.stream
+
+import org.apache.flink.table.api.TableException
+import org.apache.flink.table.planner.calcite.FlinkTypeFactory
+import org.apache.flink.table.planner.plan.nodes.common.CommonPhysicalJoin
+import org.apache.flink.table.planner.plan.nodes.exec.{ExecEdge, ExecNode}
+import org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecIntervalJoin
+import org.apache.flink.table.planner.plan.utils.IntervalJoinUtil.WindowBounds
+import org.apache.flink.table.planner.plan.utils.PythonUtil.containsPythonCall
+import org.apache.flink.table.planner.plan.utils.RelExplainUtil.preferExpressionFormat
+
+import org.apache.calcite.plan._
+import org.apache.calcite.rel.core.{Join, JoinRelType}
+import org.apache.calcite.rel.{RelNode, RelWriter}
+import org.apache.calcite.rex.RexNode
+
+import scala.collection.JavaConversions._
+
+/**
+ * Stream physical RelNode for a time interval stream join.
+ */
+class StreamPhysicalIntervalJoin(
+ cluster: RelOptCluster,
+ traitSet: RelTraitSet,
+ leftRel: RelNode,
+ rightRel: RelNode,
+ joinType: JoinRelType,
+ condition: RexNode,
Review comment:
I have added back the originalCondition in the physical node.
##########
File path: flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdUniqueKeys.scala
##########
@@ -470,11 +470,11 @@ class FlinkRelMdUniqueKeys private extends MetadataHandler[BuiltInMetadata.Uniqu
}
def getUniqueKeys(
- rel: StreamExecIntervalJoin,
+ rel: StreamPhysicalIntervalJoin,
mq: RelMetadataQuery,
ignoreNulls: Boolean): JSet[ImmutableBitSet] = {
- val joinInfo = JoinInfo.of(rel.getLeft, rel.getRight, rel.joinCondition)
- getJoinUniqueKeys(joinInfo, rel.joinType, rel.getLeft, rel.getRight, mq, ignoreNulls)
+ val joinInfo = JoinInfo.of(rel.getLeft, rel.getRight, rel.getCondition)
Review comment:
we can not use analyzeCondition here, because join condition in interval join doesn't contain time range part
##########
File path: flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecIntervalJoin.java
##########
@@ -0,0 +1,357 @@
+/*
+ * 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.planner.plan.nodes.exec.stream;
+
+import org.apache.flink.api.common.functions.FlatMapFunction;
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.dag.Transformation;
+import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
+import org.apache.flink.streaming.api.operators.StreamFlatMap;
+import org.apache.flink.streaming.api.operators.StreamMap;
+import org.apache.flink.streaming.api.operators.co.KeyedCoProcessOperator;
+import org.apache.flink.streaming.api.transformations.OneInputTransformation;
+import org.apache.flink.streaming.api.transformations.TwoInputTransformation;
+import org.apache.flink.streaming.api.transformations.UnionTransformation;
+import org.apache.flink.table.api.TableException;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.planner.delegation.PlannerBase;
+import org.apache.flink.table.planner.plan.nodes.exec.ExecEdge;
+import org.apache.flink.table.planner.plan.nodes.exec.ExecNode;
+import org.apache.flink.table.planner.plan.nodes.exec.ExecNodeBase;
+import org.apache.flink.table.planner.plan.nodes.exec.utils.IntervalJoinSpec;
+import org.apache.flink.table.planner.plan.nodes.exec.utils.JoinSpec;
+import org.apache.flink.table.planner.plan.utils.JoinUtil;
+import org.apache.flink.table.planner.plan.utils.KeySelectorUtil;
+import org.apache.flink.table.runtime.generated.GeneratedJoinCondition;
+import org.apache.flink.table.runtime.keyselector.RowDataKeySelector;
+import org.apache.flink.table.runtime.operators.join.KeyedCoProcessOperatorWithWatermarkDelay;
+import org.apache.flink.table.runtime.operators.join.OuterJoinPaddingUtil;
+import org.apache.flink.table.runtime.operators.join.interval.IntervalJoinFunction;
+import org.apache.flink.table.runtime.operators.join.interval.ProcTimeIntervalJoin;
+import org.apache.flink.table.runtime.operators.join.interval.RowTimeIntervalJoin;
+import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.util.Collector;
+
+import org.apache.flink.shaded.guava18.com.google.common.collect.Lists;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/** {@link StreamExecNode} for a time interval stream join. */
+public class StreamExecIntervalJoin extends ExecNodeBase<RowData>
+ implements StreamExecNode<RowData> {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(StreamExecIntervalJoin.class);
+
+ private final IntervalJoinSpec intervalJoinSpec;
Review comment:
I would keep IntervalJoinSpec and remove JoinSpec and windowBounds, since they are introduced for convenience
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #14699: [FLINK-21011][table-planner-blink] Separate implementation of StreamExecIntervalJoin
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #14699:
URL: https://github.com/apache/flink/pull/14699#issuecomment-763287840
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "ddd72a9067d9c7b0fed65b26947d00f530d53310",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=12258",
"triggerID" : "ddd72a9067d9c7b0fed65b26947d00f530d53310",
"triggerType" : "PUSH"
}, {
"hash" : "23ac68d1604e6f5e38af38caa73d11a8dc76ceb9",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=12268",
"triggerID" : "23ac68d1604e6f5e38af38caa73d11a8dc76ceb9",
"triggerType" : "PUSH"
}, {
"hash" : "e9d2494f67bf130a6a0da0c9a1433195b79a7c45",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=12287",
"triggerID" : "e9d2494f67bf130a6a0da0c9a1433195b79a7c45",
"triggerType" : "PUSH"
}, {
"hash" : "c463989c4c6afdc6a85485822d1350f6ab771a30",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=12304",
"triggerID" : "c463989c4c6afdc6a85485822d1350f6ab771a30",
"triggerType" : "PUSH"
}, {
"hash" : "316311b1eb3e13ae6ececc71d75c02b95403214d",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "316311b1eb3e13ae6ececc71d75c02b95403214d",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* c463989c4c6afdc6a85485822d1350f6ab771a30 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=12304)
* 316311b1eb3e13ae6ececc71d75c02b95403214d UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] godfreyhe commented on a change in pull request #14699: [FLINK-21011][table-planner-blink] Separate implementation of StreamExecIntervalJoin
Posted by GitBox <gi...@apache.org>.
godfreyhe commented on a change in pull request #14699:
URL: https://github.com/apache/flink/pull/14699#discussion_r560693866
##########
File path: flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/operators/join/interval/IntervalJoinFunction.java
##########
@@ -0,0 +1,75 @@
+package org.apache.flink.table.runtime.operators.join.interval;
+
+import org.apache.flink.api.common.functions.RichFlatJoinFunction;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.binary.BinaryRowData;
+import org.apache.flink.table.data.utils.JoinedRowData;
+import org.apache.flink.table.runtime.generated.GeneratedJoinCondition;
+import org.apache.flink.table.runtime.generated.JoinCondition;
+import org.apache.flink.table.runtime.operators.join.NullAwareJoinHelper;
+import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
+import org.apache.flink.util.Collector;
+
+/**
+ * {@link IntervalJoinFunction} is a {@link RichFlatJoinFunction} wrappers a {@link
+ * GeneratedJoinCondition}.
+ */
+public class IntervalJoinFunction extends RichFlatJoinFunction<RowData, RowData, RowData>
Review comment:
@beyond1920 could you help review this change ? thanks
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] godfreyhe commented on a change in pull request #14699: [FLINK-21011][table-planner-blink] Separate implementation of StreamExecIntervalJoin
Posted by GitBox <gi...@apache.org>.
godfreyhe commented on a change in pull request #14699:
URL: https://github.com/apache/flink/pull/14699#discussion_r560654175
##########
File path: flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecIntervalJoin.java
##########
@@ -0,0 +1,352 @@
+/*
+ * 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.planner.plan.nodes.exec.stream;
+
+import org.apache.flink.api.common.functions.FlatMapFunction;
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.dag.Transformation;
+import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
+import org.apache.flink.streaming.api.operators.StreamFlatMap;
+import org.apache.flink.streaming.api.operators.StreamMap;
+import org.apache.flink.streaming.api.operators.co.KeyedCoProcessOperator;
+import org.apache.flink.streaming.api.transformations.OneInputTransformation;
+import org.apache.flink.streaming.api.transformations.TwoInputTransformation;
+import org.apache.flink.streaming.api.transformations.UnionTransformation;
+import org.apache.flink.table.api.TableException;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.planner.delegation.PlannerBase;
+import org.apache.flink.table.planner.plan.nodes.exec.ExecEdge;
+import org.apache.flink.table.planner.plan.nodes.exec.ExecNode;
+import org.apache.flink.table.planner.plan.nodes.exec.ExecNodeBase;
+import org.apache.flink.table.planner.plan.nodes.exec.utils.JoinSpec;
+import org.apache.flink.table.planner.plan.utils.IntervalJoinUtil;
+import org.apache.flink.table.planner.plan.utils.JoinUtil;
+import org.apache.flink.table.planner.plan.utils.KeySelectorUtil;
+import org.apache.flink.table.runtime.generated.GeneratedJoinCondition;
+import org.apache.flink.table.runtime.keyselector.RowDataKeySelector;
+import org.apache.flink.table.runtime.operators.join.KeyedCoProcessOperatorWithWatermarkDelay;
+import org.apache.flink.table.runtime.operators.join.OuterJoinPaddingUtil;
+import org.apache.flink.table.runtime.operators.join.interval.IntervalJoinFunction;
+import org.apache.flink.table.runtime.operators.join.interval.ProcTimeIntervalJoin;
+import org.apache.flink.table.runtime.operators.join.interval.RowTimeIntervalJoin;
+import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.util.Collector;
+
+import org.apache.flink.shaded.guava18.com.google.common.collect.Lists;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/** {@link StreamExecNode} for a time interval stream join. */
+public class StreamExecIntervalJoin extends ExecNodeBase<RowData>
+ implements StreamExecNode<RowData> {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(StreamExecIntervalJoin.class);
+
+ private final JoinSpec joinSpec;
+ private final IntervalJoinUtil.WindowBounds windowBounds;
+
+ public StreamExecIntervalJoin(
+ JoinSpec joinSpec,
+ IntervalJoinUtil.WindowBounds windowBounds,
Review comment:
port `WindowBounds` to java ?
##########
File path: flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/operators/join/interval/IntervalJoinFunction.java
##########
@@ -0,0 +1,75 @@
+package org.apache.flink.table.runtime.operators.join.interval;
+
+import org.apache.flink.api.common.functions.RichFlatJoinFunction;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.binary.BinaryRowData;
+import org.apache.flink.table.data.utils.JoinedRowData;
+import org.apache.flink.table.runtime.generated.GeneratedJoinCondition;
+import org.apache.flink.table.runtime.generated.JoinCondition;
+import org.apache.flink.table.runtime.operators.join.NullAwareJoinHelper;
+import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
+import org.apache.flink.util.Collector;
+
+/**
+ * {@link IntervalJoinFunction} is a {@link RichFlatJoinFunction} wrappers a {@link
+ * GeneratedJoinCondition}.
+ */
+public class IntervalJoinFunction extends RichFlatJoinFunction<RowData, RowData, RowData>
Review comment:
@beyond1920 could you help to review this change ? thanks
##########
File path: flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamPhysicalIntervalJoin.scala
##########
@@ -0,0 +1,94 @@
+/*
+ * 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.planner.plan.nodes.physical.stream
+
+import org.apache.flink.table.api.TableException
+import org.apache.flink.table.planner.calcite.FlinkTypeFactory
+import org.apache.flink.table.planner.plan.nodes.common.CommonPhysicalJoin
+import org.apache.flink.table.planner.plan.nodes.exec.{ExecEdge, ExecNode}
+import org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecIntervalJoin
+import org.apache.flink.table.planner.plan.utils.IntervalJoinUtil.WindowBounds
+import org.apache.flink.table.planner.plan.utils.PythonUtil.containsPythonCall
+import org.apache.flink.table.planner.plan.utils.RelExplainUtil.preferExpressionFormat
+
+import org.apache.calcite.plan._
+import org.apache.calcite.rel.core.{Join, JoinRelType}
+import org.apache.calcite.rel.{RelNode, RelWriter}
+import org.apache.calcite.rex.RexNode
+
+import scala.collection.JavaConversions._
+
+/**
+ * Stream physical RelNode for a time interval stream join.
+ */
+class StreamPhysicalIntervalJoin(
+ cluster: RelOptCluster,
+ traitSet: RelTraitSet,
+ leftRel: RelNode,
+ rightRel: RelNode,
+ joinType: JoinRelType,
+ condition: RexNode,
Review comment:
the condition in `StreamExecIntervalJoin` (the removed class) is the whole join condition, including interval condition and remaining condition, while the condition here is only the remaining part. This change will affect explain result.
##########
File path: flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdColumnUniqueness.scala
##########
@@ -501,14 +501,14 @@ class FlinkRelMdColumnUniqueness private extends MetadataHandler[BuiltInMetadata
}
def areColumnsUnique(
- rel: StreamExecIntervalJoin,
+ rel: StreamPhysicalIntervalJoin,
mq: RelMetadataQuery,
columns: ImmutableBitSet,
ignoreNulls: Boolean): JBoolean = {
- val joinInfo = JoinInfo.of(rel.getLeft, rel.getRight, rel.joinCondition)
+ val joinInfo = JoinInfo.of(rel.getLeft, rel.getRight, rel.getCondition)
Review comment:
use `rel.analyzeCondition()`
##########
File path: flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdUniqueKeys.scala
##########
@@ -470,11 +470,11 @@ class FlinkRelMdUniqueKeys private extends MetadataHandler[BuiltInMetadata.Uniqu
}
def getUniqueKeys(
- rel: StreamExecIntervalJoin,
+ rel: StreamPhysicalIntervalJoin,
mq: RelMetadataQuery,
ignoreNulls: Boolean): JSet[ImmutableBitSet] = {
- val joinInfo = JoinInfo.of(rel.getLeft, rel.getRight, rel.joinCondition)
- getJoinUniqueKeys(joinInfo, rel.joinType, rel.getLeft, rel.getRight, mq, ignoreNulls)
+ val joinInfo = JoinInfo.of(rel.getLeft, rel.getRight, rel.getCondition)
Review comment:
use `rel.analyzeCondition()`
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #14699: [FLINK-21011][table-planner-blink] Separate implementation of StreamExecIntervalJoin
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #14699:
URL: https://github.com/apache/flink/pull/14699#issuecomment-763280085
Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.
## Automated Checks
Last check on commit 5e4cf66cd16642cec55f3ed47c413f4dbfb856fe (Fri May 28 07:12:20 UTC 2021)
**Warnings:**
* No documentation files were touched! Remember to keep the Flink docs up to date!
<sub>Mention the bot in a comment to re-run the automated checks.</sub>
## Review Progress
* ❓ 1. The [description] looks good.
* ❓ 2. There is [consensus] that the contribution should go into to Flink.
* ❓ 3. Needs [attention] from.
* ❓ 4. The change fits into the overall [architecture].
* ❓ 5. Overall code [quality] is good.
Please see the [Pull Request Review Guide](https://flink.apache.org/contributing/reviewing-prs.html) for a full explanation of the review process.<details>
The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required <summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot approve description` to approve one or more aspects (aspects: `description`, `consensus`, `architecture` and `quality`)
- `@flinkbot approve all` to approve all aspects
- `@flinkbot approve-until architecture` to approve everything until `architecture`
- `@flinkbot attention @username1 [@username2 ..]` to require somebody's attention
- `@flinkbot disapprove architecture` to remove an approval you gave earlier
</details>
--
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:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #14699: [FLINK-21011][table-planner-blink] Separate implementation of StreamExecIntervalJoin
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #14699:
URL: https://github.com/apache/flink/pull/14699#issuecomment-763287840
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "ddd72a9067d9c7b0fed65b26947d00f530d53310",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=12258",
"triggerID" : "ddd72a9067d9c7b0fed65b26947d00f530d53310",
"triggerType" : "PUSH"
}, {
"hash" : "23ac68d1604e6f5e38af38caa73d11a8dc76ceb9",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=12268",
"triggerID" : "23ac68d1604e6f5e38af38caa73d11a8dc76ceb9",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 23ac68d1604e6f5e38af38caa73d11a8dc76ceb9 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=12268)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] wenlong88 commented on a change in pull request #14699: [FLINK-21011][table-planner-blink] Separate implementation of StreamExecIntervalJoin
Posted by GitBox <gi...@apache.org>.
wenlong88 commented on a change in pull request #14699:
URL: https://github.com/apache/flink/pull/14699#discussion_r562378877
##########
File path: flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecIntervalJoin.java
##########
@@ -0,0 +1,357 @@
+/*
+ * 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.planner.plan.nodes.exec.stream;
+
+import org.apache.flink.api.common.functions.FlatMapFunction;
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.dag.Transformation;
+import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
+import org.apache.flink.streaming.api.operators.StreamFlatMap;
+import org.apache.flink.streaming.api.operators.StreamMap;
+import org.apache.flink.streaming.api.operators.co.KeyedCoProcessOperator;
+import org.apache.flink.streaming.api.transformations.OneInputTransformation;
+import org.apache.flink.streaming.api.transformations.TwoInputTransformation;
+import org.apache.flink.streaming.api.transformations.UnionTransformation;
+import org.apache.flink.table.api.TableException;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.planner.delegation.PlannerBase;
+import org.apache.flink.table.planner.plan.nodes.exec.ExecEdge;
+import org.apache.flink.table.planner.plan.nodes.exec.ExecNode;
+import org.apache.flink.table.planner.plan.nodes.exec.ExecNodeBase;
+import org.apache.flink.table.planner.plan.nodes.exec.utils.IntervalJoinSpec;
+import org.apache.flink.table.planner.plan.nodes.exec.utils.JoinSpec;
+import org.apache.flink.table.planner.plan.utils.JoinUtil;
+import org.apache.flink.table.planner.plan.utils.KeySelectorUtil;
+import org.apache.flink.table.runtime.generated.GeneratedJoinCondition;
+import org.apache.flink.table.runtime.keyselector.RowDataKeySelector;
+import org.apache.flink.table.runtime.operators.join.KeyedCoProcessOperatorWithWatermarkDelay;
+import org.apache.flink.table.runtime.operators.join.OuterJoinPaddingUtil;
+import org.apache.flink.table.runtime.operators.join.interval.IntervalJoinFunction;
+import org.apache.flink.table.runtime.operators.join.interval.ProcTimeIntervalJoin;
+import org.apache.flink.table.runtime.operators.join.interval.RowTimeIntervalJoin;
+import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.util.Collector;
+
+import org.apache.flink.shaded.guava18.com.google.common.collect.Lists;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/** {@link StreamExecNode} for a time interval stream join. */
+public class StreamExecIntervalJoin extends ExecNodeBase<RowData>
+ implements StreamExecNode<RowData> {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(StreamExecIntervalJoin.class);
+
+ private final IntervalJoinSpec intervalJoinSpec;
Review comment:
I would keep IntervalJoinSpec and remove JoinSpec and windowBounds, since they are introduced for convenience
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #14699: [FLINK-21011][table-planner-blink] Separate implementation of StreamExecIntervalJoin
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #14699:
URL: https://github.com/apache/flink/pull/14699#issuecomment-763287840
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "ddd72a9067d9c7b0fed65b26947d00f530d53310",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=12258",
"triggerID" : "ddd72a9067d9c7b0fed65b26947d00f530d53310",
"triggerType" : "PUSH"
}, {
"hash" : "23ac68d1604e6f5e38af38caa73d11a8dc76ceb9",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=12268",
"triggerID" : "23ac68d1604e6f5e38af38caa73d11a8dc76ceb9",
"triggerType" : "PUSH"
}, {
"hash" : "e9d2494f67bf130a6a0da0c9a1433195b79a7c45",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=12287",
"triggerID" : "e9d2494f67bf130a6a0da0c9a1433195b79a7c45",
"triggerType" : "PUSH"
}, {
"hash" : "c463989c4c6afdc6a85485822d1350f6ab771a30",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=12304",
"triggerID" : "c463989c4c6afdc6a85485822d1350f6ab771a30",
"triggerType" : "PUSH"
}, {
"hash" : "316311b1eb3e13ae6ececc71d75c02b95403214d",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=12314",
"triggerID" : "316311b1eb3e13ae6ececc71d75c02b95403214d",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 316311b1eb3e13ae6ececc71d75c02b95403214d Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=12314)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #14699: [FLINK-21011][table-planner-blink] Separate implementation of StreamExecIntervalJoin
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #14699:
URL: https://github.com/apache/flink/pull/14699#issuecomment-763287840
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] beyond1920 commented on a change in pull request #14699: [FLINK-21011][table-planner-blink] Separate implementation of StreamExecIntervalJoin
Posted by GitBox <gi...@apache.org>.
beyond1920 commented on a change in pull request #14699:
URL: https://github.com/apache/flink/pull/14699#discussion_r562302355
##########
File path: flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/operators/join/interval/IntervalJoinFunction.java
##########
@@ -0,0 +1,75 @@
+package org.apache.flink.table.runtime.operators.join.interval;
+
+import org.apache.flink.api.common.functions.RichFlatJoinFunction;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.binary.BinaryRowData;
+import org.apache.flink.table.data.utils.JoinedRowData;
+import org.apache.flink.table.runtime.generated.GeneratedJoinCondition;
+import org.apache.flink.table.runtime.generated.JoinCondition;
+import org.apache.flink.table.runtime.operators.join.NullAwareJoinHelper;
+import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
+import org.apache.flink.util.Collector;
+
+/**
+ * {@link IntervalJoinFunction} is a {@link RichFlatJoinFunction} wrappers a {@link
+ * GeneratedJoinCondition}.
+ */
+public class IntervalJoinFunction extends RichFlatJoinFunction<RowData, RowData, RowData>
Review comment:
The pr is good.
It's better to add a case for `IS NOT DISTINCT FROM` in IntervalJoinITCase.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #14699: [FLINK-21011][table-planner-blink] Separate implementation of StreamExecIntervalJoin
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #14699:
URL: https://github.com/apache/flink/pull/14699#issuecomment-763287840
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "ddd72a9067d9c7b0fed65b26947d00f530d53310",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=12258",
"triggerID" : "ddd72a9067d9c7b0fed65b26947d00f530d53310",
"triggerType" : "PUSH"
}, {
"hash" : "23ac68d1604e6f5e38af38caa73d11a8dc76ceb9",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=12268",
"triggerID" : "23ac68d1604e6f5e38af38caa73d11a8dc76ceb9",
"triggerType" : "PUSH"
}, {
"hash" : "e9d2494f67bf130a6a0da0c9a1433195b79a7c45",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "e9d2494f67bf130a6a0da0c9a1433195b79a7c45",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 23ac68d1604e6f5e38af38caa73d11a8dc76ceb9 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=12268)
* e9d2494f67bf130a6a0da0c9a1433195b79a7c45 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] beyond1920 commented on a change in pull request #14699: [FLINK-21011][table-planner-blink] Separate implementation of StreamExecIntervalJoin
Posted by GitBox <gi...@apache.org>.
beyond1920 commented on a change in pull request #14699:
URL: https://github.com/apache/flink/pull/14699#discussion_r562302355
##########
File path: flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/operators/join/interval/IntervalJoinFunction.java
##########
@@ -0,0 +1,75 @@
+package org.apache.flink.table.runtime.operators.join.interval;
+
+import org.apache.flink.api.common.functions.RichFlatJoinFunction;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.binary.BinaryRowData;
+import org.apache.flink.table.data.utils.JoinedRowData;
+import org.apache.flink.table.runtime.generated.GeneratedJoinCondition;
+import org.apache.flink.table.runtime.generated.JoinCondition;
+import org.apache.flink.table.runtime.operators.join.NullAwareJoinHelper;
+import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
+import org.apache.flink.util.Collector;
+
+/**
+ * {@link IntervalJoinFunction} is a {@link RichFlatJoinFunction} wrappers a {@link
+ * GeneratedJoinCondition}.
+ */
+public class IntervalJoinFunction extends RichFlatJoinFunction<RowData, RowData, RowData>
Review comment:
The pr is good.
It's better to add a case for `IS NOT DISTINCT FROM` in IntervalJoinITCase.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #14699: [FLINK-21011][table-planner-blink] Separate implementation of StreamExecIntervalJoin
Posted by GitBox <gi...@apache.org>.
flinkbot commented on pull request #14699:
URL: https://github.com/apache/flink/pull/14699#issuecomment-763280085
Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.
## Automated Checks
Last check on commit ddd72a9067d9c7b0fed65b26947d00f530d53310 (Wed Jan 20 02:15:32 UTC 2021)
**Warnings:**
* No documentation files were touched! Remember to keep the Flink docs up to date!
<sub>Mention the bot in a comment to re-run the automated checks.</sub>
## Review Progress
* ❓ 1. The [description] looks good.
* ❓ 2. There is [consensus] that the contribution should go into to Flink.
* ❓ 3. Needs [attention] from.
* ❓ 4. The change fits into the overall [architecture].
* ❓ 5. Overall code [quality] is good.
Please see the [Pull Request Review Guide](https://flink.apache.org/contributing/reviewing-prs.html) for a full explanation of the review process.<details>
The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required <summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot approve description` to approve one or more aspects (aspects: `description`, `consensus`, `architecture` and `quality`)
- `@flinkbot approve all` to approve all aspects
- `@flinkbot approve-until architecture` to approve everything until `architecture`
- `@flinkbot attention @username1 [@username2 ..]` to require somebody's attention
- `@flinkbot disapprove architecture` to remove an approval you gave earlier
</details>
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #14699: [FLINK-21011][table-planner-blink] Separate implementation of StreamExecIntervalJoin
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #14699:
URL: https://github.com/apache/flink/pull/14699#issuecomment-763287840
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "ddd72a9067d9c7b0fed65b26947d00f530d53310",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=12258",
"triggerID" : "ddd72a9067d9c7b0fed65b26947d00f530d53310",
"triggerType" : "PUSH"
}, {
"hash" : "23ac68d1604e6f5e38af38caa73d11a8dc76ceb9",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=12268",
"triggerID" : "23ac68d1604e6f5e38af38caa73d11a8dc76ceb9",
"triggerType" : "PUSH"
}, {
"hash" : "e9d2494f67bf130a6a0da0c9a1433195b79a7c45",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=12287",
"triggerID" : "e9d2494f67bf130a6a0da0c9a1433195b79a7c45",
"triggerType" : "PUSH"
}, {
"hash" : "c463989c4c6afdc6a85485822d1350f6ab771a30",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=12304",
"triggerID" : "c463989c4c6afdc6a85485822d1350f6ab771a30",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* c463989c4c6afdc6a85485822d1350f6ab771a30 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=12304)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] godfreyhe closed pull request #14699: [FLINK-21011][table-planner-blink] Separate implementation of StreamExecIntervalJoin
Posted by GitBox <gi...@apache.org>.
godfreyhe closed pull request #14699:
URL: https://github.com/apache/flink/pull/14699
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #14699: [FLINK-21011][table-planner-blink] Separate implementation of StreamExecIntervalJoin
Posted by GitBox <gi...@apache.org>.
flinkbot commented on pull request #14699:
URL: https://github.com/apache/flink/pull/14699#issuecomment-763287840
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "ddd72a9067d9c7b0fed65b26947d00f530d53310",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "ddd72a9067d9c7b0fed65b26947d00f530d53310",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* ddd72a9067d9c7b0fed65b26947d00f530d53310 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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:
users@infra.apache.org