[GitHub] [flink] wuchong commented on a change in pull request #8901: [FLINK-13003][table-planner-blink] Support Temporal TableFunction Join in blink planner
wuchong commented on a change in pull request #8901:
[FLINK-13003][table-planner-blink] Support Temporal TableFunction Join in blink
planner
URL: https://github.com/apache/flink/pull/8901#discussion_r298484227
##
File path:
flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/rules/physical/stream/StreamExecTemporalUdtfJoinRule.scala
##
@@ -0,0 +1,102 @@
+/*
+ * 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.physical.stream
+
+import org.apache.flink.table.plan.`trait`.FlinkRelDistribution
+import org.apache.flink.table.plan.nodes.FlinkConventions
+import org.apache.flink.table.plan.nodes.logical._
+import
org.apache.flink.table.plan.nodes.physical.stream.StreamExecTemporalTableJoin
+import
org.apache.flink.table.plan.util.TemporalJoinUtil.containsTemporalJoinCondition
+import org.apache.flink.table.plan.util.{FlinkRelOptUtil, WindowJoinUtil}
+
+import org.apache.calcite.plan.{RelOptRule, RelOptRuleCall, RelTraitSet}
+import org.apache.calcite.plan.RelOptRule.{any, operand}
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.core.JoinRelType
+
+import java.util
+
+class StreamExecTemporalUdtfJoinRule
+ extends RelOptRule(
+operand(
+ classOf[FlinkLogicalJoin],
Review comment:
Sure
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:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] wuchong commented on a change in pull request #8901: [FLINK-13003][table-planner-blink] Support Temporal TableFunction Join in blink planner
wuchong commented on a change in pull request #8901:
[FLINK-13003][table-planner-blink] Support Temporal TableFunction Join in blink
planner
URL: https://github.com/apache/flink/pull/8901#discussion_r298475401
##
File path:
flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/rules/physical/stream/StreamExecTemporalUdtfJoinRule.scala
##
@@ -0,0 +1,102 @@
+/*
+ * 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.physical.stream
+
+import org.apache.flink.table.plan.`trait`.FlinkRelDistribution
+import org.apache.flink.table.plan.nodes.FlinkConventions
+import org.apache.flink.table.plan.nodes.logical._
+import
org.apache.flink.table.plan.nodes.physical.stream.StreamExecTemporalTableJoin
+import
org.apache.flink.table.plan.util.TemporalJoinUtil.containsTemporalJoinCondition
+import org.apache.flink.table.plan.util.{FlinkRelOptUtil, WindowJoinUtil}
+
+import org.apache.calcite.plan.{RelOptRule, RelOptRuleCall, RelTraitSet}
+import org.apache.calcite.plan.RelOptRule.{any, operand}
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.core.JoinRelType
+
+import java.util
+
+class StreamExecTemporalUdtfJoinRule
+ extends RelOptRule(
+operand(
+ classOf[FlinkLogicalJoin],
Review comment:
I don't think `LogicalCorrelateToLookupJoinRule` is a good idea. Because it
doesn't translate Correlate to a Lookup Join here, but only convert correlate
to a join. Currently the join will be converted to lookup join in physical.
However, we may also support convert the join to temporal join in the future.
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:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] wuchong commented on a change in pull request #8901: [FLINK-13003][table-planner-blink] Support Temporal TableFunction Join in blink planner
wuchong commented on a change in pull request #8901:
[FLINK-13003][table-planner-blink] Support Temporal TableFunction Join in blink
planner
URL: https://github.com/apache/flink/pull/8901#discussion_r298473554
##
File path:
flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/util/TemporalJoinUtil.scala
##
@@ -0,0 +1,112 @@
+/*
+ * 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.util
+
+import org.apache.calcite.rex._
+import org.apache.calcite.sql.`type`.{OperandTypes, ReturnTypes}
+import org.apache.calcite.sql.{SqlFunction, SqlFunctionCategory, SqlKind}
+import org.apache.flink.util.Preconditions.checkArgument
+
+/**
+ * Utilities for temporal table join
+ */
+object TemporalJoinUtil {
+
+ //
+ // Temporal TableFunction Join Utilities
+ //
+
+ /**
+* [[TEMPORAL_JOIN_CONDITION]] is a specific condition which correctly
defines
+* references to rightTimeAttribute, rightPrimaryKeyExpression and
leftTimeAttribute.
+* The condition is used to mark this is a temporal tablefunction join.
+* Later rightTimeAttribute, rightPrimaryKeyExpression and
leftTimeAttribute will be
+* extracted from the condition.
+*/
+ val TEMPORAL_JOIN_CONDITION = new SqlFunction(
+"__TEMPORAL_JOIN_CONDITION",
+SqlKind.OTHER_FUNCTION,
+ReturnTypes.BOOLEAN_NOT_NULL,
+null,
+OperandTypes.or(
+ OperandTypes.sequence(
+"'(LEFT_TIME_ATTRIBUTE, RIGHT_TIME_ATTRIBUTE, PRIMARY_KEY)'",
+OperandTypes.DATETIME,
+OperandTypes.DATETIME,
+OperandTypes.ANY),
+ OperandTypes.sequence(
+"'(LEFT_TIME_ATTRIBUTE, PRIMARY_KEY)'",
+OperandTypes.DATETIME,
+OperandTypes.ANY)),
+SqlFunctionCategory.SYSTEM)
+
+
+ def isRowtimeCall(call: RexCall): Boolean = {
+checkArgument(call.getOperator == TEMPORAL_JOIN_CONDITION)
+call.getOperands.size() == 3
+ }
+
+ def isProctimeCall(call: RexCall): Boolean = {
+checkArgument(call.getOperator == TEMPORAL_JOIN_CONDITION)
+call.getOperands.size() == 2
+ }
+
+ def makeRowTimeTemporalJoinConditionCall(
+rexBuilder: RexBuilder,
+leftTimeAttribute: RexNode,
+rightTimeAttribute: RexNode,
+rightPrimaryKeyExpression: RexNode): RexNode = {
+rexBuilder.makeCall(
+ TEMPORAL_JOIN_CONDITION,
+ leftTimeAttribute,
+ rightTimeAttribute,
+ rightPrimaryKeyExpression)
+ }
+
+ def makeProcTimeTemporalJoinConditionCall(
+rexBuilder: RexBuilder,
+leftTimeAttribute: RexNode,
+rightPrimaryKeyExpression: RexNode): RexNode = {
+rexBuilder.makeCall(
+ TEMPORAL_JOIN_CONDITION,
+ leftTimeAttribute,
+ rightPrimaryKeyExpression)
+ }
+
+
+ def containsTemporalJoinCondition(condition: RexNode): Boolean = {
Review comment:
Sorry, I misunderstood it.
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:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] wuchong commented on a change in pull request #8901: [FLINK-13003][table-planner-blink] Support Temporal TableFunction Join in blink planner
wuchong commented on a change in pull request #8901:
[FLINK-13003][table-planner-blink] Support Temporal TableFunction Join in blink
planner
URL: https://github.com/apache/flink/pull/8901#discussion_r298468317
##
File path:
flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/plan/stream/sql/join/TemporalUdtfJoinTest.scala
##
@@ -0,0 +1,130 @@
+/*
+ * 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.stream.sql.join
+
+import java.sql.Timestamp
+
+import org.apache.flink.api.scala._
+import org.apache.flink.table.api.TableException
+import org.apache.flink.table.util.{StreamTableTestUtil, TableTestBase}
+import org.hamcrest.Matchers.containsString
+import org.junit.Test
+
+class TemporalUdtfJoinTest extends TableTestBase {
Review comment:
Rename to `TemporalJoinTest` to align with `TemporalJoinITCase`.
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:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] wuchong commented on a change in pull request #8901: [FLINK-13003][table-planner-blink] Support Temporal TableFunction Join in blink planner
wuchong commented on a change in pull request #8901:
[FLINK-13003][table-planner-blink] Support Temporal TableFunction Join in blink
planner
URL: https://github.com/apache/flink/pull/8901#discussion_r298467837
##
File path:
flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/rules/physical/stream/StreamExecTemporalUdtfJoinRule.scala
##
@@ -0,0 +1,102 @@
+/*
+ * 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.physical.stream
+
+import org.apache.flink.table.plan.`trait`.FlinkRelDistribution
+import org.apache.flink.table.plan.nodes.FlinkConventions
+import org.apache.flink.table.plan.nodes.logical._
+import
org.apache.flink.table.plan.nodes.physical.stream.StreamExecTemporalTableJoin
+import
org.apache.flink.table.plan.util.TemporalJoinUtil.containsTemporalJoinCondition
+import org.apache.flink.table.plan.util.{FlinkRelOptUtil, WindowJoinUtil}
+
+import org.apache.calcite.plan.{RelOptRule, RelOptRuleCall, RelTraitSet}
+import org.apache.calcite.plan.RelOptRule.{any, operand}
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.core.JoinRelType
+
+import java.util
+
+class StreamExecTemporalUdtfJoinRule
+ extends RelOptRule(
+operand(
+ classOf[FlinkLogicalJoin],
Review comment:
Both of lookup join and temporal tablefunction join are correlates before
optimization. We need to convert them to joins in the first step.
I renamed them to `LogicalCorrelateToJoinFromTemporalTableRule` and
`LogicalCorrelateToJoinFromTemporalTableFunctionRule`, what do you think?
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:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] wuchong commented on a change in pull request #8901: [FLINK-13003][table-planner-blink] Support Temporal TableFunction Join in blink planner
wuchong commented on a change in pull request #8901:
[FLINK-13003][table-planner-blink] Support Temporal TableFunction Join in blink
planner
URL: https://github.com/apache/flink/pull/8901#discussion_r298456137
##
File path:
flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/join/temporal/TemporalRowTimeJoinOperator.java
##
@@ -0,0 +1,376 @@
+/*
+ * 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.runtime.join.temporal;
+
+import org.apache.flink.api.common.functions.util.FunctionUtils;
+import org.apache.flink.api.common.state.MapState;
+import org.apache.flink.api.common.state.MapStateDescriptor;
+import org.apache.flink.api.common.state.ValueState;
+import org.apache.flink.api.common.state.ValueStateDescriptor;
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.state.VoidNamespace;
+import org.apache.flink.runtime.state.VoidNamespaceSerializer;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.InternalTimer;
+import org.apache.flink.streaming.api.operators.InternalTimerService;
+import org.apache.flink.streaming.api.operators.TimestampedCollector;
+import org.apache.flink.streaming.api.operators.Triggerable;
+import org.apache.flink.streaming.api.operators.TwoInputStreamOperator;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.table.dataformat.BaseRow;
+import org.apache.flink.table.dataformat.JoinedRow;
+import org.apache.flink.table.dataformat.util.BaseRowUtil;
+import org.apache.flink.table.generated.GeneratedJoinCondition;
+import org.apache.flink.table.generated.JoinCondition;
+import org.apache.flink.table.typeutils.BaseRowTypeInfo;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.Iterator;
+import java.util.List;
+import java.util.ListIterator;
+import java.util.Map;
+import java.util.Optional;
+
+/**
+ * This operator works by keeping on the state collection of probe and build
records to process
+ * on next watermark. The idea is that between watermarks we are collecting
those elements
+ * and once we are sure that there will be no updates we emit the correct
result and clean up the
+ * state.
+ *
+ * Cleaning up the state drops all of the "old" values from the probe side,
where "old" is defined
+ * as older then the current watermark. Build side is also cleaned up in the
similar fashion,
+ * however we always keep at least one record - the latest one - even if it's
past the last
+ * watermark.
+ *
+ * One more trick is how the emitting results and cleaning up is triggered.
It is achieved
+ * by registering timers for the keys. We could register a timer for every
probe and build
+ * side element's event time (when watermark exceeds this timer, that's when
we are emitting and/or
+ * cleaning up the state). However this would cause huge number of registered
timers. For example
+ * with following evenTimes of probe records accumulated: {1, 2, 5, 8, 9}, if
we
+ * had received Watermark(10), it would trigger 5 separate timers for the same
key. To avoid that
+ * we always keep only one single registered timer for any given key,
registered for the minimal
+ * value. Upon triggering it, we process all records with event times older
then or equal to
+ * currentWatermark.
+ */
+public class TemporalRowTimeJoinOperator
Review comment:
I will support state ttl in `TemporalRowTimeJoinOperator` too.
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:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] wuchong commented on a change in pull request #8901: [FLINK-13003][table-planner-blink] Support Temporal TableFunction Join in blink planner
wuchong commented on a change in pull request #8901:
[FLINK-13003][table-planner-blink] Support Temporal TableFunction Join in blink
planner
URL: https://github.com/apache/flink/pull/8901#discussion_r298454608
##
File path:
flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/join/stream/AbstractStreamingJoinOperator.java
##
@@ -117,7 +117,9 @@ public void open() throws Exception {
@Override
public void close() throws Exception {
super.close();
- joinCondition.backingJoinCondition.close();
+ if (joinCondition != null) {
Review comment:
I think we should protect it in each close() method. The compiled instance
may not be instantiated when close() is invoked (an exception happens before
instantiation).
This change is not related to this pull request.
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:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] wuchong commented on a change in pull request #8901: [FLINK-13003][table-planner-blink] Support Temporal TableFunction Join in blink planner
wuchong commented on a change in pull request #8901:
[FLINK-13003][table-planner-blink] Support Temporal TableFunction Join in blink
planner
URL: https://github.com/apache/flink/pull/8901#discussion_r298454129
##
File path:
flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/util/TemporalJoinUtil.scala
##
@@ -0,0 +1,112 @@
+/*
+ * 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.util
+
+import org.apache.calcite.rex._
+import org.apache.calcite.sql.`type`.{OperandTypes, ReturnTypes}
+import org.apache.calcite.sql.{SqlFunction, SqlFunctionCategory, SqlKind}
+import org.apache.flink.util.Preconditions.checkArgument
+
+/**
+ * Utilities for temporal table join
+ */
+object TemporalJoinUtil {
+
+ //
+ // Temporal TableFunction Join Utilities
+ //
+
+ /**
+* [[TEMPORAL_JOIN_CONDITION]] is a specific condition which correctly
defines
+* references to rightTimeAttribute, rightPrimaryKeyExpression and
leftTimeAttribute.
+* The condition is used to mark this is a temporal tablefunction join.
+* Later rightTimeAttribute, rightPrimaryKeyExpression and
leftTimeAttribute will be
+* extracted from the condition.
+*/
+ val TEMPORAL_JOIN_CONDITION = new SqlFunction(
+"__TEMPORAL_JOIN_CONDITION",
+SqlKind.OTHER_FUNCTION,
+ReturnTypes.BOOLEAN_NOT_NULL,
+null,
+OperandTypes.or(
+ OperandTypes.sequence(
+"'(LEFT_TIME_ATTRIBUTE, RIGHT_TIME_ATTRIBUTE, PRIMARY_KEY)'",
+OperandTypes.DATETIME,
+OperandTypes.DATETIME,
+OperandTypes.ANY),
+ OperandTypes.sequence(
+"'(LEFT_TIME_ATTRIBUTE, PRIMARY_KEY)'",
+OperandTypes.DATETIME,
+OperandTypes.ANY)),
+SqlFunctionCategory.SYSTEM)
+
+
+ def isRowtimeCall(call: RexCall): Boolean = {
+checkArgument(call.getOperator == TEMPORAL_JOIN_CONDITION)
+call.getOperands.size() == 3
+ }
+
+ def isProctimeCall(call: RexCall): Boolean = {
+checkArgument(call.getOperator == TEMPORAL_JOIN_CONDITION)
+call.getOperands.size() == 2
+ }
+
+ def makeRowTimeTemporalJoinConditionCall(
+rexBuilder: RexBuilder,
+leftTimeAttribute: RexNode,
+rightTimeAttribute: RexNode,
+rightPrimaryKeyExpression: RexNode): RexNode = {
+rexBuilder.makeCall(
+ TEMPORAL_JOIN_CONDITION,
+ leftTimeAttribute,
+ rightTimeAttribute,
+ rightPrimaryKeyExpression)
+ }
+
+ def makeProcTimeTemporalJoinConditionCall(
+rexBuilder: RexBuilder,
+leftTimeAttribute: RexNode,
+rightPrimaryKeyExpression: RexNode): RexNode = {
+rexBuilder.makeCall(
+ TEMPORAL_JOIN_CONDITION,
+ leftTimeAttribute,
+ rightPrimaryKeyExpression)
+ }
+
+
+ def containsTemporalJoinCondition(condition: RexNode): Boolean = {
Review comment:
Emmm. I think a separate method to check condition is more clean here.
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:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] wuchong commented on a change in pull request #8901: [FLINK-13003][table-planner-blink] Support Temporal TableFunction Join in blink planner
wuchong commented on a change in pull request #8901:
[FLINK-13003][table-planner-blink] Support Temporal TableFunction Join in blink
planner
URL: https://github.com/apache/flink/pull/8901#discussion_r298453939
##
File path:
flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/nodes/physical/stream/StreamExecTemporalTableJoin.scala
##
@@ -0,0 +1,422 @@
+/*
+ * 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.nodes.physical.stream
+
+import org.apache.flink.api.java.typeutils.ResultTypeQueryable
+import org.apache.flink.streaming.api.operators.TwoInputStreamOperator
+import org.apache.flink.streaming.api.transformations.{StreamTransformation,
TwoInputTransformation}
+import org.apache.flink.table.api.{StreamTableEnvironment, TableConfig,
TableException, ValidationException}
+import org.apache.flink.table.calcite.FlinkTypeFactory
+import
org.apache.flink.table.calcite.FlinkTypeFactory.{isProctimeIndicatorType,
isRowtimeIndicatorType}
+import org.apache.flink.table.codegen.{CodeGeneratorContext,
ExprCodeGenerator, FunctionCodeGenerator}
+import org.apache.flink.table.dataformat.BaseRow
+import org.apache.flink.table.generated.GeneratedJoinCondition
+import org.apache.flink.table.plan.nodes.common.CommonPhysicalJoin
+import org.apache.flink.table.plan.nodes.exec.{ExecNode, StreamExecNode}
+import
org.apache.flink.table.plan.util.TemporalJoinUtil.TEMPORAL_JOIN_CONDITION
+import org.apache.flink.table.plan.util.{InputRefVisitor, KeySelectorUtil,
RelExplainUtil, TemporalJoinUtil}
+import
org.apache.flink.table.runtime.join.temporal.{TemporalProcessTimeJoinOperator,
TemporalRowTimeJoinOperator}
+import org.apache.flink.table.runtime.keyselector.BaseRowKeySelector
+import org.apache.flink.table.types.logical.RowType
+import org.apache.flink.table.typeutils.BaseRowTypeInfo
+import org.apache.flink.util.Preconditions.checkState
+import org.apache.calcite.plan.{RelOptCluster, RelTraitSet}
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.core.{Join, JoinInfo, JoinRelType}
+import org.apache.calcite.rex._
+
+import java.util
+
+import scala.collection.JavaConversions._
+
+class StreamExecTemporalTableJoin(
+cluster: RelOptCluster,
+traitSet: RelTraitSet,
+leftRel: RelNode,
+rightRel: RelNode,
+condition: RexNode,
+joinType: JoinRelType)
+ extends CommonPhysicalJoin(cluster, traitSet, leftRel, rightRel, condition,
joinType)
+ with StreamPhysicalRel
+ with StreamExecNode[BaseRow] {
+
+ override def producesUpdates: Boolean = false
+
+ override def needsUpdatesAsRetraction(input: RelNode): Boolean = false
+
+ override def consumesRetractions: Boolean = false
+
+ override def producesRetractions: Boolean = false
+
+ override def requireWatermark: Boolean = {
+val nonEquiJoinRex = getJoinInfo.getRemaining(cluster.getRexBuilder)
+
+var rowtimeJoin: Boolean = false
+val visitor = new RexVisitorImpl[Unit](true) {
+ override def visitCall(call: RexCall): Unit = {
+if (call.getOperator == TEMPORAL_JOIN_CONDITION) {
+ rowtimeJoin = TemporalJoinUtil.isRowtimeCall(call)
+} else {
+ call.getOperands.foreach(node => node.accept(this))
+}
+ }
+}
+nonEquiJoinRex.accept(visitor)
+rowtimeJoin
+ }
+
+ override def copy(
+ traitSet: RelTraitSet,
+ conditionExpr: RexNode,
+ left: RelNode,
+ right: RelNode,
+ joinType: JoinRelType,
+ semiJoinDone: Boolean): Join = {
+new StreamExecTemporalTableJoin(
+ cluster,
+ traitSet,
+ left,
+ right,
+ conditionExpr,
+ joinType)
+ }
+
+ //~ ExecNode methods
---
+
+ override def getInputNodes: util.List[ExecNode[StreamTableEnvironment, _]] =
{
+getInputs.map(_.asInstanceOf[ExecNode[StreamTableEnvironment, _]])
+ }
+
+ override def replaceInputNode(
+ordinalInParent: Int,
+newInputNode: ExecNode[StreamTableEnvironment, _]): Unit = {
+replaceInput(ordinalInParent, newInputNode.asInstanceOf[RelNode])
+ }
+
+ override protected def translateToPlanInternal(
Review comment:
I
[GitHub] [flink] wuchong commented on a change in pull request #8901: [FLINK-13003][table-planner-blink] Support Temporal TableFunction Join in blink planner
wuchong commented on a change in pull request #8901: [FLINK-13003][table-planner-blink] Support Temporal TableFunction Join in blink planner URL: https://github.com/apache/flink/pull/8901#discussion_r298453583 ## File path: flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/api/TableImpl.scala ## @@ -48,8 +48,18 @@ class TableImpl(val tableEnv: TableEnvironment, operationTree: QueryOperation) e override def select(fields: Expression*): Table = ??? override def createTemporalTableFunction( Review comment: The another way (using Expression parameter) will be supported when we integrate table-common's `TableImpl` and will add some tests after that. 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: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] wuchong commented on a change in pull request #8901: [FLINK-13003][table-planner-blink] Support Temporal TableFunction Join in blink planner
wuchong commented on a change in pull request #8901:
[FLINK-13003][table-planner-blink] Support Temporal TableFunction Join in blink
planner
URL: https://github.com/apache/flink/pull/8901#discussion_r298453442
##
File path:
flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/api/StreamTableEnvironment.scala
##
@@ -488,14 +488,14 @@ abstract class StreamTableEnvironment(
}
}
-fields.zipWithIndex.foreach {
- case ("rowtime", idx) =>
-extractRowtime(idx, "rowtime", None)
-
- case ("proctime", idx) =>
-extractProctime(idx, "proctime")
-
- case (name, _) => fieldNames = name :: fieldNames
+fields.zipWithIndex.foreach { case (name, idx) =>
Review comment:
Yes. We hacked this to make the test util to support "o_rowtime" as a
rowtime attribute. Currently, we can't invoke rowtime/proctime function in
blink-planner.
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:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] wuchong commented on a change in pull request #8901: [FLINK-13003][table-planner-blink] Support Temporal TableFunction Join in blink planner
wuchong commented on a change in pull request #8901:
[FLINK-13003][table-planner-blink] Support Temporal TableFunction Join in blink
planner
URL: https://github.com/apache/flink/pull/8901#discussion_r298446897
##
File path:
flink-table/flink-sql-parser/src/test/java/org/apache/flink/sql/parser/FlinkSqlParserImplTest.java
##
@@ -73,7 +73,7 @@ public void before() {
@Test
public void testCreateTable() {
- check("CREATE TABLE tbl1 (\n" +
+ check("CREATE TABLE db.tbl1 (\n" +
Review comment:
A mistake ...
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:
us...@infra.apache.org
With regards,
Apache Git Services
