[SPARK-8319] [CORE] [SQL] Update logic related to key orderings in shuffle dependencies

[JoshRosen]

This patch updates two pieces of logic that are related to handling of keyOrderings in ShuffleDependencies:

• The Tungsten ShuffleManager falls back to regular SortShuffleManager whenever the shuffle dependency specifies a key ordering, but technically we only need to fall back when an aggregator is also specified. This patch updates the fallback logic to reflect this so that the Tungsten optimizations can apply to more workloads.

• The SQL Exchange operator performs defensive copying of shuffle inputs when a key ordering is specified, but this is unnecessary. The copying was added to guard against cases where ExternalSorter would buffer non-serialized records in memory. When ExternalSorter is configured without an aggregator, it uses the following logic to determine whether to buffer records in a serialized or deserialized format:

   private val useSerializedPairBuffer =
      ordering.isEmpty &&
      conf.getBoolean("spark.shuffle.sort.serializeMapOutputs", true) &&
      ser.supportsRelocationOfSerializedObjects

  The newOrdering.isDefined branch in ExternalSorter.needToCopyObjectsBeforeShuffle, removed by this patch, is not necessary:

  • It was checked even if we weren't using sort-based shuffle, but this was unnecessary because only SortShuffleManager performs map-side sorting.
  • Map-side sorting during shuffle writing is only performed for shuffles that perform map-side aggregation as part of the shuffle (to see this, look at how SortShuffleWriter constructs ExternalSorter). Since SQL never pushes aggregation into Spark's shuffle, we can guarantee that both the aggregator and ordering will be empty and Spark SQL always uses serializers that support relocation, so sort-shuffle will use the serialized pair buffer unless the user has explicitly disabled it via the SparkConf feature-flag. Therefore, I think my optimization in Exchange should be safe.

[JoshRosen]

/cc @yhuai for Exchange-related changes.

[yhuai]

Just try to understand the context. Is this if (newOrdering.nonEmpty) part not necessary because of our recent change?

[JoshRosen]

See comment in PR description; even if an ordering was defined I don't think it would be used to sort on the map side because we don't do map side sort in shuffle unless we specify an aggregator

[SparkQA]

Test build #34738 has finished for PR 6773 at commit 85a4628.

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

[JoshRosen]

Jenkins, retest this please.

[SparkQA]

Test build #34751 has finished for PR 6773 at commit 85a4628.

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

[yhuai]

ah, i see. Exchange related part looks good to me.

[JoshRosen]

This patch was pretty tricky to explain, so I've revised the description and have slightly extended the code comments. I think it should be good-to-go after the next Jenkins run, so I'll merge it unless there are any objections.

[JoshRosen]

This check isn't fixing any bugs / correctness issues yet, but I thought that it might guard against hard-to-find future bugs if someone defines a new shuffle manager without updating the logic here. It's extremely unlikely that anyone would do this, but in light of proposals like ParquetShuffleManager it seemed like the safest option.

[SparkQA]

Test build #34830 has finished for PR 6773 at commit 7a14129.

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

[JoshRosen]

Alright, I'm going to merge this into master.