tomtau/scalding — pull request #12 by tomtau (2013-06-28 16:19:49+00:00)
Title: Feature/optimizedmatrix

# ISSUES
- [ ] NO GENERICS

I was quite puzzled how to make it generic / make the compiler happy (complains about not knowing generic Ring/Monoid[T] and ord: Ordering[T]). Hints appreciated.

- [ ] IS THERE A MOCK TYPEDPIPE OBJECT?

I wanted to port the original prototype tests (in Matrix2OptimizationTest), but the problem I had was with abstracting the Scalding/Cascading stuff from it... The "solution" I came up with is that I wrapped TypedPipes with Option and created a constructor for Literals without a Pipe... but I'm not entirely happy with it. Again, hints appreciated

- [ ] BETTER TYPEDPIPE APPROACH

I sometimes struggled in making it work, and the current code (in Matrix2 and Matrix2Test) is likely inefficient and/or unnecessary (e.g. creating TypedPipes via Fields). Feel free to comment on it.

- [ ] BROKEN MULTIPLICATION WITH LAZY SELF-OPTIMIZATION

Lastly, when I added the self-optimization feature, the unit test broke (see the stack trace below). I suspected two things may fix it: 1) removing that Option wrapper, 2) changing the mutable HashMap to immutable or ConcurrentMap... I tried both and neither of them fixed it - any other guesses where it goes wrong?

