You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by va...@apache.org on 2016/03/18 14:48:15 UTC
[21/50] [abbrv] flink git commit: [FLINK-3537] Fix code gen for
disjunctions.
[FLINK-3537] Fix code gen for disjunctions.
This closes #1733
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/ad3a6366
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/ad3a6366
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/ad3a6366
Branch: refs/heads/master
Commit: ad3a636605de417e6e44785faf22e81138d16e74
Parents: ab16e3c
Author: Fabian Hueske <fh...@apache.org>
Authored: Mon Feb 29 15:00:06 2016 +0100
Committer: vasia <va...@apache.org>
Committed: Fri Mar 18 14:44:50 2016 +0100
----------------------------------------------------------------------
.../api/table/codegen/calls/ScalarOperators.scala | 2 +-
.../flink/api/java/table/test/FilterITCase.java | 16 ++++++++++++++++
.../flink/api/scala/table/test/FilterITCase.scala | 12 ++++++++++++
.../flink/api/scala/table/test/JoinITCase.scala | 17 +++++++++++++++++
4 files changed, 46 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/ad3a6366/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/calls/ScalarOperators.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/calls/ScalarOperators.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/calls/ScalarOperators.scala
index 8580b25..f71b643 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/calls/ScalarOperators.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/calls/ScalarOperators.scala
@@ -303,7 +303,7 @@ object ScalarOperators {
s"""
|${left.code}
|${right.code}
- |boolean $resultTerm = ${left.resultTerm} && ${right.resultTerm};
+ |boolean $resultTerm = ${left.resultTerm} || ${right.resultTerm};
|""".stripMargin
}
http://git-wip-us.apache.org/repos/asf/flink/blob/ad3a6366/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/FilterITCase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/FilterITCase.java b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/FilterITCase.java
index c69d1a7..c783524 100644
--- a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/FilterITCase.java
+++ b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/FilterITCase.java
@@ -126,6 +126,22 @@ public class FilterITCase extends TableProgramsTestBase {
}
@Test
+ public void testDisjunctivePreds() throws Exception {
+ ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+ TableEnvironment tableEnv = getJavaTableEnvironment();
+
+ DataSet<Tuple3<Integer, Long, String>> input = CollectionDataSets.get3TupleDataSet(env);
+
+ Table table = tableEnv.fromDataSet(input, "a, b, c");
+ Table result = table.filter("a < 2 || a > 20");
+
+ DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
+ List<Row> results = ds.collect();
+ String expected = "1,1,Hi\n" + "21,6,Comment#15\n";
+ compareResultAsText(results, expected);
+ }
+
+ @Test
public void testIntegerBiggerThan128() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
TableEnvironment tableEnv = getJavaTableEnvironment();
http://git-wip-us.apache.org/repos/asf/flink/blob/ad3a6366/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/FilterITCase.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/FilterITCase.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/FilterITCase.scala
index 2dfdb2c..0febd4d 100644
--- a/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/FilterITCase.scala
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/FilterITCase.scala
@@ -126,6 +126,18 @@ class FilterITCase(
}
@Test
+ def testDisjunctivePreds(): Unit = {
+
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ val ds = CollectionDataSets.get3TupleDataSet(env).as('a, 'b, 'c)
+
+ val filterDs = ds.filter( 'a < 2 || 'a > 20)
+ val expected = "1,1,Hi\n" + "21,6,Comment#15\n"
+ val results = filterDs.collect()
+ TestBaseUtils.compareResultAsText(results.asJava, expected)
+ }
+
+ @Test
def testFilterMerge(): Unit = {
// verify FilterMergeRule.
http://git-wip-us.apache.org/repos/asf/flink/blob/ad3a6366/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/JoinITCase.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/JoinITCase.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/JoinITCase.scala
index b11e4f1..bc9db59 100644
--- a/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/JoinITCase.scala
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/JoinITCase.scala
@@ -180,4 +180,21 @@ class JoinITCase(mode: TestExecutionMode) extends MultipleProgramsTestBase(mode)
val results = joinT.toDataSet[Row]collect()
TestBaseUtils.compareResultAsText(results.asJava, expected)
}
+
+ @Test
+ def testJoinWithDisjunctivePred(): Unit = {
+
+ val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
+ val ds1 = CollectionDataSets.get3TupleDataSet(env).as('a, 'b, 'c)
+ val ds2 = CollectionDataSets.get5TupleDataSet(env).as('d, 'e, 'f, 'g, 'h)
+
+ val joinT = ds1.join(ds2).filter('a === 'd && ('b === 'e || 'b === 'e - 10 )).select('c, 'g)
+
+ val expected =
+ "Hi,Hallo\n" +
+ "Hello,Hallo Welt\n" +
+ "I am fine.,IJK"
+ val results = joinT.toDataSet[Row]collect()
+ TestBaseUtils.compareResultAsText(results.asJava, expected)
+ }
}