You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by di...@apache.org on 2020/10/22 12:00:10 UTC
[flink] branch FLINK-19675 created (now 9becc81)
This is an automated email from the ASF dual-hosted git repository.
dianfu pushed a change to branch FLINK-19675
in repository https://gitbox.apache.org/repos/asf/flink.git.
at 9becc81 [FLINK-19675][python] Fix PythonCalcExpandProjectRule to handle cases when the calc node contains WHERE clause, composite fields access and Python UDF at the same time
This branch includes the following new commits:
new 9becc81 [FLINK-19675][python] Fix PythonCalcExpandProjectRule to handle cases when the calc node contains WHERE clause, composite fields access and Python UDF at the same time
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[flink] 01/01: [FLINK-19675][python] Fix
PythonCalcExpandProjectRule to handle cases when the calc node contains
WHERE clause, composite fields access and Python UDF at the same time
Posted by di...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dianfu pushed a commit to branch FLINK-19675
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 9becc81983d757b76614749a96a14a94f14aa7d6
Author: Dian Fu <di...@apache.org>
AuthorDate: Sun Oct 18 22:36:26 2020 +0800
[FLINK-19675][python] Fix PythonCalcExpandProjectRule to handle cases when the calc node contains WHERE clause, composite fields access and Python UDF at the same time
---
.../plan/rules/logical/PythonCalcSplitRule.scala | 4 ++-
.../plan/rules/logical/PythonCalcSplitRuleTest.xml | 37 ++++++++++++++++------
.../rules/logical/PythonCalcSplitRuleTest.scala | 6 ++++
.../plan/rules/logical/PythonCalcSplitRule.scala | 4 ++-
.../flink/table/plan/PythonCalcSplitRuleTest.scala | 24 ++++++++++++++
5 files changed, 64 insertions(+), 11 deletions(-)
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/PythonCalcSplitRule.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/PythonCalcSplitRule.scala
index 25f50a0..24f7abf 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/PythonCalcSplitRule.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/PythonCalcSplitRule.scala
@@ -231,7 +231,9 @@ object PythonCalcExpandProjectRule extends PythonCalcSplitRuleBase(
override def split(program: RexProgram, splitter: ScalarFunctionSplitter)
: (Option[RexNode], Option[RexNode], Seq[RexNode]) = {
- (None, None, program.getProjectList.map(program.expandLocalRef(_).accept(splitter)))
+ (Option(program.getCondition).map(program.expandLocalRef),
+ None,
+ program.getProjectList.map(program.expandLocalRef(_).accept(splitter)))
}
private def containsFieldAccessInputs(node: RexNode): Boolean = {
diff --git a/flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/PythonCalcSplitRuleTest.xml b/flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/PythonCalcSplitRuleTest.xml
index 4c17d69..5222e7a 100644
--- a/flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/PythonCalcSplitRuleTest.xml
+++ b/flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/PythonCalcSplitRuleTest.xml
@@ -387,6 +387,27 @@ FlinkLogicalCalc(select=[f0 AS EXPR$0, +(c, 1) AS EXPR$1])
]]>
</Resource>
</TestCase>
+ <TestCase name="testPythonFunctionMixedWithJavaFunctionInWhereClause">
+ <Resource name="sql">
+ <![CDATA[SELECT pyFunc1(a, b), c + 1 FROM MyTable WHERE pyFunc2(a, c) > 0]]>
+ </Resource>
+ <Resource name="planBefore">
+ <![CDATA[
+LogicalProject(EXPR$0=[pyFunc1($0, $1)], EXPR$1=[+($2, 1)])
++- LogicalFilter(condition=[>(pyFunc2($0, $2), 0)])
+ +- LogicalTableScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b, c, d)]]])
+]]>
+ </Resource>
+ <Resource name="planAfter">
+ <![CDATA[
+FlinkLogicalCalc(select=[f0 AS EXPR$0, +(c, 1) AS EXPR$1])
++- FlinkLogicalCalc(select=[c, pyFunc1(a, b) AS f0])
+ +- FlinkLogicalCalc(select=[c, a, b], where=[>(f0, 0)])
+ +- FlinkLogicalCalc(select=[a, b, c, pyFunc2(a, c) AS f0])
+ +- FlinkLogicalLegacyTableSourceScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b, c, d)]]], fields=[a, b, c, d])
+]]>
+ </Resource>
+ </TestCase>
<TestCase name="testReorderPythonCalc">
<Resource name="sql">
<![CDATA[SELECT a, pyFunc1(a, c), b FROM MyTable]]>
@@ -405,24 +426,22 @@ FlinkLogicalCalc(select=[a, f0 AS EXPR$1, b])
]]>
</Resource>
</TestCase>
- <TestCase name="testPythonFunctionMixedWithJavaFunctionInWhereClause">
+ <TestCase name="testPythonFunctionWithCompositeInputsAndWhereClause">
<Resource name="sql">
- <![CDATA[SELECT pyFunc1(a, b), c + 1 FROM MyTable WHERE pyFunc2(a, c) > 0]]>
+ <![CDATA[SELECT a, pyFunc1(b, d._1) FROM MyTable WHERE a + 1 > 0]]>
</Resource>
<Resource name="planBefore">
<![CDATA[
-LogicalProject(EXPR$0=[pyFunc1($0, $1)], EXPR$1=[+($2, 1)])
-+- LogicalFilter(condition=[>(pyFunc2($0, $2), 0)])
+LogicalProject(a=[$0], EXPR$1=[pyFunc1($1, $3._1)])
++- LogicalFilter(condition=[>(+($0, 1), 0)])
+- LogicalTableScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b, c, d)]]])
]]>
</Resource>
<Resource name="planAfter">
<![CDATA[
-FlinkLogicalCalc(select=[f0 AS EXPR$0, +(c, 1) AS EXPR$1])
-+- FlinkLogicalCalc(select=[c, pyFunc1(a, b) AS f0])
- +- FlinkLogicalCalc(select=[c, a, b], where=[>(f0, 0)])
- +- FlinkLogicalCalc(select=[a, b, c, pyFunc2(a, c) AS f0])
- +- FlinkLogicalLegacyTableSourceScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b, c, d)]]], fields=[a, b, c, d])
+FlinkLogicalCalc(select=[a, pyFunc1(b, f0) AS EXPR$1])
++- FlinkLogicalCalc(select=[a, b, d._1 AS f0], where=[>(+(a, 1), 0)])
+ +- FlinkLogicalLegacyTableSourceScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b, c, d)]]], fields=[a, b, c, d])
]]>
</Resource>
</TestCase>
diff --git a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/rules/logical/PythonCalcSplitRuleTest.scala b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/rules/logical/PythonCalcSplitRuleTest.scala
index 895c47c..67581d7 100644
--- a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/rules/logical/PythonCalcSplitRuleTest.scala
+++ b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/rules/logical/PythonCalcSplitRuleTest.scala
@@ -187,6 +187,12 @@ class PythonCalcSplitRuleTest extends TableTestBase {
}
@Test
+ def testPythonFunctionWithCompositeInputsAndWhereClause(): Unit = {
+ val sqlQuery = "SELECT a, pyFunc1(b, d._1) FROM MyTable WHERE a + 1 > 0"
+ util.verifyPlan(sqlQuery)
+ }
+
+ @Test
def testChainingPythonFunctionWithCompositeInputs(): Unit = {
val sqlQuery = "SELECT a, pyFunc1(b, pyFunc1(c, d._1)) FROM MyTable"
util.verifyPlan(sqlQuery)
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/rules/logical/PythonCalcSplitRule.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/rules/logical/PythonCalcSplitRule.scala
index 72b31de..2b73c0b 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/rules/logical/PythonCalcSplitRule.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/rules/logical/PythonCalcSplitRule.scala
@@ -231,7 +231,9 @@ object PythonCalcExpandProjectRule extends PythonCalcSplitRuleBase(
override def split(program: RexProgram, splitter: ScalarFunctionSplitter)
: (Option[RexNode], Option[RexNode], Seq[RexNode]) = {
- (None, None, program.getProjectList.map(program.expandLocalRef(_).accept(splitter)))
+ (Option(program.getCondition).map(program.expandLocalRef),
+ None,
+ program.getProjectList.map(program.expandLocalRef(_).accept(splitter)))
}
private def containsFieldAccessInputs(node: RexNode): Boolean = {
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/plan/PythonCalcSplitRuleTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/plan/PythonCalcSplitRuleTest.scala
index 973fa09..a0444fa 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/plan/PythonCalcSplitRuleTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/plan/PythonCalcSplitRuleTest.scala
@@ -584,6 +584,30 @@ class PythonCalcSplitRuleTest extends TableTestBase {
}
@Test
+ def testPythonFunctionWithCompositeInputsAndWhereClause(): Unit = {
+ val util = streamTestUtil()
+ val table = util.addTable[(Int, Int, (Int, Int))]("MyTable", 'a, 'b, 'c)
+ util.tableEnv.registerFunction("pyFunc1", new PythonScalarFunction("pyFunc1"))
+
+ val resultTable = table.select('a, 'b, 'c.flatten())
+ .select($"a", call("pyFunc1", $"b", $"c$$_1"))
+ .where($"a".plus(lit(1)).isGreater(lit(0)))
+
+ val expected = unaryNode(
+ "DataStreamPythonCalc",
+ unaryNode(
+ "DataStreamCalc",
+ streamTableNode(table),
+ term("select", "a", "b", "c._1 AS f0"),
+ term("where", ">(+(a, 1), 0)")
+ ),
+ term("select", "a", "pyFunc1(b, f0) AS _c1")
+ )
+
+ util.verifyTable(resultTable, expected)
+ }
+
+ @Test
def testPandasFunctionWithCompositeInputs(): Unit = {
val util = streamTestUtil()
val table = util.addTable[(Int, Int, (Int, Int))]("MyTable", 'a, 'b, 'c)