flink-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From aljos...@apache.org
Subject git commit: First cleanup attempt, mostly on Scala code, to follow guidelines from http://docs.scala-lang.org/style: 1. Remove return statement from Scala code where it is not necessary (end of method definition) 2. Remove extra ; from Scala and Java cod
Date Tue, 08 Jul 2014 10:09:22 GMT
Repository: incubator-flink
Updated Branches:
  refs/heads/master 59287062f -> 3f511953c


First cleanup attempt, mostly on Scala code, to follow guidelines from http://docs.scala-lang.org/style:
1. Remove return statement from Scala code where it is not necessary (end of method definition)
2. Remove extra ; from Scala and Java code
3. First drop to abide to 100 chars per line for Scala code.
   Will send another PR for other files as I have encountered.
4. Remove parentheses for empty argument methods that do not have side effect
   (see http://www.artima.com/pins1ed/composition-and-inheritance.html#i1343251059-1)
5. Remove unused import statements in Scala code as I have encountered them.

This is first drop to refactor and clean up Scala code to see comment/ response from community.
More PRs for follow up code will come.

Close #64


Project: http://git-wip-us.apache.org/repos/asf/incubator-flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-flink/commit/3f511953
Tree: http://git-wip-us.apache.org/repos/asf/incubator-flink/tree/3f511953
Diff: http://git-wip-us.apache.org/repos/asf/incubator-flink/diff/3f511953

Branch: refs/heads/master
Commit: 3f511953cde583a9735f88a600e50ed8fae029a6
Parents: 5928706
Author: Henry Saputra <hsaputra@apache.org>
Authored: Tue Jul 8 00:38:46 2014 -0700
Committer: Aljoscha Krettek <aljoscha.krettek@gmail.com>
Committed: Tue Jul 8 12:11:25 2014 +0200

----------------------------------------------------------------------
 .../examples/scala/wordcount/WordCount.scala    |   2 +-
 .../src/main/scala/Job.scala                    |   4 +-
 .../runtime/hash/HashMatchIteratorITCase.java   |   4 +-
 .../test/util/types/IntPairSerializer.java      |   2 +-
 .../stratosphere/api/scala/AnnotationUtil.scala |  14 +-
 .../stratosphere/api/scala/CompilerHints.scala  | 145 ++++++++++++-----
 .../eu/stratosphere/api/scala/DataSet.scala     |  26 ++-
 .../eu/stratosphere/api/scala/DataSource.scala  |   2 +-
 .../stratosphere/api/scala/ScalaOperator.scala  |  53 +++---
 .../api/scala/operators/MapOperator.scala       | 162 +++++++++++--------
 .../api/scala/operators/UnionOperator.scala     |  10 +-
 .../test/javaApiOperators/FilterITCase.java     |   2 +-
 12 files changed, 267 insertions(+), 159 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/3f511953/stratosphere-examples/stratosphere-scala-examples/src/main/scala/eu/stratosphere/examples/scala/wordcount/WordCount.scala
----------------------------------------------------------------------
diff --git a/stratosphere-examples/stratosphere-scala-examples/src/main/scala/eu/stratosphere/examples/scala/wordcount/WordCount.scala
b/stratosphere-examples/stratosphere-scala-examples/src/main/scala/eu/stratosphere/examples/scala/wordcount/WordCount.scala
index 791f19e..5cbcba6 100644
--- a/stratosphere-examples/stratosphere-scala-examples/src/main/scala/eu/stratosphere/examples/scala/wordcount/WordCount.scala
+++ b/stratosphere-examples/stratosphere-scala-examples/src/main/scala/eu/stratosphere/examples/scala/wordcount/WordCount.scala
@@ -33,7 +33,7 @@ class WordCount extends Program with ProgramDescription with Serializable
{
     val words = input flatMap { _.toLowerCase().split("""\W+""") filter { _ != "" } map {
(_, 1) } }
     val counts = words groupBy { case (word, _) => word } reduce { (w1, w2) => (w1._1,
w1._2 + w2._2) }
 
-    val output = counts.write(wordsOutput, CsvOutputFormat("\n", " "));
+    val output = counts.write(wordsOutput, CsvOutputFormat("\n", " "))
   
     val plan = new ScalaPlan(Seq(output), "Word Count")
     plan.setDefaultParallelism(numSubTasks)

http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/3f511953/stratosphere-quickstart/quickstart-scala/src/main/resources/archetype-resources/src/main/scala/Job.scala
----------------------------------------------------------------------
diff --git a/stratosphere-quickstart/quickstart-scala/src/main/resources/archetype-resources/src/main/scala/Job.scala
b/stratosphere-quickstart/quickstart-scala/src/main/resources/archetype-resources/src/main/scala/Job.scala
index 89a429f..dae5349 100644
--- a/stratosphere-quickstart/quickstart-scala/src/main/resources/archetype-resources/src/main/scala/Job.scala
+++ b/stratosphere-quickstart/quickstart-scala/src/main/resources/archetype-resources/src/main/scala/Job.scala
@@ -41,8 +41,8 @@ object RunJobRemote {
     // You will also need to change the name of the jar if you change the
     // project name and/or version. Before running this you also need
     // to run "mvn package" to create the jar.
-    val ex = new RemoteExecutor("localhost", 6123, "target/stratosphere-project-0.1-SNAPSHOT.jar");
-    ex.executePlan(plan);
+    val ex = new RemoteExecutor("localhost", 6123, "target/stratosphere-project-0.1-SNAPSHOT.jar")
+    ex.executePlan(plan)
   }
 }
 

http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/3f511953/stratosphere-runtime/src/test/java/eu/stratosphere/pact/runtime/hash/HashMatchIteratorITCase.java
----------------------------------------------------------------------
diff --git a/stratosphere-runtime/src/test/java/eu/stratosphere/pact/runtime/hash/HashMatchIteratorITCase.java
b/stratosphere-runtime/src/test/java/eu/stratosphere/pact/runtime/hash/HashMatchIteratorITCase.java
index e59f4a6..8ca7ebf 100644
--- a/stratosphere-runtime/src/test/java/eu/stratosphere/pact/runtime/hash/HashMatchIteratorITCase.java
+++ b/stratosphere-runtime/src/test/java/eu/stratosphere/pact/runtime/hash/HashMatchIteratorITCase.java
@@ -426,7 +426,7 @@ public class HashMatchIteratorITCase {
 			
 			while (iterator.callWithNextKey(matcher, collector));
 			
-			iterator.close();;
+			iterator.close();
 	
 			// assert that each expected match was seen
 			for (Entry<TestData.Key, Collection<RecordIntPairMatch>> entry : expectedMatchesMap.entrySet())
{
@@ -473,7 +473,7 @@ public class HashMatchIteratorITCase {
 			
 			while (iterator.callWithNextKey(matcher, collector));
 			
-			iterator.close();;
+			iterator.close();
 	
 			// assert that each expected match was seen
 			for (Entry<TestData.Key, Collection<RecordIntPairMatch>> entry : expectedMatchesMap.entrySet())
{

http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/3f511953/stratosphere-runtime/src/test/java/eu/stratosphere/pact/runtime/test/util/types/IntPairSerializer.java
----------------------------------------------------------------------
diff --git a/stratosphere-runtime/src/test/java/eu/stratosphere/pact/runtime/test/util/types/IntPairSerializer.java
b/stratosphere-runtime/src/test/java/eu/stratosphere/pact/runtime/test/util/types/IntPairSerializer.java
index b4a3137..af11748 100644
--- a/stratosphere-runtime/src/test/java/eu/stratosphere/pact/runtime/test/util/types/IntPairSerializer.java
+++ b/stratosphere-runtime/src/test/java/eu/stratosphere/pact/runtime/test/util/types/IntPairSerializer.java
@@ -69,7 +69,7 @@ public class IntPairSerializer extends TypeSerializer<IntPair> {
 
 	@Override
 	public void copy(DataInputView source, DataOutputView target) throws IOException {
-		target.write(source, 8);;
+		target.write(source, 8);
 	}
 
 	public static final class IntPairSerializerFactory implements TypeSerializerFactory<IntPair>
{

http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/3f511953/stratosphere-scala/src/main/scala/eu/stratosphere/api/scala/AnnotationUtil.scala
----------------------------------------------------------------------
diff --git a/stratosphere-scala/src/main/scala/eu/stratosphere/api/scala/AnnotationUtil.scala
b/stratosphere-scala/src/main/scala/eu/stratosphere/api/scala/AnnotationUtil.scala
index 53fdcda..abcb659 100644
--- a/stratosphere-scala/src/main/scala/eu/stratosphere/api/scala/AnnotationUtil.scala
+++ b/stratosphere-scala/src/main/scala/eu/stratosphere/api/scala/AnnotationUtil.scala
@@ -121,7 +121,7 @@ object AnnotationUtil {
 
         val fieldSet = new FieldSet(notConstantSet1Annotation.value(): _*)
 
-        for (i <- 0 until scalaOp.getUDF().getOutputLength) {
+        for (i <- 0 until scalaOp.getUDF.getOutputLength) {
           if (!fieldSet.contains(i)) {
             properties.addForwardedField1(i, i)
           }
@@ -141,7 +141,7 @@ object AnnotationUtil {
 
         val fieldSet = new FieldSet(notConstantSet2Annotation.value(): _*)
 
-        for (i <- 0 until scalaOp.getUDF().getOutputLength) {
+        for (i <- 0 until scalaOp.getUDF.getOutputLength) {
           if (!fieldSet.contains(i)) {
             properties.addForwardedField2(i, i)
           }
@@ -166,10 +166,10 @@ object AnnotationUtil {
       }
 
       // get constantSet annotation from stub
-      val constantSet: FunctionAnnotation.ConstantFields = scalaOp.getUserCodeAnnotation(classOf[FunctionAnnotation
-      .ConstantFields])
-      val notConstantSet: FunctionAnnotation.ConstantFieldsExcept = scalaOp.getUserCodeAnnotation(
-        classOf[FunctionAnnotation.ConstantFieldsExcept])
+      val constantSet: FunctionAnnotation.ConstantFields =
+        scalaOp.getUserCodeAnnotation(classOf[FunctionAnnotation.ConstantFields])
+      val notConstantSet: FunctionAnnotation.ConstantFieldsExcept =
+        scalaOp.getUserCodeAnnotation(classOf[FunctionAnnotation.ConstantFieldsExcept])
 
       if (notConstantSet != null && constantSet != null) {
         throw new RuntimeException("Either ConstantFields or ConstantFieldsExcept can be
specified, not both.")
@@ -186,7 +186,7 @@ object AnnotationUtil {
           }
         }
 
-        for (i <- 0 until scalaOp.getUDF().getOutputLength) {
+        for (i <- 0 until scalaOp.getUDF.getOutputLength) {
           if (!nonConstant.contains(i)) {
             properties.addForwardedField(i, i)
           }

http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/3f511953/stratosphere-scala/src/main/scala/eu/stratosphere/api/scala/CompilerHints.scala
----------------------------------------------------------------------
diff --git a/stratosphere-scala/src/main/scala/eu/stratosphere/api/scala/CompilerHints.scala
b/stratosphere-scala/src/main/scala/eu/stratosphere/api/scala/CompilerHints.scala
index 68afb3c..fd23e03 100644
--- a/stratosphere-scala/src/main/scala/eu/stratosphere/api/scala/CompilerHints.scala
+++ b/stratosphere-scala/src/main/scala/eu/stratosphere/api/scala/CompilerHints.scala
@@ -15,24 +15,30 @@ package eu.stratosphere.api.scala
 
 import language.experimental.macros
 import scala.util.DynamicVariable
+import scala.reflect.macros.Context
 import eu.stratosphere.api.scala.analysis._
-import eu.stratosphere.api.scala.operators.Annotations
 import eu.stratosphere.api.common.operators.util.{ FieldSet => PactFieldSet }
 import eu.stratosphere.api.common.operators.Operator
 import eu.stratosphere.api.scala.codegen.MacroContextHolder
-import scala.reflect.macros.Context
 import eu.stratosphere.types.Record
 
-case class KeyCardinality(key: FieldSelector, isUnique: Boolean, distinctCount: Option[Long],
avgNumRecords: Option[Float]) {
+case class KeyCardinality(
+    key: FieldSelector,
+    isUnique: Boolean,
+    distinctCount: Option[Long],
+    avgNumRecords: Option[Float]) {
 
-  @transient private var pactFieldSets = collection.mutable.Map[Operator[Record] with ScalaOperator[_,
_], PactFieldSet]()
+  @transient private var pactFieldSets =
+    collection.mutable.Map[Operator[Record] with ScalaOperator[_, _], PactFieldSet]()
 
   def getPactFieldSet(contract: Operator[Record] with ScalaOperator[_, _]): PactFieldSet
= {
 
-    if (pactFieldSets == null)
-      pactFieldSets = collection.mutable.Map[Operator[Record] with ScalaOperator[_, _], PactFieldSet]()
+    if (pactFieldSets == null) {
+      pactFieldSets =
+        collection.mutable.Map[Operator[Record] with ScalaOperator[_, _], PactFieldSet]()
+    }
 
-    val keyCopy = key.copy
+    val keyCopy = key.copy()
     contract.getUDF.attachOutputsToInputs(keyCopy.inputFields)
     val keySet = keyCopy.selectedFields.toIndexSet.toArray
 
@@ -57,27 +63,40 @@ trait OutputHintable[Out] { this: DataSet[Out] =>
     
   def outputSize = contract.getCompilerHints().getOutputSize()
   def outputSize_=(value: Long) = contract.getCompilerHints().setOutputSize(value)
-  def outputSize(value: Long): this.type = { contract.getCompilerHints().setOutputSize(value);
this }
+  def outputSize(value: Long): this.type = {
+    contract.getCompilerHints().setOutputSize(value)
+    this
+  }
   
   def outputCardinality = contract.getCompilerHints().getOutputCardinality()
   def outputCardinality_=(value: Long) = contract.getCompilerHints().setOutputCardinality(value)
-  def outputCardinality(value: Long): this.type = { contract.getCompilerHints().setOutputCardinality(value);
this }
+  def outputCardinality(value: Long): this.type = {
+    contract.getCompilerHints().setOutputCardinality(value)
+    this
+  }
   
   def avgBytesPerRecord = contract.getCompilerHints().getAvgOutputRecordSize()
   def avgBytesPerRecord_=(value: Float) = contract.getCompilerHints().setAvgOutputRecordSize(value)
-  def avgBytesPerRecord(value: Float): this.type = { contract.getCompilerHints().setAvgOutputRecordSize(value);
this }
+  def avgBytesPerRecord(value: Float): this.type = {
+    contract.getCompilerHints().setAvgOutputRecordSize(value)
+    this
+  }
 
   def filterFactor = contract.getCompilerHints().getFilterFactor()
   def filterFactor_=(value: Float) = contract.getCompilerHints().setFilterFactor(value)
-  def filterFactor(value: Float): this.type = { contract.getCompilerHints().setFilterFactor(value);
this }
+  def filterFactor(value: Float): this.type = {
+    contract.getCompilerHints().setFilterFactor(value)
+    this
+  }
 
   def uniqueKey[Key](fields: Out => Key) = macro OutputHintableMacros.uniqueKey[Out, Key]
 
   def applyHints(contract: Operator[Record] with ScalaOperator[_, _]): Unit = {
     val hints = contract.getCompilerHints
 
-    if (hints.getUniqueFields != null)
+    if (hints.getUniqueFields != null) {
       hints.getUniqueFields.clear()
+    }
 
     _cardinalities.foreach { card =>
 
@@ -92,7 +111,10 @@ trait OutputHintable[Out] { this: DataSet[Out] =>
 
 object OutputHintableMacros {
   
-  def uniqueKey[Out: c.WeakTypeTag, Key: c.WeakTypeTag](c: Context { type PrefixType = OutputHintable[Out]
})(fields: c.Expr[Out => Key]): c.Expr[Unit] = {
+  def uniqueKey[Out: c.WeakTypeTag, Key: c.WeakTypeTag]
+      (c: Context { type PrefixType = OutputHintable[Out] })(fields: c.Expr[Out => Key])
+    : c.Expr[Unit] = {
+
     import c.universe._
 
     val slave = MacroContextHolder.newMacroHelper(c)
@@ -109,10 +131,14 @@ object OutputHintableMacros {
       
       c.prefix.splice.addCardinality(card)
     }
-    return result
+    result
   }
   
-  def uniqueKeyWithDistinctCount[Out: c.WeakTypeTag, Key: c.WeakTypeTag](c: Context { type
PrefixType = OutputHintable[Out] })(fields: c.Expr[Out => Key], distinctCount: c.Expr[Long]):
c.Expr[Unit] = {
+  def uniqueKeyWithDistinctCount[Out: c.WeakTypeTag, Key: c.WeakTypeTag]
+      (c: Context { type PrefixType = OutputHintable[Out] })
+      (fields: c.Expr[Out => Key], distinctCount: c.Expr[Long])
+    : c.Expr[Unit] = {
+
     import c.universe._
 
     val slave = MacroContextHolder.newMacroHelper(c)
@@ -129,10 +155,13 @@ object OutputHintableMacros {
       
       c.prefix.splice.addCardinality(card)
     }
-    return result
+    result
   }
   
-  def cardinality[Out: c.WeakTypeTag, Key: c.WeakTypeTag](c: Context { type PrefixType =
OutputHintable[Out] })(fields: c.Expr[Out => Key]): c.Expr[Unit] = {
+  def cardinality[Out: c.WeakTypeTag, Key: c.WeakTypeTag]
+      (c: Context { type PrefixType = OutputHintable[Out] })(fields: c.Expr[Out => Key])
+    : c.Expr[Unit] = {
+
     import c.universe._
 
     val slave = MacroContextHolder.newMacroHelper(c)
@@ -149,10 +178,14 @@ object OutputHintableMacros {
       
       c.prefix.splice.addCardinality(card)
     }
-    return result
+    result
   }
   
-  def cardinalityWithDistinctCount[Out: c.WeakTypeTag, Key: c.WeakTypeTag](c: Context { type
PrefixType = OutputHintable[Out] })(fields: c.Expr[Out => Key], distinctCount: c.Expr[Long]):
c.Expr[Unit] = {
+  def cardinalityWithDistinctCount[Out: c.WeakTypeTag, Key: c.WeakTypeTag]
+      (c: Context { type PrefixType = OutputHintable[Out] })
+      (fields: c.Expr[Out => Key], distinctCount: c.Expr[Long])
+    : c.Expr[Unit] = {
+
     import c.universe._
 
     val slave = MacroContextHolder.newMacroHelper(c)
@@ -169,10 +202,12 @@ object OutputHintableMacros {
       
       c.prefix.splice.addCardinality(card)
     }
-    return result
+    result
   }
   
-  def cardinalityWithAvgNumRecords[Out: c.WeakTypeTag, Key: c.WeakTypeTag](c: Context { type
PrefixType = OutputHintable[Out] })(fields: c.Expr[Out => Key], avgNumRecords: c.Expr[Float]):
c.Expr[Unit] = {
+  def cardinalityWithAvgNumRecords[Out: c.WeakTypeTag, Key: c.WeakTypeTag]
+      (c: Context { type PrefixType = OutputHintable[Out] })
+      (fields: c.Expr[Out => Key], avgNumRecords: c.Expr[Float]): c.Expr[Unit] = {
     import c.universe._
 
     val slave = MacroContextHolder.newMacroHelper(c)
@@ -189,10 +224,14 @@ object OutputHintableMacros {
       
       c.prefix.splice.addCardinality(card)
     }
-    return result
+    result
   }
   
-  def cardinalityWithAll[Out: c.WeakTypeTag, Key: c.WeakTypeTag](c: Context { type PrefixType
= OutputHintable[Out] })(fields: c.Expr[Out => Key], distinctCount: c.Expr[Long], avgNumRecords:
c.Expr[Float]): c.Expr[Unit] = {
+  def cardinalityWithAll[Out: c.WeakTypeTag, Key: c.WeakTypeTag]
+      (c: Context { type PrefixType = OutputHintable[Out] })
+      (fields: c.Expr[Out => Key], distinctCount: c.Expr[Long], avgNumRecords: c.Expr[Float])
+    : c.Expr[Unit] = {
+
     import c.universe._
 
     val slave = MacroContextHolder.newMacroHelper(c)
@@ -201,15 +240,14 @@ object OutputHintableMacros {
 
     val result = reify {
       val contract = c.prefix.splice.getContract
-      val hints = contract.getCompilerHints
       
       val keySelection = generatedKeySelector.splice
-      val key = new FieldSelector(c.prefix.splice.getContract.getUDF.outputUDT, keySelection)
+      val key = new FieldSelector(contract.getUDF.outputUDT, keySelection)
       val card = KeyCardinality(key, false, Some(distinctCount.splice), Some(avgNumRecords.splice))
       
       c.prefix.splice.addCardinality(card)
     }
-    return result
+    result
   }
 }
 
@@ -220,9 +258,12 @@ trait InputHintable[In, Out] { this: DataSet[Out] =>
   def getInputUDT: UDT[In]
   def getOutputUDT: UDT[Out]
 
-  def neglects[Fields](fields: In => Fields): Unit = macro InputHintableMacros.neglects[In,
Out, Fields]
-  def observes[Fields](fields: In => Fields): Unit = macro InputHintableMacros.observes[In,
Out, Fields]
-  def preserves[Fields](from: In => Fields, to: Out => Fields) = macro InputHintableMacros.preserves[In,
Out, Fields]
+  def neglects[Fields](fields: In => Fields): Unit =
+    macro InputHintableMacros.neglects[In, Out, Fields]
+  def observes[Fields](fields: In => Fields): Unit =
+    macro InputHintableMacros.observes[In, Out, Fields]
+  def preserves[Fields](from: In => Fields, to: Out => Fields) =
+    macro InputHintableMacros.preserves[In, Out, Fields]
 }
 
 object InputHintable {
@@ -235,7 +276,11 @@ object InputHintable {
 
 object InputHintableMacros {
   
-  def neglects[In: c.WeakTypeTag, Out: c.WeakTypeTag, Fields: c.WeakTypeTag](c: Context {
type PrefixType = InputHintable[In, Out] })(fields: c.Expr[In => Fields]): c.Expr[Unit]
= {
+  def neglects[In: c.WeakTypeTag, Out: c.WeakTypeTag, Fields: c.WeakTypeTag]
+      (c: Context { type PrefixType = InputHintable[In, Out] })
+      (fields: c.Expr[In => Fields])
+    : c.Expr[Unit] = {
+
     import c.universe._
 
     val slave = MacroContextHolder.newMacroHelper(c)
@@ -248,10 +293,14 @@ object InputHintableMacros {
       val unreadFields = fieldSelector.selectedFields.map(_.localPos).toSet
       unreadFields.foreach(c.prefix.splice.markUnread(_))
     }
-    return result
+    result
   }
   
-  def observes[In: c.WeakTypeTag, Out: c.WeakTypeTag, Fields: c.WeakTypeTag](c: Context {
type PrefixType = InputHintable[In, Out] })(fields: c.Expr[In => Fields]): c.Expr[Unit]
= {
+  def observes[In: c.WeakTypeTag, Out: c.WeakTypeTag, Fields: c.WeakTypeTag]
+      (c: Context { type PrefixType = InputHintable[In, Out] })
+      (fields: c.Expr[In => Fields])
+    : c.Expr[Unit] = {
+
     import c.universe._
 
     val slave = MacroContextHolder.newMacroHelper(c)
@@ -265,14 +314,18 @@ object InputHintableMacros {
       val unreadFields = fieldSelector.inputFields.map(_.localPos).toSet.diff(fieldSet)
       unreadFields.foreach(c.prefix.splice.markUnread(_))
     }
-    return result
+    result
   }
   
-  def preserves[In: c.WeakTypeTag, Out: c.WeakTypeTag, Fields: c.WeakTypeTag](c: Context
{ type PrefixType = InputHintable[In, Out] })(from: c.Expr[In => Fields], to: c.Expr[Out
=> Fields]): c.Expr[Unit] = {
+  def preserves[In: c.WeakTypeTag, Out: c.WeakTypeTag, Fields: c.WeakTypeTag]
+      (c: Context { type PrefixType = InputHintable[In, Out] })
+      (from: c.Expr[In => Fields], to: c.Expr[Out => Fields])
+    : c.Expr[Unit] = {
+
     import c.universe._
 
-     val slave = MacroContextHolder.newMacroHelper(c)
-    
+    val slave = MacroContextHolder.newMacroHelper(c)
+
     val generatedFromFieldSelector = slave.getSelector(from)
     val generatedToFieldSelector = slave.getSelector(to)
 
@@ -281,14 +334,16 @@ object InputHintableMacros {
       val fromSelector = new FieldSelector(c.prefix.splice.getInputUDT, fromSelection)
       val toSelection = generatedToFieldSelector.splice
       val toSelector = new FieldSelector(c.prefix.splice.getOutputUDT, toSelection)
-      val pairs = fromSelector.selectedFields.map(_.localPos).zip(toSelector.selectedFields.map(_.localPos))
+      val pairs = fromSelector.selectedFields.map(_.localPos)
+        .zip(toSelector.selectedFields.map(_.localPos))
       pairs.foreach(c.prefix.splice.markCopied.tupled)
     }
-    return result
+    result
   }
 }
 
-trait OneInputHintable[In, Out] extends InputHintable[In, Out] with OutputHintable[Out] {
this: DataSet[Out] =>
+trait OneInputHintable[In, Out] extends InputHintable[In, Out] with OutputHintable[Out] {
+  this: DataSet[Out] =>
 	override def markUnread = contract.getUDF.asInstanceOf[UDF1[In, Out]].markInputFieldUnread
_ 
 	override def markCopied = contract.getUDF.asInstanceOf[UDF1[In, Out]].markFieldCopied _

 	
@@ -298,15 +353,19 @@ trait OneInputHintable[In, Out] extends InputHintable[In, Out] with
OutputHintab
 
 trait TwoInputHintable[LeftIn, RightIn, Out] extends OutputHintable[Out] { this: DataSet[Out]
=>
   val left = new DataSet[Out](contract) with OneInputHintable[LeftIn, Out] {
-	override def markUnread = { pos: Int => contract.getUDF.asInstanceOf[UDF2[LeftIn, RightIn,
Out]].markInputFieldUnread(Left(pos))}
-	override def markCopied = { (from: Int, to: Int) => contract.getUDF.asInstanceOf[UDF2[LeftIn,
RightIn, Out]].markFieldCopied(Left(from), to)} 
+	override def markUnread = { pos: Int => contract.getUDF.asInstanceOf[UDF2[LeftIn, RightIn,
Out]]
+    .markInputFieldUnread(Left(pos))}
+	override def markCopied = { (from: Int, to: Int) => contract.getUDF
+    .asInstanceOf[UDF2[LeftIn, RightIn, Out]].markFieldCopied(Left(from), to)}
 	override def getInputUDT = contract.getUDF.asInstanceOf[UDF2[LeftIn, RightIn, Out]].leftInputUDT
 	override def getOutputUDT = contract.getUDF.asInstanceOf[UDF2[LeftIn, RightIn, Out]].outputUDT
   }
   
   val right = new DataSet[Out](contract) with OneInputHintable[RightIn, Out] {
-	override def markUnread = { pos: Int => contract.getUDF.asInstanceOf[UDF2[LeftIn, RightIn,
Out]].markInputFieldUnread(Right(pos))}
-	override def markCopied = { (from: Int, to: Int) => contract.getUDF.asInstanceOf[UDF2[LeftIn,
RightIn, Out]].markFieldCopied(Right(from), to)} 
+	override def markUnread = { pos: Int => contract.getUDF.asInstanceOf[UDF2[LeftIn, RightIn,
Out]]
+    .markInputFieldUnread(Right(pos))}
+	override def markCopied = { (from: Int, to: Int) => contract.getUDF
+    .asInstanceOf[UDF2[LeftIn, RightIn, Out]].markFieldCopied(Right(from), to)}
 	override def getInputUDT = contract.getUDF.asInstanceOf[UDF2[LeftIn, RightIn, Out]].rightInputUDT
 	override def getOutputUDT = contract.getUDF.asInstanceOf[UDF2[LeftIn, RightIn, Out]].outputUDT
   }

http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/3f511953/stratosphere-scala/src/main/scala/eu/stratosphere/api/scala/DataSet.scala
----------------------------------------------------------------------
diff --git a/stratosphere-scala/src/main/scala/eu/stratosphere/api/scala/DataSet.scala b/stratosphere-scala/src/main/scala/eu/stratosphere/api/scala/DataSet.scala
index d004492..ffc91e9 100644
--- a/stratosphere-scala/src/main/scala/eu/stratosphere/api/scala/DataSet.scala
+++ b/stratosphere-scala/src/main/scala/eu/stratosphere/api/scala/DataSet.scala
@@ -28,7 +28,8 @@ import eu.stratosphere.types.Record
 
 class DataSet[T] (val contract: Operator[Record] with ScalaOperator[T, Record]) {
   
-  def cogroup[RightIn](rightInput: DataSet[RightIn]) = new CoGroupDataSet[T, RightIn](this,
rightInput)
+  def cogroup[RightIn](rightInput: DataSet[RightIn]) =
+    new CoGroupDataSet[T, RightIn](this, rightInput)
   def cross[RightIn](rightInput: DataSet[RightIn]) = new CrossDataSet[T, RightIn](this, rightInput)
   def join[RightIn](rightInput: DataSet[RightIn]) = new JoinDataSet[T, RightIn](this, rightInput)
   
@@ -42,17 +43,28 @@ class DataSet[T] (val contract: Operator[Record] with ScalaOperator[T,
Record])
   // reduce without grouping
   def reduce(fun: (T, T) => T) = macro ReduceMacros.globalReduce[T]
   def reduceAll[Out](fun: Iterator[T] => Out) = macro ReduceMacros.globalReduceGroup[T,
Out]
-  def combinableReduceAll[Out](fun: Iterator[T] => Out) = macro ReduceMacros.combinableGlobalReduceGroup[T]
+  def combinableReduceAll[Out](fun: Iterator[T] => Out) =
+    macro ReduceMacros.combinableGlobalReduceGroup[T]
 
   def union(secondInput: DataSet[T]) = UnionOperator.impl[T](this, secondInput)
   
-  def iterateWithDelta[DeltaItem](stepFunction: DataSet[T] => (DataSet[T], DataSet[DeltaItem]))
= macro IterateMacros.iterateWithDelta[T, DeltaItem]
+  def iterateWithDelta[DeltaItem](stepFunction: DataSet[T] => (DataSet[T], DataSet[DeltaItem]))
=
+    macro IterateMacros.iterateWithDelta[T, DeltaItem]
   def iterate(n: Int, stepFunction: DataSet[T] => DataSet[T])= macro IterateMacros.iterate[T]
-  def iterateWithTermination[C](n: Int, stepFunction: DataSet[T] => DataSet[T], terminationFunction:
(DataSet[T],
-    DataSet[T]) => DataSet[C]) = macro IterateMacros.iterateWithTermination[T, C]
-  def iterateWithDelta[SolutionKey, WorksetItem](workset: DataSet[WorksetItem], solutionSetKey:
T => SolutionKey, stepFunction: (DataSet[T], DataSet[WorksetItem]) => (DataSet[T], DataSet[WorksetItem]),
maxIterations: Int) = macro WorksetIterateMacros.iterateWithDelta[T, SolutionKey, WorksetItem]
+  def iterateWithTermination[C](
+      n: Int,
+      stepFunction: DataSet[T] => DataSet[T],
+      terminationFunction: (DataSet[T],DataSet[T]) => DataSet[C]) =
+    macro IterateMacros.iterateWithTermination[T, C]
+  def iterateWithDelta[SolutionKey, WorksetItem](
+      workset: DataSet[WorksetItem],
+      solutionSetKey: T => SolutionKey,
+      stepFunction: (DataSet[T], DataSet[WorksetItem]) =>
+                    (DataSet[T], DataSet[WorksetItem]), maxIterations: Int) =
+    macro WorksetIterateMacros.iterateWithDelta[T, SolutionKey, WorksetItem]
   
   def write(url: String, format: ScalaOutputFormat[T]) = DataSinkOperator.write(this, url,
format)
-  def write(url: String, format: ScalaOutputFormat[T], name: String) = DataSinkOperator.write(this,
url, format, name)
+  def write(url: String, format: ScalaOutputFormat[T], name: String) =
+    DataSinkOperator.write(this, url, format, name)
   
 }
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/3f511953/stratosphere-scala/src/main/scala/eu/stratosphere/api/scala/DataSource.scala
----------------------------------------------------------------------
diff --git a/stratosphere-scala/src/main/scala/eu/stratosphere/api/scala/DataSource.scala
b/stratosphere-scala/src/main/scala/eu/stratosphere/api/scala/DataSource.scala
index 97688dc..ff8bc41 100644
--- a/stratosphere-scala/src/main/scala/eu/stratosphere/api/scala/DataSource.scala
+++ b/stratosphere-scala/src/main/scala/eu/stratosphere/api/scala/DataSource.scala
@@ -37,7 +37,7 @@ object DataSource {
 
         override def getUDF = format.getUDF
 
-        override def persistConfiguration() = format.persistConfiguration(this.getParameters())
+        override def persistConfiguration() = format.persistConfiguration(this.getParameters)
       }
 
 //      case "ext" => new GenericDataSource[GenericInputFormat[_]](format.asInstanceOf[GenericInputFormat[_]],
uri.toString)

http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/3f511953/stratosphere-scala/src/main/scala/eu/stratosphere/api/scala/ScalaOperator.scala
----------------------------------------------------------------------
diff --git a/stratosphere-scala/src/main/scala/eu/stratosphere/api/scala/ScalaOperator.scala
b/stratosphere-scala/src/main/scala/eu/stratosphere/api/scala/ScalaOperator.scala
index 7d7c6b5..d1da8dd 100644
--- a/stratosphere-scala/src/main/scala/eu/stratosphere/api/scala/ScalaOperator.scala
+++ b/stratosphere-scala/src/main/scala/eu/stratosphere/api/scala/ScalaOperator.scala
@@ -26,8 +26,9 @@ import eu.stratosphere.api.common.operators.Union
 import eu.stratosphere.types.Record
 import eu.stratosphere.types.{Nothing => JavaNothing}
 
-trait ScalaOperator[T, UT] { this: Operator[UT] =>
-  def getUDF(): UDF[T]
+trait ScalaOperator[T, UT] {
+  this: Operator[UT] =>
+  def getUDF: UDF[T]
   def getKeys: Seq[FieldSelector] = Seq()
   def persistConfiguration(): Unit = {}
   
@@ -41,14 +42,17 @@ trait ScalaOperator[T, UT] { this: Operator[UT] =>
         for ((key, inputNum) <- getKeys.zipWithIndex) {
           
           val source = key.selectedFields.toSerializerIndexArray
-          val target = optimizerNode map { _.getRemappedKeys(inputNum) } getOrElse { contract.getKeyColumns(inputNum)
}
+          val target = optimizerNode map { _.getRemappedKeys(inputNum) } getOrElse {
+            contract.getKeyColumns(inputNum) }
 
-          assert(source.length == target.length, "Attempt to write " + source.length + "
key indexes to an array of size " + target.length)
+          assert(source.length == target.length, "Attempt to write " + source.length +
+            " key indexes to an array of size " + target.length)
           System.arraycopy(source, 0, target, 0, source.length)
         }
       }
       
-      case _ if getKeys.size > 0 => throw new UnsupportedOperationException("Attempted
to set keys on a contract that doesn't support them")
+      case _ if getKeys.size > 0 => throw new UnsupportedOperationException("Attempted
to set " +
+        "keys on a contract that doesn't support them")
       
       case _ =>
     }
@@ -60,11 +64,14 @@ trait ScalaOperator[T, UT] { this: Operator[UT] =>
   protected def annotations: Seq[Annotation] = Seq()
 
   def getUserCodeAnnotation[A <: Annotation](annotationClass: Class[A]): A = {
-    val res = annotations find { _.annotationType().equals(annotationClass) } map { _.asInstanceOf[A]
} getOrElse null.asInstanceOf[A]
+    val res = annotations find { _.annotationType().equals(annotationClass) } map {
+      _.asInstanceOf[A] } getOrElse null.asInstanceOf[A]
 //    println("returning ANOOT: " + res + " FOR: " + annotationClass.toString)
 //    res match {
-//      case r : FunctionAnnotation.ConstantFieldsFirst => println("CONSTANT FIELDS FIRST:
" + r.value().mkString(","))
-//      case r : FunctionAnnotation.ConstantFieldsSecond => println("CONSTANT FIELDS SECOND:
" + r.value().mkString(","))
+//      case r : FunctionAnnotation.ConstantFieldsFirst => println("CONSTANT FIELDS FIRST:
" +
+//        r.value().mkString(","))
+//      case r : FunctionAnnotation.ConstantFieldsSecond => println("CONSTANT FIELDS SECOND:
" +
+//        r.value().mkString(","))
 //      case _ =>
 //    }
     res
@@ -75,38 +82,46 @@ trait NoOpScalaOperator[In, Out] extends ScalaOperator[Out, Record] {
this: Oper
 }
 
 trait HigherOrderScalaOperator[T] extends ScalaOperator[T, Record] { this: Operator[Record]
=>
-  override def getUDF(): UDF0[T]
+  override def getUDF: UDF0[T]
 }
 
 trait BulkIterationScalaOperator[T] extends HigherOrderScalaOperator[T] { this: Operator[Record]
=>
 }
 
-trait DeltaIterationScalaOperator[T] extends HigherOrderScalaOperator[T] { this: Operator[Record]
=>
+trait DeltaIterationScalaOperator[T] extends HigherOrderScalaOperator[T] {
+  this: Operator[Record] =>
   val key: FieldSelector
 }
 
-trait ScalaOutputOperator[In] extends ScalaOperator[Nothing, JavaNothing] { this: Operator[JavaNothing]
=>
-  override def getUDF(): UDF1[In, Nothing]
+trait ScalaOutputOperator[In] extends ScalaOperator[Nothing, JavaNothing] {
+  this: Operator[JavaNothing] =>
+  override def getUDF: UDF1[In, Nothing]
 }
 
 trait OneInputScalaOperator[In, Out] extends ScalaOperator[Out, Record] { this: Operator[Record]
=>
-  override def getUDF(): UDF1[In, Out]
+  override def getUDF: UDF1[In, Out]
 }
 
-trait TwoInputScalaOperator[In1, In2, Out] extends ScalaOperator[Out, Record] { this: Operator[Record]
=>
-  override def getUDF(): UDF2[In1, In2, Out]
+trait TwoInputScalaOperator[In1, In2, Out] extends ScalaOperator[Out, Record] {
+  this: Operator[Record] =>
+  override def getUDF: UDF2[In1, In2, Out]
 }
 
-trait UnionScalaOperator[In] extends TwoInputScalaOperator[In, In, In] { this: Union[Record]
=>
-  override def getUDF(): UDF2[In, In, In]
+trait UnionScalaOperator[In] extends TwoInputScalaOperator[In, In, In] {
+  this: Union[Record] =>
+  override def getUDF: UDF2[In, In, In]
 }
 
-trait OneInputKeyedScalaOperator[In, Out] extends OneInputScalaOperator[In, Out] { this:
Operator[Record] =>
+trait OneInputKeyedScalaOperator[In, Out] extends OneInputScalaOperator[In, Out] {
+  this: Operator[Record] =>
   val key: FieldSelector
   override def getKeys = Seq(key)
 }
 
-trait TwoInputKeyedScalaOperator[LeftIn, RightIn, Out] extends TwoInputScalaOperator[LeftIn,
RightIn, Out] { this: Operator[Record] =>
+trait TwoInputKeyedScalaOperator[LeftIn, RightIn, Out]
+  extends TwoInputScalaOperator[LeftIn, RightIn, Out] {
+
+  this: Operator[Record] =>
   val leftKey: FieldSelector
   val rightKey: FieldSelector
   override def getKeys = Seq(leftKey, rightKey)

http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/3f511953/stratosphere-scala/src/main/scala/eu/stratosphere/api/scala/operators/MapOperator.scala
----------------------------------------------------------------------
diff --git a/stratosphere-scala/src/main/scala/eu/stratosphere/api/scala/operators/MapOperator.scala
b/stratosphere-scala/src/main/scala/eu/stratosphere/api/scala/operators/MapOperator.scala
index 766ae0d..d60b552 100644
--- a/stratosphere-scala/src/main/scala/eu/stratosphere/api/scala/operators/MapOperator.scala
+++ b/stratosphere-scala/src/main/scala/eu/stratosphere/api/scala/operators/MapOperator.scala
@@ -18,49 +18,54 @@ import scala.reflect.macros.Context
 
 import eu.stratosphere.api.java.record.operators.MapOperator
 import eu.stratosphere.types.Record
-import eu.stratosphere.api.common.operators.Operator
-import eu.stratosphere.configuration.Configuration;
 import eu.stratosphere.util.Collector
 
 import eu.stratosphere.api.scala.codegen.{MacroContextHolder, Util}
 import eu.stratosphere.api.scala._
 import eu.stratosphere.api.scala.analysis._
-import eu.stratosphere.api.scala.functions.{MapFunction, FlatMapFunction, FilterFunction,
MapFunctionBase}
+import eu.stratosphere.api.scala.functions.{MapFunction, MapFunctionBase}
+import eu.stratosphere.api.scala.functions.{FlatMapFunction, FilterFunction}
 
 object MapMacros {
 
-  def map[In: c.WeakTypeTag, Out: c.WeakTypeTag](c: Context { type PrefixType = DataSet[In]
})(fun: c.Expr[In => Out]): c.Expr[DataSet[Out] with OneInputHintable[In, Out]] = {
+  def map[In: c.WeakTypeTag, Out: c.WeakTypeTag]
+      (c: Context { type PrefixType = DataSet[In] })
+      (fun: c.Expr[In => Out]): c.Expr[DataSet[Out]
+    with OneInputHintable[In, Out]] = {
+
     import c.universe._
 
     val slave = MacroContextHolder.newMacroHelper(c)
     
 //    val (paramName, udfBody) = slave.extractOneInputUdf(fun.tree)
 
-    val (udtIn, createUdtIn) = slave.mkUdtClass[In]
-    val (udtOut, createUdtOut) = slave.mkUdtClass[Out]
-
-    val stub: c.Expr[MapFunctionBase[In, Out]] = if (fun.actualType <:< weakTypeOf[MapFunction[In,
Out]])
-      reify { fun.splice.asInstanceOf[MapFunctionBase[In, Out]] }
-    else reify {
-      implicit val inputUDT: UDT[In] = c.Expr[UDT[In]](createUdtIn).splice
-      implicit val outputUDT: UDT[Out] = c.Expr[UDT[Out]](createUdtOut).splice
-      new MapFunctionBase[In, Out] {
-//        val userFun = ClosureCleaner.clean(fun.splice)
-//        val userFun = fun.splice
-        override def map(record: Record, out: Collector[Record]) = {
-          val input = deserializer.deserializeRecyclingOn(record)
-          val output = fun.splice.apply(input)
-
-          record.setNumFields(outputLength)
-
-          for (field <- discard)
-            record.setNull(field)
-
-          serializer.serialize(output, record)
-          out.collect(record)
+    val (udtIn, createUdtIn) = slave.mkUdtClass[In]()
+    val (udtOut, createUdtOut) = slave.mkUdtClass[Out]()
+
+    val stub: c.Expr[MapFunctionBase[In, Out]] =
+      if (fun.actualType <:< weakTypeOf[MapFunction[In, Out]])
+        reify { fun.splice.asInstanceOf[MapFunctionBase[In, Out]] }
+      else
+        reify {
+          implicit val inputUDT: UDT[In] = c.Expr[UDT[In]](createUdtIn).splice
+          implicit val outputUDT: UDT[Out] = c.Expr[UDT[Out]](createUdtOut).splice
+          new MapFunctionBase[In, Out] {
+            // val userFun = ClosureCleaner.clean(fun.splice)
+            // val userFun = fun.splice
+            override def map(record: Record, out: Collector[Record]) = {
+              val input = deserializer.deserializeRecyclingOn(record)
+              val output = fun.splice.apply(input)
+
+              record.setNumFields(outputLength)
+
+              for (field <- discard)
+                record.setNull(field)
+
+              serializer.serialize(output, record)
+              out.collect(record)
+            }
+          }
         }
-      }
-    }
 
     val contract = reify {
       val input = c.prefix.splice.contract
@@ -78,47 +83,53 @@ object MapMacros {
       stream
     }
 
-    val result = c.Expr[DataSet[Out] with OneInputHintable[In, Out]](Block(List(udtIn, udtOut),
contract.tree))
+    val result = c.Expr[DataSet[Out]
+      with OneInputHintable[In, Out]](Block(List(udtIn, udtOut), contract.tree))
 
-    return result
+    result
   }
   
-  def flatMap[In: c.WeakTypeTag, Out: c.WeakTypeTag](c: Context { type PrefixType = DataSet[In]
})(fun: c.Expr[In => Iterator[Out]]): c.Expr[DataSet[Out] with OneInputHintable[In, Out]]
= {
+  def flatMap[In: c.WeakTypeTag, Out: c.WeakTypeTag]
+      (c: Context { type PrefixType = DataSet[In] })
+      (fun: c.Expr[In => Iterator[Out]]): c.Expr[DataSet[Out]
+    with OneInputHintable[In, Out]] = {
+
     import c.universe._
 
     val slave = MacroContextHolder.newMacroHelper(c)
     
 //    val (paramName, udfBody) = slave.extractOneInputUdf(fun.tree)
 
-    val (udtIn, createUdtIn) = slave.mkUdtClass[In]
-    val (udtOut, createUdtOut) = slave.mkUdtClass[Out]
+    val (udtIn, createUdtIn) = slave.mkUdtClass[In]()
+    val (udtOut, createUdtOut) = slave.mkUdtClass[Out]()
 
-    val stub: c.Expr[MapFunctionBase[In, Out]] = if (fun.actualType <:< weakTypeOf[FlatMapFunction[In,
Out]])
-      reify { fun.splice.asInstanceOf[MapFunctionBase[In, Out]] }
-    else reify {
-      implicit val inputUDT: UDT[In] = c.Expr[UDT[In]](createUdtIn).splice
-      implicit val outputUDT: UDT[Out] = c.Expr[UDT[Out]](createUdtOut).splice
-      new MapFunctionBase[In, Out] {
-        override def map(record: Record, out: Collector[Record]) = {
-          val input = deserializer.deserializeRecyclingOn(record)
-          val output = fun.splice.apply(input)
+    val stub: c.Expr[MapFunctionBase[In, Out]] =
+      if (fun.actualType <:< weakTypeOf[FlatMapFunction[In, Out]])
+        reify { fun.splice.asInstanceOf[MapFunctionBase[In, Out]] }
+      else reify {
+        implicit val inputUDT: UDT[In] = c.Expr[UDT[In]](createUdtIn).splice
+        implicit val outputUDT: UDT[Out] = c.Expr[UDT[Out]](createUdtOut).splice
+        new MapFunctionBase[In, Out] {
+          override def map(record: Record, out: Collector[Record]) = {
+            val input = deserializer.deserializeRecyclingOn(record)
+            val output = fun.splice.apply(input)
 
-          if (output.nonEmpty) {
+            if (output.nonEmpty) {
 
-            record.setNumFields(outputLength)
+              record.setNumFields(outputLength)
 
-            for (field <- discard)
-              record.setNull(field)
+              for (field <- discard)
+                record.setNull(field)
 
-            for (item <- output) {
+              for (item <- output) {
 
-              serializer.serialize(item, record)
-              out.collect(record)
+                serializer.serialize(item, record)
+                out.collect(record)
+              }
             }
           }
         }
       }
-    }
     val contract = reify {
       val input = c.prefix.splice.contract
       val generatedStub = ClosureCleaner.clean(stub.splice)
@@ -135,33 +146,41 @@ object MapMacros {
       stream
     }
 
-    val result = c.Expr[DataSet[Out] with OneInputHintable[In, Out]](Block(List(udtIn, udtOut),
contract.tree))
+    val result = c.Expr[DataSet[Out] with OneInputHintable[In, Out]](Block(List(udtIn, udtOut),
+      contract.tree))
 
-    return result
+    result
   }
   
-  def filter[In: c.WeakTypeTag](c: Context { type PrefixType = DataSet[In] })(fun: c.Expr[In
=> Boolean]): c.Expr[DataSet[In] with OneInputHintable[In, In]] = {
+  def filter[In: c.WeakTypeTag]
+      (c: Context { type PrefixType = DataSet[In] })
+      (fun: c.Expr[In => Boolean]): c.Expr[DataSet[In]
+
+    with OneInputHintable[In, In]] = {
+
     import c.universe._
 
     val slave = MacroContextHolder.newMacroHelper(c)
-    
-//    val (paramName, udfBody) = slave.extractOneInputUdf(fun.tree)
 
-    val (udtIn, createUdtIn) = slave.mkUdtClass[In]
-
-    val stub: c.Expr[MapFunctionBase[In, In]] = if (fun.actualType <:< weakTypeOf[FilterFunction[In,
In]])
-      reify { fun.splice.asInstanceOf[MapFunctionBase[In, In]] }
-    else reify {
-      implicit val inputUDT: UDT[In] = c.Expr[UDT[In]](createUdtIn).splice
-      new MapFunctionBase[In, In] {
-        override def map(record: Record, out: Collector[Record]) = {
-          val input = deserializer.deserializeRecyclingOn(record)
-          if (fun.splice.apply(input)) {
-        	  out.collect(record)
+    //    val (paramName, udfBody) = slave.extractOneInputUdf(fun.tree)
+
+    val (udtIn, createUdtIn) = slave.mkUdtClass[In]()
+
+    val stub: c.Expr[MapFunctionBase[In, In]] =
+      if (fun.actualType <:< weakTypeOf[FilterFunction[In, In]])
+        reify { fun.splice.asInstanceOf[MapFunctionBase[In, In]] }
+      else
+        reify {
+          implicit val inputUDT: UDT[In] = c.Expr[UDT[In]](createUdtIn).splice
+          new MapFunctionBase[In, In] {
+            override def map(record: Record, out: Collector[Record]) = {
+              val input = deserializer.deserializeRecyclingOn(record)
+              if (fun.splice.apply(input)) {
+        	      out.collect(record)
+              }
+            }
           }
         }
-      }
-    }
     val contract = reify {
       val input = c.prefix.splice.contract
       val generatedStub = ClosureCleaner.clean(stub.splice)
@@ -175,14 +194,15 @@ object MapMacros {
       }
       val stream = new DataSet[In](contract) with OneInputHintable[In, In] {}
       contract.persistHints = { () =>
-        stream.applyHints(contract);
+        stream.applyHints(contract)
         0 until generatedStub.udf.getOutputLength foreach { i => stream.markCopied(i,
i) }
       }
       stream
     }
 
-    val result = c.Expr[DataSet[In] with OneInputHintable[In, In]](Block(List(udtIn), contract.tree))
+    val result = c.Expr[DataSet[In]
+      with OneInputHintable[In, In]](Block(List(udtIn), contract.tree))
 
-    return result
+    result
   }
 }

http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/3f511953/stratosphere-scala/src/main/scala/eu/stratosphere/api/scala/operators/UnionOperator.scala
----------------------------------------------------------------------
diff --git a/stratosphere-scala/src/main/scala/eu/stratosphere/api/scala/operators/UnionOperator.scala
b/stratosphere-scala/src/main/scala/eu/stratosphere/api/scala/operators/UnionOperator.scala
index af58d21..d3f2e85 100644
--- a/stratosphere-scala/src/main/scala/eu/stratosphere/api/scala/operators/UnionOperator.scala
+++ b/stratosphere-scala/src/main/scala/eu/stratosphere/api/scala/operators/UnionOperator.scala
@@ -23,12 +23,14 @@ import eu.stratosphere.types.Record
 object UnionOperator {
 
   def impl[In](firstInput: DataSet[In], secondInput: DataSet[In]): DataSet[In] = {
-    val union = new Union[Record](firstInput.contract, secondInput.contract) with UnionScalaOperator[In]
{
-      private val inputUDT = firstInput.contract.getUDF().outputUDT
+    val union = new Union[Record](firstInput.contract, secondInput.contract)
+      with UnionScalaOperator[In] {
+
+      private val inputUDT = firstInput.contract.getUDF.outputUDT
       private val udf: UDF2[In, In, In] = new UDF2(inputUDT, inputUDT, inputUDT)
 
-      override def getUDF = udf;
+      override def getUDF = udf
     }
-    return new DataSet(union)
+    new DataSet(union)
   }
 }

http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/3f511953/stratosphere-tests/src/test/java/eu/stratosphere/test/javaApiOperators/FilterITCase.java
----------------------------------------------------------------------
diff --git a/stratosphere-tests/src/test/java/eu/stratosphere/test/javaApiOperators/FilterITCase.java
b/stratosphere-tests/src/test/java/eu/stratosphere/test/javaApiOperators/FilterITCase.java
index 6d4aed8..ffa804e 100644
--- a/stratosphere-tests/src/test/java/eu/stratosphere/test/javaApiOperators/FilterITCase.java
+++ b/stratosphere-tests/src/test/java/eu/stratosphere/test/javaApiOperators/FilterITCase.java
@@ -319,7 +319,7 @@ public class FilterITCase extends JavaProgramTestBase {
 							public boolean filter(Tuple3<Integer, Long, String> value) throws Exception
{
 								return (value.f1 == (broadcastSum / 11));
 							}
-						}).withBroadcastSet(intDs, "ints");;
+						}).withBroadcastSet(intDs, "ints");
 				filterDs.writeAsCsv(resultPath);
 				env.execute();
 				


Mime
View raw message