You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by rm...@apache.org on 2016/01/14 17:16:15 UTC
[18/50] flink git commit: [FLINK-1712] Remove "flink-staging" module
http://git-wip-us.apache.org/repos/asf/flink/blob/e9bf13d8/flink-staging/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/CastingITCase.scala
----------------------------------------------------------------------
diff --git a/flink-staging/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/CastingITCase.scala b/flink-staging/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/CastingITCase.scala
deleted file mode 100644
index c177184..0000000
--- a/flink-staging/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/CastingITCase.scala
+++ /dev/null
@@ -1,125 +0,0 @@
-/*
- * 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.api.scala.table.test
-
-import java.util.Date
-
-import org.junit._
-import org.junit.runner.RunWith
-import org.junit.runners.Parameterized
-
-import org.apache.flink.api.common.typeinfo.BasicTypeInfo
-import org.apache.flink.api.scala._
-import org.apache.flink.api.scala.table._
-import org.apache.flink.api.table.Row
-import org.apache.flink.test.util.{TestBaseUtils, MultipleProgramsTestBase}
-import org.apache.flink.test.util.MultipleProgramsTestBase.TestExecutionMode
-
-import scala.collection.JavaConverters._
-
-@RunWith(classOf[Parameterized])
-class CastingITCase(mode: TestExecutionMode) extends MultipleProgramsTestBase(mode) {
-
- @Test
- def testAutoCastToString(): Unit = {
-
- val env = ExecutionEnvironment.getExecutionEnvironment
- val ds = env.fromElements((1: Byte, 1: Short, 1, 1L, 1.0f, 1.0d, new Date(0))).toTable
- .select('_1 + "b", '_2 + "s", '_3 + "i", '_4 + "L", '_5 + "f", '_6 + "d", '_7 + "Date")
- .toDataSet[Row]
- val expected = "1b,1s,1i,1L,1.0f,1.0d,1970-01-01 00:00:00.000Date"
- val results = ds.collect()
- TestBaseUtils.compareResultAsText(results.asJava, expected)
- }
-
- @Test
- def testNumericAutoCastInArithmetic(): Unit = {
-
- // don't test everything, just some common cast directions
-
- val env = ExecutionEnvironment.getExecutionEnvironment
- val ds = env.fromElements((1: Byte, 1: Short, 1, 1L, 1.0f, 1.0d)).toTable
- .select('_1 + 1, '_2 + 1, '_3 + 1L, '_4 + 1.0f, '_5 + 1.0d, '_6 + 1)
- .toDataSet[Row]
- val expected = "2,2,2,2.0,2.0,2.0"
- val results = ds.collect()
- TestBaseUtils.compareResultAsText(results.asJava, expected)
- }
-
- @Test
- def testNumericAutoCastInComparison(): Unit = {
-
- // don't test everything, just some common cast directions
-
- val env = ExecutionEnvironment.getExecutionEnvironment
- val ds = env.fromElements(
- (1: Byte, 1: Short, 1, 1L, 1.0f, 1.0d),
- (2: Byte, 2: Short, 2, 2L, 2.0f, 2.0d)).as('a, 'b, 'c, 'd, 'e, 'f)
- .filter('a > 1 && 'b > 1 && 'c > 1L && 'd > 1.0f && 'e > 1.0d && 'f > 1)
- .toDataSet[Row]
- val expected = "2,2,2,2,2.0,2.0"
- val results = ds.collect()
- TestBaseUtils.compareResultAsText(results.asJava, expected)
- }
-
- @Test
- def testCastFromString: Unit = {
-
- val env = ExecutionEnvironment.getExecutionEnvironment
- val ds = env.fromElements(("1", "true", "2.0",
- "2011-05-03", "15:51:36", "2011-05-03 15:51:36.000", "1446473775"))
- .toTable
- .select(
- '_1.cast(BasicTypeInfo.BYTE_TYPE_INFO),
- '_1.cast(BasicTypeInfo.SHORT_TYPE_INFO),
- '_1.cast(BasicTypeInfo.INT_TYPE_INFO),
- '_1.cast(BasicTypeInfo.LONG_TYPE_INFO),
- '_3.cast(BasicTypeInfo.DOUBLE_TYPE_INFO),
- '_3.cast(BasicTypeInfo.FLOAT_TYPE_INFO),
- '_2.cast(BasicTypeInfo.BOOLEAN_TYPE_INFO),
- '_4.cast(BasicTypeInfo.DATE_TYPE_INFO).cast(BasicTypeInfo.STRING_TYPE_INFO),
- '_5.cast(BasicTypeInfo.DATE_TYPE_INFO).cast(BasicTypeInfo.STRING_TYPE_INFO),
- '_6.cast(BasicTypeInfo.DATE_TYPE_INFO).cast(BasicTypeInfo.STRING_TYPE_INFO),
- '_7.cast(BasicTypeInfo.DATE_TYPE_INFO).cast(BasicTypeInfo.STRING_TYPE_INFO))
- .toDataSet[Row]
- val expected = "1,1,1,1,2.0,2.0,true," +
- "2011-05-03 00:00:00.000,1970-01-01 15:51:36.000,2011-05-03 15:51:36.000," +
- "1970-01-17 17:47:53.775\n"
- val results = ds.collect()
- TestBaseUtils.compareResultAsText(results.asJava, expected)
- }
-
- @Test
- def testCastDateToStringAndLong {
- val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
- val ds = env.fromElements(("2011-05-03 15:51:36.000", "1304437896000"))
- val result = ds.toTable
- .select('_1.cast(BasicTypeInfo.DATE_TYPE_INFO).as('f0),
- '_2.cast(BasicTypeInfo.DATE_TYPE_INFO).as('f1))
- .select('f0.cast(BasicTypeInfo.STRING_TYPE_INFO),
- 'f0.cast(BasicTypeInfo.LONG_TYPE_INFO),
- 'f1.cast(BasicTypeInfo.STRING_TYPE_INFO),
- 'f1.cast(BasicTypeInfo.LONG_TYPE_INFO))
- .toDataSet[Row]
- .collect
- val expected = "2011-05-03 15:51:36.000,1304437896000," +
- "2011-05-03 15:51:36.000,1304437896000\n"
- TestBaseUtils.compareResultAsText(result.asJava, expected)
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/e9bf13d8/flink-staging/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/ExpressionsITCase.scala
----------------------------------------------------------------------
diff --git a/flink-staging/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/ExpressionsITCase.scala b/flink-staging/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/ExpressionsITCase.scala
deleted file mode 100644
index 017cbf1..0000000
--- a/flink-staging/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/ExpressionsITCase.scala
+++ /dev/null
@@ -1,134 +0,0 @@
-/*
- * 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.api.scala.table.test
-
-import java.util.Date
-
-import org.apache.flink.api.common.typeinfo.BasicTypeInfo
-import org.apache.flink.api.table.expressions.Literal
-import org.apache.flink.api.table.{Row, ExpressionException}
-import org.apache.flink.api.scala._
-import org.apache.flink.api.scala.table._
-import org.apache.flink.test.util.{TestBaseUtils, MultipleProgramsTestBase}
-import org.apache.flink.test.util.MultipleProgramsTestBase.TestExecutionMode
-import org.junit._
-import org.junit.runner.RunWith
-import org.junit.runners.Parameterized
-
-import scala.collection.JavaConverters._
-
-@RunWith(classOf[Parameterized])
-class ExpressionsITCase(mode: TestExecutionMode) extends MultipleProgramsTestBase(mode) {
-
- @Test
- def testArithmetic(): Unit = {
-
- val env = ExecutionEnvironment.getExecutionEnvironment
- val ds = env.fromElements((5, 10)).as('a, 'b)
- .select('a - 5, 'a + 5, 'a / 2, 'a * 2, 'a % 2, -'a).toDataSet[Row]
- val expected = "0,10,2,10,1,-5"
- val results = ds.collect()
- TestBaseUtils.compareResultAsText(results.asJava, expected)
- }
-
- @Test
- def testLogic(): Unit = {
-
- val env = ExecutionEnvironment.getExecutionEnvironment
- val ds = env.fromElements((5, true)).as('a, 'b)
- .select('b && true, 'b && false, 'b || false, !'b).toDataSet[Row]
- val expected = "true,false,true,false"
- val results = ds.collect()
- TestBaseUtils.compareResultAsText(results.asJava, expected)
- }
-
- @Test
- def testComparisons(): Unit = {
-
- val env = ExecutionEnvironment.getExecutionEnvironment
- val ds = env.fromElements((5, 5, 4)).as('a, 'b, 'c)
- .select('a > 'c, 'a >= 'b, 'a < 'c, 'a.isNull, 'a.isNotNull).toDataSet[Row]
- val expected = "true,true,false,false,true"
- val results = ds.collect()
- TestBaseUtils.compareResultAsText(results.asJava, expected)
- }
-
- @Test
- def testBitwiseOperations(): Unit = {
-
- val env = ExecutionEnvironment.getExecutionEnvironment
-
- val ds = env.fromElements((3.toByte, 5.toByte)).as('a, 'b)
- .select('a & 'b, 'a | 'b, 'a ^ 'b, ~'a).toDataSet[Row]
- val expected = "1,7,6,-4"
- val results = ds.collect()
- TestBaseUtils.compareResultAsText(results.asJava, expected)
- }
-
- @Test
- def testBitwiseWithAutocast(): Unit = {
-
- val env = ExecutionEnvironment.getExecutionEnvironment
-
- val ds = env.fromElements((3, 5.toByte)).as('a, 'b)
- .select('a & 'b, 'a | 'b, 'a ^ 'b, ~'a).toDataSet[Row]
- val expected = "1,7,6,-4"
- val results = ds.collect()
- TestBaseUtils.compareResultAsText(results.asJava, expected)
- }
-
- @Test(expected = classOf[ExpressionException])
- def testBitwiseWithNonWorkingAutocast(): Unit = {
-
- val env = ExecutionEnvironment.getExecutionEnvironment
-
- val ds = env.fromElements((3.0, 5)).as('a, 'b)
- .select('a & 'b, 'a | 'b, 'a ^ 'b, ~'a).toDataSet[Row]
- val expected = "1,7,6,-4"
- val results = ds.collect()
- TestBaseUtils.compareResultAsText(results.asJava, expected)
- }
-
- @Test
- def testCaseInsensitiveForAs(): Unit = {
-
- val env = ExecutionEnvironment.getExecutionEnvironment
-
- val ds = env.fromElements((3, 5.toByte)).as('a, 'b)
- .groupBy("a").select("a, a.count As cnt").toDataSet[Row]
- val expected = "3,1"
- val results = ds.collect()
- TestBaseUtils.compareResultAsText(results.asJava, expected)
- }
-
- @Test
- def testDateLiteral(): Unit = {
- val env = ExecutionEnvironment.getExecutionEnvironment
-
- val ds = env.fromElements((0L, "test")).as('a, 'b)
- .select('a,
- Literal(new Date(0)).cast(BasicTypeInfo.STRING_TYPE_INFO),
- 'a.cast(BasicTypeInfo.DATE_TYPE_INFO).cast(BasicTypeInfo.STRING_TYPE_INFO))
- .toDataSet[Row]
- val expected = "0,1970-01-01 00:00:00.000,1970-01-01 00:00:00.000"
- val results = ds.collect()
- TestBaseUtils.compareResultAsText(results.asJava, expected)
- }
-
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/e9bf13d8/flink-staging/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/FilterITCase.scala
----------------------------------------------------------------------
diff --git a/flink-staging/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/FilterITCase.scala b/flink-staging/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/FilterITCase.scala
deleted file mode 100644
index c0b86f8..0000000
--- a/flink-staging/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/FilterITCase.scala
+++ /dev/null
@@ -1,132 +0,0 @@
-/*
- * 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.api.scala.table.test
-
-import org.apache.flink.api.table.expressions.Literal
-import org.apache.flink.api.scala._
-import org.apache.flink.api.scala.table._
-import org.apache.flink.api.scala.util.CollectionDataSets
-import org.apache.flink.test.util.{TestBaseUtils, MultipleProgramsTestBase}
-import org.apache.flink.test.util.MultipleProgramsTestBase.TestExecutionMode
-import org.junit._
-import org.junit.runner.RunWith
-import org.junit.runners.Parameterized
-
-import scala.collection.JavaConverters._
-
-
-@RunWith(classOf[Parameterized])
-class FilterITCase(mode: TestExecutionMode) extends MultipleProgramsTestBase(mode) {
-
- @Test
- def testAllRejectingFilter(): Unit = {
- /*
- * Test all-rejecting filter.
- */
- val env = ExecutionEnvironment.getExecutionEnvironment
- val ds = CollectionDataSets.get3TupleDataSet(env).as('a, 'b, 'c)
-
- val filterDs = ds.filter( Literal(false) )
- val expected = "\n"
- val results = filterDs.collect()
- TestBaseUtils.compareResultAsText(results.asJava, expected)
- }
-
- @Test
- def testAllPassingFilter(): Unit = {
- /*
- * Test all-passing filter.
- */
- val env = ExecutionEnvironment.getExecutionEnvironment
- val ds = CollectionDataSets.get3TupleDataSet(env).as('a, 'b, 'c)
-
- val filterDs = ds.filter( Literal(true) )
- val expected = "1,1,Hi\n" + "2,2,Hello\n" + "3,2,Hello world\n" + "4,3,Hello world, " +
- "how are you?\n" + "5,3,I am fine.\n" + "6,3,Luke Skywalker\n" + "7,4," +
- "Comment#1\n" + "8,4,Comment#2\n" + "9,4,Comment#3\n" + "10,4,Comment#4\n" + "11,5," +
- "Comment#5\n" + "12,5,Comment#6\n" + "13,5,Comment#7\n" + "14,5,Comment#8\n" + "15,5," +
- "Comment#9\n" + "16,6,Comment#10\n" + "17,6,Comment#11\n" + "18,6,Comment#12\n" + "19," +
- "6,Comment#13\n" + "20,6,Comment#14\n" + "21,6,Comment#15\n"
- val results = filterDs.collect()
- TestBaseUtils.compareResultAsText(results.asJava, expected)
- }
-
- @Test
- def testFilterOnStringTupleField(): Unit = {
- /*
- * Test filter on String tuple field.
- */
- val env = ExecutionEnvironment.getExecutionEnvironment
- val ds = CollectionDataSets.get3TupleDataSet(env)
- val filterDs = ds.filter( _._3.contains("world") )
- val expected = "(3,2,Hello world)\n" + "(4,3,Hello world, how are you?)\n"
- val results = filterDs.collect()
- TestBaseUtils.compareResultAsText(results.asJava, expected)
- }
-
- @Test
- def testFilterOnIntegerTupleField(): Unit = {
- /*
- * Test filter on Integer tuple field.
- */
- val env = ExecutionEnvironment.getExecutionEnvironment
- val ds = CollectionDataSets.get3TupleDataSet(env).as('a, 'b, 'c)
-
- val filterDs = ds.filter( 'a % 2 === 0 )
- val expected = "2,2,Hello\n" + "4,3,Hello world, how are you?\n" +
- "6,3,Luke Skywalker\n" + "8,4," + "Comment#2\n" + "10,4,Comment#4\n" +
- "12,5,Comment#6\n" + "14,5,Comment#8\n" + "16,6," +
- "Comment#10\n" + "18,6,Comment#12\n" + "20,6,Comment#14\n"
- val results = filterDs.collect()
- TestBaseUtils.compareResultAsText(results.asJava, expected)
- }
-
- // These two not yet done, but are planned
-
- @Ignore
- @Test
- def testFilterBasicType(): Unit = {
- /*
- * Test filter on basic type
- */
-
- val env = ExecutionEnvironment.getExecutionEnvironment
- val ds = CollectionDataSets.getStringDataSet(env)
-
- val filterDs = ds.filter( _.startsWith("H") )
- val expected = "Hi\n" + "Hello\n" + "Hello world\n" + "Hello world, how are you?\n"
- val results = filterDs.collect()
- TestBaseUtils.compareResultAsText(results.asJava, expected)
- }
-
- @Ignore
- @Test
- def testFilterOnCustomType(): Unit = {
- /*
- * Test filter on custom type
- */
- val env = ExecutionEnvironment.getExecutionEnvironment
- val ds = CollectionDataSets.getCustomTypeDataSet(env)
- val filterDs = ds.filter( _.myString.contains("a") )
- val expected = "3,3,Hello world, how are you?\n" + "3,4,I am fine.\n" + "3,5,Luke Skywalker\n"
- val results = filterDs.collect()
- TestBaseUtils.compareResultAsText(results.asJava, expected)
- }
-
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/e9bf13d8/flink-staging/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/GroupedAggreagationsITCase.scala
----------------------------------------------------------------------
diff --git a/flink-staging/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/GroupedAggreagationsITCase.scala b/flink-staging/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/GroupedAggreagationsITCase.scala
deleted file mode 100644
index fb76507..0000000
--- a/flink-staging/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/GroupedAggreagationsITCase.scala
+++ /dev/null
@@ -1,115 +0,0 @@
-/*
- * 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.api.scala.table.test
-
-import org.apache.flink.api.table.{Row, ExpressionException}
-import org.apache.flink.api.scala._
-import org.apache.flink.api.scala.table._
-import org.apache.flink.api.scala.util.CollectionDataSets
-import org.apache.flink.core.fs.FileSystem.WriteMode
-import org.apache.flink.test.util.{TestBaseUtils, MultipleProgramsTestBase}
-import org.apache.flink.test.util.MultipleProgramsTestBase.TestExecutionMode
-import org.junit._
-import org.junit.runner.RunWith
-import org.junit.runners.Parameterized
-
-import scala.collection.JavaConverters._
-
-@RunWith(classOf[Parameterized])
-class GroupedAggreagationsITCase(mode: TestExecutionMode) extends MultipleProgramsTestBase(mode) {
-
- @Test(expected = classOf[ExpressionException])
- def testGroupingOnNonExistentField(): Unit = {
-
- val env = ExecutionEnvironment.getExecutionEnvironment
- val ds = CollectionDataSets.get3TupleDataSet(env).as('a, 'b, 'c)
- .groupBy('_foo)
- .select('a.avg).toDataSet[Row]
- val expected = ""
- val results = ds.collect()
- TestBaseUtils.compareResultAsText(results.asJava, expected)
- }
-
- @Test
- def testGroupedAggregate(): Unit = {
-
- // the grouping key needs to be forwarded to the intermediate DataSet, even
- // if we don't want the key in the output
-
- val env = ExecutionEnvironment.getExecutionEnvironment
- val ds = CollectionDataSets.get3TupleDataSet(env).as('a, 'b, 'c)
- .groupBy('b)
- .select('b, 'a.sum).toDataSet[Row]
- val expected = "1,1\n" + "2,5\n" + "3,15\n" + "4,34\n" + "5,65\n" + "6,111\n"
- val results = ds.collect()
- TestBaseUtils.compareResultAsText(results.asJava, expected)
- }
-
- @Test
- def testGroupingKeyForwardIfNotUsed(): Unit = {
-
- // the grouping key needs to be forwarded to the intermediate DataSet, even
- // if we don't want the key in the output
-
- val env = ExecutionEnvironment.getExecutionEnvironment
- val ds = CollectionDataSets.get3TupleDataSet(env).as('a, 'b, 'c)
- .groupBy('b)
- .select('a.sum).toDataSet[Row]
- val expected = "1\n" + "5\n" + "15\n" + "34\n" + "65\n" + "111\n"
- val results = ds.collect()
- TestBaseUtils.compareResultAsText(results.asJava, expected)
- }
-
- @Test
- def testSQLStyleAggregations(): Unit = {
-
- // the grouping key needs to be forwarded to the intermediate DataSet, even
- // if we don't want the key in the output
-
- val env = ExecutionEnvironment.getExecutionEnvironment
- val ds = CollectionDataSets.get3TupleDataSet(env).as('a, 'b, 'c)
- .select(
- """Sum( a) as a1, a.sum as a2,
- |Min (a) as b1, a.min as b2,
- |Max (a ) as c1, a.max as c2,
- |Avg ( a ) as d1, a.avg as d2,
- |Count(a) as e1, a.count as e2
- """.stripMargin).toDataSet[Row]
- val expected = "231,231,1,1,21,21,11,11,21,21"
- val results = ds.collect()
- TestBaseUtils.compareResultAsText(results.asJava, expected)
- }
-
- @Test
- def testGroupNoAggregation(): Unit = {
-
- val env = ExecutionEnvironment.getExecutionEnvironment
- val ds = CollectionDataSets.get3TupleDataSet(env)
- .as('a, 'b, 'c)
- .groupBy('b)
- .select('a.sum as 'd, 'b)
- .groupBy('b, 'd)
- .select('b)
- .toDataSet[Row]
-
- val expected = "1\n" + "2\n" + "3\n" + "4\n" + "5\n" + "6\n"
- val results = ds.collect()
- TestBaseUtils.compareResultAsText(results.asJava, expected)
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/e9bf13d8/flink-staging/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/JoinITCase.scala
----------------------------------------------------------------------
diff --git a/flink-staging/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/JoinITCase.scala b/flink-staging/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/JoinITCase.scala
deleted file mode 100644
index e12c9d6..0000000
--- a/flink-staging/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/JoinITCase.scala
+++ /dev/null
@@ -1,122 +0,0 @@
-/*
- * 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.api.scala.table.test
-
-import org.apache.flink.api.table.{Row, ExpressionException}
-import org.apache.flink.api.scala._
-import org.apache.flink.api.scala.table._
-import org.apache.flink.api.scala.util.CollectionDataSets
-import org.apache.flink.test.util.{TestBaseUtils, MultipleProgramsTestBase}
-import org.apache.flink.test.util.MultipleProgramsTestBase.TestExecutionMode
-import org.junit._
-import org.junit.runner.RunWith
-import org.junit.runners.Parameterized
-
-import scala.collection.JavaConverters._
-
-@RunWith(classOf[Parameterized])
-class JoinITCase(mode: TestExecutionMode) extends MultipleProgramsTestBase(mode) {
-
- @Test
- def testJoin(): Unit = {
- val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
- val ds1 = CollectionDataSets.getSmall3TupleDataSet(env).as('a, 'b, 'c)
- val ds2 = CollectionDataSets.get5TupleDataSet(env).as('d, 'e, 'f, 'g, 'h)
-
- val joinDs = ds1.join(ds2).where('b === 'e).select('c, 'g).toDataSet[Row]
- val expected = "Hi,Hallo\n" + "Hello,Hallo Welt\n" + "Hello world,Hallo Welt\n"
- val results = joinDs.collect()
- TestBaseUtils.compareResultAsText(results.asJava, expected)
- }
-
- @Test
- def testJoinWithFilter(): Unit = {
- val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
- val ds1 = CollectionDataSets.getSmall3TupleDataSet(env).as('a, 'b, 'c)
- val ds2 = CollectionDataSets.get5TupleDataSet(env).as('d, 'e, 'f, 'g, 'h)
-
- val joinDs = ds1.join(ds2).where('b === 'e && 'b < 2).select('c, 'g).toDataSet[Row]
- val expected = "Hi,Hallo\n"
- val results = joinDs.collect()
- TestBaseUtils.compareResultAsText(results.asJava, expected)
- }
-
- @Test
- def testJoinWithMultipleKeys(): 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 joinDs = ds1.join(ds2).filter('a === 'd && 'b === 'h).select('c, 'g).toDataSet[Row]
- val expected = "Hi,Hallo\n" + "Hello,Hallo Welt\n" + "Hello world,Hallo Welt wie gehts?\n" +
- "Hello world,ABC\n" + "I am fine.,HIJ\n" + "I am fine.,IJK\n"
- val results = joinDs.collect()
- TestBaseUtils.compareResultAsText(results.asJava, expected)
- }
-
- @Test(expected = classOf[ExpressionException])
- def testJoinNonExistingKey(): Unit = {
- val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
- val ds1 = CollectionDataSets.getSmall3TupleDataSet(env).as('a, 'b, 'c)
- val ds2 = CollectionDataSets.get5TupleDataSet(env).as('d, 'e, 'f, 'g, 'h)
-
- val joinDs = ds1.join(ds2).where('foo === 'e).select('c, 'g).toDataSet[Row]
- val expected = ""
- val results = joinDs.collect()
- TestBaseUtils.compareResultAsText(results.asJava, expected)
- }
-
- @Test(expected = classOf[ExpressionException])
- def testJoinWithNonMatchingKeyTypes(): Unit = {
- val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
- val ds1 = CollectionDataSets.getSmall3TupleDataSet(env).as('a, 'b, 'c)
- val ds2 = CollectionDataSets.get5TupleDataSet(env).as('d, 'e, 'f, 'g, 'h)
-
- val joinDs = ds1.join(ds2).where('a === 'g).select('c, 'g).toDataSet[Row]
- val expected = ""
- val results = joinDs.collect()
- TestBaseUtils.compareResultAsText(results.asJava, expected)
- }
-
- @Test(expected = classOf[ExpressionException])
- def testJoinWithAmbiguousFields(): Unit = {
- val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
- val ds1 = CollectionDataSets.getSmall3TupleDataSet(env).as('a, 'b, 'c)
- val ds2 = CollectionDataSets.get5TupleDataSet(env).as('d, 'e, 'f, 'g, 'c)
-
- val joinDs = ds1.join(ds2).where('a === 'd).select('c, 'g).toDataSet[Row]
- val expected = ""
- val results = joinDs.collect()
- TestBaseUtils.compareResultAsText(results.asJava, expected)
- }
-
- @Test
- def testJoinWithAggregation(): Unit = {
- val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
- val ds1 = CollectionDataSets.getSmall3TupleDataSet(env).as('a, 'b, 'c)
- val ds2 = CollectionDataSets.get5TupleDataSet(env).as('d, 'e, 'f, 'g, 'h)
-
- val joinDs = ds1.join(ds2).where('a === 'd).select('g.count).toDataSet[Row]
- val expected = "6"
- val results = joinDs.collect()
- TestBaseUtils.compareResultAsText(results.asJava, expected)
- }
-
-
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/e9bf13d8/flink-staging/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/SelectITCase.scala
----------------------------------------------------------------------
diff --git a/flink-staging/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/SelectITCase.scala b/flink-staging/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/SelectITCase.scala
deleted file mode 100644
index fa3f283..0000000
--- a/flink-staging/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/SelectITCase.scala
+++ /dev/null
@@ -1,122 +0,0 @@
-/*
- * 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.api.scala.table.test
-
-import org.apache.flink.api.table.{Row, ExpressionException}
-import org.apache.flink.api.scala._
-import org.apache.flink.api.scala.table._
-import org.apache.flink.api.scala.util.CollectionDataSets
-import org.apache.flink.test.util.{TestBaseUtils, MultipleProgramsTestBase}
-import org.apache.flink.test.util.MultipleProgramsTestBase.TestExecutionMode
-import org.junit._
-import org.junit.runner.RunWith
-import org.junit.runners.Parameterized
-
-import scala.collection.JavaConverters._
-
-@RunWith(classOf[Parameterized])
-class SelectITCase(mode: TestExecutionMode) extends MultipleProgramsTestBase(mode) {
-
- @Test
- def testSimpleSelectAll(): Unit = {
-
- val env = ExecutionEnvironment.getExecutionEnvironment
- val ds = CollectionDataSets.get3TupleDataSet(env).toTable.select('_1, '_2, '_3)
- .toDataSet[Row]
- val expected = "1,1,Hi\n" + "2,2,Hello\n" + "3,2,Hello world\n" + "4,3,Hello world, " +
- "how are you?\n" + "5,3,I am fine.\n" + "6,3,Luke Skywalker\n" + "7,4," +
- "Comment#1\n" + "8,4,Comment#2\n" + "9,4,Comment#3\n" + "10,4,Comment#4\n" + "11,5," +
- "Comment#5\n" + "12,5,Comment#6\n" + "13,5,Comment#7\n" + "14,5,Comment#8\n" + "15,5," +
- "Comment#9\n" + "16,6,Comment#10\n" + "17,6,Comment#11\n" + "18,6,Comment#12\n" + "19," +
- "6,Comment#13\n" + "20,6,Comment#14\n" + "21,6,Comment#15\n"
- val results = ds.collect()
- TestBaseUtils.compareResultAsText(results.asJava, expected)
- }
-
- @Test
- def testSimpleSelectAllWithAs(): Unit = {
-
- val env = ExecutionEnvironment.getExecutionEnvironment
- val ds = CollectionDataSets.get3TupleDataSet(env).as('a, 'b, 'c).select('a, 'b, 'c)
- .toDataSet[Row]
- val expected = "1,1,Hi\n" + "2,2,Hello\n" + "3,2,Hello world\n" + "4,3,Hello world, " +
- "how are you?\n" + "5,3,I am fine.\n" + "6,3,Luke Skywalker\n" + "7,4," +
- "Comment#1\n" + "8,4,Comment#2\n" + "9,4,Comment#3\n" + "10,4,Comment#4\n" + "11,5," +
- "Comment#5\n" + "12,5,Comment#6\n" + "13,5,Comment#7\n" + "14,5,Comment#8\n" + "15,5," +
- "Comment#9\n" + "16,6,Comment#10\n" + "17,6,Comment#11\n" + "18,6,Comment#12\n" + "19," +
- "6,Comment#13\n" + "20,6,Comment#14\n" + "21,6,Comment#15\n"
- val results = ds.collect()
- TestBaseUtils.compareResultAsText(results.asJava, expected)
- }
-
- @Test
- def testSimpleSelectWithNaming(): Unit = {
-
- val env = ExecutionEnvironment.getExecutionEnvironment
- val ds = CollectionDataSets.get3TupleDataSet(env).toTable
- .select('_1 as 'a, '_2 as 'b)
- .select('a, 'b).toDataSet[Row]
- val expected = "1,1\n" + "2,2\n" + "3,2\n" + "4,3\n" + "5,3\n" + "6,3\n" + "7,4\n" +
- "8,4\n" + "9,4\n" + "10,4\n" + "11,5\n" + "12,5\n" + "13,5\n" + "14,5\n" + "15,5\n" +
- "16,6\n" + "17,6\n" + "18,6\n" + "19,6\n" + "20,6\n" + "21,6\n"
- val results = ds.collect()
- TestBaseUtils.compareResultAsText(results.asJava, expected)
- }
-
- @Test(expected = classOf[ExpressionException])
- def testAsWithToFewFields(): Unit = {
-
- val env = ExecutionEnvironment.getExecutionEnvironment
- val ds = CollectionDataSets.get3TupleDataSet(env).as('a, 'b).toDataSet[Row]
- val expected = "no"
- val results = ds.collect()
- TestBaseUtils.compareResultAsText(results.asJava, expected)
- }
-
- @Test(expected = classOf[ExpressionException])
- def testAsWithToManyFields(): Unit = {
-
- val env = ExecutionEnvironment.getExecutionEnvironment
- val ds = CollectionDataSets.get3TupleDataSet(env).as('a, 'b, 'c, 'd).toDataSet[Row]
- val expected = "no"
- val results = ds.collect()
- TestBaseUtils.compareResultAsText(results.asJava, expected)
- }
-
- @Test(expected = classOf[ExpressionException])
- def testAsWithAmbiguousFields(): Unit = {
-
- val env = ExecutionEnvironment.getExecutionEnvironment
- val ds = CollectionDataSets.get3TupleDataSet(env).as('a, 'b, 'b).toDataSet[Row]
- val expected = "no"
- val results = ds.collect()
- TestBaseUtils.compareResultAsText(results.asJava, expected)
- }
-
-
- @Test(expected = classOf[ExpressionException])
- def testOnlyFieldRefInAs(): Unit = {
-
- val env = ExecutionEnvironment.getExecutionEnvironment
- val ds = CollectionDataSets.get3TupleDataSet(env).as('a, 'b as 'c, 'd).toDataSet[Row]
- val expected = "no"
- val results = ds.collect()
- TestBaseUtils.compareResultAsText(results.asJava, expected)
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/e9bf13d8/flink-staging/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/SqlExplainITCase.scala
----------------------------------------------------------------------
diff --git a/flink-staging/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/SqlExplainITCase.scala b/flink-staging/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/SqlExplainITCase.scala
deleted file mode 100644
index bead02f..0000000
--- a/flink-staging/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/SqlExplainITCase.scala
+++ /dev/null
@@ -1,96 +0,0 @@
-/*
- * 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.api.scala.table.test
-
-import org.apache.flink.api.scala._
-import org.apache.flink.api.scala.table._
-
-import org.junit._
-import org.junit.Assert.assertEquals
-
-case class WC(count: Int, word: String)
-
-class SqlExplainITCase {
-
- val testFilePath = SqlExplainITCase.this.getClass.getResource("/").getFile
-
- @Test
- def testGroupByWithoutExtended() : Unit = {
- val env = ExecutionEnvironment.createLocalEnvironment()
- val expr = env.fromElements(WC(1, "hello"), WC(2, "hello"), WC(3, "ciao")).toTable.as('a, 'b)
- val result = expr.filter("a % 2 = 0").explain()
- val source = scala.io.Source.fromFile(testFilePath +
- "../../src/test/scala/resources/testFilter0.out").mkString
- assertEquals(result, source)
- }
-
- @Test
- def testGroupByWithExtended() : Unit = {
- val env = ExecutionEnvironment.createLocalEnvironment()
- val expr = env.fromElements(WC(1, "hello"), WC(2, "hello"), WC(3, "ciao")).toTable.as('a, 'b)
- val result = expr.filter("a % 2 = 0").explain(true)
- val source = scala.io.Source.fromFile(testFilePath +
- "../../src/test/scala/resources/testFilter1.out").mkString
- assertEquals(result, source)
- }
-
- @Test
- def testJoinWithoutExtended() : Unit = {
- val env = ExecutionEnvironment.createLocalEnvironment()
- val expr1 = env.fromElements(WC(1, "hello"), WC(1, "hello"), WC(1, "ciao")).toTable.as('a, 'b)
- val expr2 = env.fromElements(WC(1, "hello"), WC(1, "hello"), WC(1, "java")).toTable.as('c, 'd)
- val result = expr1.join(expr2).where("b = d").select("a, c").explain()
- val source = scala.io.Source.fromFile(testFilePath +
- "../../src/test/scala/resources/testJoin0.out").mkString
- assertEquals(result, source)
- }
-
- @Test
- def testJoinWithExtended() : Unit = {
- val env = ExecutionEnvironment.createLocalEnvironment()
- val expr1 = env.fromElements(WC(1, "hello"), WC(1, "hello"), WC(1, "ciao")).toTable.as('a, 'b)
- val expr2 = env.fromElements(WC(1, "hello"), WC(1, "hello"), WC(1, "java")).toTable.as('c, 'd)
- val result = expr1.join(expr2).where("b = d").select("a, c").explain(true)
- val source = scala.io.Source.fromFile(testFilePath +
- "../../src/test/scala/resources/testJoin1.out").mkString
- assertEquals(result, source)
- }
-
- @Test
- def testUnionWithoutExtended() : Unit = {
- val env = ExecutionEnvironment.createLocalEnvironment()
- val expr1 = env.fromElements(WC(1, "hello"), WC(1, "hello"), WC(1, "ciao")).toTable
- val expr2 = env.fromElements(WC(1, "hello"), WC(1, "hello"), WC(1, "java")).toTable
- val result = expr1.unionAll(expr2).explain()
- val source = scala.io.Source.fromFile(testFilePath +
- "../../src/test/scala/resources/testUnion0.out").mkString
- assertEquals(result, source)
- }
-
- @Test
- def testUnionWithExtended() : Unit = {
- val env = ExecutionEnvironment.createLocalEnvironment()
- val expr1 = env.fromElements(WC(1, "hello"), WC(1, "hello"), WC(1, "ciao")).toTable
- val expr2 = env.fromElements(WC(1, "hello"), WC(1, "hello"), WC(1, "java")).toTable
- val result = expr1.unionAll(expr2).explain(true)
- val source = scala.io.Source.fromFile(testFilePath +
- "../../src/test/scala/resources/testUnion1.out").mkString
- assertEquals(result, source)
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/e9bf13d8/flink-staging/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/StringExpressionsITCase.scala
----------------------------------------------------------------------
diff --git a/flink-staging/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/StringExpressionsITCase.scala b/flink-staging/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/StringExpressionsITCase.scala
deleted file mode 100644
index 10bc8fd..0000000
--- a/flink-staging/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/StringExpressionsITCase.scala
+++ /dev/null
@@ -1,78 +0,0 @@
-/*
- * 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.api.scala.table.test
-
-import org.apache.flink.api.table.{Row, ExpressionException}
-import org.apache.flink.api.scala._
-import org.apache.flink.api.scala.table._
-import org.apache.flink.test.util.{TestBaseUtils, MultipleProgramsTestBase}
-import org.apache.flink.test.util.MultipleProgramsTestBase.TestExecutionMode
-import org.junit._
-import org.junit.runner.RunWith
-import org.junit.runners.Parameterized
-
-import scala.collection.JavaConverters._
-
-@RunWith(classOf[Parameterized])
-class StringExpressionsITCase(mode: TestExecutionMode) extends MultipleProgramsTestBase(mode) {
-
- @Test
- def testSubstring(): Unit = {
- val env = ExecutionEnvironment.getExecutionEnvironment
- val ds = env.fromElements(("AAAA", 2), ("BBBB", 1)).as('a, 'b)
- .select('a.substring(0, 'b)).toDataSet[Row]
- val expected = "AA\nB"
- val results = ds.collect()
- TestBaseUtils.compareResultAsText(results.asJava, expected)
- }
-
- @Test
- def testSubstringWithMaxEnd(): Unit = {
- val env = ExecutionEnvironment.getExecutionEnvironment
- val ds = env.fromElements(("ABCD", 2), ("ABCD", 1)).as('a, 'b)
- .select('a.substring('b)).toDataSet[Row]
- val expected = "CD\nBCD"
- val results = ds.collect()
- TestBaseUtils.compareResultAsText(results.asJava, expected)
- }
-
- @Test(expected = classOf[ExpressionException])
- def testNonWorkingSubstring1(): Unit = {
-
- val env = ExecutionEnvironment.getExecutionEnvironment
- val ds = env.fromElements(("AAAA", 2.0), ("BBBB", 1.0)).as('a, 'b)
- .select('a.substring(0, 'b)).toDataSet[Row]
- val expected = "AAA\nBB"
- val results = ds.collect()
- TestBaseUtils.compareResultAsText(results.asJava, expected)
- }
-
- @Test(expected = classOf[ExpressionException])
- def testNonWorkingSubstring2(): Unit = {
-
- val env = ExecutionEnvironment.getExecutionEnvironment
- val ds = env.fromElements(("AAAA", "c"), ("BBBB", "d")).as('a, 'b)
- .select('a.substring('b, 15)).toDataSet[Row]
- val expected = "AAA\nBB"
- val results = ds.collect()
- TestBaseUtils.compareResultAsText(results.asJava, expected)
- }
-
-
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/e9bf13d8/flink-staging/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/UnionITCase.scala
----------------------------------------------------------------------
diff --git a/flink-staging/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/UnionITCase.scala b/flink-staging/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/UnionITCase.scala
deleted file mode 100644
index a47d4b7..0000000
--- a/flink-staging/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/UnionITCase.scala
+++ /dev/null
@@ -1,100 +0,0 @@
-/*
- * 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.api.scala.table.test
-
-import org.apache.flink.api.scala._
-import org.apache.flink.api.scala.table._
-import org.apache.flink.api.scala.util.CollectionDataSets
-import org.apache.flink.api.table.{ExpressionException, Row}
-import org.apache.flink.test.util.MultipleProgramsTestBase.TestExecutionMode
-import org.apache.flink.test.util.{MultipleProgramsTestBase, TestBaseUtils}
-import org.junit._
-import org.junit.runner.RunWith
-import org.junit.runners.Parameterized
-
-import scala.collection.JavaConverters._
-
-@RunWith(classOf[Parameterized])
-class UnionITCase(mode: TestExecutionMode) extends MultipleProgramsTestBase(mode) {
-
- @Test
- def testUnion(): Unit = {
- val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
- val ds1 = CollectionDataSets.getSmall3TupleDataSet(env).as('a, 'b, 'c)
- val ds2 = CollectionDataSets.getSmall3TupleDataSet(env).as('a, 'b, 'c)
-
- val unionDs = ds1.unionAll(ds2).select('c)
-
- val results = unionDs.toDataSet[Row].collect()
- val expected = "Hi\n" + "Hello\n" + "Hello world\n" + "Hi\n" + "Hello\n" + "Hello world\n"
- TestBaseUtils.compareResultAsText(results.asJava, expected)
- }
-
- @Test
- def testUnionWithFilter(): Unit = {
- val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
- val ds1 = CollectionDataSets.getSmall3TupleDataSet(env).as('a, 'b, 'c)
- val ds2 = CollectionDataSets.get5TupleDataSet(env).as('a, 'b, 'd, 'c, 'e)
-
- val joinDs = ds1.unionAll(ds2.select('a, 'b, 'c)).filter('b < 2).select('c)
-
- val results = joinDs.toDataSet[Row].collect()
- val expected = "Hi\n" + "Hallo\n"
- TestBaseUtils.compareResultAsText(results.asJava, expected)
- }
-
- @Test(expected = classOf[ExpressionException])
- def testUnionFieldsNameNotOverlap1(): Unit = {
- val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
- val ds1 = CollectionDataSets.getSmall3TupleDataSet(env).as('a, 'b, 'c)
- val ds2 = CollectionDataSets.get5TupleDataSet(env).as('a, 'b, 'd, 'c, 'e)
-
- val unionDs = ds1.unionAll(ds2)
-
- val results = unionDs.toDataSet[Row].collect()
- val expected = ""
- TestBaseUtils.compareResultAsText(results.asJava, expected)
- }
-
- @Test(expected = classOf[ExpressionException])
- def testUnionFieldsNameNotOverlap2(): Unit = {
- val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
- val ds1 = CollectionDataSets.getSmall3TupleDataSet(env).as('a, 'b, 'c)
- val ds2 = CollectionDataSets.get5TupleDataSet(env).as('a, 'b, 'c, 'd, 'e).select('a, 'b, 'c)
-
- val unionDs = ds1.unionAll(ds2)
-
- val results = unionDs.toDataSet[Row].collect()
- val expected = ""
- TestBaseUtils.compareResultAsText(results.asJava, expected)
- }
-
- @Test
- def testUnionWithAggregation(): Unit = {
- val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
- val ds1 = CollectionDataSets.getSmall3TupleDataSet(env).as('a, 'b, 'c)
- val ds2 = CollectionDataSets.get5TupleDataSet(env).as('a, 'b, 'd, 'c, 'e)
-
- val unionDs = ds1.unionAll(ds2.select('a, 'b, 'c)).select('c.count)
-
- val results = unionDs.toDataSet[Row].collect()
- val expected = "18"
- TestBaseUtils.compareResultAsText(results.asJava, expected)
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/e9bf13d8/flink-staging/flink-table/src/test/scala/org/apache/flink/api/table/typeinfo/RenamingProxyTypeInfoTest.scala
----------------------------------------------------------------------
diff --git a/flink-staging/flink-table/src/test/scala/org/apache/flink/api/table/typeinfo/RenamingProxyTypeInfoTest.scala b/flink-staging/flink-table/src/test/scala/org/apache/flink/api/table/typeinfo/RenamingProxyTypeInfoTest.scala
deleted file mode 100644
index ef616a9..0000000
--- a/flink-staging/flink-table/src/test/scala/org/apache/flink/api/table/typeinfo/RenamingProxyTypeInfoTest.scala
+++ /dev/null
@@ -1,75 +0,0 @@
-/*
- * 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.api.table.typeinfo
-
-import org.apache.flink.api.common.typeutils.CompositeType
-import org.apache.flink.api.java.typeutils.TypeExtractor
-import org.apache.flink.util.TestLogger
-import org.junit.Test
-import org.scalatest.junit.JUnitSuiteLike
-
-class RenamingProxyTypeInfoTest extends TestLogger with JUnitSuiteLike {
-
- @Test
- def testRenamingProxyTypeEquality(): Unit = {
- val pojoTypeInfo1 = TypeExtractor.createTypeInfo(classOf[TestPojo])
- .asInstanceOf[CompositeType[TestPojo]]
-
- val tpeInfo1 = new RenamingProxyTypeInfo[TestPojo](
- pojoTypeInfo1,
- Array("someInt", "aString", "doubleArray"))
-
- val tpeInfo2 = new RenamingProxyTypeInfo[TestPojo](
- pojoTypeInfo1,
- Array("someInt", "aString", "doubleArray"))
-
- assert(tpeInfo1.equals(tpeInfo2))
- assert(tpeInfo1.hashCode() == tpeInfo2.hashCode())
- }
-
- @Test
- def testRenamingProxyTypeInequality(): Unit = {
- val pojoTypeInfo1 = TypeExtractor.createTypeInfo(classOf[TestPojo])
- .asInstanceOf[CompositeType[TestPojo]]
-
- val tpeInfo1 = new RenamingProxyTypeInfo[TestPojo](
- pojoTypeInfo1,
- Array("someInt", "aString", "doubleArray"))
-
- val tpeInfo2 = new RenamingProxyTypeInfo[TestPojo](
- pojoTypeInfo1,
- Array("foobar", "aString", "doubleArray"))
-
- assert(!tpeInfo1.equals(tpeInfo2))
- }
-}
-
-final class TestPojo {
- var someInt: Int = 0
- private var aString: String = null
- var doubleArray: Array[Double] = null
-
- def setaString(aString: String) {
- this.aString = aString
- }
-
- def getaString: String = {
- return aString
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/e9bf13d8/flink-staging/flink-table/src/test/scala/resources/testFilter0.out
----------------------------------------------------------------------
diff --git a/flink-staging/flink-table/src/test/scala/resources/testFilter0.out b/flink-staging/flink-table/src/test/scala/resources/testFilter0.out
deleted file mode 100644
index 062fc90..0000000
--- a/flink-staging/flink-table/src/test/scala/resources/testFilter0.out
+++ /dev/null
@@ -1,28 +0,0 @@
-== Abstract Syntax Tree ==
-Filter(As(Root(ArraySeq((count,Integer), (word,String))), a,b), ('a * 2) === 0)
-
-== Physical Execution Plan ==
-Stage 3 : Data Source
- content : collect elements with CollectionInputFormat
- Partitioning : RANDOM_PARTITIONED
-
- Stage 2 : Map
- content : Map at select('count as 'count,'word as 'word)
- ship_strategy : Forward
- exchange_mode : PIPELINED
- driver_strategy : Map
- Partitioning : RANDOM_PARTITIONED
-
- Stage 1 : Filter
- content : ('a * 2) === 0
- ship_strategy : Forward
- exchange_mode : PIPELINED
- driver_strategy : FlatMap
- Partitioning : RANDOM_PARTITIONED
-
- Stage 0 : Data Sink
- content : org.apache.flink.api.java.io.DiscardingOutputFormat
- ship_strategy : Forward
- exchange_mode : PIPELINED
- Partitioning : RANDOM_PARTITIONED
-
http://git-wip-us.apache.org/repos/asf/flink/blob/e9bf13d8/flink-staging/flink-table/src/test/scala/resources/testFilter1.out
----------------------------------------------------------------------
diff --git a/flink-staging/flink-table/src/test/scala/resources/testFilter1.out b/flink-staging/flink-table/src/test/scala/resources/testFilter1.out
deleted file mode 100644
index 83378e6..0000000
--- a/flink-staging/flink-table/src/test/scala/resources/testFilter1.out
+++ /dev/null
@@ -1,96 +0,0 @@
-== Abstract Syntax Tree ==
-Filter(As(Root(ArraySeq((count,Integer), (word,String))), a,b), ('a * 2) === 0)
-
-== Physical Execution Plan ==
-Stage 3 : Data Source
- content : collect elements with CollectionInputFormat
- Partitioning : RANDOM_PARTITIONED
- Partitioning Order : (none)
- Uniqueness : not unique
- Order : (none)
- Grouping : not grouped
- Uniqueness : not unique
- Est. Output Size : (unknown)
- Est. Cardinality : (unknown)
- Network : 0.0
- Disk I/O : 0.0
- CPU : 0.0
- Cumulative Network : 0.0
- Cumulative Disk I/O : 0.0
- Cumulative CPU : 0.0
- Output Size (bytes) : (none)
- Output Cardinality : (none)
- Avg. Output Record Size (bytes) : (none)
- Filter Factor : (none)
-
- Stage 2 : Map
- content : Map at select('count as 'count,'word as 'word)
- ship_strategy : Forward
- exchange_mode : PIPELINED
- driver_strategy : Map
- Partitioning : RANDOM_PARTITIONED
- Partitioning Order : (none)
- Uniqueness : not unique
- Order : (none)
- Grouping : not grouped
- Uniqueness : not unique
- Est. Output Size : (unknown)
- Est. Cardinality : (unknown)
- Network : 0.0
- Disk I/O : 0.0
- CPU : 0.0
- Cumulative Network : 0.0
- Cumulative Disk I/O : 0.0
- Cumulative CPU : 0.0
- Output Size (bytes) : (none)
- Output Cardinality : (none)
- Avg. Output Record Size (bytes) : (none)
- Filter Factor : (none)
-
- Stage 1 : Filter
- content : ('a * 2) === 0
- ship_strategy : Forward
- exchange_mode : PIPELINED
- driver_strategy : FlatMap
- Partitioning : RANDOM_PARTITIONED
- Partitioning Order : (none)
- Uniqueness : not unique
- Order : (none)
- Grouping : not grouped
- Uniqueness : not unique
- Est. Output Size : 0.0
- Est. Cardinality : 0.0
- Network : 0.0
- Disk I/O : 0.0
- CPU : 0.0
- Cumulative Network : 0.0
- Cumulative Disk I/O : 0.0
- Cumulative CPU : 0.0
- Output Size (bytes) : (none)
- Output Cardinality : (none)
- Avg. Output Record Size (bytes) : (none)
- Filter Factor : (none)
-
- Stage 0 : Data Sink
- content : org.apache.flink.api.java.io.DiscardingOutputFormat
- ship_strategy : Forward
- exchange_mode : PIPELINED
- Partitioning : RANDOM_PARTITIONED
- Partitioning Order : (none)
- Uniqueness : not unique
- Order : (none)
- Grouping : not grouped
- Uniqueness : not unique
- Est. Output Size : 0.0
- Est. Cardinality : 0.0
- Network : 0.0
- Disk I/O : 0.0
- CPU : 0.0
- Cumulative Network : 0.0
- Cumulative Disk I/O : 0.0
- Cumulative CPU : 0.0
- Output Size (bytes) : (none)
- Output Cardinality : (none)
- Avg. Output Record Size (bytes) : (none)
- Filter Factor : (none)
-
http://git-wip-us.apache.org/repos/asf/flink/blob/e9bf13d8/flink-staging/flink-table/src/test/scala/resources/testJoin0.out
----------------------------------------------------------------------
diff --git a/flink-staging/flink-table/src/test/scala/resources/testJoin0.out b/flink-staging/flink-table/src/test/scala/resources/testJoin0.out
deleted file mode 100644
index e6e30be..0000000
--- a/flink-staging/flink-table/src/test/scala/resources/testJoin0.out
+++ /dev/null
@@ -1,39 +0,0 @@
-== Abstract Syntax Tree ==
-Select(Filter(Join(As(Root(ArraySeq((count,Integer), (word,String))), a,b), As(Root(ArraySeq((count,Integer), (word,String))), c,d)), 'b === 'd), 'a,'c)
-
-== Physical Execution Plan ==
-Stage 3 : Data Source
- content : collect elements with CollectionInputFormat
- Partitioning : RANDOM_PARTITIONED
-
- Stage 2 : Map
- content : Map at select('count as 'count,'word as 'word)
- ship_strategy : Forward
- exchange_mode : PIPELINED
- driver_strategy : Map
- Partitioning : RANDOM_PARTITIONED
-
-Stage 5 : Data Source
- content : collect elements with CollectionInputFormat
- Partitioning : RANDOM_PARTITIONED
-
- Stage 4 : Map
- content : Map at select('count as 'count,'word as 'word)
- ship_strategy : Forward
- exchange_mode : PIPELINED
- driver_strategy : Map
- Partitioning : RANDOM_PARTITIONED
-
- Stage 1 : Join
- content : Join at 'b === 'd
- ship_strategy : Hash Partition on [1]
- exchange_mode : PIPELINED
- driver_strategy : Hybrid Hash (build: Map at select('count as 'count,'word as 'word))
- Partitioning : RANDOM_PARTITIONED
-
- Stage 0 : Data Sink
- content : org.apache.flink.api.java.io.DiscardingOutputFormat
- ship_strategy : Forward
- exchange_mode : PIPELINED
- Partitioning : RANDOM_PARTITIONED
-
http://git-wip-us.apache.org/repos/asf/flink/blob/e9bf13d8/flink-staging/flink-table/src/test/scala/resources/testJoin1.out
----------------------------------------------------------------------
diff --git a/flink-staging/flink-table/src/test/scala/resources/testJoin1.out b/flink-staging/flink-table/src/test/scala/resources/testJoin1.out
deleted file mode 100644
index a8f05dd..0000000
--- a/flink-staging/flink-table/src/test/scala/resources/testJoin1.out
+++ /dev/null
@@ -1,141 +0,0 @@
-== Abstract Syntax Tree ==
-Select(Filter(Join(As(Root(ArraySeq((count,Integer), (word,String))), a,b), As(Root(ArraySeq((count,Integer), (word,String))), c,d)), 'b === 'd), 'a,'c)
-
-== Physical Execution Plan ==
-Stage 3 : Data Source
- content : collect elements with CollectionInputFormat
- Partitioning : RANDOM_PARTITIONED
- Partitioning Order : (none)
- Uniqueness : not unique
- Order : (none)
- Grouping : not grouped
- Uniqueness : not unique
- Est. Output Size : (unknown)
- Est. Cardinality : (unknown)
- Network : 0.0
- Disk I/O : 0.0
- CPU : 0.0
- Cumulative Network : 0.0
- Cumulative Disk I/O : 0.0
- Cumulative CPU : 0.0
- Output Size (bytes) : (none)
- Output Cardinality : (none)
- Avg. Output Record Size (bytes) : (none)
- Filter Factor : (none)
-
- Stage 2 : Map
- content : Map at select('count as 'count,'word as 'word)
- ship_strategy : Forward
- exchange_mode : PIPELINED
- driver_strategy : Map
- Partitioning : RANDOM_PARTITIONED
- Partitioning Order : (none)
- Uniqueness : not unique
- Order : (none)
- Grouping : not grouped
- Uniqueness : not unique
- Est. Output Size : (unknown)
- Est. Cardinality : (unknown)
- Network : 0.0
- Disk I/O : 0.0
- CPU : 0.0
- Cumulative Network : 0.0
- Cumulative Disk I/O : 0.0
- Cumulative CPU : 0.0
- Output Size (bytes) : (none)
- Output Cardinality : (none)
- Avg. Output Record Size (bytes) : (none)
- Filter Factor : (none)
-
-Stage 5 : Data Source
- content : collect elements with CollectionInputFormat
- Partitioning : RANDOM_PARTITIONED
- Partitioning Order : (none)
- Uniqueness : not unique
- Order : (none)
- Grouping : not grouped
- Uniqueness : not unique
- Est. Output Size : (unknown)
- Est. Cardinality : (unknown)
- Network : 0.0
- Disk I/O : 0.0
- CPU : 0.0
- Cumulative Network : 0.0
- Cumulative Disk I/O : 0.0
- Cumulative CPU : 0.0
- Output Size (bytes) : (none)
- Output Cardinality : (none)
- Avg. Output Record Size (bytes) : (none)
- Filter Factor : (none)
-
- Stage 4 : Map
- content : Map at select('count as 'count,'word as 'word)
- ship_strategy : Forward
- exchange_mode : PIPELINED
- driver_strategy : Map
- Partitioning : RANDOM_PARTITIONED
- Partitioning Order : (none)
- Uniqueness : not unique
- Order : (none)
- Grouping : not grouped
- Uniqueness : not unique
- Est. Output Size : (unknown)
- Est. Cardinality : (unknown)
- Network : 0.0
- Disk I/O : 0.0
- CPU : 0.0
- Cumulative Network : 0.0
- Cumulative Disk I/O : 0.0
- Cumulative CPU : 0.0
- Output Size (bytes) : (none)
- Output Cardinality : (none)
- Avg. Output Record Size (bytes) : (none)
- Filter Factor : (none)
-
- Stage 1 : Join
- content : Join at 'b === 'd
- ship_strategy : Hash Partition on [1]
- exchange_mode : PIPELINED
- driver_strategy : Hybrid Hash (build: Map at select('count as 'count,'word as 'word))
- Partitioning : RANDOM_PARTITIONED
- Partitioning Order : (none)
- Uniqueness : not unique
- Order : (none)
- Grouping : not grouped
- Uniqueness : not unique
- Est. Output Size : (unknown)
- Est. Cardinality : (unknown)
- Network : (unknown)
- Disk I/O : (unknown)
- CPU : (unknown)
- Cumulative Network : (unknown)
- Cumulative Disk I/O : (unknown)
- Cumulative CPU : (unknown)
- Output Size (bytes) : (none)
- Output Cardinality : (none)
- Avg. Output Record Size (bytes) : (none)
- Filter Factor : (none)
-
- Stage 0 : Data Sink
- content : org.apache.flink.api.java.io.DiscardingOutputFormat
- ship_strategy : Forward
- exchange_mode : PIPELINED
- Partitioning : RANDOM_PARTITIONED
- Partitioning Order : (none)
- Uniqueness : not unique
- Order : (none)
- Grouping : not grouped
- Uniqueness : not unique
- Est. Output Size : (unknown)
- Est. Cardinality : (unknown)
- Network : 0.0
- Disk I/O : 0.0
- CPU : 0.0
- Cumulative Network : (unknown)
- Cumulative Disk I/O : (unknown)
- Cumulative CPU : (unknown)
- Output Size (bytes) : (none)
- Output Cardinality : (none)
- Avg. Output Record Size (bytes) : (none)
- Filter Factor : (none)
-
http://git-wip-us.apache.org/repos/asf/flink/blob/e9bf13d8/flink-staging/flink-table/src/test/scala/resources/testUnion0.out
----------------------------------------------------------------------
diff --git a/flink-staging/flink-table/src/test/scala/resources/testUnion0.out b/flink-staging/flink-table/src/test/scala/resources/testUnion0.out
deleted file mode 100644
index db9d2f9..0000000
--- a/flink-staging/flink-table/src/test/scala/resources/testUnion0.out
+++ /dev/null
@@ -1,38 +0,0 @@
-== Abstract Syntax Tree ==
-Union(Root(ArraySeq((count,Integer), (word,String))), Root(ArraySeq((count,Integer), (word,String))))
-
-== Physical Execution Plan ==
-Stage 3 : Data Source
- content : collect elements with CollectionInputFormat
- Partitioning : RANDOM_PARTITIONED
-
- Stage 2 : Map
- content : Map at select('count as 'count,'word as 'word)
- ship_strategy : Forward
- exchange_mode : PIPELINED
- driver_strategy : Map
- Partitioning : RANDOM_PARTITIONED
-
-Stage 5 : Data Source
- content : collect elements with CollectionInputFormat
- Partitioning : RANDOM_PARTITIONED
-
- Stage 4 : Map
- content : Map at select('count as 'count,'word as 'word)
- ship_strategy : Forward
- exchange_mode : PIPELINED
- driver_strategy : Map
- Partitioning : RANDOM_PARTITIONED
-
- Stage 1 : Union
- content :
- ship_strategy : Redistribute
- exchange_mode : PIPELINED
- Partitioning : RANDOM_PARTITIONED
-
- Stage 0 : Data Sink
- content : org.apache.flink.api.java.io.DiscardingOutputFormat
- ship_strategy : Forward
- exchange_mode : PIPELINED
- Partitioning : RANDOM_PARTITIONED
-
http://git-wip-us.apache.org/repos/asf/flink/blob/e9bf13d8/flink-staging/flink-table/src/test/scala/resources/testUnion1.out
----------------------------------------------------------------------
diff --git a/flink-staging/flink-table/src/test/scala/resources/testUnion1.out b/flink-staging/flink-table/src/test/scala/resources/testUnion1.out
deleted file mode 100644
index 8dc1e53..0000000
--- a/flink-staging/flink-table/src/test/scala/resources/testUnion1.out
+++ /dev/null
@@ -1,140 +0,0 @@
-== Abstract Syntax Tree ==
-Union(Root(ArraySeq((count,Integer), (word,String))), Root(ArraySeq((count,Integer), (word,String))))
-
-== Physical Execution Plan ==
-Stage 3 : Data Source
- content : collect elements with CollectionInputFormat
- Partitioning : RANDOM_PARTITIONED
- Partitioning Order : (none)
- Uniqueness : not unique
- Order : (none)
- Grouping : not grouped
- Uniqueness : not unique
- Est. Output Size : (unknown)
- Est. Cardinality : (unknown)
- Network : 0.0
- Disk I/O : 0.0
- CPU : 0.0
- Cumulative Network : 0.0
- Cumulative Disk I/O : 0.0
- Cumulative CPU : 0.0
- Output Size (bytes) : (none)
- Output Cardinality : (none)
- Avg. Output Record Size (bytes) : (none)
- Filter Factor : (none)
-
- Stage 2 : Map
- content : Map at select('count as 'count,'word as 'word)
- ship_strategy : Forward
- exchange_mode : PIPELINED
- driver_strategy : Map
- Partitioning : RANDOM_PARTITIONED
- Partitioning Order : (none)
- Uniqueness : not unique
- Order : (none)
- Grouping : not grouped
- Uniqueness : not unique
- Est. Output Size : (unknown)
- Est. Cardinality : (unknown)
- Network : 0.0
- Disk I/O : 0.0
- CPU : 0.0
- Cumulative Network : 0.0
- Cumulative Disk I/O : 0.0
- Cumulative CPU : 0.0
- Output Size (bytes) : (none)
- Output Cardinality : (none)
- Avg. Output Record Size (bytes) : (none)
- Filter Factor : (none)
-
-Stage 5 : Data Source
- content : collect elements with CollectionInputFormat
- Partitioning : RANDOM_PARTITIONED
- Partitioning Order : (none)
- Uniqueness : not unique
- Order : (none)
- Grouping : not grouped
- Uniqueness : not unique
- Est. Output Size : (unknown)
- Est. Cardinality : (unknown)
- Network : 0.0
- Disk I/O : 0.0
- CPU : 0.0
- Cumulative Network : 0.0
- Cumulative Disk I/O : 0.0
- Cumulative CPU : 0.0
- Output Size (bytes) : (none)
- Output Cardinality : (none)
- Avg. Output Record Size (bytes) : (none)
- Filter Factor : (none)
-
- Stage 4 : Map
- content : Map at select('count as 'count,'word as 'word)
- ship_strategy : Forward
- exchange_mode : PIPELINED
- driver_strategy : Map
- Partitioning : RANDOM_PARTITIONED
- Partitioning Order : (none)
- Uniqueness : not unique
- Order : (none)
- Grouping : not grouped
- Uniqueness : not unique
- Est. Output Size : (unknown)
- Est. Cardinality : (unknown)
- Network : 0.0
- Disk I/O : 0.0
- CPU : 0.0
- Cumulative Network : 0.0
- Cumulative Disk I/O : 0.0
- Cumulative CPU : 0.0
- Output Size (bytes) : (none)
- Output Cardinality : (none)
- Avg. Output Record Size (bytes) : (none)
- Filter Factor : (none)
-
- Stage 1 : Union
- content :
- ship_strategy : Redistribute
- exchange_mode : PIPELINED
- Partitioning : RANDOM_PARTITIONED
- Partitioning Order : (none)
- Uniqueness : not unique
- Order : (none)
- Grouping : not grouped
- Uniqueness : not unique
- Est. Output Size : (unknown)
- Est. Cardinality : (unknown)
- Network : 0.0
- Disk I/O : 0.0
- CPU : 0.0
- Cumulative Network : (unknown)
- Cumulative Disk I/O : 0.0
- Cumulative CPU : 0.0
- Output Size (bytes) : (none)
- Output Cardinality : (none)
- Avg. Output Record Size (bytes) : (none)
- Filter Factor : (none)
-
- Stage 0 : Data Sink
- content : org.apache.flink.api.java.io.DiscardingOutputFormat
- ship_strategy : Forward
- exchange_mode : PIPELINED
- Partitioning : RANDOM_PARTITIONED
- Partitioning Order : (none)
- Uniqueness : not unique
- Order : (none)
- Grouping : not grouped
- Uniqueness : not unique
- Est. Output Size : (unknown)
- Est. Cardinality : (unknown)
- Network : 0.0
- Disk I/O : 0.0
- CPU : 0.0
- Cumulative Network : (unknown)
- Cumulative Disk I/O : 0.0
- Cumulative CPU : 0.0
- Output Size (bytes) : (none)
- Output Cardinality : (none)
- Avg. Output Record Size (bytes) : (none)
- Filter Factor : (none)
-
http://git-wip-us.apache.org/repos/asf/flink/blob/e9bf13d8/flink-staging/flink-tez/pom.xml
----------------------------------------------------------------------
diff --git a/flink-staging/flink-tez/pom.xml b/flink-staging/flink-tez/pom.xml
deleted file mode 100644
index 6083f7f..0000000
--- a/flink-staging/flink-tez/pom.xml
+++ /dev/null
@@ -1,224 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<!--
-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.
--->
-<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
- xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
-
- <modelVersion>4.0.0</modelVersion>
-
- <parent>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-staging</artifactId>
- <version>1.0-SNAPSHOT</version>
- <relativePath>..</relativePath>
- </parent>
-
- <artifactId>flink-tez</artifactId>
- <name>flink-tez</name>
-
- <packaging>jar</packaging>
-
- <dependencies>
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-java</artifactId>
- <version>${project.version}</version>
- <exclusions>
- <exclusion>
- <groupId>org.apache.flink</groupId>
- <artifactId>${shading-artifact.name}</artifactId>
- </exclusion>
- </exclusions>
- </dependency>
-
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-core</artifactId>
- <version>${project.version}</version>
- <exclusions>
- <exclusion>
- <groupId>org.apache.flink</groupId>
- <artifactId>${shading-artifact.name}</artifactId>
- </exclusion>
- </exclusions>
- </dependency>
-
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-runtime</artifactId>
- <version>${project.version}</version>
- <exclusions>
- <exclusion>
- <groupId>org.apache.flink</groupId>
- <artifactId>${shading-artifact.name}</artifactId>
- </exclusion>
- </exclusions>
- </dependency>
-
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-optimizer</artifactId>
- <version>${project.version}</version>
- <exclusions>
- <exclusion>
- <groupId>org.apache.flink</groupId>
- <artifactId>${shading-artifact.name}</artifactId>
- </exclusion>
- </exclusions>
- </dependency>
-
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-examples-batch</artifactId>
- <version>${project.version}</version>
- <exclusions>
- <exclusion>
- <groupId>org.apache.flink</groupId>
- <artifactId>${shading-artifact.name}</artifactId>
- </exclusion>
- </exclusions>
- </dependency>
-
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-test-utils</artifactId>
- <version>${project.version}</version>
- <exclusions>
- <exclusion>
- <groupId>org.apache.flink</groupId>
- <artifactId>${shading-artifact.name}</artifactId>
- </exclusion>
- </exclusions>
- </dependency>
-
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-tests</artifactId>
- <version>${project.version}</version>
- <exclusions>
- <exclusion>
- <groupId>org.apache.flink</groupId>
- <artifactId>${shading-artifact.name}</artifactId>
- </exclusion>
- </exclusions>
- </dependency>
-
- <dependency>
- <groupId>org.apache.tez</groupId>
- <artifactId>tez-api</artifactId>
- <version>${tez.version}</version>
- </dependency>
-
- <dependency>
- <groupId>org.apache.tez</groupId>
- <artifactId>tez-common</artifactId>
- <version>${tez.version}</version>
- </dependency>
-
- <dependency>
- <groupId>org.apache.tez</groupId>
- <artifactId>tez-dag</artifactId>
- <version>${tez.version}</version>
- </dependency>
-
- <dependency>
- <groupId>org.apache.tez</groupId>
- <artifactId>tez-runtime-library</artifactId>
- <version>${tez.version}</version>
- </dependency>
- <dependency>
- <groupId>org.apache.hadoop</groupId>
- <artifactId>hadoop-common</artifactId>
- <version>${hadoop.version}</version>
- </dependency>
-
- <dependency>
- <groupId>org.apache.hadoop</groupId>
- <artifactId>hadoop-hdfs</artifactId>
- <version>${hadoop.version}</version>
- </dependency>
-
- <dependency>
- <groupId>org.apache.hadoop</groupId>
- <artifactId>hadoop-mapreduce-client-core</artifactId>
- <version>${hadoop.version}</version>
- </dependency>
-
- <dependency>
- <groupId>org.apache.hadoop</groupId>
- <artifactId>hadoop-yarn-client</artifactId>
- <version>${hadoop.version}</version>
- </dependency>
-
- <dependency>
- <groupId>org.apache.hadoop</groupId>
- <artifactId>hadoop-yarn-common</artifactId>
- <version>${hadoop.version}</version>
- </dependency>
-
- <dependency>
- <groupId>org.apache.hadoop</groupId>
- <artifactId>hadoop-yarn-api</artifactId>
- <version>${hadoop.version}</version>
- </dependency>
-
- <dependency>
- <groupId>commons-codec</groupId>
- <artifactId>commons-codec</artifactId>
- <version>1.4</version>
- </dependency>
-
-
- <dependency>
- <groupId>com.google.guava</groupId>
- <artifactId>guava</artifactId>
- <version>${guava.version}</version>
- </dependency>
-
- </dependencies>
-
- <build>
- <plugins>
- <plugin>
- <artifactId>maven-assembly-plugin</artifactId>
- <version>2.4.1</version>
- <configuration>
- <descriptors>
- <descriptor>${basedir}/src/assembly/flink-fat-jar.xml</descriptor>
- </descriptors>
- <archive>
- <manifest>
- <mainClass>org.apache.flink.tez.examples.ExampleDriver</mainClass>
- </manifest>
- </archive>
- </configuration>
- <executions>
- <execution>
- <!--<id>assemble-all</id>-->
- <id>make-assembly</id>
- <phase>package</phase>
- <goals>
- <goal>single</goal>
- </goals>
- </execution>
- </executions>
- </plugin>
- </plugins>
- </build>
-</project>
http://git-wip-us.apache.org/repos/asf/flink/blob/e9bf13d8/flink-staging/flink-tez/src/assembly/flink-fat-jar.xml
----------------------------------------------------------------------
diff --git a/flink-staging/flink-tez/src/assembly/flink-fat-jar.xml b/flink-staging/flink-tez/src/assembly/flink-fat-jar.xml
deleted file mode 100644
index 504761a..0000000
--- a/flink-staging/flink-tez/src/assembly/flink-fat-jar.xml
+++ /dev/null
@@ -1,42 +0,0 @@
-<!--
- ~ 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.
- -->
-
-<assembly xmlns="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.2"
- xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
- xsi:schemaLocation="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.2 http://maven.apache.org/xsd/assembly-1.1.2.xsd">
- <id>flink-fat-jar</id>
- <formats>
- <format>jar</format>
- </formats>
- <includeBaseDirectory>false</includeBaseDirectory>
- <dependencySets>
- <dependencySet>
- <outputDirectory>/</outputDirectory>
- <useProjectArtifact>true</useProjectArtifact>
- <!--<excludes>
- <exclude>org.apache.flink:*</exclude>
- </excludes>-->
- <useTransitiveFiltering>true</useTransitiveFiltering>
- <unpack>true</unpack>
- <scope>runtime</scope>
- <excludes>
- <exclude>com.google.guava:guava</exclude>
- </excludes>
- </dependencySet>
- </dependencySets>
-</assembly>
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/flink/blob/e9bf13d8/flink-staging/flink-tez/src/main/java/org/apache/flink/tez/client/LocalTezEnvironment.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-tez/src/main/java/org/apache/flink/tez/client/LocalTezEnvironment.java b/flink-staging/flink-tez/src/main/java/org/apache/flink/tez/client/LocalTezEnvironment.java
deleted file mode 100644
index 4c091e5..0000000
--- a/flink-staging/flink-tez/src/main/java/org/apache/flink/tez/client/LocalTezEnvironment.java
+++ /dev/null
@@ -1,76 +0,0 @@
-/*
- * 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.tez.client;
-
-import org.apache.flink.api.common.JobExecutionResult;
-import org.apache.flink.api.common.Plan;
-import org.apache.flink.api.java.ExecutionEnvironment;
-import org.apache.flink.api.java.ExecutionEnvironmentFactory;
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.optimizer.DataStatistics;
-import org.apache.flink.optimizer.Optimizer;
-import org.apache.flink.optimizer.costs.DefaultCostEstimator;
-import org.apache.tez.dag.api.TezConfiguration;
-import org.apache.tez.runtime.library.api.TezRuntimeConfiguration;
-
-public class LocalTezEnvironment extends ExecutionEnvironment {
-
- TezExecutor executor;
- Optimizer compiler;
-
- private LocalTezEnvironment() {
- compiler = new Optimizer(new DataStatistics(), new DefaultCostEstimator(), new Configuration());
- executor = new TezExecutor(compiler, this.getParallelism());
- }
-
- public static LocalTezEnvironment create() {
- return new LocalTezEnvironment();
- }
-
- @Override
- public JobExecutionResult execute(String jobName) throws Exception {
- TezConfiguration tezConf = new TezConfiguration();
- tezConf.setBoolean(TezConfiguration.TEZ_LOCAL_MODE, true);
- tezConf.set("fs.defaultFS", "file:///");
- tezConf.setBoolean(TezRuntimeConfiguration.TEZ_RUNTIME_OPTIMIZE_LOCAL_FETCH, true);
- executor.setConfiguration(tezConf);
- return executor.executePlan(createProgramPlan(jobName));
- }
-
- @Override
- public String getExecutionPlan() throws Exception {
- Plan p = createProgramPlan(null, false);
- return executor.getOptimizerPlanAsJSON(p);
- }
-
- public void setAsContext() {
- ExecutionEnvironmentFactory factory = new ExecutionEnvironmentFactory() {
- @Override
- public ExecutionEnvironment createExecutionEnvironment() {
- return LocalTezEnvironment.this;
- }
- };
- initializeContextEnvironment(factory);
- }
-
- @Override
- public void startNewSession() throws Exception {
- throw new UnsupportedOperationException("Session management is not implemented in Flink on Tez.");
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/e9bf13d8/flink-staging/flink-tez/src/main/java/org/apache/flink/tez/client/RemoteTezEnvironment.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-tez/src/main/java/org/apache/flink/tez/client/RemoteTezEnvironment.java b/flink-staging/flink-tez/src/main/java/org/apache/flink/tez/client/RemoteTezEnvironment.java
deleted file mode 100644
index 131937e..0000000
--- a/flink-staging/flink-tez/src/main/java/org/apache/flink/tez/client/RemoteTezEnvironment.java
+++ /dev/null
@@ -1,83 +0,0 @@
-/*
- * 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.tez.client;
-
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.apache.flink.api.common.JobExecutionResult;
-import org.apache.flink.api.common.Plan;
-import org.apache.flink.api.java.ExecutionEnvironment;
-import org.apache.flink.optimizer.DataStatistics;
-import org.apache.flink.optimizer.Optimizer;
-import org.apache.flink.optimizer.costs.DefaultCostEstimator;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.util.ClassUtil;
-import org.apache.hadoop.util.ToolRunner;
-
-
-public class RemoteTezEnvironment extends ExecutionEnvironment {
-
- private static final Log LOG = LogFactory.getLog(RemoteTezEnvironment.class);
-
- private Optimizer compiler;
- private TezExecutor executor;
- private Path jarPath = null;
-
-
- public void registerMainClass (Class mainClass) {
- jarPath = new Path(ClassUtil.findContainingJar(mainClass));
- LOG.info ("Registering main class " + mainClass.getName() + " contained in " + jarPath.toString());
- }
-
- @Override
- public JobExecutionResult execute(String jobName) throws Exception {
- TezExecutorTool tool = new TezExecutorTool(executor, createProgramPlan());
- if (jarPath != null) {
- tool.setJobJar(jarPath);
- }
- try {
- int executionResult = ToolRunner.run(new Configuration(), tool, new String[]{jobName});
- }
- finally {
- return new JobExecutionResult(null, -1, null);
- }
-
- }
-
- @Override
- public String getExecutionPlan() throws Exception {
- Plan p = createProgramPlan(null, false);
- return executor.getOptimizerPlanAsJSON(p);
- }
-
- public static RemoteTezEnvironment create () {
- return new RemoteTezEnvironment();
- }
-
- public RemoteTezEnvironment() {
- compiler = new Optimizer(new DataStatistics(), new DefaultCostEstimator(), new org.apache.flink.configuration.Configuration());
- executor = new TezExecutor(compiler, getParallelism());
- }
-
- @Override
- public void startNewSession() throws Exception {
- throw new UnsupportedOperationException("Session management is not implemented in Flink on Tez.");
- }
-}