GitHub PullRequestReviewCommentEvent at 2014-07-25 01:43:26
Repository: scalding
Actor: jcoveney
Payload: {"action":"created","comment":{"url":"https://api.github.com/repos/twitter/scalding/pulls/comments/15382965","id":15382965,"diff_hunk":"@@ -512,117 +527,142 @@ final case class IterablePipe[T](iterable: Iterable[T],\n     tiny.flatMap { u => iterable.map { (_, u) } }\n \n   override def filter(f: T => Boolean): TypedPipe[T] =\n-    IterablePipe(iterable.filter(f), fd, mode)\n+    IterablePipe(iterable.filter(f))\n \n   override def flatMap[U](f: T => TraversableOnce[U]) =\n-    IterablePipe(iterable.flatMap(f), fd, mode)\n+    IterablePipe(iterable.flatMap(f))\n \n   override def fork: TypedPipe[T] = this\n \n-  override def limit(count: Int): TypedPipe[T] = IterablePipe(iterable.take(count), fd, mode)\n+  override def forceToDisk = this\n \n-  private def defaultSeed: Long = System.identityHashCode(this) * 2654435761L ^ System.currentTimeMillis\n-  def sample(percent: Double): TypedPipe[T] = sample(percent, defaultSeed)\n-  def sample(percent: Double, seed: Long): TypedPipe[T] = {\n-    val rand = new Random(seed)\n-    IterablePipe(iterable.filter(_ => rand.nextDouble < percent), fd, mode)\n+  override def limit(count: Int): TypedPipe[T] = IterablePipe(iterable.take(count))\n+\n+  override def map[U](f: T => U): TypedPipe[U] = IterablePipe(iterable.map(f))\n+\n+  override def sum[U >: T](implicit plus: Semigroup[U]): ValuePipe[U] =\n+    Semigroup.sumOption[U](iterable).map(LiteralValue(_))\n+      .getOrElse(EmptyValue)\n+\n+  override def sumByLocalKeys[K, V](implicit ev: T <:< (K, V), sg: Semigroup[V]) =\n+    // TODO This is pretty inefficient\n+    IterablePipe(iterable.map(ev(_)).groupBy(_._1).mapValues(_.map(_._2).reduce(sg.plus(_, _))))\n+\n+  override def toPipe[U >: T](fieldNames: Fields)(implicit flowDef: FlowDef, mode: Mode, setter: TupleSetter[U]): Pipe =\n+    IterableSource[U](iterable, fieldNames)(setter, singleConverter[U]).read(flowDef, mode)\n+}\n+\n+object TypedPipeFactory {\n+  def apply[T](next: (FlowDef, Mode) => TypedPipe[T]): TypedPipeFactory[T] = {\n+    val memo = new java.util.WeakHashMap[FlowDef, (Mode, TypedPipe[T])]()\n+    val fn = { (fd: FlowDef, m: Mode) =>\n+      memo.synchronized {\n+        memo.get(fd) match {\n+          case null =>\n+            val res = next(fd, m)\n+            memo.put(fd, (m, res))\n+            res\n+          case (memoMode, pipe) if memoMode == m => pipe\n+          case (memoMode, pipe) =>\n+            sys.error(\"FlowDef reused on different Mode. Original: %s, now: %s\".format(memoMode, m))\n+        }\n+      }\n+    }\n+    new TypedPipeFactory(NoStackAndThen(fn.tupled))\n   }\n+  def unapply[T](tp: TypedPipe[T]): Option[NoStackAndThen[(FlowDef, Mode), TypedPipe[T]]] =\n+    tp match {\n+      case tp: TypedPipeFactory[_] =>\n+        Some(tp.asInstanceOf[TypedPipeFactory[T]].next)\n+      case _ => None\n+    }\n+}\n \n-  override def map[U](f: T => U): TypedPipe[U] =\n-    IterablePipe(iterable.map(f), fd, mode)\n+/**\n+ * This is a TypedPipe that delays having access\n+ * to the FlowDef and Mode until toPipe is called\n+ */\n+class TypedPipeFactory[T] private (@transient val next: NoStackAndThen[(FlowDef, Mode), TypedPipe[T]]) extends TypedPipe[T] {\n+  private[this] def andThen[U](fn: TypedPipe[T] => TypedPipe[U]): TypedPipe[U] =\n+    new TypedPipeFactory(next.andThen(fn))\n \n-  override def toPipe[U >: T](fieldNames: Fields)(implicit setter: TupleSetter[U]): Pipe =\n-    IterableSource[U](iterable, fieldNames)(setter, singleConverter[U]).read(fd, mode)\n+  def cross[U](tiny: TypedPipe[U]) = andThen(_.cross(tiny))\n+  override def filter(f: T => Boolean): TypedPipe[T] = andThen(_.filter(f))\n+  def flatMap[U](f: T => TraversableOnce[U]): TypedPipe[U] = andThen(_.flatMap(f))\n+  override def map[U](f: T => U): TypedPipe[U] = andThen(_.map(f))\n \n-  override def sum[U >: T](implicit plus: Semigroup[U]): ValuePipe[U] =\n-    Semigroup.sumOption[U](iterable).map(LiteralValue(_)(fd, mode))\n-      .getOrElse(EmptyValue()(fd, mode))\n+  override def limit(count: Int) = andThen(_.limit(count))\n \n   override def sumByLocalKeys[K, V](implicit ev: T <:< (K, V), sg: Semigroup[V]) =\n-    IterablePipe(iterable.map(ev(_)).groupBy(_._1).mapValues(_.map(_._2).reduce(sg.plus(_, _))), fd, mode)\n+    andThen(_.sumByLocalKeys[K, V])\n+\n+  def toPipe[U >: T](fieldNames: Fields)(implicit flowDef: FlowDef, mode: Mode, setter: TupleSetter[U]) = {\n+    @annotation.tailrec\n+    def unwrap(pipe: TypedPipe[T]): TypedPipe[T] = pipe match {\n+      case TypedPipeFactory(n) => unwrap(n(flowDef, mode))\n+      case tp => tp\n+    }\n+    // unwrap in a loop, without recursing\n+    unwrap(next((flowDef, mode))).toPipe[U](fieldNames)(flowDef, mode, setter)\n+  }\n }\n \n /**\n  * This is an instance of a TypedPipe that wraps a cascading Pipe\n  */\n-final case class TypedPipeInst[T](@transient inpipe: Pipe,\n+class TypedPipeInst[T] private[scalding] (@transient inpipe: Pipe,\n   fields: Fields,\n+  @transient localFlowDef: FlowDef,\n+  @transient val mode: Mode,\n   flatMapFn: FlatMapFn[T]) extends TypedPipe[T] {\n \n-  import Dsl._\n-\n-  // The output pipe has a single item CTuple with an object of type T in position 0\n-  @transient protected lazy val pipe: Pipe = toPipe(0)(singleSetter[T])\n+  /**\n+   * If this TypedPipeInst represents a Source that was opened with no","path":"scalding-core/src/main/scala/com/twitter/scalding/typed/TypedPipe.scala","position":533,"original_position":533,"commit_id":"069749a9f4fe7361f36785b7b424fbbfc7bfabc8","original_commit_id":"069749a9f4fe7361f36785b7b424fbbfc7bfabc8","user":{"login":"jcoveney","id":830087,"avatar_url":"https://avatars.githubusercontent.com/u/830087?","gravatar_id":"c31517bb2497ca20710a294d3f8d1144","url":"https://api.github.com/users/jcoveney","html_url":"https://github.com/jcoveney","followers_url":"https://api.github.com/users/jcoveney/followers","following_url":"https://api.github.com/users/jcoveney/following{/other_user}","gists_url":"https://api.github.com/users/jcoveney/gists{/gist_id}","starred_url":"https://api.github.com/users/jcoveney/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/jcoveney/subscriptions","organizations_url":"https://api.github.com/users/jcoveney/orgs","repos_url":"https://api.github.com/users/jcoveney/repos","events_url":"https://api.github.com/users/jcoveney/events{/privacy}","received_events_url":"https://api.github.com/users/jcoveney/received_events","type":"User","site_admin":false},"body":"yeah, it's a grammar error. you repeat no twice","created_at":"2014-07-25T01:43:26Z","updated_at":"2014-07-25T01:43:26Z","html_url":"https://github.com/twitter/scalding/pull/968#discussion_r15382965","pull_request_url":"https://api.github.com/repos/twitter/scalding/pulls/968","_links":{"self":{"href":"https://api.github.com/repos/twitter/scalding/pulls/comments/15382965"},"html":{"href":"https://github.com/twitter/scalding/pull/968#discussion_r15382965"},"pull_request":{"href":"https://api.github.com/repos/twitter/scalding/pulls/968"}}},"pull_request":{"url":"https://api.github.com/repos/twitter/scalding/pulls/968","id":18691723,"html_url":"https://github.com/twitter/scalding/pull/968","diff_url":"https://github.com/twitter/scalding/pull/968.diff","patch_url":"https://github.com/twitter/scalding/pull/968.patch","issue_url":"https://api.github.com/repos/twitter/scalding/issues/968","number":968,"state":"open","title":"Make TypedPipe immutable","user":{"login":"johnynek","id":67958,"avatar_url":"https://avatars.githubusercontent.com/u/67958?","gravatar_id":"fa94a62940f16a371fdd3e6e8e12c3a1","url":"https://api.github.com/users/johnynek","html_url":"https://github.com/johnynek","followers_url":"https://api.github.com/users/johnynek/followers","following_url":"https://api.github.com/users/johnynek/following{/other_user}","gists_url":"https://api.github.com/users/johnynek/gists{/gist_id}","starred_url":"https://api.github.com/users/johnynek/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/johnynek/subscriptions","organizations_url":"https://api.github.com/users/johnynek/orgs","repos_url":"https://api.github.com/users/johnynek/repos","events_url":"https://api.github.com/users/johnynek/events{/privacy}","received_events_url":"https://api.github.com/users/johnynek/received_events","type":"User","site_admin":false},"body":"Let's really look carefully at this. This was an attempt at a minimal change that would be source-code compatible.\r\n\r\nSome questions:\r\n\r\n1. should TypedPipeInst be a case class? It seems like we don't want to pattern match and get the FlowDef back out, as it should never be mutated. Tend to think it should not be a case class (and we can add a custom unapply if needed to match the old cases, which does not expose the flowDef).\r\n2. Should we bite the bullet and make an ImmutableFlowDef, which is just exactly cascading's FlowDef with scala Map/Vector/List? Then have functions to copy between? We could have a Monoid of this class to handle the merging.\r\n3. Be on the lookout for any way to make this change smaller, while keeping the TypedPipe immutable. Verify that despite the fact that FlowDefs are mutable, the TypedPipe is still immutable.\r\n4. Can we move the snapshot/save/toIterator/dump methods from the repl directly on to TypedPipe?\r\n\r\nTODO:\r\n- [x] audit the Matrix2 code to remove unneeded FlowDefs from the code.\r\n- [x] consider if we can add implicit Mode to `def toPipe` so that we can remove the Mode bound to IterablePipe, EmptyPipe, LiteralValue, EmptyValue. Seems like a possibly harder case.","created_at":"2014-07-22T02:17:59Z","updated_at":"2014-07-25T01:43:26Z","closed_at":null,"merged_at":null,"merge_commit_sha":"e0717679b6dcb016a92b51e7557e6c7ee0bf7474","assignee":null,"milestone":null,"commits_url":"https://api.github.com/repos/twitter/scalding/pulls/968/commits","review_comments_url":"https://api.github.com/repos/twitter/scalding/pulls/968/comments","review_comment_url":"https://api.github.com/repos/twitter/scalding/pulls/comments/{number}","comments_url":"https://api.github.com/repos/twitter/scalding/issues/968/comments","statuses_url":"https://api.github.com/repos/twitter/scalding/statuses/069749a9f4fe7361f36785b7b424fbbfc7bfabc8","head":{"label":"twitter:ref_trans_typedpipe","ref":"ref_trans_typedpipe","sha":"069749a9f4fe7361f36785b7b424fbbfc7bfabc8","user":{"login":"twitter","id":50278,"avatar_url":"https://avatars.githubusercontent.com/u/50278?","gravatar_id":"2f4a8254d032a8ec5e4c48d461e54fcc","url":"https://api.github.com/users/twitter","html_url":"https://github.com/twitter","followers_url":"https://api.github.com/users/twitter/followers","following_url":"https://api.github.com/users/twitter/following{/other_user}","gists_url":"https://api.github.com/users/twitter/gists{/gist_id}","starred_url":"https://api.github.com/users/twitter/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/twitter/subscriptions","organizations_url":"https://api.github.com/users/twitter/orgs","repos_url":"https://api.github.com/users/twitter/repos","events_url":"https://api.github.com/users/twitter/events{/privacy}","received_events_url":"https://api.github.com/users/twitter/received_events","type":"Organization","site_admin":false},"repo":{"id":3146558,"name":"scalding","full_name":"twitter/scalding","owner":{"login":"twitter","id":50278,"avatar_url":"https://avatars.githubusercontent.com/u/50278?","gravatar_id":"2f4a8254d032a8ec5e4c48d461e54fcc","url":"https://api.github.com/users/twitter","html_url":"https://github.com/twitter","followers_url":"https://api.github.com/users/twitter/followers","following_url":"https://api.github.com/users/twitter/following{/other_user}","gists_url":"https://api.github.com/users/twitter/gists{/gist_id}","starred_url":"https://api.github.com/users/twitter/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/twitter/subscriptions","organizations_url":"https://api.github.com/users/twitter/orgs","repos_url":"https://api.github.com/users/twitter/repos","events_url":"https://api.github.com/users/twitter/events{/privacy}","received_events_url":"https://api.github.com/users/twitter/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/twitter/scalding","description":"A Scala API for Cascading","fork":false,"url":"https://api.github.com/repos/twitter/scalding","forks_url":"https://api.github.com/repos/twitter/scalding/forks","keys_url":"https://api.github.com/repos/twitter/scalding/keys{/key_id}","collaborators_url":"https://api.github.com/repos/twitter/scalding/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/twitter/scalding/teams","hooks_url":"https://api.github.com/repos/twitter/scalding/hooks","issue_events_url":"https://api.github.com/repos/twitter/scalding/issues/events{/number}","events_url":"https://api.github.com/repos/twitter/scalding/events","assignees_url":"https://api.github.com/repos/twitter/scalding/assignees{/user}","branches_url":"https://api.github.com/repos/twitter/scalding/branches{/branch}","tags_url":"https://api.github.com/repos/twitter/scalding/tags","blobs_url":"https://api.github.com/repos/twitter/scalding/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/twitter/scalding/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/twitter/scalding/git/refs{/sha}","trees_url":"https://api.github.com/repos/twitter/scalding/git/trees{/sha}","statuses_url":"https://api.github.com/repos/twitter/scalding/statuses/{sha}","languages_url":"https://api.github.com/repos/twitter/scalding/languages","stargazers_url":"https://api.github.com/repos/twitter/scalding/stargazers","contributors_url":"https://api.github.com/repos/twitter/scalding/contributors","subscribers_url":"https://api.github.com/repos/twitter/scalding/subscribers","subscription_url":"https://api.github.com/repos/twitter/scalding/subscription","commits_url":"https://api.github.com/repos/twitter/scalding/commits{/sha}","git_commits_url":"https://api.github.com/repos/twitter/scalding/git/commits{/sha}","comments_url":"https://api.github.com/repos/twitter/scalding/comments{/number}","issue_comment_url":"https://api.github.com/repos/twitter/scalding/issues/comments/{number}","contents_url":"https://api.github.com/repos/twitter/scalding/contents/{+path}","compare_url":"https://api.github.com/repos/twitter/scalding/compare/{base}...{head}","merges_url":"https://api.github.com/repos/twitter/scalding/merges","archive_url":"https://api.github.com/repos/twitter/scalding/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/twitter/scalding/downloads","issues_url":"https://api.github.com/repos/twitter/scalding/issues{/number}","pulls_url":"https://api.github.com/repos/twitter/scalding/pulls{/number}","milestones_url":"https://api.github.com/repos/twitter/scalding/milestones{/number}","notifications_url":"https://api.github.com/repos/twitter/scalding/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/twitter/scalding/labels{/name}","releases_url":"https://api.github.com/repos/twitter/scalding/releases{/id}","created_at":"2012-01-10T16:22:08Z","updated_at":"2014-07-25T00:02:55Z","pushed_at":"2014-07-25T01:28:28Z","git_url":"git://github.com/twitter/scalding.git","ssh_url":"git@github.com:twitter/scalding.git","clone_url":"https://github.com/twitter/scalding.git","svn_url":"https://github.com/twitter/scalding","homepage":"http://twitter.com/scalding","size":25034,"stargazers_count":1643,"watchers_count":1643,"language":"Scala","has_issues":true,"has_downloads":true,"has_wiki":true,"forks_count":373,"mirror_url":null,"open_issues_count":165,"forks":373,"open_issues":165,"watchers":1643,"default_branch":"develop"}},"base":{"label":"twitter:develop","ref":"develop","sha":"685ccd81c8fc1f71f0cbc74ee22743f01abcc9f6","user":{"login":"twitter","id":50278,"avatar_url":"https://avatars.githubusercontent.com/u/50278?","gravatar_id":"2f4a8254d032a8ec5e4c48d461e54fcc","url":"https://api.github.com/users/twitter","html_url":"https://github.com/twitter","followers_url":"https://api.github.com/users/twitter/followers","following_url":"https://api.github.com/users/twitter/following{/other_user}","gists_url":"https://api.github.com/users/twitter/gists{/gist_id}","starred_url":"https://api.github.com/users/twitter/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/twitter/subscriptions","organizations_url":"https://api.github.com/users/twitter/orgs","repos_url":"https://api.github.com/users/twitter/repos","events_url":"https://api.github.com/users/twitter/events{/privacy}","received_events_url":"https://api.github.com/users/twitter/received_events","type":"Organization","site_admin":false},"repo":{"id":3146558,"name":"scalding","full_name":"twitter/scalding","owner":{"login":"twitter","id":50278,"avatar_url":"https://avatars.githubusercontent.com/u/50278?","gravatar_id":"2f4a8254d032a8ec5e4c48d461e54fcc","url":"https://api.github.com/users/twitter","html_url":"https://github.com/twitter","followers_url":"https://api.github.com/users/twitter/followers","following_url":"https://api.github.com/users/twitter/following{/other_user}","gists_url":"https://api.github.com/users/twitter/gists{/gist_id}","starred_url":"https://api.github.com/users/twitter/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/twitter/subscriptions","organizations_url":"https://api.github.com/users/twitter/orgs","repos_url":"https://api.github.com/users/twitter/repos","events_url":"https://api.github.com/users/twitter/events{/privacy}","received_events_url":"https://api.github.com/users/twitter/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/twitter/scalding","description":"A Scala API for Cascading","fork":false,"url":"https://api.github.com/repos/twitter/scalding","forks_url":"https://api.github.com/repos/twitter/scalding/forks","keys_url":"https://api.github.com/repos/twitter/scalding/keys{/key_id}","collaborators_url":"https://api.github.com/repos/twitter/scalding/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/twitter/scalding/teams","hooks_url":"https://api.github.com/repos/twitter/scalding/hooks","issue_events_url":"https://api.github.com/repos/twitter/scalding/issues/events{/number}","events_url":"https://api.github.com/repos/twitter/scalding/events","assignees_url":"https://api.github.com/repos/twitter/scalding/assignees{/user}","branches_url":"https://api.github.com/repos/twitter/scalding/branches{/branch}","tags_url":"https://api.github.com/repos/twitter/scalding/tags","blobs_url":"https://api.github.com/repos/twitter/scalding/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/twitter/scalding/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/twitter/scalding/git/refs{/sha}","trees_url":"https://api.github.com/repos/twitter/scalding/git/trees{/sha}","statuses_url":"https://api.github.com/repos/twitter/scalding/statuses/{sha}","languages_url":"https://api.github.com/repos/twitter/scalding/languages","stargazers_url":"https://api.github.com/repos/twitter/scalding/stargazers","contributors_url":"https://api.github.com/repos/twitter/scalding/contributors","subscribers_url":"https://api.github.com/repos/twitter/scalding/subscribers","subscription_url":"https://api.github.com/repos/twitter/scalding/subscription","commits_url":"https://api.github.com/repos/twitter/scalding/commits{/sha}","git_commits_url":"https://api.github.com/repos/twitter/scalding/git/commits{/sha}","comments_url":"https://api.github.com/repos/twitter/scalding/comments{/number}","issue_comment_url":"https://api.github.com/repos/twitter/scalding/issues/comments/{number}","contents_url":"https://api.github.com/repos/twitter/scalding/contents/{+path}","compare_url":"https://api.github.com/repos/twitter/scalding/compare/{base}...{head}","merges_url":"https://api.github.com/repos/twitter/scalding/merges","archive_url":"https://api.github.com/repos/twitter/scalding/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/twitter/scalding/downloads","issues_url":"https://api.github.com/repos/twitter/scalding/issues{/number}","pulls_url":"https://api.github.com/repos/twitter/scalding/pulls{/number}","milestones_url":"https://api.github.com/repos/twitter/scalding/milestones{/number}","notifications_url":"https://api.github.com/repos/twitter/scalding/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/twitter/scalding/labels{/name}","releases_url":"https://api.github.com/repos/twitter/scalding/releases{/id}","created_at":"2012-01-10T16:22:08Z","updated_at":"2014-07-25T00:02:55Z","pushed_at":"2014-07-25T01:28:28Z","git_url":"git://github.com/twitter/scalding.git","ssh_url":"git@github.com:twitter/scalding.git","clone_url":"https://github.com/twitter/scalding.git","svn_url":"https://github.com/twitter/scalding","homepage":"http://twitter.com/scalding","size":25034,"stargazers_count":1643,"watchers_count":1643,"language":"Scala","has_issues":true,"has_downloads":true,"has_wiki":true,"forks_count":373,"mirror_url":null,"open_issues_count":165,"forks":373,"open_issues":165,"watchers":1643,"default_branch":"develop"}},"_links":{"self":{"href":"https://api.github.com/repos/twitter/scalding/pulls/968"},"html":{"href":"https://github.com/twitter/scalding/pull/968"},"issue":{"href":"https://api.github.com/repos/twitter/scalding/issues/968"},"comments":{"href":"https://api.github.com/repos/twitter/scalding/issues/968/comments"},"review_comments":{"href":"https://api.github.com/repos/twitter/scalding/pulls/968/comments"},"review_comment":{"href":"https://api.github.com/repos/twitter/scalding/pulls/comments/{number}"},"commits":{"href":"https://api.github.com/repos/twitter/scalding/pulls/968/commits"},"statuses":{"href":"https://api.github.com/repos/twitter/scalding/statuses/069749a9f4fe7361f36785b7b424fbbfc7bfabc8"}}}}