[SPARK-9054] [SQL] Rename RowOrdering to InterpretedOrdering; use newOrdering in more places

[JoshRosen]

This patch renames RowOrdering to InterpretedOrdering and updates a few operators to use SparkPlan.newOrdering instead of manually constructing a RowOrdering.

[JoshRosen]

/cc @yhuai @rxin

[rxin]

hm since copy is a verb, it is better to keep the () i think. also we defined copy with ()

[JoshRosen]

I didn't mean to make this change in this PR, will back out this one line.

[rxin]

LGTM other than that nit.

[yhuai]

LGTM

[JoshRosen]

Ah, looks like I may have to move some calls so that we don't try to serialize the generated ordering:

[info] - sort followed by limit *** FAILED *** (97 milliseconds)
[info]   Exception thrown while executing Spark plan to calculate expected answer:
[info]    Limit 10
[info]    !Sort ['a ASC], true
[info]     LocalTableScan [a#457], [[1],[2],[3],[4],[5],[6],[7],[8],[9],[10],[11],[12],[13],[14],[15],[16],[17],[18],[19],[20],[21],[22],[23],[24],[25],[26],[27],[28],[29],[30],[31],[32],[33],[34],[35],[36],[37],[38],[39],[40],[41],[42],[43],[44],[45],[46],[47],[48],[49],[50],[51],[52],[53],[54],[55],[56],[57],[58],[59],[60],[61],[62],[63],[64],[65],[66],[67],[68],[69],[70],[71],[72],[73],[74],[75],[76],[77],[78],[79],[80],[81],[82],[83],[84],[85],[86],[87],[88],[89],[90],[91],[92],[93],[94],[95],[96],[97],[98],[99],[100]]
[info]   
[info]    == Exception ==
[info]    org.apache.spark.SparkException: Job aborted due to stage failure: Task not serializable: java.io.NotSerializableException: SC
[info]      - field (class "SC$SpecificOrdering", name: "this$0", type: "class SC")
[info]      - object (class "SC$SpecificOrdering", SC$SpecificOrdering@1edd238a)
[info]      - field (class "org.apache.spark.util.CollectionsUtils$$anonfun$makeBinarySearch$8", name: "comparator$1", type: "interface java.util.Comparator")
[info]      - object (class "org.apache.spark.util.CollectionsUtils$$anonfun$makeBinarySearch$8", <function2>)
[info]      - field (class "org.apache.spark.RangePartitioner", name: "org$apache$spark$RangePartitioner$$binarySearch", type: "interface scala.Function2")
[info]      - custom writeObject data (class "org.apache.spark.RangePartitioner")
[info]      - object (class "org.apache.spark.RangePartitioner", org.apache.spark.RangePartitioner@50cf16c6)
[info]      - field (class "org.apache.spark.ShuffleDependency", name: "partitioner", type: "class org.apache.spark.Partitioner")
[info]      - object (class "org.apache.spark.ShuffleDependency", org.apache.spark.ShuffleDependency@5d3944b0)
[info]      - field (class "scala.Tuple2", name: "_2", type: "class java.lang.Object")
[info]      - root object (class "scala.Tuple2", (MapPartitionsRDD[424] at apply at Transformer.scala:22,org.apache.spark.ShuffleDependency@5d3944b0))
[info]    org.apache.spark.SparkException: Job aborted due to stage failure: Task not serializable: java.io.NotSerializableException: SC
[info]      - field (class "SC$SpecificOrdering", name: "this$0", type: "class SC")
[info]      - object (class "SC$SpecificOrdering", SC$SpecificOrdering@1edd238a)
[info]      - field (class "org.apache.spark.util.CollectionsUtils$$anonfun$makeBinarySearch$8", name: "comparator$1", type: "interface java.util.Comparator")

[JoshRosen]

Aha, I see the problem: we're trying to serialize the GeneratedOrdering when passing it into the RangePartitioner. I'll see if I can come up with a workaround for this.

[JoshRosen]

I came up with a workaround which allows GeneratedOrdering to be used as the ordering when computing bounds in the RangePartitioner. Using a generated ordering might not be a huge win here, so I'd be okay with backing out of this change if someone thinks I should.

[SparkQA]

Test build #37298 timed out for PR 7408 at commit 5eb7222 after a configured wait of 175m.

[SparkQA]

Test build #37311 has finished for PR 7408 at commit 8060594.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[JoshRosen]

Ah, looks like we have the same problem in TakeOrderedAndProject: we have to pass the generated ordering into Spark Core code which accepts the ordering on the driver rather than a closure to create the ordering.

[JoshRosen]

Alright, I've updated this to add a similar wrapper in TakeOderedAndProject. PTAL and let me know if this looks okay.

[SparkQA]

Test build #37672 has finished for PR 7408 at commit 15b7d92.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
  • class NaiveBayes(override val uid: String)
  • class PCA(JavaEstimator, HasInputCol, HasOutputCol):
  • class PCAModel(JavaModel):

[cloud-fan]

@JoshRosen how about make the generated Ordering serializable? We can make the generated class implements Serializable by evaluator.setImplementedInterfaces(Array(classOf[Serializable])) at here.
cc @davies , does this work?

[davies]

@cloud-fan I think this should not work, because the generated class only exists in the JVM that generate it.

[davies]

Or we can always return an wrapper for generated ordering, not sure what's the overhead of the wrapper.

[JoshRosen]

I'm going to close this for now; this doesn't seem like a big win and is actually kind of tricky to implement.

[JoshRosen]

I'm going to revive this since I've found a few problems caused by our use of RowOrdering in a few operators.

[JoshRosen]

This is so conflicted that I'm just going to start over and will force-push; for posterity, the old version's HEAD is 15b7d92

[JoshRosen]

Hmm, GitHub won't let me re-open after a force-push. I'll make a new PR.