introduce new scalding-quotation module

Author: fwbrasil

scalding-avro/src/main/scala/com/twitter/scalding/avro/package.scala

@@ -15,6 +15,7 @@
 
 package com.twitter.scalding
 
+import com.twitter.scalding.quotation.Quoted
 import cascading.flow.FlowDef
 import org.apache.avro.Schema
 import collection.JavaConverters._
@@ -26,7 +27,8 @@ package object avro {
     conv: TupleConverter[T],
     set: TupleSetter[T],
     flow: FlowDef,
-    mode: Mode): Unit = {
+    mode: Mode,
+    q: Quoted): Unit = {
     val sink = PackedAvroSource[T](path)
     pipe.write(sink)
   }
@@ -35,7 +37,8 @@ package object avro {
     conv: TupleConverter[T],
     set: TupleSetter[T],
     flow: FlowDef,
-    mode: Mode): Unit = {
+    mode: Mode,
+    q: Quoted): Unit = {
     import Dsl._
     val sink = UnpackedAvroSource[T](path, Some(schema))
     val outFields = {

scalding-commons/src/main/scala/com/twitter/scalding/commons/extensions/Checkpoint.scala

@@ -17,6 +17,7 @@ limitations under the License.
 package com.twitter.scalding.commons.extensions
 
 import com.twitter.scalding._
+import com.twitter.scalding.quotation.Quoted
 import com.twitter.scalding.Dsl._
 
 import cascading.flow.FlowDef
@@ -111,7 +112,7 @@ object Checkpoint {
 
   // Wrapper for Checkpoint when using a TypedPipe
   def apply[A](checkpointName: String)(flow: => TypedPipe[A])(implicit args: Args, mode: Mode, flowDef: FlowDef,
-    conv: TupleConverter[A], setter: TupleSetter[A]): TypedPipe[A] = {
+    conv: TupleConverter[A], setter: TupleSetter[A], q: Quoted): TypedPipe[A] = {
     val rPipe = apply(checkpointName, Dsl.intFields(0 until conv.arity)) {
       flow.toPipe(Dsl.intFields(0 until conv.arity))
     }

scalding-commons/src/main/scala/com/twitter/scalding/commons/source/VersionedKeyValSource.scala

@@ -32,6 +32,7 @@ import com.twitter.scalding.commons.tap.VersionedTap.TapMode
 import com.twitter.scalding.source.{ CheckedInversion, MaxFailuresCheck }
 import com.twitter.scalding.typed.KeyedListLike
 import com.twitter.scalding.typed.TypedSink
+import com.twitter.scalding.quotation.Quoted
 import org.apache.hadoop.mapred.JobConf
 import scala.collection.JavaConverters._
 
@@ -224,7 +225,7 @@ class TypedRichPipeEx[K: Ordering, V: Monoid](pipe: TypedPipe[(K, V)]) extends j
   // the pipe in using an implicit `Monoid[V]` and sinks all results
   // into the `sinkVersion` of data (or a new version) specified by
   // `src`.
-  def writeIncremental(src: VersionedKeyValSource[K, V], reducers: Int = 1)(implicit flowDef: FlowDef, mode: Mode): TypedPipe[(K, V)] = {
+  def writeIncremental(src: VersionedKeyValSource[K, V], reducers: Int = 1)(implicit flowDef: FlowDef, mode: Mode, q: Quoted): TypedPipe[(K, V)] = {
     val outPipe =
       if (!src.resourceExists(mode))
         pipe

scalding-commons/src/main/scala/com/twitter/scalding/examples/KMeans.scala

@@ -2,6 +2,7 @@ package com.twitter.scalding.examples
 
 import com.twitter.scalding._
 import com.twitter.scalding.typed.ComputedValue
+import com.twitter.scalding.quotation.Quoted
 
 object KMeans {
 
@@ -88,12 +89,12 @@ object KMeans {
     }
   }
 
-  def initializeClusters(k: Int, points: TypedPipe[Vector[Double]]): (ValuePipe[List[LabeledVector]], TypedPipe[LabeledVector]) = {
+  def initializeClusters(k: Int, points: TypedPipe[Vector[Double]])(implicit q: Quoted): (ValuePipe[List[LabeledVector]], TypedPipe[LabeledVector]) = {
     val rng = new java.util.Random(123)
     // take a random k vectors:
     val clusters = points.map { v => (rng.nextDouble, v) }
       .groupAll
-      .sortedTake(k)(Ordering.by(_._1))
+      .sortedTake(k)(Ordering.by(_._1), q)
       .mapValues { randk =>
         randk.iterator
           .zipWithIndex

scalding-core/src/main/scala/com/twitter/scalding/Config.scala

@@ -431,6 +431,11 @@ abstract class Config extends Serializable {
    */
   def setVerboseFileSourceLogging(b: Boolean): Config =
     this + (VerboseFileSourceLoggingKey -> b.toString)
+    
+  def getAutomaticProjectionPushdown: Boolean =
+    get(ExperimentalAutomaticProjectionPushdown)
+      .map(_.toBoolean)
+      .getOrElse(false)
 
   def getSkipNullCounters: Boolean =
     get(SkipNullCounters)
@@ -515,6 +520,11 @@ object Config {
    */
   val HashJoinAutoForceRight: String = "scalding.hashjoin.autoforceright"
 
+  /**
+   * Enables automatic projection pushdown (experimental)
+   */
+  val ExperimentalAutomaticProjectionPushdown: String = "scalding.experimental.automatic_projection_pushdown"
+  
   val empty: Config = Config(Map.empty)
 
   /*

scalding-core/src/main/scala/com/twitter/scalding/CumulativeSum.scala

@@ -1,6 +1,7 @@
 package com.twitter.scalding.typed
 
 import com.twitter.algebird._
+import com.twitter.scalding.quotation.Quoted
 
 /**
  * Extension for TypedPipe to add a cumulativeSum method.
@@ -39,7 +40,8 @@ object CumulativeSum {
     def cumulativeSum(
       implicit sg: Semigroup[V],
       ordU: Ordering[U],
-      ordK: Ordering[K]): SortedGrouped[K, (U, V)] = {
+      ordK: Ordering[K],
+      quoted: Quoted): SortedGrouped[K, (U, V)] = {
       pipe.group
         .sortBy { case (u, _) => u }
         .scanLeft(Nil: List[(U, V)]) {
@@ -62,7 +64,8 @@ object CumulativeSum {
       implicit ordS: Ordering[S],
       sg: Semigroup[V],
       ordU: Ordering[U],
-      ordK: Ordering[K]): TypedPipe[(K, (U, V))] = {
+      ordK: Ordering[K],
+      q: Quoted): TypedPipe[(K, (U, V))] = {
 
       val sumPerS = pipe
         .map { case (k, (u, v)) => (k, partition(u)) -> v }

scalding-core/src/main/scala/com/twitter/scalding/bdd/TBddDsl.scala

@@ -5,9 +5,12 @@ import com.twitter.scalding._
 import com.twitter.scalding.source.TypedText
 import scala.collection.mutable.Buffer
 import TDsl._
+import com.twitter.scalding.quotation.Quoted
 
 trait TBddDsl extends FieldConversions with TypedPipeOperationsConversions {
 
+  private implicit val q: Quoted = Quoted.internal
+
   def Given[TypeIn](source: TypedTestSource[TypeIn]): TestCaseGiven1[TypeIn] = new TestCaseGiven1[TypeIn](source)
 
   def GivenSources(sources: List[TypedTestSource[_]]): TestCaseGivenList = new TestCaseGivenList(sources)

scalding-core/src/main/scala/com/twitter/scalding/mathematics/Matrix.scala

@@ -30,6 +30,7 @@ import cascading.tap._
 import com.twitter.scalding.Dsl._
 import scala.math.max
 import scala.annotation.tailrec
+import com.twitter.scalding.quotation.Quoted
 
 /**
  * Matrix class - represents an infinite (hopefully sparse) matrix.
@@ -138,10 +139,11 @@ class MatrixMappableExtensions[T](mappable: Mappable[T])(implicit fd: FlowDef, m
   }
 
   def toBlockMatrix[Group, Row, Col, Val](implicit ev: <:<[T, (Group, Row, Col, Val)], ord: Ordering[(Group, Row)],
-    setter: TupleSetter[(Group, Row, Col, Val)]): BlockMatrix[Group, Row, Col, Val] =
+    setter: TupleSetter[(Group, Row, Col, Val)],
+    q: Quoted): BlockMatrix[Group, Row, Col, Val] =
     mapToBlockMatrix { _.asInstanceOf[(Group, Row, Col, Val)] }
 
-  def mapToBlockMatrix[Group, Row, Col, Val](fn: (T) => (Group, Row, Col, Val))(implicit ord: Ordering[(Group, Row)]): BlockMatrix[Group, Row, Col, Val] = {
+  def mapToBlockMatrix[Group, Row, Col, Val](fn: (T) => (Group, Row, Col, Val))(implicit ord: Ordering[(Group, Row)], q: Quoted): BlockMatrix[Group, Row, Col, Val] = {
     val matPipe = TypedPipe
       .from(mappable)
       .map(fn)