[SPARK-20920][SQL] ForkJoinPool pools are leaked when writing hive tables with many partitions

[srowen]

What changes were proposed in this pull request?

Don't leave thread pool running from AlterTableRecoverPartitionsCommand DDL command

How was this patch tested?

Existing tests.

[srowen]

@davies could you help me review as you wrote the original code?
CC @viirya because it's similar to #18100

[SparkQA]

Test build #77773 has finished for PR 18216 at commit 27a76c1.

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

[SparkQA]

Test build #77774 has finished for PR 18216 at commit ca6d3cd.

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

[SparkQA]

Test build #3779 has finished for PR 18216 at commit ca6d3cd.

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

[viirya]

The change looks good. Shall we add a test for it?

[viirya]

Ah, I remember that it is a bit hard to automatically test it.

[viirya]

The fix looks straightforward as previous fix. LGTM. Should be better if we can manually verify it like before. cc @cloud-fan

[cloud-fan]

LGTM, can we add a manual test like #18100 ?

[srowen]

@viirya / @cloud-fan what do you mean here by a manual test?

[viirya]

@srowen In #18073, I ran the code snippet which can leak the pools, and checked the number of threads before/after the fix by using some tools like jconsole.

[srowen]

OK, I tried this as a test:

import scala.collection.JavaConverters._
spark.sql("DROP TABLE IF EXISTS my_tab")
spark.sql("CREATE TABLE my_tab(a INT, b STRING, c INT, d String) USING parquet PARTITIONED BY (a, b, c)")
spark.sql("INSERT INTO TABLE my_tab VALUES (1, 'foo', 4, 'bing')")
spark.sql("INSERT INTO TABLE my_tab VALUES (2, 'bar', 5, 'bing')")
spark.sql("INSERT INTO TABLE my_tab VALUES (3, 'baz', 6, 'bing')")
Thread.getAllStackTraces.keySet().asScala.map(_.getName).filter(_.contains("ForkJoinPool"))
res5: scala.collection.mutable.Set[String] = Set(ForkJoinPool-1-worker-13)
spark.sql("ALTER TABLE my_tab RECOVER PARTITIONS")
Thread.getAllStackTraces.keySet().asScala.map(_.getName).filter(_.contains("ForkJoinPool"))
res10: scala.collection.mutable.Set[String] = Set(ForkJoinPool-1-worker-13, ForkJoinPool-1-worker-9, ForkJoinPool-2-worker-13)
spark.sql("ALTER TABLE my_tab RECOVER PARTITIONS")
spark.sql("ALTER TABLE my_tab RECOVER PARTITIONS")
spark.sql("ALTER TABLE my_tab RECOVER PARTITIONS")
res23: scala.collection.mutable.Set[String] = Set(ForkJoinPool-5-worker-13, ForkJoinPool-3-worker-13, ForkJoinPool-2-worker-13, ForkJoinPool-1-worker-11, ForkJoinPool-4-worker-13, ForkJoinPool-1-worker-3)

Before the change you can see that it does look like one thread is left from each of several ForkJoinPools.

After the change, the result is Set(ForkJoinPool-1-worker-13) every time. I think that does suggest this fixes the issue.

[viirya]

Thanks @srowen. I think it looks good.

[srowen]

Merged to master/2.2/2.1