Improve performance of Sort for the common single batch use case

[revans2]

This fixes #10570

The performance speed up is not that huge, but it is there.

I ran

spark.time(spark.range(0, 100000000L, 1, 12).selectExpr("id as oc2", "id DIV 10 as oc3", "CAST(id * 10 AS STRING) as oc", "CAST(id % 4 AS STRING) as pc", "id").selectExpr("*", "row_number() over (PARTITION BY pc ORDER BY oc, oc2, oc3) as rn", "max(id) OVER (PARTITION BY pc ORDER BY oc, oc2, oc3 ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) as m").orderBy(desc("oc")).show())

Both with this patch and without it. I captured the metric for sort time and the total run time of the query.

With this patch, on my desktop, the median run time was 6968 ms, and from the Spark UI the median sort time was 5.5 seconds.
Without this patch the runtime was 7051 and the sort time was 5.9 seconds. That saves 83 ms (about 1% which is not really that huge), but the sort time metric showed about 0.4 seconds of savings or about 6% less time, which is a lot better.

[revans2]

build

[gerashegalov]

If single batch is the common case it does not seem to matter but if we needed to save intermediate Option generation we could use an explicit PartialFunction instance:

Suggested change

	val spillableIter = iter.flatMap { cb =>
	// Filter out empty batches and make them spillable
	if (cb.numRows() > 0) {
	Some(SpillableColumnarBatch(cb, SpillPriorities.ACTIVE_ON_DECK_PRIORITY))
	} else {
	cb.close()
	None
	}
	}
	val spillableIter = iter.collect {
	// Filter out empty batches and make them spillable
	new PartialFunction[ColumnarBatch, SpillableColumnarBatch] {
	override def isDefinedAt(cb: ColumnarBatch): Boolean = if (cb.numRows() > 0) {
	true
	} else {
	cb.close()
	false
	}
	override def apply(cb: ColumnarBatch): SpillableColumnarBatch =
	SpillableColumnarBatch(cb, SpillPriorities.ACTIVE_ON_DECK_PRIORITY)
	}
	}

[revans2]

I'll keep it in mind, but I'm not sure it matters that much here.

[gerashegalov]

What if this throws?

[revans2]

When someone calls next the exception is thrown and they would need to handle it.

https://github.com/scala/scala/blob/1bdf362b0d8344927460116380b67fcaa766435b/src/library/scala/collection/Iterator.scala#L587-L620

[gerashegalov]

nit: space

Suggested change

	while(alreadySortedIter.hasNext) {
	while (alreadySortedIter.hasNext) {

[revans2]

build