introduce new scalding-quotation module

Author: fwbrasil

.gitignore

@@ -13,6 +13,12 @@ project/plugins/lib_managed/
 project/plugins/src_managed/
 /.idea/
 /.idea_modules/
+.project
+.classpath
+.cache-main
+.cache-tests
+.tmpBin
+bin
 *.iml
 sonatype.sbt
 tutorial/data/cofollows.tsv

build.sbt

@@ -216,6 +216,7 @@ lazy val scalding = Project(
  .aggregate(
   scaldingArgs,
   scaldingDate,
+  scaldingQuotation,
   scaldingCore,
   scaldingCommons,
   scaldingAvro,
@@ -242,6 +243,7 @@ lazy val scaldingAssembly = Project(
  .aggregate(
   scaldingArgs,
   scaldingDate,
+  scaldingQuotation,
   scaldingCore,
   scaldingCommons,
   scaldingAvro,
@@ -312,6 +314,13 @@ lazy val scaldingBenchmarks = module("benchmarks")
     parallelExecution in Test := false
   ).dependsOn(scaldingCore)
 
+lazy val scaldingQuotation = module("quotation").settings(
+  libraryDependencies ++= Seq(
+    "org.scala-lang" % "scala-reflect" % scalaVersion.value % "provided",
+    "org.scala-lang" % "scala-compiler" % scalaVersion.value % "provided"
+  )
+)
+
 lazy val scaldingCore = module("core").settings(
   libraryDependencies ++= Seq(
     "cascading" % "cascading-core" % cascadingVersion,
@@ -333,7 +342,7 @@ lazy val scaldingCore = module("core").settings(
     "org.slf4j" % "slf4j-api" % slf4jVersion,
     "org.slf4j" % "slf4j-log4j12" % slf4jVersion % "provided"),
   addCompilerPlugin("org.scalamacros" % "paradise" % paradiseVersion cross CrossVersion.full)
-).dependsOn(scaldingArgs, scaldingDate, scaldingSerialization, maple)
+).dependsOn(scaldingArgs, scaldingDate, scaldingSerialization, maple, scaldingQuotation)
 
 lazy val scaldingCommons = module("commons").settings(
   libraryDependencies ++= Seq(

project/plugins.sbt

@@ -6,16 +6,17 @@ resolvers ++= Seq(
   "Twitter Maven" at "http://maven.twttr.com"
 )
 
-addSbtPlugin("com.eed3si9n"       % "sbt-assembly"        % "0.10.2")
-addSbtPlugin("com.eed3si9n"       % "sbt-unidoc"          % "0.3.3")
-addSbtPlugin("com.fortysevendeg"  % "sbt-microsites"      % "0.3.3")
-addSbtPlugin("com.github.gseitz"  % "sbt-release"         % "1.0.0")
-addSbtPlugin("com.jsuereth"       % "sbt-pgp"             % "1.0.0")
-addSbtPlugin("com.twitter"        %% "scrooge-sbt-plugin" % "4.14.0")
-addSbtPlugin("com.typesafe"       % "sbt-mima-plugin"     % "0.1.14")
-addSbtPlugin("com.typesafe.sbt"   % "sbt-ghpages"         % "0.5.4")
-addSbtPlugin("com.typesafe.sbt"   % "sbt-git"             % "0.6.2")
-addSbtPlugin("com.typesafe.sbt"   % "sbt-scalariform"     % "1.3.0")
-addSbtPlugin("org.scoverage"      % "sbt-scoverage"       % "1.5.0")
-addSbtPlugin("org.xerial.sbt"     % "sbt-sonatype"        % "1.0")
-addSbtPlugin("org.wartremover"    % "sbt-wartremover"     % "2.1.1")
+addSbtPlugin("com.eed3si9n"             % "sbt-assembly"        % "0.10.2")
+addSbtPlugin("com.eed3si9n"             % "sbt-unidoc"          % "0.3.3")
+addSbtPlugin("com.fortysevendeg"        % "sbt-microsites"      % "0.3.3")
+addSbtPlugin("com.github.gseitz"        % "sbt-release"         % "1.0.0")
+addSbtPlugin("com.jsuereth"             % "sbt-pgp"             % "1.0.0")
+addSbtPlugin("com.twitter"              %% "scrooge-sbt-plugin" % "4.14.0")
+addSbtPlugin("com.typesafe"             % "sbt-mima-plugin"     % "0.1.14")
+addSbtPlugin("com.typesafe.sbt"         % "sbt-ghpages"         % "0.5.4")
+addSbtPlugin("com.typesafe.sbt"         % "sbt-git"             % "0.6.2")
+addSbtPlugin("com.typesafe.sbt"         % "sbt-scalariform"     % "1.3.0")
+addSbtPlugin("org.scoverage"            % "sbt-scoverage"       % "1.5.0")
+addSbtPlugin("org.xerial.sbt"           % "sbt-sonatype"        % "1.0")
+addSbtPlugin("org.wartremover"          % "sbt-wartremover"     % "2.1.1")
+addSbtPlugin("com.typesafe.sbteclipse"  % "sbteclipse-plugin"   % "5.2.3")

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

@@ -417,6 +417,11 @@ trait Config extends Serializable {
    */
   def setVerboseFileSourceLogging(b: Boolean): Config =
     this + (VerboseFileSourceLoggingKey -> b.toString)
+    
+  def getAutomaticProjectionPushdown: Boolean =
+    get(ExperimentalAutomaticProjectionPushdown)
+      .map(_.toBoolean)
+      .getOrElse(false)
 
   override def hashCode = toMap.hashCode
   override def equals(that: Any) = that match {
@@ -487,6 +492,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)