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 2014/10/18 17:17:18 UTC
[3/9] [FLINK-1171] Move Scala API tests to flink-tests project
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/a0ad9031/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/JoinOperatorTest.scala
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/JoinOperatorTest.scala b/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/JoinOperatorTest.scala
new file mode 100644
index 0000000..0219154
--- /dev/null
+++ b/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/JoinOperatorTest.scala
@@ -0,0 +1,276 @@
+/*
+ * 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.operators
+
+import org.apache.flink.api.java.operators.Keys.IncompatibleKeysException
+import org.apache.flink.api.scala.util.CollectionDataSets.CustomType
+import org.junit.Assert
+import org.apache.flink.api.common.InvalidProgramException
+import org.junit.Ignore
+import org.junit.Test
+
+import org.apache.flink.api.scala._
+
+class JoinOperatorTest {
+
+ private val emptyTupleData = Array[(Int, Long, String, Long, Int)]()
+ private val customTypeData = Array[CustomType](new CustomType())
+ private val emptyLongData = Array[Long]()
+
+ @Test
+ def testJoinKeyIndices1(): Unit = {
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ val ds1 = env.fromCollection(emptyTupleData)
+ val ds2 = env.fromCollection(emptyTupleData)
+
+ // should work
+ try {
+ ds1.join(ds2).where(0).equalTo(0)
+ }
+ catch {
+ case e: Exception => Assert.fail()
+ }
+ }
+
+ @Test(expected = classOf[IncompatibleKeysException])
+ def testJoinKeyIndices2(): Unit = {
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ val ds1 = env.fromCollection(emptyTupleData)
+ val ds2 = env.fromCollection(emptyTupleData)
+
+ // should not work, incompatible key types
+ ds1.join(ds2).where(0).equalTo(2)
+ }
+
+ @Test(expected = classOf[IncompatibleKeysException])
+ def testJoinKeyIndices3(): Unit = {
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ val ds1 = env.fromCollection(emptyTupleData)
+ val ds2 = env.fromCollection(emptyTupleData)
+
+ // should not work, non-matching number of key indices
+ ds1.join(ds2).where(0, 1).equalTo(2)
+ }
+
+ @Test(expected = classOf[IllegalArgumentException])
+ def testJoinKeyIndices4(): Unit = {
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ val ds1 = env.fromCollection(emptyTupleData)
+ val ds2 = env.fromCollection(emptyTupleData)
+
+ // should not work, index out of range
+ ds1.join(ds2).where(5).equalTo(0)
+ }
+
+ @Test(expected = classOf[IllegalArgumentException])
+ def testJoinKeyIndices5(): Unit = {
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ val ds1 = env.fromCollection(emptyTupleData)
+ val ds2 = env.fromCollection(emptyTupleData)
+
+ // should not work, negative position
+ ds1.join(ds2).where(-1).equalTo(-1)
+ }
+
+ @Test(expected = classOf[IllegalArgumentException])
+ def testJoinKeyIndices6(): Unit = {
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ val ds1 = env.fromCollection(emptyTupleData)
+ val ds2 = env.fromCollection(customTypeData)
+
+ // should not work, key index on custom type
+ ds1.join(ds2).where(5).equalTo(0)
+ }
+
+ @Test
+ def testJoinKeyFields1(): Unit = {
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ val ds1 = env.fromCollection(emptyTupleData)
+ val ds2 = env.fromCollection(emptyTupleData)
+
+ // should work
+ try {
+ ds1.join(ds2).where("_1").equalTo("_1")
+ }
+ catch {
+ case e: Exception => Assert.fail()
+ }
+ }
+
+ @Test(expected = classOf[IncompatibleKeysException])
+ def testJoinKeyFields2(): Unit = {
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ val ds1 = env.fromCollection(emptyTupleData)
+ val ds2 = env.fromCollection(emptyTupleData)
+
+ // should not work, incompatible field types
+ ds1.join(ds2).where("_1").equalTo("_3")
+ }
+
+ @Test(expected = classOf[IncompatibleKeysException])
+ def testJoinKeyFields3(): Unit = {
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ val ds1 = env.fromCollection(emptyTupleData)
+ val ds2 = env.fromCollection(emptyTupleData)
+
+ // should not work, non-matching number of key indices
+
+ ds1.join(ds2).where("_1", "_2").equalTo("_3")
+ }
+
+ @Test(expected = classOf[RuntimeException])
+ def testJoinKeyFields4(): Unit = {
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ val ds1 = env.fromCollection(emptyTupleData)
+ val ds2 = env.fromCollection(emptyTupleData)
+
+ // should not work, non-existent key
+ ds1.join(ds2).where("foo").equalTo("_1")
+ }
+
+ @Test(expected = classOf[RuntimeException])
+ def testJoinKeyFields5(): Unit = {
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ val ds1 = env.fromCollection(emptyTupleData)
+ val ds2 = env.fromCollection(emptyTupleData)
+
+ // should not work, non-matching number of key indices
+ ds1.join(ds2).where("_1").equalTo("bar")
+ }
+
+ @Test(expected = classOf[IllegalArgumentException])
+ def testJoinKeyFields6(): Unit = {
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ val ds1 = env.fromCollection(emptyTupleData)
+ val ds2 = env.fromCollection(customTypeData)
+
+ // should not work, field key on custom type
+ ds1.join(ds2).where("_2").equalTo("_1")
+ }
+
+ @Test
+ def testJoinKeyExpressions1(): Unit = {
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ val ds1 = env.fromCollection(customTypeData)
+ val ds2 = env.fromCollection(customTypeData)
+
+ // should work
+ try {
+ ds1.join(ds2).where("myInt").equalTo("myInt")
+ }
+ catch {
+ case e: Exception => Assert.fail()
+ }
+ }
+
+ @Test(expected = classOf[IncompatibleKeysException])
+ def testJoinKeyExpressions2(): Unit = {
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ val ds1 = env.fromCollection(customTypeData)
+ val ds2 = env.fromCollection(customTypeData)
+
+ // should not work, incompatible join key types
+ ds1.join(ds2).where("myInt").equalTo("myString")
+ }
+
+ @Test(expected = classOf[IncompatibleKeysException])
+ def testJoinKeyExpressions3(): Unit = {
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ val ds1 = env.fromCollection(customTypeData)
+ val ds2 = env.fromCollection(customTypeData)
+
+ // should not work, incompatible number of keys
+ ds1.join(ds2).where("myInt", "myString").equalTo("myInt")
+ }
+
+ @Test(expected = classOf[IllegalArgumentException])
+ def testJoinKeyExpressions4(): Unit = {
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ val ds1 = env.fromCollection(customTypeData)
+ val ds2 = env.fromCollection(customTypeData)
+
+ // should not work, join key non-existent
+ ds1.join(ds2).where("myNonExistent").equalTo("i")
+ }
+
+ @Test
+ def testJoinKeySelectors1(): Unit = {
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ val ds1 = env.fromCollection(customTypeData)
+ val ds2 = env.fromCollection(customTypeData)
+
+ // should work
+ try {
+ ds1.join(ds2).where { _.myLong} equalTo { _.myLong }
+ }
+ catch {
+ case e: Exception => Assert.fail()
+ }
+ }
+
+ @Test
+ def testJoinKeyMixing1(): Unit = {
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ val ds1 = env.fromCollection(customTypeData)
+ val ds2 = env.fromCollection(emptyTupleData)
+
+ // should work
+ try {
+ ds1.join(ds2).where { _.myLong }.equalTo(3)
+ }
+ catch {
+ case e: Exception => Assert.fail()
+ }
+ }
+
+ @Test
+ def testJoinKeyMixing2(): Unit = {
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ val ds1 = env.fromCollection(emptyTupleData)
+ val ds2 = env.fromCollection(customTypeData)
+
+ // should work
+ try {
+ ds1.join(ds2).where(3).equalTo { _.myLong }
+ }
+ catch {
+ case e: Exception => Assert.fail()
+ }
+ }
+
+ @Test(expected = classOf[IncompatibleKeysException])
+ def testJoinKeyMixing3(): Unit = {
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ val ds1 = env.fromCollection(emptyTupleData)
+ val ds2 = env.fromCollection(customTypeData)
+
+ // should not work, incompatible types
+ ds1.join(ds2).where(2).equalTo { _.myLong }
+ }
+
+ @Test(expected = classOf[IncompatibleKeysException])
+ def testJoinKeyMixing4(): Unit = {
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ val ds1 = env.fromCollection(emptyTupleData)
+ val ds2 = env.fromCollection(customTypeData)
+
+ // should not work, more than one field position key
+ ds1.join(ds2).where(1, 3) equalTo { _.myLong }
+ }
+}
+
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/a0ad9031/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/MapITCase.scala
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/MapITCase.scala b/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/MapITCase.scala
new file mode 100644
index 0000000..71c6b8b
--- /dev/null
+++ b/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/MapITCase.scala
@@ -0,0 +1,242 @@
+/*
+ * 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.operators
+
+import org.apache.flink.api.common.functions.RichMapFunction
+import org.apache.flink.api.scala.util.CollectionDataSets
+import org.apache.flink.api.scala.util.CollectionDataSets.MutableTuple3
+import org.apache.flink.configuration.Configuration
+import org.apache.flink.test.util.JavaProgramTestBase
+import org.apache.flink.util.Collector
+import org.junit.Assert
+import org.junit.runner.RunWith
+import org.junit.runners.Parameterized
+import org.junit.runners.Parameterized.Parameters
+
+import scala.collection.JavaConverters._
+import scala.collection.mutable
+
+import org.apache.flink.api.scala._
+
+
+object MapProgs {
+ var NUM_PROGRAMS: Int = 9
+
+ def runProgram(progId: Int, resultPath: String): String = {
+ progId match {
+ case 1 =>
+ /*
+ * Test identity map with basic type
+ */
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ val ds = CollectionDataSets.getStringDataSet(env)
+ val identityMapDs = ds.map( t => t)
+ identityMapDs.writeAsText(resultPath)
+ env.execute()
+ "Hi\n" + "Hello\n" + "Hello world\n" + "Hello world, how are you?\n" + "I am fine" +
+ ".\n" + "Luke Skywalker\n" + "Random comment\n" + "LOL\n"
+
+ case 2 =>
+ /*
+ * Test identity map with a tuple
+ */
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ val ds = CollectionDataSets.get3TupleDataSet(env)
+ val identityMapDs = ds.map( t => t )
+ identityMapDs.writeAsCsv(resultPath)
+ env.execute()
+ "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"
+
+ case 3 =>
+ /*
+ * Test type conversion mapper (Custom -> Tuple)
+ */
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ val ds = CollectionDataSets.getCustomTypeDataSet(env)
+ val typeConversionMapDs = ds.map( c => (c.myInt, c.myLong, c.myString) )
+ typeConversionMapDs.writeAsCsv(resultPath)
+ env.execute()
+ "1,0,Hi\n" + "2,1,Hello\n" + "2,2,Hello world\n" + "3,3,Hello world, " +
+ "how are you?\n" + "3,4,I am fine.\n" + "3,5,Luke Skywalker\n" + "4,6," +
+ "Comment#1\n" + "4,7,Comment#2\n" + "4,8,Comment#3\n" + "4,9,Comment#4\n" + "5,10," +
+ "Comment#5\n" + "5,11,Comment#6\n" + "5,12,Comment#7\n" + "5,13,Comment#8\n" + "5,14," +
+ "Comment#9\n" + "6,15,Comment#10\n" + "6,16,Comment#11\n" + "6,17,Comment#12\n" + "6," +
+ "18,Comment#13\n" + "6,19,Comment#14\n" + "6,20,Comment#15\n"
+
+ case 4 =>
+ /*
+ * Test type conversion mapper (Tuple -> Basic)
+ */
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ val ds = CollectionDataSets.get3TupleDataSet(env)
+ val typeConversionMapDs = ds.map(_._3)
+ typeConversionMapDs.writeAsText(resultPath)
+ env.execute()
+ "Hi\n" + "Hello\n" + "Hello world\n" + "Hello world, how are you?\n" + "I am fine" +
+ ".\n" + "Luke Skywalker\n" + "Comment#1\n" + "Comment#2\n" + "Comment#3\n" +
+ "Comment#4\n" + "Comment#5\n" + "Comment#6\n" + "Comment#7\n" + "Comment#8\n" +
+ "Comment#9\n" + "Comment#10\n" + "Comment#11\n" + "Comment#12\n" + "Comment#13\n" +
+ "Comment#14\n" + "Comment#15\n"
+
+ case 5 =>
+ /*
+ * Test mapper on tuple - Increment Integer field, reorder second and third fields
+ */
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ val ds = CollectionDataSets.get3TupleDataSet(env)
+ val tupleMapDs = ds.map( t => (t._1 + 1, t._3, t._2) )
+ tupleMapDs.writeAsCsv(resultPath)
+ env.execute()
+ "2,Hi,1\n" + "3,Hello,2\n" + "4,Hello world,2\n" + "5,Hello world, how are you?," +
+ "3\n" + "6,I am fine.,3\n" + "7,Luke Skywalker,3\n" + "8,Comment#1,4\n" + "9,Comment#2," +
+ "4\n" + "10,Comment#3,4\n" + "11,Comment#4,4\n" + "12,Comment#5,5\n" + "13,Comment#6," +
+ "5\n" + "14,Comment#7,5\n" + "15,Comment#8,5\n" + "16,Comment#9,5\n" + "17,Comment#10," +
+ "6\n" + "18,Comment#11,6\n" + "19,Comment#12,6\n" + "20,Comment#13,6\n" + "21," +
+ "Comment#14,6\n" + "22,Comment#15,6\n"
+
+ case 6 =>
+ /*
+ * Test mapper on Custom - lowercase myString
+ */
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ val ds = CollectionDataSets.getCustomTypeDataSet(env)
+ val customMapDs = ds.map { c => c.myString = c.myString.toLowerCase; c }
+ customMapDs.writeAsText(resultPath)
+ env.execute()
+ "1,0,hi\n" + "2,1,hello\n" + "2,2,hello world\n" + "3,3,hello world, " +
+ "how are you?\n" + "3,4,i am fine.\n" + "3,5,luke skywalker\n" + "4,6," +
+ "comment#1\n" + "4,7,comment#2\n" + "4,8,comment#3\n" + "4,9,comment#4\n" + "5,10," +
+ "comment#5\n" + "5,11,comment#6\n" + "5,12,comment#7\n" + "5,13,comment#8\n" + "5,14," +
+ "comment#9\n" + "6,15,comment#10\n" + "6,16,comment#11\n" + "6,17,comment#12\n" + "6," +
+ "18,comment#13\n" + "6,19,comment#14\n" + "6,20,comment#15\n"
+
+ case 7 =>
+ /*
+ * Test mapper if UDF returns input object - increment first field of a tuple
+ */
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ val ds = CollectionDataSets.get3TupleDataSet(env).map {
+ t => MutableTuple3(t._1, t._2, t._3)
+ }
+ val inputObjMapDs = ds.map { t => t._1 = t._1 + 1; t }
+ inputObjMapDs.writeAsCsv(resultPath)
+ env.execute()
+ "2,1,Hi\n" + "3,2,Hello\n" + "4,2,Hello world\n" + "5,3,Hello world, " +
+ "how are you?\n" + "6,3,I am fine.\n" + "7,3,Luke Skywalker\n" + "8,4," +
+ "Comment#1\n" + "9,4,Comment#2\n" + "10,4,Comment#3\n" + "11,4,Comment#4\n" + "12,5," +
+ "Comment#5\n" + "13,5,Comment#6\n" + "14,5,Comment#7\n" + "15,5,Comment#8\n" + "16,5," +
+ "Comment#9\n" + "17,6,Comment#10\n" + "18,6,Comment#11\n" + "19,6,Comment#12\n" + "20," +
+ "6,Comment#13\n" + "21,6,Comment#14\n" + "22,6,Comment#15\n"
+
+ case 8 =>
+ /*
+ * Test map with broadcast set
+ */
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ val ints = CollectionDataSets.getIntDataSet(env)
+ val ds = CollectionDataSets.get3TupleDataSet(env)
+ val bcMapDs = ds.map(
+ new RichMapFunction[(Int, Long, String), (Int, Long, String)] {
+ var f2Replace = 0
+ override def open(config: Configuration): Unit = {
+ val ints = getRuntimeContext.getBroadcastVariable[Int]("ints").asScala
+ f2Replace = ints.sum
+ }
+ override def map(in: (Int, Long, String)): (Int, Long, String) = {
+ in.copy(_1 = f2Replace)
+ }
+ }).withBroadcastSet(ints, "ints")
+ bcMapDs.writeAsCsv(resultPath)
+ env.execute()
+ "55,1,Hi\n" + "55,2,Hello\n" + "55,2,Hello world\n" + "55,3,Hello world, " +
+ "how are you?\n" + "55,3,I am fine.\n" + "55,3,Luke Skywalker\n" + "55,4," +
+ "Comment#1\n" + "55,4,Comment#2\n" + "55,4,Comment#3\n" + "55,4,Comment#4\n" + "55,5," +
+ "Comment#5\n" + "55,5,Comment#6\n" + "55,5,Comment#7\n" + "55,5,Comment#8\n" + "55,5," +
+ "Comment#9\n" + "55,6,Comment#10\n" + "55,6,Comment#11\n" + "55,6,Comment#12\n" + "55," +
+ "6,Comment#13\n" + "55,6,Comment#14\n" + "55,6,Comment#15\n"
+
+ case 9 =>
+ /*
+ * Test passing configuration object.
+ */
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ val ds = CollectionDataSets.getSmall3TupleDataSet(env)
+ val conf = new Configuration
+ val testKey = "testVariable"
+ val testValue = 666
+ conf.setInteger(testKey, testValue)
+ val bcMapDs = ds.map(
+ new RichMapFunction[(Int, Long, String), (Int, Long, String)] {
+ override def open(config: Configuration): Unit = {
+ val fromConfig = config.getInteger(testKey, -1)
+ Assert.assertEquals(testValue, fromConfig)
+ }
+ override def map(in: (Int, Long, String)): (Int, Long, String) = {
+ in
+ }
+ }).withParameters(conf)
+ bcMapDs.writeAsCsv(resultPath)
+ env.execute()
+ "1,1,Hi\n" + "2,2,Hello\n" + "3,2,Hello world"
+
+ case _ =>
+ throw new IllegalArgumentException("Invalid program id")
+ }
+ }
+}
+
+
+@RunWith(classOf[Parameterized])
+class MapITCase(config: Configuration) extends JavaProgramTestBase(config) {
+
+ private var curProgId: Int = config.getInteger("ProgramId", -1)
+ private var resultPath: String = null
+ private var expectedResult: String = null
+
+ protected override def preSubmit(): Unit = {
+ resultPath = getTempDirPath("result")
+ }
+
+ protected def testProgram(): Unit = {
+ expectedResult = MapProgs.runProgram(curProgId, resultPath)
+ }
+
+ protected override def postSubmit(): Unit = {
+ compareResultsByLinesInMemory(expectedResult, resultPath)
+ }
+}
+
+object MapITCase {
+ @Parameters
+ def getConfigurations: java.util.Collection[Array[AnyRef]] = {
+ val configs = mutable.MutableList[Array[AnyRef]]()
+ for (i <- 1 to MapProgs.NUM_PROGRAMS) {
+ val config = new Configuration()
+ config.setInteger("ProgramId", i)
+ configs += Array(config)
+ }
+
+ configs.asJavaCollection
+ }
+}
+
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/a0ad9031/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/PartitionITCase.scala
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/PartitionITCase.scala b/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/PartitionITCase.scala
new file mode 100644
index 0000000..07d4897
--- /dev/null
+++ b/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/PartitionITCase.scala
@@ -0,0 +1,218 @@
+/*
+ * 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.operators
+
+import org.apache.flink.api.common.functions.{RichFilterFunction, RichMapFunction}
+import org.apache.flink.api.scala.ExecutionEnvironment
+import org.apache.flink.api.scala.util.CollectionDataSets
+import org.apache.flink.configuration.Configuration
+import org.apache.flink.test.util.JavaProgramTestBase
+import org.junit.runner.RunWith
+import org.junit.runners.Parameterized
+import org.junit.runners.Parameterized.Parameters
+
+import scala.collection.JavaConverters._
+import scala.collection.mutable
+
+import org.apache.flink.api.scala._
+
+
+object PartitionProgs {
+ var NUM_PROGRAMS: Int = 7
+
+ def runProgram(progId: Int, resultPath: String, onCollection: Boolean): String = {
+ progId match {
+ case 1 =>
+ /*
+ * Test hash partition by tuple field
+ */
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ val ds = CollectionDataSets.get3TupleDataSet(env)
+
+ val unique = ds.partitionByHash(1).mapPartition( _.map(_._2).toSet )
+
+ unique.writeAsText(resultPath)
+ env.execute()
+
+ "1\n" + "2\n" + "3\n" + "4\n" + "5\n" + "6\n"
+
+ case 2 =>
+ /*
+ * Test hash partition by key selector
+ */
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ val ds = CollectionDataSets.get3TupleDataSet(env)
+ val unique = ds.partitionByHash( _._2 ).mapPartition( _.map(_._2).toSet )
+
+ unique.writeAsText(resultPath)
+ env.execute()
+ "1\n" + "2\n" + "3\n" + "4\n" + "5\n" + "6\n"
+
+ case 3 =>
+ /*
+ * Test forced rebalancing
+ */
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ val ds = env.generateSequence(1, 3000)
+
+ val skewed = ds.filter(_ > 780)
+ val rebalanced = skewed.rebalance()
+
+ val countsInPartition = rebalanced.map( new RichMapFunction[Long, (Int, Long)] {
+ def map(in: Long) = {
+ (getRuntimeContext.getIndexOfThisSubtask, 1)
+ }
+ })
+ .groupBy(0)
+ .reduce { (v1, v2) => (v1._1, v1._2 + v2._2) }
+ // round counts to mitigate runtime scheduling effects (lazy split assignment)
+ .map { in => (in._1, in._2 / 10) }
+
+ countsInPartition.writeAsText(resultPath)
+ env.execute()
+
+ val numPerPartition : Int = 2220 / env.getDegreeOfParallelism / 10
+ var result = ""
+ for (i <- 0 until env.getDegreeOfParallelism) {
+ result += "(" + i + "," + numPerPartition + ")\n"
+ }
+ result
+
+ case 4 =>
+ // Verify that mapPartition operation after repartition picks up correct
+ // DOP
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ val ds = CollectionDataSets.get3TupleDataSet(env)
+ env.setDegreeOfParallelism(1)
+
+ val unique = ds.partitionByHash(1)
+ .setParallelism(4)
+ .mapPartition( _.map(_._2).toSet )
+
+ unique.writeAsText(resultPath)
+ env.execute()
+
+ "1\n" + "2\n" + "3\n" + "4\n" + "5\n" + "6\n"
+
+ case 5 =>
+ // Verify that map operation after repartition picks up correct
+ // DOP
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ val ds = CollectionDataSets.get3TupleDataSet(env)
+ env.setDegreeOfParallelism(1)
+
+ val count = ds.partitionByHash(0).setParallelism(4).map(
+ new RichMapFunction[(Int, Long, String), Tuple1[Int]] {
+ var first = true
+ override def map(in: (Int, Long, String)): Tuple1[Int] = {
+ // only output one value with count 1
+ if (first) {
+ first = false
+ Tuple1(1)
+ } else {
+ Tuple1(0)
+ }
+ }
+ }).sum(0)
+
+ count.writeAsText(resultPath)
+ env.execute()
+
+ if (onCollection) "(1)\n" else "(4)\n"
+
+ case 6 =>
+ // Verify that filter operation after repartition picks up correct
+ // DOP
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ val ds = CollectionDataSets.get3TupleDataSet(env)
+ env.setDegreeOfParallelism(1)
+
+ val count = ds.partitionByHash(0).setParallelism(4).filter(
+ new RichFilterFunction[(Int, Long, String)] {
+ var first = true
+ override def filter(in: (Int, Long, String)): Boolean = {
+ // only output one value with count 1
+ if (first) {
+ first = false
+ true
+ } else {
+ false
+ }
+ }
+ })
+ .map( _ => Tuple1(1)).sum(0)
+
+ count.writeAsText(resultPath)
+ env.execute()
+
+ if (onCollection) "(1)\n" else "(4)\n"
+
+ case 7 =>
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ env.setDegreeOfParallelism(3)
+ val ds = CollectionDataSets.getDuplicatePojoDataSet(env)
+ val uniqLongs = ds
+ .partitionByHash("nestedPojo.longNumber")
+ .setParallelism(4)
+ .mapPartition( _.map(_.nestedPojo.longNumber).toSet )
+
+ uniqLongs.writeAsText(resultPath)
+ env.execute()
+ "10000\n" + "20000\n" + "30000\n"
+
+ case _ =>
+ throw new IllegalArgumentException("Invalid program id")
+ }
+ }
+}
+
+
+@RunWith(classOf[Parameterized])
+class PartitionITCase(config: Configuration) extends JavaProgramTestBase(config) {
+
+ private var curProgId: Int = config.getInteger("ProgramId", -1)
+ private var resultPath: String = null
+ private var expectedResult: String = null
+
+ protected override def preSubmit(): Unit = {
+ resultPath = getTempDirPath("result")
+ }
+
+ protected def testProgram(): Unit = {
+ expectedResult = PartitionProgs.runProgram(curProgId, resultPath, isCollectionExecution)
+ }
+
+ protected override def postSubmit(): Unit = {
+ compareResultsByLinesInMemory(expectedResult, resultPath)
+ }
+}
+
+object PartitionITCase {
+ @Parameters
+ def getConfigurations: java.util.Collection[Array[AnyRef]] = {
+ val configs = mutable.MutableList[Array[AnyRef]]()
+ for (i <- 1 to PartitionProgs.NUM_PROGRAMS) {
+ val config = new Configuration()
+ config.setInteger("ProgramId", i)
+ configs += Array(config)
+ }
+
+ configs.asJavaCollection
+ }
+}
+
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/a0ad9031/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/ReduceITCase.scala
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/ReduceITCase.scala b/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/ReduceITCase.scala
new file mode 100644
index 0000000..867eb81
--- /dev/null
+++ b/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/ReduceITCase.scala
@@ -0,0 +1,236 @@
+/*
+ * 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.operators
+
+import org.apache.flink.api.common.functions.RichReduceFunction
+import org.apache.flink.api.scala.util.CollectionDataSets
+import org.apache.flink.api.scala.util.CollectionDataSets.MutableTuple3
+import org.apache.flink.configuration.Configuration
+import org.apache.flink.test.util.JavaProgramTestBase
+import org.junit.runner.RunWith
+import org.junit.runners.Parameterized
+import org.junit.runners.Parameterized.Parameters
+
+import scala.collection.JavaConverters._
+import scala.collection.mutable
+
+import org.apache.flink.api.scala._
+
+
+object ReduceProgs {
+ var NUM_PROGRAMS: Int = 10
+
+ def runProgram(progId: Int, resultPath: String): String = {
+ progId match {
+ case 1 =>
+ /*
+ * Reduce on tuples with key field selector
+ */
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ val ds = CollectionDataSets.get3TupleDataSet(env)
+ val reduceDs = ds.groupBy(1)
+ .reduce { (in1, in2) => (in1._1 + in2._1, in1._2, "B-)") }
+ reduceDs.writeAsCsv(resultPath)
+ env.execute()
+ "1,1,Hi\n" + "5,2,B-)\n" + "15,3,B-)\n" + "34,4,B-)\n" + "65,5,B-)\n" + "111,6,B-)\n"
+
+ case 2 =>
+ /*
+ * Reduce on tuples with multiple key field selectors
+ */
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ val ds = CollectionDataSets.get5TupleDataSet(env)
+ val reduceDs = ds.groupBy(4, 0)
+ .reduce { (in1, in2) => (in1._1, in1._2 + in2._2, 0, "P-)", in1._5) }
+ reduceDs.writeAsCsv(resultPath)
+ env.execute()
+ "1,1,0,Hallo,1\n" + "2,3,2,Hallo Welt wie,1\n" + "2,2,1,Hallo Welt,2\n" + "3,9,0," +
+ "P-),2\n" + "3,6,5,BCD,3\n" + "4,17,0,P-),1\n" + "4,17,0,P-),2\n" + "5,11,10,GHI," +
+ "1\n" + "5,29,0,P-),2\n" + "5,25,0,P-),3\n"
+
+ case 3 =>
+ /*
+ * Reduce on tuples with key extractor
+ */
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ val ds = CollectionDataSets.get3TupleDataSet(env)
+ val reduceDs = ds.groupBy(_._2)
+ .reduce { (in1, in2) => (in1._1 + in2._1, in1._2, "B-)") }
+ reduceDs.writeAsCsv(resultPath)
+ env.execute()
+ "1,1,Hi\n" + "5,2,B-)\n" + "15,3,B-)\n" + "34,4,B-)\n" + "65,5,B-)\n" + "111,6,B-)\n"
+
+ case 4 =>
+ /*
+ * Reduce on custom type with key extractor
+ */
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ val ds = CollectionDataSets.getCustomTypeDataSet(env)
+ val reduceDs = ds.groupBy(_.myInt)
+ .reduce { (in1, in2) =>
+ in1.myLong += in2.myLong
+ in1.myString = "Hello!"
+ in1
+ }
+ reduceDs.writeAsText(resultPath)
+ env.execute()
+ "1,0,Hi\n" + "2,3,Hello!\n" + "3,12,Hello!\n" + "4,30,Hello!\n" + "5,60," +
+ "Hello!\n" + "6,105,Hello!\n"
+
+ case 5 =>
+ /*
+ * All-reduce for tuple
+ */
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ val ds = CollectionDataSets.get3TupleDataSet(env)
+ val reduceDs =
+ ds.reduce { (in1, in2) => (in1._1 + in2._1, in1._2 + in2._2, "Hello World") }
+ reduceDs.writeAsCsv(resultPath)
+ env.execute()
+ "231,91,Hello World\n"
+
+ case 6 =>
+ /*
+ * All-reduce for custom types
+ */
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ val ds = CollectionDataSets.getCustomTypeDataSet(env)
+ val reduceDs = ds
+ .reduce { (in1, in2) =>
+ in1.myInt += in2.myInt
+ in1.myLong += in2.myLong
+ in1.myString = "Hello!"
+ in1
+ }
+ reduceDs.writeAsText(resultPath)
+ env.execute()
+ "91,210,Hello!"
+
+ case 7 =>
+ /*
+ * Reduce with broadcast set
+ */
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ val intDs = CollectionDataSets.getIntDataSet(env)
+ val ds = CollectionDataSets.get3TupleDataSet(env)
+ val reduceDs = ds.groupBy(1).reduce(
+ new RichReduceFunction[(Int, Long, String)] {
+ private var f2Replace = ""
+
+ override def open(config: Configuration) {
+ val ints = this.getRuntimeContext.getBroadcastVariable[Int]("ints").asScala
+ f2Replace = ints.sum + ""
+ }
+
+ override def reduce(
+ in1: (Int, Long, String),
+ in2: (Int, Long, String)): (Int, Long, String) = {
+ (in1._1 + in2._1, in1._2, f2Replace)
+ }
+ }).withBroadcastSet(intDs, "ints")
+ reduceDs.writeAsCsv(resultPath)
+ env.execute()
+ "1,1,Hi\n" + "5,2,55\n" + "15,3,55\n" + "34,4,55\n" + "65,5,55\n" + "111,6,55\n"
+
+ case 8 =>
+ /*
+ * Reduce with UDF that returns the second input object (check mutable object handling)
+ */
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ val ds = CollectionDataSets.get3TupleDataSet(env).map (t => MutableTuple3(t._1, t._2, t._3))
+ val reduceDs = ds.groupBy(1).reduce(
+ new RichReduceFunction[MutableTuple3[Int, Long, String]] {
+ override def reduce(
+ in1: MutableTuple3[Int, Long, String],
+ in2: MutableTuple3[Int, Long, String]): MutableTuple3[Int, Long, String] = {
+ in2._1 = in1._1 + in2._1
+ in2._3 = "Hi again!"
+ in2
+ }
+ })
+ reduceDs.writeAsCsv(resultPath)
+ env.execute()
+ "1,1,Hi\n" + "5,2,Hi again!\n" + "15,3,Hi again!\n" + "34,4,Hi again!\n" + "65,5," +
+ "Hi again!\n" + "111,6,Hi again!\n"
+
+ case 9 =>
+ /*
+ * Reduce with a Tuple-returning KeySelector
+ */
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ val ds = CollectionDataSets.get5TupleDataSet(env)
+ val reduceDs = ds.groupBy(t => (t._1, t._5))
+ .reduce { (in1, in2) => (in1._1, in1._2 + in2._2, 0, "P-)", in1._5) }
+ reduceDs.writeAsCsv(resultPath)
+ env.execute()
+ "1,1,0,Hallo,1\n" + "2,3,2,Hallo Welt wie,1\n" + "2,2,1,Hallo Welt,2\n" + "3,9,0," +
+ "P-),2\n" + "3,6,5,BCD,3\n" + "4,17,0,P-),1\n" + "4,17,0,P-),2\n" + "5,11,10,GHI," +
+ "1\n" + "5,29,0,P-),2\n" + "5,25,0,P-),3\n"
+
+ case 10 =>
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ val ds = CollectionDataSets.get5TupleDataSet(env)
+ val reduceDs = ds.groupBy("_5", "_1")
+ .reduce { (in1, in2) => (in1._1, in1._2 + in2._2, 0, "P-)", in1._5) }
+ reduceDs.writeAsCsv(resultPath)
+ env.execute()
+ "1,1,0,Hallo,1\n" + "2,3,2,Hallo Welt wie,1\n" + "2,2,1,Hallo Welt,2\n" + "3,9,0," +
+ "P-),2\n" + "3,6,5,BCD,3\n" + "4,17,0,P-),1\n" + "4,17,0,P-),2\n" + "5,11,10,GHI," +
+ "1\n" + "5,29,0,P-),2\n" + "5,25,0,P-),3\n"
+
+ case id =>
+ throw new IllegalArgumentException(s"Invalid program id $id")
+ }
+ }
+}
+
+
+@RunWith(classOf[Parameterized])
+class ReduceITCase(config: Configuration) extends JavaProgramTestBase(config) {
+
+ private var curProgId: Int = config.getInteger("ProgramId", -1)
+ private var resultPath: String = null
+ private var expectedResult: String = null
+
+ protected override def preSubmit(): Unit = {
+ resultPath = getTempDirPath("result")
+ }
+
+ protected def testProgram(): Unit = {
+ expectedResult = ReduceProgs.runProgram(curProgId, resultPath)
+ }
+
+ protected override def postSubmit(): Unit = {
+ compareResultsByLinesInMemory(expectedResult, resultPath)
+ }
+}
+
+object ReduceITCase {
+ @Parameters
+ def getConfigurations: java.util.Collection[Array[AnyRef]] = {
+ val configs = mutable.MutableList[Array[AnyRef]]()
+ for (i <- 1 to ReduceProgs.NUM_PROGRAMS) {
+ val config = new Configuration()
+ config.setInteger("ProgramId", i)
+ configs += Array(config)
+ }
+
+ configs.asJavaCollection
+ }
+}
+
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/a0ad9031/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/SumMinMaxITCase.scala
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/SumMinMaxITCase.scala b/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/SumMinMaxITCase.scala
new file mode 100644
index 0000000..488520e
--- /dev/null
+++ b/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/SumMinMaxITCase.scala
@@ -0,0 +1,146 @@
+/*
+ * 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.operators
+
+import org.apache.flink.api.java.aggregation.Aggregations
+import org.apache.flink.api.scala.ExecutionEnvironment
+import org.apache.flink.api.scala.util.CollectionDataSets
+import org.apache.flink.configuration.Configuration
+import org.apache.flink.test.util.JavaProgramTestBase
+import org.junit.runner.RunWith
+import org.junit.runners.Parameterized
+import org.junit.runners.Parameterized.Parameters
+
+import scala.collection.JavaConverters._
+import scala.collection.mutable
+
+import org.apache.flink.api.scala._
+
+
+
+/**
+ * These tests are copied from [[AggregateITCase]] replacing calls to aggregate with calls to sum,
+ * min, and max
+ */
+object SumMinMaxProgs {
+ var NUM_PROGRAMS: Int = 3
+
+ def runProgram(progId: Int, resultPath: String): String = {
+ progId match {
+ case 1 =>
+ // Full aggregate
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ env.setDegreeOfParallelism(10)
+ // val ds = CollectionDataSets.get3TupleDataSet(env)
+ val ds = CollectionDataSets.get3TupleDataSet(env)
+
+ val aggregateDs = ds
+ .sum(0)
+ .andMax(1)
+ // Ensure aggregate operator correctly copies other fields
+ .filter(_._3 != null)
+ .map{ t => (t._1, t._2) }
+
+ aggregateDs.writeAsCsv(resultPath)
+
+ env.execute()
+
+ // return expected result
+ "231,6\n"
+
+ case 2 =>
+ // Grouped aggregate
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ val ds = CollectionDataSets.get3TupleDataSet(env)
+
+ val aggregateDs = ds
+ .groupBy(1)
+ .sum(0)
+ // Ensure aggregate operator correctly copies other fields
+ .filter(_._3 != null)
+ .map { t => (t._2, t._1) }
+
+ aggregateDs.writeAsCsv(resultPath)
+
+ env.execute()
+
+ // return expected result
+ "1,1\n" + "2,5\n" + "3,15\n" + "4,34\n" + "5,65\n" + "6,111\n"
+
+ case 3 =>
+ // Nested aggregate
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ val ds = CollectionDataSets.get3TupleDataSet(env)
+
+ val aggregateDs = ds
+ .groupBy(1)
+ .min(0)
+ .min(0)
+ // Ensure aggregate operator correctly copies other fields
+ .filter(_._3 != null)
+ .map { t => new Tuple1(t._1) }
+
+ aggregateDs.writeAsCsv(resultPath)
+
+ env.execute()
+
+ // return expected result
+ "1\n"
+
+
+ case _ =>
+ throw new IllegalArgumentException("Invalid program id")
+ }
+ }
+}
+
+
+@RunWith(classOf[Parameterized])
+class SumMinMaxITCase(config: Configuration) extends JavaProgramTestBase(config) {
+
+ private var curProgId: Int = config.getInteger("ProgramId", -1)
+ private var resultPath: String = null
+ private var expectedResult: String = null
+
+ protected override def preSubmit(): Unit = {
+ resultPath = getTempDirPath("result")
+ }
+
+ protected def testProgram(): Unit = {
+ expectedResult = SumMinMaxProgs.runProgram(curProgId, resultPath)
+ }
+
+ protected override def postSubmit(): Unit = {
+ compareResultsByLinesInMemory(expectedResult, resultPath)
+ }
+}
+
+object SumMinMaxITCase {
+ @Parameters
+ def getConfigurations: java.util.Collection[Array[AnyRef]] = {
+ val configs = mutable.MutableList[Array[AnyRef]]()
+ for (i <- 1 to SumMinMaxProgs.NUM_PROGRAMS) {
+ val config = new Configuration()
+ config.setInteger("ProgramId", i)
+ configs += Array(config)
+ }
+
+ configs.asJavaCollection
+ }
+}
+
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/a0ad9031/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/UnionITCase.scala
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/UnionITCase.scala b/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/UnionITCase.scala
new file mode 100644
index 0000000..bf29b58
--- /dev/null
+++ b/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/UnionITCase.scala
@@ -0,0 +1,126 @@
+/*
+ * 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.operators
+
+import org.apache.flink.api.scala.util.CollectionDataSets
+import org.apache.flink.api.scala.util.CollectionDataSets.MutableTuple3
+import org.apache.flink.configuration.Configuration
+import org.apache.flink.test.util.JavaProgramTestBase
+import org.junit.Assert
+import org.junit.runner.RunWith
+import org.junit.runners.Parameterized
+import org.junit.runners.Parameterized.Parameters
+
+import scala.collection.JavaConverters._
+import scala.collection.mutable
+
+import org.apache.flink.api.scala._
+
+
+object UnionProgs {
+ var NUM_PROGRAMS: Int = 3
+
+ private final val FULL_TUPLE_3_STRING: String = "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"
+
+ def runProgram(progId: Int, resultPath: String): String = {
+ progId match {
+ case 1 =>
+ /*
+ * Union of 2 Same Data Sets
+ */
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ val ds = CollectionDataSets.get3TupleDataSet(env)
+ val unionDs = ds.union(CollectionDataSets.get3TupleDataSet(env))
+ unionDs.writeAsCsv(resultPath)
+ env.execute()
+ FULL_TUPLE_3_STRING + FULL_TUPLE_3_STRING
+
+ case 2 =>
+ /*
+ * Union of 5 same Data Sets, with multiple unions
+ */
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ val ds = CollectionDataSets.get3TupleDataSet(env)
+ val unionDs = ds
+ .union(CollectionDataSets.get3TupleDataSet(env))
+ .union(CollectionDataSets.get3TupleDataSet(env))
+ .union(CollectionDataSets.get3TupleDataSet(env))
+ .union(CollectionDataSets.get3TupleDataSet(env))
+ unionDs.writeAsCsv(resultPath)
+ env.execute()
+ FULL_TUPLE_3_STRING + FULL_TUPLE_3_STRING + FULL_TUPLE_3_STRING +
+ FULL_TUPLE_3_STRING + FULL_TUPLE_3_STRING
+
+ case 3 =>
+ /*
+ * Test on union with empty dataset
+ */
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ // Don't know how to make an empty result in an other way than filtering it
+ val empty = CollectionDataSets.get3TupleDataSet(env).filter( t => false )
+ val unionDs = CollectionDataSets.get3TupleDataSet(env).union(empty)
+ unionDs.writeAsCsv(resultPath)
+ env.execute()
+ FULL_TUPLE_3_STRING
+
+ case _ =>
+ throw new IllegalArgumentException("Invalid program id")
+ }
+ }
+}
+
+
+@RunWith(classOf[Parameterized])
+class UnionITCase(config: Configuration) extends JavaProgramTestBase(config) {
+
+ private var curProgId: Int = config.getInteger("ProgramId", -1)
+ private var resultPath: String = null
+ private var expectedResult: String = null
+
+ protected override def preSubmit(): Unit = {
+ resultPath = getTempDirPath("result")
+ }
+
+ protected def testProgram(): Unit = {
+ expectedResult = UnionProgs.runProgram(curProgId, resultPath)
+ }
+
+ protected override def postSubmit(): Unit = {
+ compareResultsByLinesInMemory(expectedResult, resultPath)
+ }
+}
+
+object UnionITCase {
+ @Parameters
+ def getConfigurations: java.util.Collection[Array[AnyRef]] = {
+ val configs = mutable.MutableList[Array[AnyRef]]()
+ for (i <- 1 to UnionProgs.NUM_PROGRAMS) {
+ val config = new Configuration()
+ config.setInteger("ProgramId", i)
+ configs += Array(config)
+ }
+
+ configs.asJavaCollection
+ }
+}
+
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/a0ad9031/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/translation/AggregateTranslationTest.scala
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/translation/AggregateTranslationTest.scala b/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/translation/AggregateTranslationTest.scala
new file mode 100644
index 0000000..c4d7dc8
--- /dev/null
+++ b/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/translation/AggregateTranslationTest.scala
@@ -0,0 +1,61 @@
+/*
+ * 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.operators.translation
+
+import org.apache.flink.api.common.Plan
+import org.apache.flink.api.common.operators.GenericDataSourceBase
+import org.apache.flink.api.common.operators.base.GroupReduceOperatorBase
+
+import org.apache.flink.api.java.aggregation.Aggregations
+import org.apache.flink.api.scala._
+import org.junit.Assert.{assertEquals, assertTrue, fail}
+import org.junit.Test
+
+class AggregateTranslationTest {
+ @Test
+ def translateAggregate(): Unit = {
+ try {
+ val DOP = 8
+
+ val env = ExecutionEnvironment.createLocalEnvironment(DOP)
+
+ val initialData = env.fromElements((3.141592, "foobar", 77L))
+
+ initialData.groupBy(0).aggregate(Aggregations.MIN, 1).and(Aggregations.SUM, 2).print()
+
+ val p: Plan = env.createProgramPlan()
+ val sink = p.getDataSinks.iterator.next
+
+ val reducer= sink.getInput.asInstanceOf[GroupReduceOperatorBase[_, _, _]]
+
+ assertEquals(1, reducer.getKeyColumns(0).length)
+ assertEquals(0, reducer.getKeyColumns(0)(0))
+ assertEquals(-1, reducer.getDegreeOfParallelism)
+ assertTrue(reducer.isCombinable)
+ assertTrue(reducer.getInput.isInstanceOf[GenericDataSourceBase[_, _]])
+ }
+ catch {
+ case e: Exception => {
+ System.err.println(e.getMessage)
+ e.printStackTrace()
+ fail("Test caused an error: " + e.getMessage)
+ }
+ }
+ }
+}
+
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/a0ad9031/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/translation/DeltaIterationTranslationTest.scala
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/translation/DeltaIterationTranslationTest.scala b/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/translation/DeltaIterationTranslationTest.scala
new file mode 100644
index 0000000..4a60154
--- /dev/null
+++ b/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/translation/DeltaIterationTranslationTest.scala
@@ -0,0 +1,254 @@
+/*
+ * 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.operators.translation
+
+import org.apache.flink.api.common.functions.{RichCoGroupFunction, RichMapFunction,
+RichJoinFunction}
+import org.apache.flink.api.common.operators.GenericDataSinkBase
+import org.apache.flink.api.java.operators.translation.WrappingFunction
+import org.junit.Assert.assertArrayEquals
+import org.junit.Assert.assertEquals
+import org.junit.Assert.fail
+import org.apache.flink.api.common.{InvalidProgramException, Plan}
+import org.apache.flink.api.common.aggregators.LongSumAggregator
+import org.apache.flink.api.common.operators.base.DeltaIterationBase
+import org.apache.flink.api.common.operators.base.JoinOperatorBase
+import org.apache.flink.api.common.operators.base.MapOperatorBase
+import org.junit.Test
+
+import org.apache.flink.util.Collector
+
+import org.apache.flink.api.scala._
+
+class DeltaIterationTranslationTest {
+
+ @Test
+ def testCorrectTranslation(): Unit = {
+ try {
+ val JOB_NAME = "Test JobName"
+ val ITERATION_NAME = "Test Name"
+ val BEFORE_NEXT_WORKSET_MAP = "Some Mapper"
+ val AGGREGATOR_NAME = "AggregatorName"
+ val ITERATION_KEYS = Array(2)
+ val NUM_ITERATIONS = 13
+ val DEFAULT_DOP = 133
+ val ITERATION_DOP = 77
+
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ env.setDegreeOfParallelism(DEFAULT_DOP)
+
+ val initialSolutionSet = env.fromElements((3.44, 5L, "abc"))
+ val initialWorkSet = env.fromElements((1.23, "abc"))
+
+ val result = initialSolutionSet.iterateDelta(initialWorkSet, NUM_ITERATIONS, ITERATION_KEYS) {
+ (s, ws) =>
+ val wsSelfJoin = ws.map(new IdentityMapper[(Double, String)]())
+ .join(ws).where(1).equalTo(1) { (l, r) => l }
+
+ val joined = wsSelfJoin.join(s).where(1).equalTo(2).apply(new SolutionWorksetJoin)
+ (joined, joined.map(new NextWorksetMapper).name(BEFORE_NEXT_WORKSET_MAP))
+ }
+ result.name(ITERATION_NAME)
+ .setParallelism(ITERATION_DOP)
+ .registerAggregator(AGGREGATOR_NAME, new LongSumAggregator)
+
+ result.print()
+ result.writeAsText("/dev/null")
+
+ val p: Plan = env.createProgramPlan(JOB_NAME)
+ assertEquals(JOB_NAME, p.getJobName)
+ assertEquals(DEFAULT_DOP, p.getDefaultParallelism)
+ var sink1: GenericDataSinkBase[_] = null
+ var sink2: GenericDataSinkBase[_] = null
+ val sinks = p.getDataSinks.iterator
+ sink1 = sinks.next
+ sink2 = sinks.next
+
+ val iteration: DeltaIterationBase[_, _] =
+ sink1.getInput.asInstanceOf[DeltaIterationBase[_,_]]
+
+ assertEquals(iteration, sink2.getInput)
+ assertEquals(NUM_ITERATIONS, iteration.getMaximumNumberOfIterations)
+ assertArrayEquals(ITERATION_KEYS, iteration.getSolutionSetKeyFields)
+ assertEquals(ITERATION_DOP, iteration.getDegreeOfParallelism)
+ assertEquals(ITERATION_NAME, iteration.getName)
+
+ val nextWorksetMapper: MapOperatorBase[_, _, _] =
+ iteration.getNextWorkset.asInstanceOf[MapOperatorBase[_, _, _]]
+ val solutionSetJoin: JoinOperatorBase[_, _, _, _] =
+ iteration.getSolutionSetDelta.asInstanceOf[JoinOperatorBase[_, _, _, _]]
+ val worksetSelfJoin: JoinOperatorBase[_, _, _, _] =
+ solutionSetJoin.getFirstInput.asInstanceOf[JoinOperatorBase[_, _, _, _]]
+ val worksetMapper: MapOperatorBase[_, _, _] =
+ worksetSelfJoin.getFirstInput.asInstanceOf[MapOperatorBase[_, _, _]]
+
+ assertEquals(classOf[IdentityMapper[_]], worksetMapper.getUserCodeWrapper.getUserCodeClass)
+
+
+ assertEquals(
+ classOf[NextWorksetMapper],
+ nextWorksetMapper.getUserCodeWrapper.getUserCodeClass)
+
+
+ if (solutionSetJoin.getUserCodeWrapper.getUserCodeObject.isInstanceOf[WrappingFunction[_]]) {
+ val wf: WrappingFunction[_] = solutionSetJoin.getUserCodeWrapper.getUserCodeObject
+ .asInstanceOf[WrappingFunction[_]]
+ assertEquals(classOf[SolutionWorksetJoin],
+ wf.getWrappedFunction.getClass)
+ }
+ else {
+ assertEquals(classOf[SolutionWorksetJoin],
+ solutionSetJoin.getUserCodeWrapper.getUserCodeClass)
+ }
+
+ assertEquals(BEFORE_NEXT_WORKSET_MAP, nextWorksetMapper.getName)
+ assertEquals(AGGREGATOR_NAME, iteration.getAggregators.getAllRegisteredAggregators.iterator
+ .next.getName)
+ }
+ catch {
+ case e: Exception => {
+ System.err.println(e.getMessage)
+ e.printStackTrace()
+ fail(e.getMessage)
+ }
+ }
+ }
+
+ @Test
+ def testRejectWhenSolutionSetKeysDontMatchJoin(): Unit = {
+ try {
+ val env = ExecutionEnvironment.getExecutionEnvironment
+
+ val initialSolutionSet = env.fromElements((3.44, 5L, "abc"))
+ val initialWorkSet = env.fromElements((1.23, "abc"))
+
+ val iteration = initialSolutionSet.iterateDelta(initialWorkSet, 10, Array(0)) {
+ (s, ws) =>
+ try {
+ ws.join(s).where(1).equalTo(2)
+ fail("Accepted invalid program.")
+ } catch {
+ case e: InvalidProgramException => // all good
+ }
+ try {
+ s.join(ws).where(2).equalTo(1)
+ fail("Accepted invalid program.")
+ } catch {
+ case e: InvalidProgramException => // all good
+ }
+ (s, ws)
+ }
+ } catch {
+ case e: Exception => {
+ System.err.println(e.getMessage)
+ e.printStackTrace()
+ fail(e.getMessage)
+ }
+ }
+ }
+
+ @Test
+ def testRejectWhenSolutionSetKeysDontMatchCoGroup(): Unit = {
+ try {
+ val env = ExecutionEnvironment.getExecutionEnvironment
+
+ val initialSolutionSet = env.fromElements((3.44, 5L, "abc"))
+ val initialWorkSet = env.fromElements((1.23, "abc"))
+
+ val iteration = initialSolutionSet.iterateDelta(initialWorkSet, 10, Array(0)) {
+ (s, ws) =>
+ try {
+ ws.coGroup(s).where(1).equalTo(2)
+ fail("Accepted invalid program.")
+ } catch {
+ case e: InvalidProgramException => // all good
+ }
+ try {
+ s.coGroup(ws).where(2).equalTo(1)
+ fail("Accepted invalid program.")
+ } catch {
+ case e: InvalidProgramException => // all good
+ }
+ (s, ws)
+ }
+ } catch {
+ case e: Exception => {
+ System.err.println(e.getMessage)
+ e.printStackTrace()
+ fail(e.getMessage)
+ }
+ }
+ }
+//
+// @Test def testRejectWhenSolutionSetKeysDontMatchCoGroup {
+// try {
+// val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
+// @SuppressWarnings(Array("unchecked")) val initialSolutionSet: DataSet[Tuple3[Double, Long,
+// String]] = env.fromElements(new Tuple3[Double, Long, String](3.44, 5L, "abc"))
+// @SuppressWarnings(Array("unchecked")) val initialWorkSet: DataSet[Tuple2[Double,
+// String]] = env.fromElements(new Tuple2[Double, String](1.23, "abc"))
+// val iteration: DeltaIteration[Tuple3[Double, Long, String], Tuple2[Double,
+// String]] = initialSolutionSet.iterateDelta(initialWorkSet, 10, 1)
+// try {
+// iteration.getWorkset.coGroup(iteration.getSolutionSet).where(1).equalTo(2).`with`(
+// new DeltaIterationTranslationTest.SolutionWorksetCoGroup1)
+// fail("Accepted invalid program.")
+// }
+// catch {
+// case e: InvalidProgramException => {
+// }
+// }
+// try {
+// iteration.getSolutionSet.coGroup(iteration.getWorkset).where(2).equalTo(1).`with`(
+// new DeltaIterationTranslationTest.SolutionWorksetCoGroup2)
+// fail("Accepted invalid program.")
+// }
+// catch {
+// case e: InvalidProgramException => {
+// }
+// }
+// }
+// catch {
+// case e: Exception => {
+// System.err.println(e.getMessage)
+// e.printStackTrace
+// fail(e.getMessage)
+// }
+// }
+// }
+}
+
+class SolutionWorksetJoin
+ extends RichJoinFunction[(Double, String), (Double, Long, String), (Double, Long, String)] {
+ def join(first: (Double, String), second: (Double, Long, String)): (Double, Long, String) = {
+ null
+ }
+}
+
+class NextWorksetMapper extends RichMapFunction[(Double, Long, String), (Double, String)] {
+ def map(value: (Double, Long, String)): (Double, String) = {
+ null
+ }
+}
+
+class IdentityMapper[T] extends RichMapFunction[T, T] {
+ def map(value: T): T = {
+ value
+ }
+}
+
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/a0ad9031/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/translation/DistinctTranslationTest.scala
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/translation/DistinctTranslationTest.scala b/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/translation/DistinctTranslationTest.scala
new file mode 100644
index 0000000..931d34d
--- /dev/null
+++ b/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/translation/DistinctTranslationTest.scala
@@ -0,0 +1,51 @@
+/*
+ * 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.operators.translation
+
+import org.apache.flink.api.common.operators.base.GroupReduceOperatorBase
+import org.junit.Assert
+import org.junit.Test
+
+import org.apache.flink.api.scala._
+
+class DistinctTranslationTest {
+ @Test
+ def testCombinable(): Unit = {
+ try {
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ val input = env.fromElements("1", "2", "1", "3")
+
+ val op = input.distinct { x => x}
+ op.print()
+
+ val p = env.createProgramPlan()
+
+ val reduceOp =
+ p.getDataSinks.iterator.next.getInput.asInstanceOf[GroupReduceOperatorBase[_, _, _]]
+
+ Assert.assertTrue(reduceOp.isCombinable)
+ }
+ catch {
+ case e: Exception => {
+ e.printStackTrace()
+ Assert.fail(e.getMessage)
+ }
+ }
+ }
+}
+
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/a0ad9031/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/translation/ReduceTranslationTest.scala
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/translation/ReduceTranslationTest.scala b/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/translation/ReduceTranslationTest.scala
new file mode 100644
index 0000000..22e4914
--- /dev/null
+++ b/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/translation/ReduceTranslationTest.scala
@@ -0,0 +1,138 @@
+/*
+ * 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.operators.translation
+
+import org.apache.flink.api.common.operators.{GenericDataSourceBase, GenericDataSinkBase}
+import org.apache.flink.api.java.operators.translation.{KeyExtractingMapper,
+PlanUnwrappingReduceOperator}
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo
+import org.apache.flink.api.java.typeutils.TupleTypeInfo
+import org.junit.Assert._
+import org.apache.flink.api.common.operators.base.MapOperatorBase
+import org.apache.flink.api.common.operators.base.ReduceOperatorBase
+import org.junit.Test
+
+import org.apache.flink.api.scala._
+
+class ReduceTranslationTest {
+ @Test
+ def translateNonGroupedReduce(): Unit = {
+ try {
+ val DOP = 8
+ val env = ExecutionEnvironment.createLocalEnvironment(DOP)
+
+ val initialData = env.fromElements((3.141592, "foobar", 77L)).setParallelism(1)
+
+
+ initialData reduce { (v1, v2) => v1 } print()
+
+ val p = env.createProgramPlan(
+
+)
+ val sink: GenericDataSinkBase[_] = p.getDataSinks.iterator.next
+ val reducer: ReduceOperatorBase[_, _] = sink.getInput.asInstanceOf[ReduceOperatorBase[_, _]]
+
+ assertEquals(initialData.javaSet.getType, reducer.getOperatorInfo.getInputType)
+ assertEquals(initialData.javaSet.getType, reducer.getOperatorInfo.getOutputType)
+ assertTrue(reducer.getKeyColumns(0) == null || reducer.getKeyColumns(0).length == 0)
+ assertTrue(reducer.getDegreeOfParallelism == 1 || reducer.getDegreeOfParallelism == -1)
+ assertTrue(reducer.getInput.isInstanceOf[GenericDataSourceBase[_, _]])
+ }
+ catch {
+ case e: Exception => {
+ System.err.println(e.getMessage)
+ e.printStackTrace()
+ fail("Test caused an error: " + e.getMessage)
+ }
+ }
+ }
+
+ @Test
+ def translateGroupedReduceNoMapper(): Unit = {
+ try {
+ val DOP: Int = 8
+ val env = ExecutionEnvironment.createLocalEnvironment(DOP)
+
+ val initialData = env.fromElements((3.141592, "foobar", 77L)).setParallelism(1)
+
+ initialData.groupBy(2) reduce { (v1, v2) => v1 } print()
+
+ val p = env.createProgramPlan()
+
+ val sink: GenericDataSinkBase[_] = p.getDataSinks.iterator.next
+ val reducer: ReduceOperatorBase[_, _] = sink.getInput.asInstanceOf[ReduceOperatorBase[_, _]]
+ assertEquals(initialData.javaSet.getType, reducer.getOperatorInfo.getInputType)
+ assertEquals(initialData.javaSet.getType, reducer.getOperatorInfo.getOutputType)
+ assertTrue(reducer.getDegreeOfParallelism == DOP || reducer.getDegreeOfParallelism == -1)
+ assertArrayEquals(Array[Int](2), reducer.getKeyColumns(0))
+ assertTrue(reducer.getInput.isInstanceOf[GenericDataSourceBase[_, _]])
+ }
+ catch {
+ case e: Exception => {
+ System.err.println(e.getMessage)
+ e.printStackTrace()
+ fail("Test caused an error: " + e.getMessage)
+ }
+ }
+ }
+
+ @Test
+ def translateGroupedReduceWithKeyExtractor(): Unit = {
+ try {
+ val DOP: Int = 8
+ val env = ExecutionEnvironment.createLocalEnvironment(DOP)
+
+ val initialData = env.fromElements((3.141592, "foobar", 77L)).setParallelism(1)
+
+ initialData.groupBy { _._2 }. reduce { (v1, v2) => v1 } setParallelism(4) print()
+
+ val p = env.createProgramPlan()
+ val sink: GenericDataSinkBase[_] = p.getDataSinks.iterator.next
+ val keyProjector: MapOperatorBase[_, _, _] = sink.getInput.asInstanceOf[MapOperatorBase[_,
+ _, _]]
+ val reducer: PlanUnwrappingReduceOperator[_, _] = keyProjector.getInput
+ .asInstanceOf[PlanUnwrappingReduceOperator[_, _]]
+ val keyExtractor: MapOperatorBase[_, _, _] = reducer.getInput
+ .asInstanceOf[MapOperatorBase[_, _, _]]
+ assertEquals(1, keyExtractor.getDegreeOfParallelism)
+ assertEquals(4, reducer.getDegreeOfParallelism)
+ assertEquals(4, keyProjector.getDegreeOfParallelism)
+ val keyValueInfo = new TupleTypeInfo(
+ BasicTypeInfo.STRING_TYPE_INFO,
+ createTypeInformation[(Double, String, Long)])
+ assertEquals(initialData.javaSet.getType, keyExtractor.getOperatorInfo.getInputType)
+ assertEquals(keyValueInfo, keyExtractor.getOperatorInfo.getOutputType)
+ assertEquals(keyValueInfo, reducer.getOperatorInfo.getInputType)
+ assertEquals(keyValueInfo, reducer.getOperatorInfo.getOutputType)
+ assertEquals(keyValueInfo, keyProjector.getOperatorInfo.getInputType)
+ assertEquals(initialData.javaSet.getType, keyProjector.getOperatorInfo.getOutputType)
+ assertEquals(
+ classOf[KeyExtractingMapper[_, _]],
+ keyExtractor.getUserCodeWrapper.getUserCodeClass)
+ assertTrue(keyExtractor.getInput.isInstanceOf[GenericDataSourceBase[_, _]])
+ }
+ catch {
+ case e: Exception => {
+ System.err.println(e.getMessage)
+ e.printStackTrace()
+ fail("Test caused an error: " + e.getMessage)
+ }
+ }
+ }
+}
+
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/a0ad9031/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/GenericPairComparatorTest.scala
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/GenericPairComparatorTest.scala b/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/GenericPairComparatorTest.scala
new file mode 100644
index 0000000..9fb0980
--- /dev/null
+++ b/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/GenericPairComparatorTest.scala
@@ -0,0 +1,76 @@
+/*
+ * 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.runtime
+
+import org.apache.flink.api.common.typeutils.{GenericPairComparator, TypeComparator, TypeSerializer}
+import org.apache.flink.api.common.typeutils.base.{DoubleComparator, DoubleSerializer, IntComparator, IntSerializer}
+
+import org.apache.flink.api.java.typeutils.runtime.TupleComparator
+import org.apache.flink.api.scala.runtime.tuple.base.PairComparatorTestBase
+import org.apache.flink.api.scala.typeutils.CaseClassComparator
+
+class GenericPairComparatorTest
+ extends PairComparatorTestBase[(Int, String, Double), (Int, Float, Long, Double)] {
+
+ override protected def createComparator(ascending: Boolean):
+ GenericPairComparator[(Int, String, Double), (Int, Float, Long, Double)] = {
+ val fields1 = Array[Int](0, 2)
+ val fields2 = Array[Int](0, 3)
+
+ val comps1 =
+ Array[TypeComparator[_]](new IntComparator(ascending), new DoubleComparator(ascending))
+ val comps2 =
+ Array[TypeComparator[_]](new IntComparator(ascending), new DoubleComparator(ascending))
+
+ val sers1 =
+ Array[TypeSerializer[_]](IntSerializer.INSTANCE, DoubleSerializer.INSTANCE)
+ val sers2 =
+ Array[TypeSerializer[_]](IntSerializer.INSTANCE, DoubleSerializer.INSTANCE)
+
+ val comp1 = new CaseClassComparator[(Int, String, Double)](fields1, comps1, sers1)
+ val comp2 = new CaseClassComparator[(Int, Float, Long, Double)](fields2, comps2, sers2)
+
+ new GenericPairComparator[(Int, String, Double), (Int, Float, Long, Double)](comp1, comp2)
+ }
+
+ protected def getSortedTestData:
+ (Array[(Int, String, Double)], Array[(Int, Float, Long, Double)]) = {
+ (dataISD, dataIDL)
+ }
+
+ private val dataISD = Array(
+ (4, "hello", 20.0),
+ (4, "world", 23.2),
+ (5, "hello", 18.0),
+ (5, "world", 19.2),
+ (6, "hello", 16.0),
+ (6, "world", 17.2),
+ (7, "hello", 14.0),
+ (7,"world", 15.2))
+
+ private val dataIDL = Array(
+ (4, 0.11f, 14L, 20.0),
+ (4, 0.221f, 15L, 23.2),
+ (5, 0.33f, 15L, 18.0),
+ (5, 0.44f, 20L, 19.2),
+ (6, 0.55f, 20L, 16.0),
+ (6, 0.66f, 29L, 17.2),
+ (7, 0.77f, 29L, 14.0),
+ (7, 0.88f, 34L, 15.2))
+}
+
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/a0ad9031/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/TupleComparatorILD2Test.scala
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/TupleComparatorILD2Test.scala b/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/TupleComparatorILD2Test.scala
new file mode 100644
index 0000000..adc46ac
--- /dev/null
+++ b/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/TupleComparatorILD2Test.scala
@@ -0,0 +1,56 @@
+/*
+ * 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.runtime
+
+import org.apache.flink.api.common.typeutils.TypeComparator
+import org.apache.flink.api.common.typeutils.TypeSerializer
+import org.apache.flink.api.java.typeutils.TupleTypeInfoBase
+import org.apache.flink.api.scala.runtime.tuple.base.TupleComparatorTestBase
+
+import org.apache.flink.api.scala._
+
+
+class TupleComparatorILD2Test extends TupleComparatorTestBase[(Int, Long, Double)] {
+
+ protected def createComparator(ascending: Boolean): TypeComparator[(Int, Long, Double)] = {
+ val ti = createTypeInformation[(Int, Long, Double)]
+ ti.asInstanceOf[TupleTypeInfoBase[(Int, Long, Double)]]
+ .createComparator(Array(0, 1), Array(ascending, ascending), 0)
+ }
+
+ protected def createSerializer: TypeSerializer[(Int, Long, Double)] = {
+ val ti = createTypeInformation[(Int, Long, Double)]
+ ti.createSerializer()
+ }
+
+ protected def getSortedTestData: Array[(Int, Long, Double)] = {
+ dataISD
+ }
+
+ private val dataISD = Array(
+ (4, 14L, 20.0),
+ (4, 15L, 23.2),
+ (5, 15L, 20.0),
+ (5, 20L, 20.0),
+ (6, 20L, 23.2),
+ (6, 29L, 20.0),
+ (7, 29L, 20.0),
+ (7, 34L, 23.2)
+ )
+}
+
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/a0ad9031/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/TupleComparatorILD3Test.scala
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/TupleComparatorILD3Test.scala b/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/TupleComparatorILD3Test.scala
new file mode 100644
index 0000000..377f985
--- /dev/null
+++ b/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/TupleComparatorILD3Test.scala
@@ -0,0 +1,54 @@
+/*
+ * 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.runtime
+
+import org.apache.flink.api.common.typeutils.{TypeComparator, TypeSerializer}
+import org.apache.flink.api.java.typeutils.TupleTypeInfoBase
+import org.apache.flink.api.scala._
+import org.apache.flink.api.scala.runtime.tuple.base.TupleComparatorTestBase
+
+
+class TupleComparatorILD3Test extends TupleComparatorTestBase[(Int, Long, Double)] {
+
+ protected def createComparator(ascending: Boolean): TypeComparator[(Int, Long, Double)] = {
+ val ti = createTypeInformation[(Int, Long, Double)]
+ ti.asInstanceOf[TupleTypeInfoBase[(Int, Long, Double)]]
+ .createComparator(Array(0, 1, 2), Array(ascending, ascending, ascending), 0)
+ }
+
+ protected def createSerializer: TypeSerializer[(Int, Long, Double)] = {
+ val ti = createTypeInformation[(Int, Long, Double)]
+ ti.createSerializer()
+ }
+
+ protected def getSortedTestData: Array[(Int, Long, Double)] = {
+ dataISD
+ }
+
+ private val dataISD = Array(
+ (4, 4L, 20.0),
+ (4, 4L, 23.2),
+ (4, 9L, 20.0),
+ (5, 4L, 20.0),
+ (5, 4L, 23.2),
+ (5, 9L, 20.0),
+ (6, 4L, 20.0),
+ (6, 4L, 23.2)
+ )
+}
+
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/a0ad9031/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/TupleComparatorILDC3Test.scala
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/TupleComparatorILDC3Test.scala b/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/TupleComparatorILDC3Test.scala
new file mode 100644
index 0000000..1578951
--- /dev/null
+++ b/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/TupleComparatorILDC3Test.scala
@@ -0,0 +1,54 @@
+/*
+ * 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.runtime
+
+import org.apache.flink.api.common.typeutils.{TypeSerializer, TypeComparator}
+import org.apache.flink.api.java.typeutils.TupleTypeInfoBase
+import org.apache.flink.api.scala._
+import org.apache.flink.api.scala.runtime.tuple.base.TupleComparatorTestBase
+
+
+class TupleComparatorILDC3Test extends TupleComparatorTestBase[(Int, Long, Double)] {
+
+ protected def createComparator(ascending: Boolean): TypeComparator[(Int, Long, Double)] = {
+ val ti = createTypeInformation[(Int, Long, Double)]
+ ti.asInstanceOf[TupleTypeInfoBase[(Int, Long, Double)]]
+ .createComparator(Array(2, 0, 1), Array(ascending, ascending, ascending), 0)
+ }
+
+ protected def createSerializer: TypeSerializer[(Int, Long, Double)] = {
+ val ti = createTypeInformation[(Int, Long, Double)]
+ ti.createSerializer()
+ }
+
+ protected def getSortedTestData: Array[(Int, Long, Double)] = {
+ dataISD
+ }
+
+ private val dataISD = Array(
+ (4, 4L, 20.0),
+ (5, 1L, 20.0),
+ (5, 2L, 20.0),
+ (5, 10L, 23.0),
+ (5, 19L, 24.0),
+ (5, 20L, 24.0),
+ (5, 24L, 25.0),
+ (5, 25L, 25.0)
+ )
+}
+
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/a0ad9031/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/TupleComparatorILDX1Test.scala
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/TupleComparatorILDX1Test.scala b/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/TupleComparatorILDX1Test.scala
new file mode 100644
index 0000000..51e08cc
--- /dev/null
+++ b/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/TupleComparatorILDX1Test.scala
@@ -0,0 +1,54 @@
+/*
+ * 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.runtime
+
+import org.apache.flink.api.common.typeutils.{TypeSerializer, TypeComparator}
+import org.apache.flink.api.java.typeutils.TupleTypeInfoBase
+import org.apache.flink.api.scala.runtime.tuple.base.TupleComparatorTestBase
+
+import org.apache.flink.api.scala._
+
+class TupleComparatorILDX1Test extends TupleComparatorTestBase[(Int, Long, Double)] {
+
+ protected def createComparator(ascending: Boolean): TypeComparator[(Int, Long, Double)] = {
+ val ti = createTypeInformation[(Int, Long, Double)]
+ ti.asInstanceOf[TupleTypeInfoBase[(Int, Long, Double)]]
+ .createComparator(Array(1), Array(ascending), 0)
+ }
+
+ protected def createSerializer: TypeSerializer[(Int, Long, Double)] = {
+ val ti = createTypeInformation[(Int, Long, Double)]
+ ti.createSerializer()
+ }
+
+ protected def getSortedTestData: Array[(Int, Long, Double)] = {
+ dataISD
+ }
+
+ private val dataISD = Array(
+ (4, 4L, 20.0),
+ (4, 5L, 23.2),
+ (4, 9L, 20.0),
+ (4, 10L, 24.0),
+ (4, 19L, 23.2),
+ (4, 20L, 24.0),
+ (4, 24L, 20.0),
+ (4, 25L, 23.2)
+ )
+}
+
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/a0ad9031/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/TupleComparatorILDXC2Test.scala
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/TupleComparatorILDXC2Test.scala b/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/TupleComparatorILDXC2Test.scala
new file mode 100644
index 0000000..3dbaabf
--- /dev/null
+++ b/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/TupleComparatorILDXC2Test.scala
@@ -0,0 +1,54 @@
+/*
+ * 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.runtime
+
+import org.apache.flink.api.common.typeutils.{TypeSerializer, TypeComparator}
+import org.apache.flink.api.java.typeutils.TupleTypeInfoBase
+import org.apache.flink.api.scala.runtime.tuple.base.TupleComparatorTestBase
+
+import org.apache.flink.api.scala._
+
+class TupleComparatorILDXC2Test extends TupleComparatorTestBase[(Int, Long, Double)] {
+
+ protected def createComparator(ascending: Boolean): TypeComparator[(Int, Long, Double)] = {
+ val ti = createTypeInformation[(Int, Long, Double)]
+ ti.asInstanceOf[TupleTypeInfoBase[(Int, Long, Double)]]
+ .createComparator(Array(2, 1), Array(ascending, ascending), 0)
+ }
+
+ protected def createSerializer: TypeSerializer[(Int, Long, Double)] = {
+ val ti = createTypeInformation[(Int, Long, Double)]
+ ti.createSerializer()
+ }
+
+ protected def getSortedTestData: Array[(Int, Long, Double)] = {
+ dataISD
+ }
+
+ private val dataISD = Array(
+ (4, 4L, 20.0),
+ (4, 5L, 20.0),
+ (4, 3L, 23.0),
+ (4, 19L, 23.0),
+ (4, 17L, 24.0),
+ (4, 18L, 24.0),
+ (4, 24L, 25.0),
+ (4, 25L, 25.0)
+ )
+}
+
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/a0ad9031/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/TupleComparatorISD1Test.scala
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/TupleComparatorISD1Test.scala b/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/TupleComparatorISD1Test.scala
new file mode 100644
index 0000000..252ae79
--- /dev/null
+++ b/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/TupleComparatorISD1Test.scala
@@ -0,0 +1,53 @@
+/*
+ * 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.runtime
+
+import org.apache.flink.api.common.typeutils.{TypeSerializer, TypeComparator}
+import org.apache.flink.api.java.typeutils.TupleTypeInfoBase
+import org.apache.flink.api.scala._
+import org.apache.flink.api.scala.runtime.tuple.base.TupleComparatorTestBase
+
+class TupleComparatorISD1Test extends TupleComparatorTestBase[(Int, String, Double)] {
+
+ protected def createComparator(ascending: Boolean): TypeComparator[(Int, String, Double)] = {
+ val ti = createTypeInformation[(Int, String, Double)]
+ ti.asInstanceOf[TupleTypeInfoBase[(Int, String, Double)]]
+ .createComparator(Array(0), Array(ascending),0)
+ }
+
+ protected def createSerializer: TypeSerializer[(Int, String, Double)] = {
+ val ti = createTypeInformation[(Int, String, Double)]
+ ti.createSerializer()
+ }
+
+ protected def getSortedTestData: Array[(Int, String, Double)] = {
+ dataISD
+ }
+
+ private val dataISD = Array(
+ (4, "hello", 20.0),
+ (5, "hello", 23.2),
+ (6, "world", 20.0),
+ (7, "hello", 20.0),
+ (8, "hello", 23.2),
+ (9, "world", 20.0),
+ (10, "hello", 20.0),
+ (11, "hello", 23.2)
+ )
+}
+