[SPARK-46937][SQL] Improve concurrency performance for FunctionRegistry #44976

beliefer · 2024-02-01T05:52:05Z

What changes were proposed in this pull request?

This PR propose to improve concurrency performance for FunctionRegistry.

Why are the changes needed?

Currently, SimpleFunctionRegistryBase adopted the mutable.Map caching function infos. The SimpleFunctionRegistryBase guarded by this so as ensure security under multithreading.
Because all the mutable state are related to functionBuilders, we can delegate security to ConcurrentHashMap.
ConcurrentHashMap has higher concurrency activity and responsiveness.
After this change, FunctionRegistry have better perf than before.

Does this PR introduce any user-facing change?

'No'.

How was this patch tested?

GA.
The benchmark test.

object FunctionRegistryBenchmark extends BenchmarkBase {

  override def runBenchmarkSuite(mainArgs: Array[String]): Unit = {
    runBenchmark("FunctionRegistry") {
      val iters = 1000000
      val threadNum = 4
      val functionRegistry = FunctionRegistry.builtin
      val names = FunctionRegistry.expressions.keys.toSeq
      val barrier = new CyclicBarrier(threadNum + 1)
      val threadPool = ThreadUtils.newDaemonFixedThreadPool(threadNum, "test-function-registry")
      val benchmark = new Benchmark("SimpleFunctionRegistry", iters, output = output)

      benchmark.addCase("only read") { _ =>
        for (_ <- 1 to threadNum) {
          threadPool.execute(new Runnable {
            val random = new Random()
            override def run(): Unit = {
              barrier.await()
              for (_ <- 1 to iters) {
                val name = names(random.nextInt(names.size))
                val fun = functionRegistry.lookupFunction(new FunctionIdentifier(name))
                assert(fun.map(_.getName).get == name)
                functionRegistry.listFunction()
              }
              barrier.await()
            }
          })
        }
        barrier.await()
        barrier.await()
      }

      benchmark.run()
    }
  }
}

The benchmark output before this PR.

Java HotSpot(TM) 64-Bit Server VM 17.0.9+11-LTS-201 on Mac OS X 10.14.6
Intel(R) Core(TM) i5-5350U CPU @ 1.80GHz
SimpleFunctionRegistry:                   Best Time(ms)   Avg Time(ms)   Stdev(ms)    Rate(M/s)   Per Row(ns)   Relative
------------------------------------------------------------------------------------------------------------------------
only read                                         54858          55043         261          0.0       54858.1       1.0X

The benchmark output after this PR.

Java HotSpot(TM) 64-Bit Server VM 17.0.9+11-LTS-201 on Mac OS X 10.14.6
Intel(R) Core(TM) i5-5350U CPU @ 1.80GHz
SimpleFunctionRegistry:                   Best Time(ms)   Avg Time(ms)   Stdev(ms)    Rate(M/s)   Per Row(ns)   Relative
------------------------------------------------------------------------------------------------------------------------
only read                                         20202          20264          88          0.0       20202.1       1.0X

Was this patch authored or co-authored using generative AI tooling?

'No'.

beliefer · 2024-02-01T13:03:36Z

ping @cloud-fan cc @MaxGekk @viirya

cloud-fan · 2024-02-01T15:10:18Z

What's the level of concurrency you expect for function registration/lookup? Do you have perf numbers?

beliefer · 2024-02-03T03:07:52Z

What's the level of concurrency you expect for function registration/lookup? Do you have perf numbers?

The perf data has been added into PR description.

dongjoon-hyun

Could you rebase this to the master branch once more, @beliefer ?

This approach (and the benchmark result) looks reasonable to me for Apache Spark 4.0.0.

Do you still have some concern, @cloud-fan ?

dongjoon-hyun · 2024-05-10T15:28:04Z

Could you resolve the conflicts, @beliefer ?

dongjoon-hyun · 2024-05-10T16:31:44Z

I resolved the conflicts for you~

dongjoon-hyun

+1, LGTM (Pending CIs).

Also, cc @cloud-fan once more because he has a concern last time.

beliefer · 2024-05-11T02:15:36Z

Let me rebase again.

dongjoon-hyun · 2024-05-11T02:35:05Z

Thank you, @beliefer

cloud-fan

interesting, even with read-only cases, this change also helps. LGTM then.

beliefer · 2024-06-04T08:27:04Z

Merged into master.
Thank you @dongjoon-hyun @cloud-fan @viirya

…cement ### What changes were proposed in this pull request? A followup of #44976 . `ConcurrentHashMap#put` has a different semantic than the scala map, and it returns null if the key is new. We should update the checking code accordingly. ### Why are the changes needed? avoid wrong warning messages ### Does this PR introduce _any_ user-facing change? no ### How was this patch tested? manual ### Was this patch authored or co-authored using generative AI tooling? no Closes #46876 from cloud-fan/log. Authored-by: Wenchen Fan <[email protected]> Signed-off-by: Kent Yao <[email protected]>

