[SPARK-27778][PYTHON] Fix toPandas conversion of empty DataFrame with Arrow enabled

[dvogelbacher]

What changes were proposed in this pull request?

#22275 introduced a performance improvement where we send partitions out of order to python and then, as a last step, send the partition order as well.
However, if there are no partitions we will never send the partition order and we will get an "EofError" on the python side.
This PR fixes this by also sending the partition order if there are no partitions present.

How was this patch tested?

New unit test added.

[dvogelbacher]

@BryanCutler can you take a look at this one

[BryanCutler]

Thanks for catching this @dvogelbacher !

[BryanCutler]

ok to test

[SparkQA]

Test build #105580 has finished for PR 24650 at commit d635a74.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[HyukjinKwon]

This method is well-commented. Can you add another comment that we should end stream when partitions are empty?

Also, I would do:

partitions = 0 until numPartitions
sparkSession.sparkContext.runJob(
  arrowBatchRdd,
  (ctx: TaskContext, it: Iterator[Array[Byte]]) => it.toArray,
  partitions,
  handlePartitionBatches)

if (partitions.isEmpty) {
  // Currently result handler is not called when given partitions are empty.
  // Therefore, we should end stream here.
  doAfterLastPartition()
}

[HyukjinKwon]

Looks fine given skimming the codes.

[SparkQA]

Test build #105588 has finished for PR 24650 at commit 52a51bf.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[BryanCutler]

I had another thought about this, the stuff in doAfterLastPartition could be removed from handlePartitionBatches and called after runJob regardless if partitions are empty or not. This really wouldn't make any difference performance-wise because it's just moving it outside the callback function and Python is waiting on it anyway.

It also has the benefit that the number of partitions would not have to be kept track of, so the variables partitionCount and numPartitions could be removed. It would be a lot clearer then too.

What do you think @dvogelbacher and @HyukjinKwon ?

[HyukjinKwon]

Yea, SGTM.

[dvogelbacher]

yes, that's a good idea @BryanCutler, it is much clearer. I've made the change.

[SparkQA]

Test build #105619 has finished for PR 24650 at commit 9f4bc3e.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[BryanCutler]

Looks pretty good now, just a couple more minor things that could be done to clean it up a bit more if you wouldn't mind.

[dvogelbacher]

of course, I addressed the comments @BryanCutler

[HyukjinKwon]

Yea, this way looks better. Looks good to me too

[SparkQA]

Test build #105647 has finished for PR 24650 at commit db6f4b1.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[HyukjinKwon]

Merged to master.