[SPARK-9860][SQL] Join: Determine the join strategy (broadcast join or shuffle join) at runtime

[lianhuiwang]

What changes were proposed in this pull request?

1. add query fragment for execute stage according to exchange.
2. before running no leaf query fragment, optimize plan using mapOutputStatistics of its child fragments.
   if size of one exchange is less than autoBroadcastJoinThreshold, it transforms sort merge join to broadcast join.
   I will support optimization for skew join in another PR.

How was this patch tested?

unit tests

[SparkQA]

Test build #58239 has finished for PR 13028 at commit 73de7b2.

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

[SparkQA]

Test build #58240 has finished for PR 13028 at commit 6cc187e.

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

[SparkQA]

Test build #58241 has finished for PR 13028 at commit 6f1105c.

• This patch fails Scala style tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #58376 has finished for PR 13028 at commit 5de46b0.

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

[SparkQA]

Test build #58378 has finished for PR 13028 at commit a18be55.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
  • class QueryFragmentSuite extends QueryTest with SQLTestUtils with SharedSQLContext

[lianhuiwang]

cc @rxin @davies

[SparkQA]

Test build #58481 has finished for PR 13028 at commit d9c66d7.

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

[sameeragarwal]

@lianhuiwang could this regress performance for existing queries? Can you please share some benchmarks with us (if any)?

[gatorsmile]

We are closing it due to inactivity. please do reopen if you want to push it forward. Thanks!

[meteorchenwu]

It is great for BI scene. Why not continue?