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 2020/02/27 13:30:40 UTC
[GitHub] [flink] HuangXingBo opened a new pull request #11242:
[FLINK-16007][table-planner][table-planner-blink][python] Add
PythonCorrelateSplitRule to push down the Java Calls contained in Python
Correlate node
HuangXingBo opened a new pull request #11242: [FLINK-16007][table-planner][table-planner-blink][python] Add PythonCorrelateSplitRule to push down the Java Calls contained in Python Correlate node
URL: https://github.com/apache/flink/pull/11242
## What is the purpose of the change
*This rule will add PythonCorrelateSplitRule to support java calls in Python UDTF*
## Brief change log
- *Add PythonCorrelateSplitRule*
## Verifying this change
This change added tests and can be verified as follows:
- *UT in PythonCorrelateSplitRuleTest*
- *IT in test_table_function_with_java_calls of test_udtf*
## 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? (yes)
- If yes, how is the feature documented? (not applicable)
----------------------------------------------------------------
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
With regards,
Apache Git Services
[GitHub] [flink] flinkbot commented on issue #11242:
[FLINK-16007][table-planner][table-planner-blink][python] Add
PythonCorrelateSplitRule to push down the Java Calls contained in Python
Correlate node
Posted by GitBox <gi...@apache.org>.
flinkbot commented on issue #11242: [FLINK-16007][table-planner][table-planner-blink][python] Add PythonCorrelateSplitRule to push down the Java Calls contained in Python Correlate node
URL: https://github.com/apache/flink/pull/11242#issuecomment-591981740
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "2de60095d8046648dd942cf0915a18c8b4a3a854",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "2de60095d8046648dd942cf0915a18c8b4a3a854",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 2de60095d8046648dd942cf0915a18c8b4a3a854 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
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11242:
[FLINK-16007][table-planner][table-planner-blink][python] Add
PythonCorrelateSplitRule to push down the Java Calls contained in Python
Correlate node
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11242: [FLINK-16007][table-planner][table-planner-blink][python] Add PythonCorrelateSplitRule to push down the Java Calls contained in Python Correlate node
URL: https://github.com/apache/flink/pull/11242#issuecomment-591981740
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "2de60095d8046648dd942cf0915a18c8b4a3a854",
"status" : "FAILURE",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5694",
"triggerID" : "2de60095d8046648dd942cf0915a18c8b4a3a854",
"triggerType" : "PUSH"
}, {
"hash" : "2de60095d8046648dd942cf0915a18c8b4a3a854",
"status" : "FAILURE",
"url" : "https://travis-ci.com/flink-ci/flink/builds/150841822",
"triggerID" : "2de60095d8046648dd942cf0915a18c8b4a3a854",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 2de60095d8046648dd942cf0915a18c8b4a3a854 Travis: [FAILURE](https://travis-ci.com/flink-ci/flink/builds/150841822) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5694)
<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
With regards,
Apache Git Services
[GitHub] [flink] hequn8128 commented on a change in pull request #11242:
[FLINK-16007][table-planner][table-planner-blink][python] Add
PythonCorrelateSplitRule to push down the Java Calls contained in Python
Correlate node
Posted by GitBox <gi...@apache.org>.
hequn8128 commented on a change in pull request #11242: [FLINK-16007][table-planner][table-planner-blink][python] Add PythonCorrelateSplitRule to push down the Java Calls contained in Python Correlate node
URL: https://github.com/apache/flink/pull/11242#discussion_r385992816
##########
File path: flink-table/flink-table-planner/src/main/java/org/apache/flink/table/plan/rules/logical/PythonCorrelateSplitRule.java
##########
@@ -0,0 +1,216 @@
+/*
+ * 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.logical;
+
+import org.apache.flink.table.plan.nodes.logical.FlinkLogicalCalc;
+import org.apache.flink.table.plan.nodes.logical.FlinkLogicalCorrelate;
+import org.apache.flink.table.plan.nodes.logical.FlinkLogicalTableFunctionScan;
+import org.apache.flink.table.plan.util.CorrelateUtil;
+import org.apache.flink.table.plan.util.PythonUtil;
+
+import org.apache.calcite.plan.RelOptRule;
+import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.plan.hep.HepRelVertex;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rex.RexBuilder;
+import org.apache.calcite.rex.RexCall;
+import org.apache.calcite.rex.RexInputRef;
+import org.apache.calcite.rex.RexLocalRef;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexProgram;
+import org.apache.calcite.rex.RexProgramBuilder;
+import org.apache.calcite.sql.validate.SqlValidatorUtil;
+
+import java.util.LinkedList;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import scala.Option;
+import scala.collection.Iterator;
+import scala.collection.mutable.ArrayBuffer;
+
+/**
+ * Rule will split the {@link FlinkLogicalTableFunctionScan} which includes java calls into a
Review comment:
Rule will split the => Rule will split the Python ..., making it more explicit.
----------------------------------------------------------------
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
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11242:
[FLINK-16007][table-planner][table-planner-blink][python] Add
PythonCorrelateSplitRule to push down the Java Calls contained in Python
Correlate node
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11242: [FLINK-16007][table-planner][table-planner-blink][python] Add PythonCorrelateSplitRule to push down the Java Calls contained in Python Correlate node
URL: https://github.com/apache/flink/pull/11242#issuecomment-591981740
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "2de60095d8046648dd942cf0915a18c8b4a3a854",
"status" : "FAILURE",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5694",
"triggerID" : "2de60095d8046648dd942cf0915a18c8b4a3a854",
"triggerType" : "PUSH"
}, {
"hash" : "2de60095d8046648dd942cf0915a18c8b4a3a854",
"status" : "FAILURE",
"url" : "https://travis-ci.com/flink-ci/flink/builds/150841822",
"triggerID" : "2de60095d8046648dd942cf0915a18c8b4a3a854",
"triggerType" : "PUSH"
}, {
"hash" : "ad21752d6e9ac8c6d2dc1c0d2d824f86d77d69c0",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "ad21752d6e9ac8c6d2dc1c0d2d824f86d77d69c0",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 2de60095d8046648dd942cf0915a18c8b4a3a854 Travis: [FAILURE](https://travis-ci.com/flink-ci/flink/builds/150841822) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5694)
* ad21752d6e9ac8c6d2dc1c0d2d824f86d77d69c0 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
With regards,
Apache Git Services
[GitHub] [flink] hequn8128 commented on a change in pull request #11242:
[FLINK-16007][table-planner][table-planner-blink][python] Add
PythonCorrelateSplitRule to push down the Java Calls contained in Python
Correlate node
Posted by GitBox <gi...@apache.org>.
hequn8128 commented on a change in pull request #11242: [FLINK-16007][table-planner][table-planner-blink][python] Add PythonCorrelateSplitRule to push down the Java Calls contained in Python Correlate node
URL: https://github.com/apache/flink/pull/11242#discussion_r385990328
##########
File path: flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/rules/FlinkBatchRuleSets.scala
##########
@@ -362,6 +362,8 @@ object FlinkBatchRuleSets {
// Rule that splits python ScalarFunctions from
// java/scala ScalarFunctions in correlate conditions
SplitPythonConditionFromCorrelateRule.INSTANCE,
+ // Rule that splits java calls from python TableFunction
+ PythonCorrelateSplitRule.INSTANCE,
Review comment:
We haven't supported python udtf for blink planner under batch mode.
----------------------------------------------------------------
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
With regards,
Apache Git Services
[GitHub] [flink] HuangXingBo commented on issue #11242:
[FLINK-16007][table-planner][table-planner-blink][python] Add
PythonCorrelateSplitRule to push down the Java Calls contained in Python
Correlate node
Posted by GitBox <gi...@apache.org>.
HuangXingBo commented on issue #11242: [FLINK-16007][table-planner][table-planner-blink][python] Add PythonCorrelateSplitRule to push down the Java Calls contained in Python Correlate node
URL: https://github.com/apache/flink/pull/11242#issuecomment-593224536
Thanks a lot for @hequn8128 review, I have addressed the comments at the latest commit.
----------------------------------------------------------------
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
With regards,
Apache Git Services
[GitHub] [flink] hequn8128 commented on a change in pull request #11242:
[FLINK-16007][table-planner][table-planner-blink][python] Add
PythonCorrelateSplitRule to push down the Java Calls contained in Python
Correlate node
Posted by GitBox <gi...@apache.org>.
hequn8128 commented on a change in pull request #11242: [FLINK-16007][table-planner][table-planner-blink][python] Add PythonCorrelateSplitRule to push down the Java Calls contained in Python Correlate node
URL: https://github.com/apache/flink/pull/11242#discussion_r385995634
##########
File path: flink-table/flink-table-planner/src/main/java/org/apache/flink/table/plan/rules/logical/PythonCorrelateSplitRule.java
##########
@@ -0,0 +1,216 @@
+/*
+ * 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.logical;
+
+import org.apache.flink.table.plan.nodes.logical.FlinkLogicalCalc;
+import org.apache.flink.table.plan.nodes.logical.FlinkLogicalCorrelate;
+import org.apache.flink.table.plan.nodes.logical.FlinkLogicalTableFunctionScan;
+import org.apache.flink.table.plan.util.CorrelateUtil;
+import org.apache.flink.table.plan.util.PythonUtil;
+
+import org.apache.calcite.plan.RelOptRule;
+import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.plan.hep.HepRelVertex;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rex.RexBuilder;
+import org.apache.calcite.rex.RexCall;
+import org.apache.calcite.rex.RexInputRef;
+import org.apache.calcite.rex.RexLocalRef;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexProgram;
+import org.apache.calcite.rex.RexProgramBuilder;
+import org.apache.calcite.sql.validate.SqlValidatorUtil;
+
+import java.util.LinkedList;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import scala.Option;
+import scala.collection.Iterator;
+import scala.collection.mutable.ArrayBuffer;
+
+/**
+ * Rule will split the {@link FlinkLogicalTableFunctionScan} which includes java calls into a
+ * {@link FlinkLogicalCalc} which will be the left input of the new {@link FlinkLogicalCorrelate}
+ * and a new {@link FlinkLogicalTableFunctionScan} without java calls.
+ */
+public class PythonCorrelateSplitRule extends RelOptRule {
+
+ public static final PythonCorrelateSplitRule INSTANCE = new PythonCorrelateSplitRule();
+
+ private PythonCorrelateSplitRule() {
+ super(operand(FlinkLogicalCorrelate.class, any()), "PythonCorrelateSplitRule");
+ }
+
+ private FlinkLogicalTableFunctionScan createNewScan(
+ FlinkLogicalTableFunctionScan scan,
+ ScalarFunctionSplitter splitter) {
+ RexCall rightRexCall = (RexCall) scan.getCall();
+ // extract java funcs.
+ List<RexNode> rightCalcProjects = rightRexCall
+ .getOperands()
+ .stream()
+ .map(x -> x.accept(splitter))
+ .collect(Collectors.toList());
+
+ RexCall newRightRexCall = rightRexCall.clone(rightRexCall.getType(), rightCalcProjects);
+ return new FlinkLogicalTableFunctionScan(
+ scan.getCluster(),
+ scan.getTraitSet(),
+ scan.getInputs(),
+ newRightRexCall,
+ scan.getElementType(),
+ scan.getRowType(),
+ scan.getColumnMappings());
+ }
+
+ @Override
+ public boolean matches(RelOptRuleCall call) {
+ FlinkLogicalCorrelate correlate = call.rel(0);
+ RelNode right = ((HepRelVertex) correlate.getRight()).getCurrentRel();
+ FlinkLogicalTableFunctionScan pythonTableFuncScan;
+ if (right instanceof FlinkLogicalTableFunctionScan) {
+ pythonTableFuncScan = (FlinkLogicalTableFunctionScan) right;
+ } else if (right instanceof FlinkLogicalCalc) {
+ Option<FlinkLogicalTableFunctionScan> scan = CorrelateUtil
+ .getTableFunctionScan((FlinkLogicalCalc) right);
+ if (scan.isEmpty()) {
+ return false;
+ }
+ pythonTableFuncScan = scan.get();
+ } else {
+ return false;
+ }
+ RexNode rexNode = pythonTableFuncScan.getCall();
+ if (rexNode instanceof RexCall) {
+ RexCall pythonTableFuncRexCall = (RexCall) rexNode;
+ return PythonUtil.isPythonCall(pythonTableFuncRexCall) &&
+ pythonTableFuncRexCall.getOperands().stream().anyMatch(PythonUtil::isNonPythonCall);
+ }
+ return false;
+ }
+
+ @Override
+ public void onMatch(RelOptRuleCall call) {
+ FlinkLogicalCorrelate correlate = call.rel(0);
+ RexBuilder rexBuilder = call.builder().getRexBuilder();
+ RelNode left = ((HepRelVertex) correlate.getLeft()).getCurrentRel();
+ RelNode right = ((HepRelVertex) correlate.getRight()).getCurrentRel();
+ int primitiveLeftFieldCount = left.getRowType().getFieldCount();
+ ArrayBuffer<RexCall> extractedJavaRexCalls = new ArrayBuffer<>();
+ ScalarFunctionSplitter splitter = new ScalarFunctionSplitter(
+ primitiveLeftFieldCount,
+ extractedJavaRexCalls,
+ false
+ );
+
+ RelNode rightNewInput;
+ if (right instanceof FlinkLogicalTableFunctionScan) {
+ FlinkLogicalTableFunctionScan scan = (FlinkLogicalTableFunctionScan) right;
+ rightNewInput = createNewScan(scan, splitter);
+ } else {
+ FlinkLogicalCalc calc = (FlinkLogicalCalc) right;
+ FlinkLogicalTableFunctionScan scan = CorrelateUtil.getTableFunctionScan(calc).get();
+ FlinkLogicalCalc mergedCalc = CorrelateUtil.getMergedCalc(calc);
+ RexProgram program = mergedCalc.getProgram();
+ FlinkLogicalTableFunctionScan newScan = createNewScan(scan, splitter);
+ RexLocalRef condition = program.getCondition();
+ rightNewInput = new FlinkLogicalCalc(
+ correlate.getCluster(),
+ correlate.getTraitSet(),
+ newScan,
+ RexProgram.create(
+ newScan.getRowType(),
+ mergedCalc.getProgram().getProjectList(),
+ condition == null ? null : program.expandLocalRef(condition),
+ mergedCalc.getRowType(),
+ rexBuilder));
Review comment:
Do we need to merge Calcs here? This can be done in the CalcMergeRule.
----------------------------------------------------------------
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
With regards,
Apache Git Services
[GitHub] [flink] hequn8128 commented on a change in pull request #11242:
[FLINK-16007][table-planner][table-planner-blink][python] Add
PythonCorrelateSplitRule to push down the Java Calls contained in Python
Correlate node
Posted by GitBox <gi...@apache.org>.
hequn8128 commented on a change in pull request #11242: [FLINK-16007][table-planner][table-planner-blink][python] Add PythonCorrelateSplitRule to push down the Java Calls contained in Python Correlate node
URL: https://github.com/apache/flink/pull/11242#discussion_r386241502
##########
File path: flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/rules/logical/PythonCorrelateSplitRule.java
##########
@@ -0,0 +1,238 @@
+/*
+ * 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.rules.logical;
+
+import org.apache.flink.table.planner.plan.nodes.logical.FlinkLogicalCalc;
+import org.apache.flink.table.planner.plan.nodes.logical.FlinkLogicalCorrelate;
+import org.apache.flink.table.planner.plan.nodes.logical.FlinkLogicalTableFunctionScan;
+import org.apache.flink.table.planner.plan.rules.physical.stream.StreamExecCorrelateRule;
+import org.apache.flink.table.planner.plan.utils.PythonUtil;
+
+import org.apache.calcite.plan.RelOptRule;
+import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.plan.hep.HepRelVertex;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rex.RexBuilder;
+import org.apache.calcite.rex.RexCall;
+import org.apache.calcite.rex.RexInputRef;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexProgram;
+import org.apache.calcite.rex.RexProgramBuilder;
+import org.apache.calcite.sql.validate.SqlValidatorUtil;
+
+import java.util.LinkedList;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import scala.collection.Iterator;
+import scala.collection.mutable.ArrayBuffer;
+
+/**
+ * Rule will split the Python {@link FlinkLogicalTableFunctionScan} which includes java calls into a
+ * {@link FlinkLogicalCalc} which will be the left input of the new {@link FlinkLogicalCorrelate}
Review comment:
Remove the blank after `new`.
----------------------------------------------------------------
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
With regards,
Apache Git Services
[GitHub] [flink] hequn8128 closed pull request #11242:
[FLINK-16007][table-planner][table-planner-blink][python] Add
PythonCorrelateSplitRule to push down the Java Calls contained in Python
Correlate node
Posted by GitBox <gi...@apache.org>.
hequn8128 closed pull request #11242: [FLINK-16007][table-planner][table-planner-blink][python] Add PythonCorrelateSplitRule to push down the Java Calls contained in Python Correlate node
URL: https://github.com/apache/flink/pull/11242
----------------------------------------------------------------
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
With regards,
Apache Git Services
[GitHub] [flink] hequn8128 commented on a change in pull request #11242:
[FLINK-16007][table-planner][table-planner-blink][python] Add
PythonCorrelateSplitRule to push down the Java Calls contained in Python
Correlate node
Posted by GitBox <gi...@apache.org>.
hequn8128 commented on a change in pull request #11242: [FLINK-16007][table-planner][table-planner-blink][python] Add PythonCorrelateSplitRule to push down the Java Calls contained in Python Correlate node
URL: https://github.com/apache/flink/pull/11242#discussion_r385996106
##########
File path: flink-table/flink-table-planner/src/main/java/org/apache/flink/table/plan/rules/logical/PythonCorrelateSplitRule.java
##########
@@ -0,0 +1,216 @@
+/*
+ * 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.logical;
+
+import org.apache.flink.table.plan.nodes.logical.FlinkLogicalCalc;
+import org.apache.flink.table.plan.nodes.logical.FlinkLogicalCorrelate;
+import org.apache.flink.table.plan.nodes.logical.FlinkLogicalTableFunctionScan;
+import org.apache.flink.table.plan.util.CorrelateUtil;
+import org.apache.flink.table.plan.util.PythonUtil;
+
+import org.apache.calcite.plan.RelOptRule;
+import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.plan.hep.HepRelVertex;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rex.RexBuilder;
+import org.apache.calcite.rex.RexCall;
+import org.apache.calcite.rex.RexInputRef;
+import org.apache.calcite.rex.RexLocalRef;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexProgram;
+import org.apache.calcite.rex.RexProgramBuilder;
+import org.apache.calcite.sql.validate.SqlValidatorUtil;
+
+import java.util.LinkedList;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import scala.Option;
+import scala.collection.Iterator;
+import scala.collection.mutable.ArrayBuffer;
+
+/**
+ * Rule will split the {@link FlinkLogicalTableFunctionScan} which includes java calls into a
+ * {@link FlinkLogicalCalc} which will be the left input of the new {@link FlinkLogicalCorrelate}
+ * and a new {@link FlinkLogicalTableFunctionScan} without java calls.
+ */
+public class PythonCorrelateSplitRule extends RelOptRule {
+
+ public static final PythonCorrelateSplitRule INSTANCE = new PythonCorrelateSplitRule();
+
+ private PythonCorrelateSplitRule() {
+ super(operand(FlinkLogicalCorrelate.class, any()), "PythonCorrelateSplitRule");
+ }
+
+ private FlinkLogicalTableFunctionScan createNewScan(
+ FlinkLogicalTableFunctionScan scan,
+ ScalarFunctionSplitter splitter) {
+ RexCall rightRexCall = (RexCall) scan.getCall();
+ // extract java funcs.
+ List<RexNode> rightCalcProjects = rightRexCall
+ .getOperands()
+ .stream()
+ .map(x -> x.accept(splitter))
+ .collect(Collectors.toList());
+
+ RexCall newRightRexCall = rightRexCall.clone(rightRexCall.getType(), rightCalcProjects);
+ return new FlinkLogicalTableFunctionScan(
+ scan.getCluster(),
+ scan.getTraitSet(),
+ scan.getInputs(),
+ newRightRexCall,
+ scan.getElementType(),
+ scan.getRowType(),
+ scan.getColumnMappings());
+ }
+
+ @Override
+ public boolean matches(RelOptRuleCall call) {
+ FlinkLogicalCorrelate correlate = call.rel(0);
+ RelNode right = ((HepRelVertex) correlate.getRight()).getCurrentRel();
+ FlinkLogicalTableFunctionScan pythonTableFuncScan;
+ if (right instanceof FlinkLogicalTableFunctionScan) {
+ pythonTableFuncScan = (FlinkLogicalTableFunctionScan) right;
+ } else if (right instanceof FlinkLogicalCalc) {
+ Option<FlinkLogicalTableFunctionScan> scan = CorrelateUtil
+ .getTableFunctionScan((FlinkLogicalCalc) right);
+ if (scan.isEmpty()) {
+ return false;
+ }
+ pythonTableFuncScan = scan.get();
+ } else {
+ return false;
+ }
+ RexNode rexNode = pythonTableFuncScan.getCall();
+ if (rexNode instanceof RexCall) {
+ RexCall pythonTableFuncRexCall = (RexCall) rexNode;
+ return PythonUtil.isPythonCall(pythonTableFuncRexCall) &&
+ pythonTableFuncRexCall.getOperands().stream().anyMatch(PythonUtil::isNonPythonCall);
+ }
+ return false;
+ }
+
+ @Override
+ public void onMatch(RelOptRuleCall call) {
Review comment:
Could you extract some smaller methods from this method? The code will be more readable.
----------------------------------------------------------------
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
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11242:
[FLINK-16007][table-planner][table-planner-blink][python] Add
PythonCorrelateSplitRule to push down the Java Calls contained in Python
Correlate node
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11242: [FLINK-16007][table-planner][table-planner-blink][python] Add PythonCorrelateSplitRule to push down the Java Calls contained in Python Correlate node
URL: https://github.com/apache/flink/pull/11242#issuecomment-591981740
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "2de60095d8046648dd942cf0915a18c8b4a3a854",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5694",
"triggerID" : "2de60095d8046648dd942cf0915a18c8b4a3a854",
"triggerType" : "PUSH"
}, {
"hash" : "2de60095d8046648dd942cf0915a18c8b4a3a854",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/150841822",
"triggerID" : "2de60095d8046648dd942cf0915a18c8b4a3a854",
"triggerType" : "PUSH"
}, {
"hash" : "ad21752d6e9ac8c6d2dc1c0d2d824f86d77d69c0",
"status" : "SUCCESS",
"url" : "https://travis-ci.com/flink-ci/flink/builds/151299714",
"triggerID" : "ad21752d6e9ac8c6d2dc1c0d2d824f86d77d69c0",
"triggerType" : "PUSH"
}, {
"hash" : "ad21752d6e9ac8c6d2dc1c0d2d824f86d77d69c0",
"status" : "FAILURE",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5796",
"triggerID" : "ad21752d6e9ac8c6d2dc1c0d2d824f86d77d69c0",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* ad21752d6e9ac8c6d2dc1c0d2d824f86d77d69c0 Travis: [SUCCESS](https://travis-ci.com/flink-ci/flink/builds/151299714) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5796)
<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
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11242:
[FLINK-16007][table-planner][table-planner-blink][python] Add
PythonCorrelateSplitRule to push down the Java Calls contained in Python
Correlate node
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11242: [FLINK-16007][table-planner][table-planner-blink][python] Add PythonCorrelateSplitRule to push down the Java Calls contained in Python Correlate node
URL: https://github.com/apache/flink/pull/11242#issuecomment-591981740
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "2de60095d8046648dd942cf0915a18c8b4a3a854",
"status" : "PENDING",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5694",
"triggerID" : "2de60095d8046648dd942cf0915a18c8b4a3a854",
"triggerType" : "PUSH"
}, {
"hash" : "2de60095d8046648dd942cf0915a18c8b4a3a854",
"status" : "PENDING",
"url" : "https://travis-ci.com/flink-ci/flink/builds/150841822",
"triggerID" : "2de60095d8046648dd942cf0915a18c8b4a3a854",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 2de60095d8046648dd942cf0915a18c8b4a3a854 Travis: [PENDING](https://travis-ci.com/flink-ci/flink/builds/150841822) Azure: [PENDING](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5694)
<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
With regards,
Apache Git Services
[GitHub] [flink] hequn8128 commented on a change in pull request #11242:
[FLINK-16007][table-planner][table-planner-blink][python] Add
PythonCorrelateSplitRule to push down the Java Calls contained in Python
Correlate node
Posted by GitBox <gi...@apache.org>.
hequn8128 commented on a change in pull request #11242: [FLINK-16007][table-planner][table-planner-blink][python] Add PythonCorrelateSplitRule to push down the Java Calls contained in Python Correlate node
URL: https://github.com/apache/flink/pull/11242#discussion_r385994214
##########
File path: flink-table/flink-table-planner/src/main/java/org/apache/flink/table/plan/rules/logical/PythonCorrelateSplitRule.java
##########
@@ -0,0 +1,216 @@
+/*
+ * 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.logical;
+
+import org.apache.flink.table.plan.nodes.logical.FlinkLogicalCalc;
+import org.apache.flink.table.plan.nodes.logical.FlinkLogicalCorrelate;
+import org.apache.flink.table.plan.nodes.logical.FlinkLogicalTableFunctionScan;
+import org.apache.flink.table.plan.util.CorrelateUtil;
+import org.apache.flink.table.plan.util.PythonUtil;
+
+import org.apache.calcite.plan.RelOptRule;
+import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.plan.hep.HepRelVertex;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rex.RexBuilder;
+import org.apache.calcite.rex.RexCall;
+import org.apache.calcite.rex.RexInputRef;
+import org.apache.calcite.rex.RexLocalRef;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexProgram;
+import org.apache.calcite.rex.RexProgramBuilder;
+import org.apache.calcite.sql.validate.SqlValidatorUtil;
+
+import java.util.LinkedList;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import scala.Option;
+import scala.collection.Iterator;
+import scala.collection.mutable.ArrayBuffer;
+
+/**
+ * Rule will split the {@link FlinkLogicalTableFunctionScan} which includes java calls into a
+ * {@link FlinkLogicalCalc} which will be the left input of the new {@link FlinkLogicalCorrelate}
+ * and a new {@link FlinkLogicalTableFunctionScan} without java calls.
+ */
+public class PythonCorrelateSplitRule extends RelOptRule {
+
+ public static final PythonCorrelateSplitRule INSTANCE = new PythonCorrelateSplitRule();
+
+ private PythonCorrelateSplitRule() {
+ super(operand(FlinkLogicalCorrelate.class, any()), "PythonCorrelateSplitRule");
+ }
+
+ private FlinkLogicalTableFunctionScan createNewScan(
+ FlinkLogicalTableFunctionScan scan,
+ ScalarFunctionSplitter splitter) {
+ RexCall rightRexCall = (RexCall) scan.getCall();
+ // extract java funcs.
+ List<RexNode> rightCalcProjects = rightRexCall
+ .getOperands()
+ .stream()
+ .map(x -> x.accept(splitter))
+ .collect(Collectors.toList());
+
+ RexCall newRightRexCall = rightRexCall.clone(rightRexCall.getType(), rightCalcProjects);
+ return new FlinkLogicalTableFunctionScan(
+ scan.getCluster(),
+ scan.getTraitSet(),
+ scan.getInputs(),
+ newRightRexCall,
+ scan.getElementType(),
+ scan.getRowType(),
+ scan.getColumnMappings());
+ }
+
+ @Override
+ public boolean matches(RelOptRuleCall call) {
+ FlinkLogicalCorrelate correlate = call.rel(0);
+ RelNode right = ((HepRelVertex) correlate.getRight()).getCurrentRel();
+ FlinkLogicalTableFunctionScan pythonTableFuncScan;
+ if (right instanceof FlinkLogicalTableFunctionScan) {
+ pythonTableFuncScan = (FlinkLogicalTableFunctionScan) right;
+ } else if (right instanceof FlinkLogicalCalc) {
+ Option<FlinkLogicalTableFunctionScan> scan = CorrelateUtil
+ .getTableFunctionScan((FlinkLogicalCalc) right);
+ if (scan.isEmpty()) {
+ return false;
+ }
+ pythonTableFuncScan = scan.get();
+ } else {
+ return false;
+ }
+ RexNode rexNode = pythonTableFuncScan.getCall();
+ if (rexNode instanceof RexCall) {
+ RexCall pythonTableFuncRexCall = (RexCall) rexNode;
+ return PythonUtil.isPythonCall(pythonTableFuncRexCall) &&
+ pythonTableFuncRexCall.getOperands().stream().anyMatch(PythonUtil::isNonPythonCall);
+ }
Review comment:
Change to
```
RexNode rexNode = pythonTableFuncScan.getCall();
if (rexNode instanceof RexCall) {
return PythonUtil.isPythonCall(rexNode) && PythonUtil.containsNonPythonCall(rexNode);
}
```
----------------------------------------------------------------
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
With regards,
Apache Git Services
[GitHub] [flink] hequn8128 commented on a change in pull request #11242:
[FLINK-16007][table-planner][table-planner-blink][python] Add
PythonCorrelateSplitRule to push down the Java Calls contained in Python
Correlate node
Posted by GitBox <gi...@apache.org>.
hequn8128 commented on a change in pull request #11242: [FLINK-16007][table-planner][table-planner-blink][python] Add PythonCorrelateSplitRule to push down the Java Calls contained in Python Correlate node
URL: https://github.com/apache/flink/pull/11242#discussion_r385988151
##########
File path: flink-python/pyflink/table/tests/test_udtf.py
##########
@@ -50,6 +50,23 @@ def test_table_function(self):
["1,0,null", "1,1,null", "2,0,null", "2,1,null", "3,0,0", "3,0,1",
"3,0,2", "3,1,1", "3,1,2", "3,2,2", "3,3,null"])
+ def test_table_function_with_java_calls(self):
Review comment:
Since we have Rule tests, we don't need the IT test.
----------------------------------------------------------------
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
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11242:
[FLINK-16007][table-planner][table-planner-blink][python] Add
PythonCorrelateSplitRule to push down the Java Calls contained in Python
Correlate node
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11242: [FLINK-16007][table-planner][table-planner-blink][python] Add PythonCorrelateSplitRule to push down the Java Calls contained in Python Correlate node
URL: https://github.com/apache/flink/pull/11242#issuecomment-591981740
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "2de60095d8046648dd942cf0915a18c8b4a3a854",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5694",
"triggerID" : "2de60095d8046648dd942cf0915a18c8b4a3a854",
"triggerType" : "PUSH"
}, {
"hash" : "2de60095d8046648dd942cf0915a18c8b4a3a854",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/150841822",
"triggerID" : "2de60095d8046648dd942cf0915a18c8b4a3a854",
"triggerType" : "PUSH"
}, {
"hash" : "ad21752d6e9ac8c6d2dc1c0d2d824f86d77d69c0",
"status" : "PENDING",
"url" : "https://travis-ci.com/flink-ci/flink/builds/151299714",
"triggerID" : "ad21752d6e9ac8c6d2dc1c0d2d824f86d77d69c0",
"triggerType" : "PUSH"
}, {
"hash" : "ad21752d6e9ac8c6d2dc1c0d2d824f86d77d69c0",
"status" : "FAILURE",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5796",
"triggerID" : "ad21752d6e9ac8c6d2dc1c0d2d824f86d77d69c0",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* ad21752d6e9ac8c6d2dc1c0d2d824f86d77d69c0 Travis: [PENDING](https://travis-ci.com/flink-ci/flink/builds/151299714) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5796)
<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
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11242:
[FLINK-16007][table-planner][table-planner-blink][python] Add
PythonCorrelateSplitRule to push down the Java Calls contained in Python
Correlate node
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11242: [FLINK-16007][table-planner][table-planner-blink][python] Add PythonCorrelateSplitRule to push down the Java Calls contained in Python Correlate node
URL: https://github.com/apache/flink/pull/11242#issuecomment-591981740
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "2de60095d8046648dd942cf0915a18c8b4a3a854",
"status" : "FAILURE",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5694",
"triggerID" : "2de60095d8046648dd942cf0915a18c8b4a3a854",
"triggerType" : "PUSH"
}, {
"hash" : "2de60095d8046648dd942cf0915a18c8b4a3a854",
"status" : "FAILURE",
"url" : "https://travis-ci.com/flink-ci/flink/builds/150841822",
"triggerID" : "2de60095d8046648dd942cf0915a18c8b4a3a854",
"triggerType" : "PUSH"
}, {
"hash" : "ad21752d6e9ac8c6d2dc1c0d2d824f86d77d69c0",
"status" : "PENDING",
"url" : "https://travis-ci.com/flink-ci/flink/builds/151299714",
"triggerID" : "ad21752d6e9ac8c6d2dc1c0d2d824f86d77d69c0",
"triggerType" : "PUSH"
}, {
"hash" : "ad21752d6e9ac8c6d2dc1c0d2d824f86d77d69c0",
"status" : "PENDING",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5796",
"triggerID" : "ad21752d6e9ac8c6d2dc1c0d2d824f86d77d69c0",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 2de60095d8046648dd942cf0915a18c8b4a3a854 Travis: [FAILURE](https://travis-ci.com/flink-ci/flink/builds/150841822) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5694)
* ad21752d6e9ac8c6d2dc1c0d2d824f86d77d69c0 Travis: [PENDING](https://travis-ci.com/flink-ci/flink/builds/151299714) Azure: [PENDING](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5796)
<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
With regards,
Apache Git Services
[GitHub] [flink] flinkbot commented on issue #11242:
[FLINK-16007][table-planner][table-planner-blink][python] Add
PythonCorrelateSplitRule to push down the Java Calls contained in Python
Correlate node
Posted by GitBox <gi...@apache.org>.
flinkbot commented on issue #11242: [FLINK-16007][table-planner][table-planner-blink][python] Add PythonCorrelateSplitRule to push down the Java Calls contained in Python Correlate node
URL: https://github.com/apache/flink/pull/11242#issuecomment-591972899
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 2de60095d8046648dd942cf0915a18c8b4a3a854 (Thu Feb 27 13:33:56 UTC 2020)
**Warnings:**
* No documentation files were touched! Remember to keep the Flink docs up to date!
* **This pull request references an unassigned [Jira ticket](https://issues.apache.org/jira/browse/FLINK-16007).** According to the [code contribution guide](https://flink.apache.org/contributing/contribute-code.html), tickets need to be assigned before starting with the implementation work.
<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
With regards,
Apache Git Services