[error] x A Matrix2Prod job should
[error] x system error
[error] null (NativeConstructorAccessorImpl.java:-2)
[error] sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
[error] sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:57)
[error] sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
[error] java.lang.reflect.Constructor.newInstance(Constructor.java:532)
[error] com.twitter.scalding.Job$.apply(Job.scala:37)
[error] com.twitter.scalding.JobTest$$anonfun$apply$1.apply(JobTest.scala:11)
[error] com.twitter.scalding.JobTest$$anonfun$apply$1.apply(JobTest.scala:11)
[error] com.twitter.scalding.JobTest.initJob(JobTest.scala:123)
[error] com.twitter.scalding.JobTest.run(JobTest.scala:89)
[error] com.twitter.scalding.mathematics.Matrix2Test$$anonfun$3$$anonfun$apply$mcV$sp$3.apply$mcV$sp(Matrix2Test.scala:98)
[error] com.twitter.scalding.mathematics.TUtil$.printStack(MatrixTest.scala:25)
[error] com.twitter.scalding.mathematics.Matrix2Test$$anonfun$3.apply$mcV$sp(Matrix2Test.scala:89)
[error] com.twitter.scalding.mathematics.Matrix2Test$$anonfun$3.apply(Matrix2Test.scala:89)
[error] com.twitter.scalding.mathematics.Matrix2Test$$anonfun$3.apply(Matrix2Test.scala:89)
[error] org.specs.specification.LifeCycle$class.withCurrent(ExampleLifeCycle.scala:66)
[error] org.specs.specification.Examples.withCurrent(Examples.scala:52)
[error] org.specs.specification.Examples$$anonfun$specifyExample$1.apply(Examples.scala:114)
[error] org.specs.specification.Examples$$anonfun$specifyExample$1.apply(Examples.scala:114)
[error] org.specs.specification.ExampleExecution$$anonfun$3$$anonfun$apply$5.apply(ExampleLifeCycle.scala:219)
[error] scala.Option.getOrElse(Option.scala:108)
[error] org.specs.specification.LifeCycle$class.executeExpectations(ExampleLifeCycle.scala:90)
[error] org.specs.specification.BaseSpecification.executeExpectations(BaseSpecification.scala:58)
[error] org.specs.specification.LifeCycle$$anonfun$executeExpectations$1.apply(ExampleLifeCycle.scala:90)
[error] org.specs.specification.LifeCycle$$anonfun$executeExpectations$1.apply(ExampleLifeCycle.scala:90)
[error] scala.Option.map(Option.scala:133)
[error] org.specs.specification.LifeCycle$class.executeExpectations(ExampleLifeCycle.scala:90)
[error] org.specs.specification.BaseSpecification.executeExpectations(BaseSpecification.scala:58)
[error] org.specs.specification.ExampleContext$$anonfun$executeExpectations$1.apply(ExampleContext.scala:78)
[error] org.specs.specification.ExampleContext$$anonfun$executeExpectations$1.apply(ExampleContext.scala:78)
[error] scala.Option.map(Option.scala:133)
[error] org.specs.specification.ExampleContext$class.executeExpectations(ExampleContext.scala:78)
[error] org.specs.specification.Examples.executeExpectations(Examples.scala:52)
[error] org.specs.specification.ExampleExecution$$anonfun$3.apply(ExampleLifeCycle.scala:219)
[error] org.specs.specification.ExampleExecution$$anonfun$3.apply(ExampleLifeCycle.scala:198)
[error] org.specs.specification.ExampleExecution$$anonfun$2.apply(ExampleLifeCycle.scala:181)
[error] org.specs.specification.ExampleExecution.execute(ExampleLifeCycle.scala:252)
[error] org.specs.specification.SpecificationExecutor$$anonfun$executeExample$2.apply(SpecificationExecutor.scala:55)
[error] org.specs.specification.SpecificationExecutor$$anonfun$executeExample$2.apply(SpecificationExecutor.scala:55)
[error] scala.Option.map(Option.scala:133)
[error] org.specs.specification.SpecificationExecutor$class.executeExample(SpecificationExecutor.scala:55)
[error] org.specs.specification.BaseSpecification.executeExample(BaseSpecification.scala:58)
[error] org.specs.specification.BaseSpecification.executeExample(BaseSpecification.scala:58)
[error] org.specs.specification.Examples$$anonfun$executeExamples$2.apply(Examples.scala:80)
[error] org.specs.specification.Examples$$anonfun$executeExamples$2.apply(Examples.scala:80)
[error] scala.Option.map(Option.scala:133)
[error] org.specs.specification.Examples.executeExamples(Examples.scala:80)
[error] org.specs.specification.ExampleStructure$class.ownFailures(ExampleStructure.scala:58)
[error] org.specs.specification.Examples.ownFailures(Examples.scala:52)
[error] org.specs.specification.ExampleStructure$class.failures(ExampleStructure.scala:64)
[error] org.specs.specification.Examples.failures(Examples.scala:52)
[error] org.specs.specification.Examples.failures(Examples.scala:52)
[error] org.specs.execute.HasResults$class.failureAndErrors(HasResults.scala:61)
[error] org.specs.specification.Examples.failureAndErrors(Examples.scala:52)
[error] org.specs.execute.HasResults$class.isOk(HasResults.scala:69)
[error] org.specs.specification.Examples.isOk(Examples.scala:52)
[error] org.specs.runner.NotifierRunner.reportSystem(NotifierRunner.scala:81)
[error] org.specs.runner.NotifierRunner$$anonfun$reportASpecification$3.apply(NotifierRunner.scala:72)
[error] org.specs.runner.NotifierRunner$$anonfun$reportASpecification$3.apply(NotifierRunner.scala:68)
[error] scala.collection.LinearSeqOptimized$class.foreach(LinearSeqOptimized.scala:59)
[error] scala.collection.immutable.List.foreach(List.scala:76)
[error] org.specs.runner.NotifierRunner.reportASpecification(NotifierRunner.scala:68)
[error] org.specs.runner.NotifierRunner.report(NotifierRunner.scala:58)
[error] org.specs.runner.NotifierRunner.report(NotifierRunner.scala:45)
[error] org.specs.runner.Reporter$class.reportSpecs(Reporter.scala:195)
[error] org.specs.runner.NotifierRunner.reportSpecs(NotifierRunner.scala:45)
[error] org.specs.runner.TestInterfaceRunner$$anonfun$run$3.apply(TestInterfaceRunner.scala:72)
[error] org.specs.runner.TestInterfaceRunner$$anonfun$run$3.apply(TestInterfaceRunner.scala:72)
[error] scala.Option.map(Option.scala:133)
[error] org.specs.runner.TestInterfaceRunner.run(TestInterfaceRunner.scala:72)
[error] org.specs.runner.TestInterfaceRunner.run(TestInterfaceRunner.scala:65)
[error] sbt.TestRunner.delegateRun(TestFramework.scala:57)
[error] sbt.TestRunner.run(TestFramework.scala:51)
[error] sbt.TestRunner.runTest$1(TestFramework.scala:71)
[error] sbt.TestRunner.run(TestFramework.scala:80)
[error] sbt.TestFramework$$anonfun$9$$anonfun$apply$9$$anonfun$10$$anonfun$apply$10.apply(TestFramework.scala:188)
[error] sbt.TestFramework$$anonfun$9$$anonfun$apply$9$$anonfun$10$$anonfun$apply$10.apply(TestFramework.scala:188)
[error] sbt.TestFramework$.sbt$TestFramework$$withContextLoader(TestFramework.scala:200)
[error] sbt.TestFramework$$anonfun$9$$anonfun$apply$9$$anonfun$10.apply(TestFramework.scala:188)
[error] sbt.TestFramework$$anonfun$9$$anonfun$apply$9$$anonfun$10.apply(TestFramework.scala:188)
[error] sbt.Tests$$anonfun$makeSerial$1$$anonfun$apply$8.apply(Tests.scala:121)
[error] sbt.Tests$$anonfun$makeSerial$1$$anonfun$apply$8.apply(Tests.scala:121)
[error] scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:233)
[error] scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:233)
[error] scala.collection.LinearSeqOptimized$class.foreach(LinearSeqOptimized.scala:59)
[error] scala.collection.immutable.List.foreach(List.scala:76)
[error] scala.collection.TraversableLike$class.map(TraversableLike.scala:233)
[error] scala.collection.immutable.List.map(List.scala:76)
[error] sbt.Tests$$anonfun$makeSerial$1.apply(Tests.scala:121)
[error] sbt.Tests$$anonfun$makeSerial$1.apply(Tests.scala:121)
[error] sbt.std.Transform$$anon$3$$anonfun$apply$2.apply(System.scala:47)
[error] sbt.std.Transform$$anon$3$$anonfun$apply$2.apply(System.scala:47)
[error] sbt.std.Transform$$anon$5.work(System.scala:71)
[error] sbt.Execute$$anonfun$submit$1$$anonfun$apply$1.apply(Execute.scala:232)
[error] sbt.Execute$$anonfun$submit$1$$anonfun$apply$1.apply(Execute.scala:232)
[error] sbt.ErrorHandling$.wideConvert(ErrorHandling.scala:18)
[error] sbt.Execute.work(Execute.scala:238)
[error] sbt.Execute$$anonfun$submit$1.apply(Execute.scala:232)
[error] sbt.Execute$$anonfun$submit$1.apply(Execute.scala:232)
[error] sbt.ConcurrentRestrictions$$anon$4$$anonfun$1.apply(ConcurrentRestrictions.scala:160)
[error] sbt.CompletionService$$anon$2.call(CompletionService.scala:30)
[error] java.util.concurrent.FutureTask$Sync.innerRun(FutureTask.java:334)
[error] java.util.concurrent.FutureTask.run(FutureTask.java:166)
[error] java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:471)
[error] java.util.concurrent.FutureTask$Sync.innerRun(FutureTask.java:334)
[error] java.util.concurrent.FutureTask.run(FutureTask.java:166)
[error] java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1110)
[error] java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:603)
[error] java.lang.Thread.run(Thread.java:679)
[error] None.get (Option.scala:274)
[error] scala.None$.get(Option.scala:274)
[error] scala.None$.get(Option.scala:272)
[error] com.twitter.scalding.mathematics.Matrix2$$anonfun$computeCosts$1$1.apply(Matrix2.scala:78)
[error] com.twitter.scalding.mathematics.Matrix2$$anonfun$computeCosts$1$1.apply(Matrix2.scala:76)
[error] scala.collection.immutable.Range.foreach(Range.scala:81)
[error] com.twitter.scalding.mathematics.Matrix2$.computeCosts$1(Matrix2.scala:76)
[error] com.twitter.scalding.mathematics.Matrix2$.optimizeProductChain(Matrix2.scala:101)
[error] com.twitter.scalding.mathematics.Matrix2$.chainOrLast$1(Matrix2.scala:120)
[error] com.twitter.scalding.mathematics.Matrix2$.optimize(Matrix2.scala:186)
[error] com.twitter.scalding.mathematics.Matrix2$Matrix2.optimizedSelf(Matrix2.scala:16)
[error] com.twitter.scalding.mathematics.Matrix2$Product.toPipe(Matrix2.scala:28)
[error] com.twitter.scalding.mathematics.Matrix2$Product.tpipe(Matrix2.scala:32)
[error] com.twitter.scalding.mathematics.Matrix2$Product.tpipe(Matrix2.scala:19)
[error] com.twitter.scalding.mathematics.Matrix2Prod.(Matrix2Test.scala:42)
[error] sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
[error] sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:57)
[error] sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
[error] java.lang.reflect.Constructor.newInstance(Constructor.java:532)
[error] com.twitter.scalding.Job$.apply(Job.scala:37)
[error] com.twitter.scalding.JobTest$$anonfun$apply$1.apply(JobTest.scala:11)
[error] com.twitter.scalding.JobTest$$anonfun$apply$1.apply(JobTest.scala:11)
[error] com.twitter.scalding.JobTest.initJob(JobTest.scala:123)
[error] com.twitter.scalding.JobTest.run(JobTest.scala:89)
[error] com.twitter.scalding.mathematics.Matrix2Test$$anonfun$3$$anonfun$apply$mcV$sp$3.apply$mcV$sp(Matrix2Test.scala:98)
[error] com.twitter.scalding.mathematics.TUtil$.printStack(MatrixTest.scala:25)
[error] com.twitter.scalding.mathematics.Matrix2Test$$anonfun$3.apply$mcV$sp(Matrix2Test.scala:89)
[error] com.twitter.scalding.mathematics.Matrix2Test$$anonfun$3.apply(Matrix2Test.scala:89)
[error] com.twitter.scalding.mathematics.Matrix2Test$$anonfun$3.apply(Matrix2Test.scala:89)
[error] org.specs.specification.LifeCycle$class.withCurrent(ExampleLifeCycle.scala:66)
[error] org.specs.specification.Examples.withCurrent(Examples.scala:52)
[error] org.specs.specification.Examples$$anonfun$specifyExample$1.apply(Examples.scala:114)
[error] org.specs.specification.Examples$$anonfun$specifyExample$1.apply(Examples.scala:114)
[error] org.specs.specification.ExampleExecution$$anonfun$3$$anonfun$apply$5.apply(ExampleLifeCycle.scala:219)
[error] scala.Option.getOrElse(Option.scala:108)
[error] org.specs.specification.LifeCycle$class.executeExpectations(ExampleLifeCycle.scala:90)
[error] org.specs.specification.BaseSpecification.executeExpectations(BaseSpecification.scala:58)
[error] org.specs.specification.LifeCycle$$anonfun$executeExpectations$1.apply(ExampleLifeCycle.scala:90)
[error] org.specs.specification.LifeCycle$$anonfun$executeExpectations$1.apply(ExampleLifeCycle.scala:90)
[error] scala.Option.map(Option.scala:133)
[error] org.specs.specification.LifeCycle$class.executeExpectations(ExampleLifeCycle.scala:90)
[error] org.specs.specification.BaseSpecification.executeExpectations(BaseSpecification.scala:58)
[error] org.specs.specification.ExampleContext$$anonfun$executeExpectations$1.apply(ExampleContext.scala:78)
[error] org.specs.specification.ExampleContext$$anonfun$executeExpectations$1.apply(ExampleContext.scala:78)
[error] scala.Option.map(Option.scala:133)
[error] org.specs.specification.ExampleContext$class.executeExpectations(ExampleContext.scala:78)
[error] org.specs.specification.Examples.executeExpectations(Examples.scala:52)
[error] org.specs.specification.ExampleExecution$$anonfun$3.apply(ExampleLifeCycle.scala:219)
[error] org.specs.specification.ExampleExecution$$anonfun$3.apply(ExampleLifeCycle.scala:198)
[error] org.specs.specification.ExampleExecution$$anonfun$2.apply(ExampleLifeCycle.scala:181)
[error] org.specs.specification.ExampleExecution.execute(ExampleLifeCycle.scala:252)
[error] org.specs.specification.SpecificationExecutor$$anonfun$executeExample$2.apply(SpecificationExecutor.scala:55)
[error] org.specs.specification.SpecificationExecutor$$anonfun$executeExample$2.apply(SpecificationExecutor.scala:55)
[error] scala.Option.map(Option.scala:133)
[error] org.specs.specification.SpecificationExecutor$class.executeExample(SpecificationExecutor.scala:55)
[error] org.specs.specification.BaseSpecification.executeExample(BaseSpecification.scala:58)
[error] org.specs.specification.BaseSpecification.executeExample(BaseSpecification.scala:58)
[error] org.specs.specification.Examples$$anonfun$executeExamples$2.apply(Examples.scala:80)
[error] org.specs.specification.Examples$$anonfun$executeExamples$2.apply(Examples.scala:80)
[error] scala.Option.map(Option.scala:133)
[error] org.specs.specification.Examples.executeExamples(Examples.scala:80)
[error] org.specs.specification.ExampleStructure$class.ownFailures(ExampleStructure.scala:58)
[error] org.specs.specification.Examples.ownFailures(Examples.scala:52)
[error] org.specs.specification.ExampleStructure$class.failures(ExampleStructure.scala:64)
[error] org.specs.specification.Examples.failures(Examples.scala:52)
[error] org.specs.specification.Examples.failures(Examples.scala:52)
[error] org.specs.execute.HasResults$class.failureAndErrors(HasResults.scala:61)
[error] org.specs.specification.Examples.failureAndErrors(Examples.scala:52)
[error] org.specs.execute.HasResults$class.isOk(HasResults.scala:69)
[error] org.specs.specification.Examples.isOk(Examples.scala:52)
[error] org.specs.runner.NotifierRunner.reportSystem(NotifierRunner.scala:81)
[error] org.specs.runner.NotifierRunner$$anonfun$reportASpecification$3.apply(NotifierRunner.scala:72)
[error] org.specs.runner.NotifierRunner$$anonfun$reportASpecification$3.apply(NotifierRunner.scala:68)
[error] scala.collection.LinearSeqOptimized$class.foreach(LinearSeqOptimized.scala:59)
[error] scala.collection.immutable.List.foreach(List.scala:76)
[error] org.specs.runner.NotifierRunner.reportASpecification(NotifierRunner.scala:68)
[error] org.specs.runner.NotifierRunner.report(NotifierRunner.scala:58)
[error] org.specs.runner.NotifierRunner.report(NotifierRunner.scala:45)
[error] org.specs.runner.Reporter$class.reportSpecs(Reporter.scala:195)
[error] org.specs.runner.NotifierRunner.reportSpecs(NotifierRunner.scala:45)
[error] org.specs.runner.TestInterfaceRunner$$anonfun$run$3.apply(TestInterfaceRunner.scala:72)
[error] org.specs.runner.TestInterfaceRunner$$anonfun$run$3.apply(TestInterfaceRunner.scala:72)
[error] scala.Option.map(Option.scala:133)
[error] org.specs.runner.TestInterfaceRunner.run(TestInterfaceRunner.scala:72)
[error] org.specs.runner.TestInterfaceRunner.run(TestInterfaceRunner.scala:65)
[error] sbt.TestRunner.delegateRun(TestFramework.scala:57)
[error] sbt.TestRunner.run(TestFramework.scala:51)
[error] sbt.TestRunner.runTest$1(TestFramework.scala:71)
[error] sbt.TestRunner.run(TestFramework.scala:80)
[error] sbt.TestFramework$$anonfun$9$$anonfun$apply$9$$anonfun$10$$anonfun$apply$10.apply(TestFramework.scala:188)
[error] sbt.TestFramework$$anonfun$9$$anonfun$apply$9$$anonfun$10$$anonfun$apply$10.apply(TestFramework.scala:188)
[error] sbt.TestFramework$.sbt$TestFramework$$withContextLoader(TestFramework.scala:200)
[error] sbt.TestFramework$$anonfun$9$$anonfun$apply$9$$anonfun$10.apply(TestFramework.scala:188)
[error] sbt.TestFramework$$anonfun$9$$anonfun$apply$9$$anonfun$10.apply(TestFramework.scala:188)
[error] sbt.Tests$$anonfun$makeSerial$1$$anonfun$apply$8.apply(Tests.scala:121)
[error] sbt.Tests$$anonfun$makeSerial$1$$anonfun$apply$8.apply(Tests.scala:121)
[error] scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:233)
[error] scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:233)
[error] scala.collection.LinearSeqOptimized$class.foreach(LinearSeqOptimized.scala:59)
[error] scala.collection.immutable.List.foreach(List.scala:76)
[error] scala.collection.TraversableLike$class.map(TraversableLike.scala:233)
[error] scala.collection.immutable.List.map(List.scala:76)
[error] sbt.Tests$$anonfun$makeSerial$1.apply(Tests.scala:121)
[error] sbt.Tests$$anonfun$makeSerial$1.apply(Tests.scala:121)
[error] sbt.std.Transform$$anon$3$$anonfun$apply$2.apply(System.scala:47)
[error] sbt.std.Transform$$anon$3$$anonfun$apply$2.apply(System.scala:47)
[error] sbt.std.Transform$$anon$5.work(System.scala:71)
[error] sbt.Execute$$anonfun$submit$1$$anonfun$apply$1.apply(Execute.scala:232)
[error] sbt.Execute$$anonfun$submit$1$$anonfun$apply$1.apply(Execute.scala:232)
[error] sbt.ErrorHandling$.wideConvert(ErrorHandling.scala:18)
[error] sbt.Execute.work(Execute.scala:238)
[error] sbt.Execute$$anonfun$submit$1.apply(Execute.scala:232)
[error] sbt.Execute$$anonfun$submit$1.apply(Execute.scala:232)
[error] sbt.ConcurrentRestrictions$$anon$4$$anonfun$1.apply(ConcurrentRestrictions.scala:160)
[error] sbt.CompletionService$$anon$2.call(CompletionService.scala:30)
[error] java.util.concurrent.FutureTask$Sync.innerRun(FutureTask.java:334)
[error] java.util.concurrent.FutureTask.run(FutureTask.java:166)
[error] java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:471)
[error] java.util.concurrent.FutureTask$Sync.innerRun(FutureTask.java:334)
[error] java.util.concurrent.FutureTask.run(FutureTask.java:166)
[error] java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1110)
[error] java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:603)
[error] java.lang.Thread.run(Thread.java:679)