cloud-fan · 2024-06-07T00:07:14Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/FunctionRegistry.scala

 class SimpleTableFunctionRegistry extends SimpleFunctionRegistryBase[LogicalPlan]
    with TableFunctionRegistry {

  override def clone(): SimpleTableFunctionRegistry = synchronized {


I think there is a problem here. We don't synchronize the write path, how can we safely clone the ConcurrentHashMap?

We can introduce a ReentrantReadWriteLock guarding the ConcurrentHashMap.
clone and clear would take the write lock on the ConcurrentHashMap.
The other methods take read lock on the map.

It may work, but makes the code complicated. We should only do it if this does make a difference to real-world workloads.

In fact, It's safe here.
Because the entrySet() of ConcurrentHashMap is thread safe. We don't need synchronized.

cloud-fan · 2024-06-07T00:11:03Z

Shall we revert this if #44976 (comment) is a real issue? I don't think this is a critical path for performance (how much parallelism do you expect for function lookups in a Spark session?), and synchronizing this seems simpler and good enough.

yaooqinn · 2024-06-07T08:54:53Z

+1 for #44976 (comment)

cloud-fan · 2024-06-11T17:48:40Z

I've sent out the revert PR: #46940

…ctionRegistry" ### What changes were proposed in this pull request? Reverts #44976 as it breaks thread-safety ### Why are the changes needed? Fix thread-safety ### Does this PR introduce _any_ user-facing change? no ### How was this patch tested? N/A ### Was this patch authored or co-authored using generative AI tooling? no Closes #46940 from cloud-fan/revert. Authored-by: Wenchen Fan <[email protected]> Signed-off-by: Wenchen Fan <[email protected]>

github-actions bot added the SQL label Feb 1, 2024

beliefer changed the title ~~[WIP][SPARK-46937][SQL] Improve concurrency performance for FunctionRegistry~~ [SPARK-46937][SQL] Improve concurrency performance for FunctionRegistry Feb 1, 2024

beliefer requested a review from MaxGekk February 1, 2024 13:03

dongjoon-hyun reviewed May 10, 2024

View reviewed changes

dongjoon-hyun approved these changes May 10, 2024

View reviewed changes

[SPARK-46937][SQL] Improve concurrency performance for FunctionRegistry

975d8d5

beliefer force-pushed the SPARK-46937 branch from b7ad286 to 975d8d5 Compare May 11, 2024 03:16

cloud-fan approved these changes Jun 3, 2024

View reviewed changes

viirya approved these changes Jun 3, 2024

View reviewed changes

beliefer closed this in 8cebb9b Jun 4, 2024

cloud-fan mentioned this pull request Jun 5, 2024

[SPARK-46937][SQL][FOLLOWUP] Properly check registered function replacement #46876

Closed

cloud-fan reviewed Jun 7, 2024

View reviewed changes

cloud-fan mentioned this pull request Jun 11, 2024

Revert "[SPARK-46937][SQL] Improve concurrency performance for FunctionRegistry" #46940

Closed

[SPARK-46937][SQL] Improve concurrency performance for FunctionRegistry #44976

[SPARK-46937][SQL] Improve concurrency performance for FunctionRegistry #44976

Uh oh!

Conversation

beliefer commented Feb 1, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

Was this patch authored or co-authored using generative AI tooling?

Uh oh!

beliefer commented Feb 1, 2024

Uh oh!

cloud-fan commented Feb 1, 2024

Uh oh!

beliefer commented Feb 3, 2024

Uh oh!

dongjoon-hyun left a comment

Choose a reason for hiding this comment

Uh oh!

dongjoon-hyun commented May 10, 2024

Uh oh!

dongjoon-hyun commented May 10, 2024

Uh oh!

dongjoon-hyun left a comment

Choose a reason for hiding this comment

Uh oh!

beliefer commented May 11, 2024

Uh oh!

dongjoon-hyun commented May 11, 2024

Uh oh!

cloud-fan left a comment

Choose a reason for hiding this comment

Uh oh!

beliefer commented Jun 4, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

cloud-fan Jun 7, 2024

Choose a reason for hiding this comment

Uh oh!

tedyu Jun 13, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

cloud-fan Jun 13, 2024

Choose a reason for hiding this comment

Uh oh!

beliefer Jun 25, 2024

Choose a reason for hiding this comment

Uh oh!

cloud-fan commented Jun 7, 2024

Uh oh!

yaooqinn commented Jun 7, 2024

Uh oh!

cloud-fan commented Jun 11, 2024

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

6 participants

beliefer commented Feb 1, 2024 •

edited

Loading

beliefer commented Jun 4, 2024 •

edited

Loading

tedyu Jun 13, 2024 •

edited

Loading