[SPARK-33935][SQL] Fix CBO cost function

[tanelk]

What changes were proposed in this pull request?

Changed the cost function in CBO to match documentation.

Why are the changes needed?

The parameter spark.sql.cbo.joinReorder.card.weight is documented as:

The weight of cardinality (number of rows) for plan cost comparison in join reorder: rows * weight + size * (1 - weight).

The implementation in JoinReorderDP.betterThan does not match this documentaiton:

def betterThan(other: JoinPlan, conf: SQLConf): Boolean = {
      if (other.planCost.card == 0 || other.planCost.size == 0) {
        false
      } else {
        val relativeRows = BigDecimal(this.planCost.card) / BigDecimal(other.planCost.card)
        val relativeSize = BigDecimal(this.planCost.size) / BigDecimal(other.planCost.size)
        relativeRows * conf.joinReorderCardWeight +
          relativeSize * (1 - conf.joinReorderCardWeight) < 1
      }
    }

This different implementation has an unfortunate consequence:
given two plans A and B, both A betterThan B and B betterThan A might give the same results. This happes when one has many rows with small sizes and other has few rows with large sizes.

A example values, that have this fenomen with the default weight value (0.7):
A.card = 500, B.card = 300
A.size = 30, B.size = 80
Both A betterThan B and B betterThan A would have score above 1 and would return false.

This happens with several of the TPCDS queries.

The new implementation does not have this behavior.

Does this PR introduce any user-facing change?

No

How was this patch tested?

New and existing UTs

[tanelk]

A bit more information:

The unstability of CBO has been noted before (#29638) and I think, that this is the main reason for this. There is also #29871, that tackles another reason for the unstability. That one should only impact, when the costs are equal, this here can impact more plans (see the example in the description)

An important thing to note is that this could change the the behavior of the spark.sql.cbo.joinReorder.card.weight config value, but luckily it seems, that it does so minimally.
I generated random values for the plan row counts and sizes, and found that the new cost function agrees most with the old cost function at the same weight value. This holds true for all the weight value, not only the default (0.7).

[tanelk]

@LuciferYang and @maropu
Could you take a look at this or perhaps point this to someone, who has worked more with the CBO.

[SparkQA]

Test build #133502 has finished for PR 30965 at commit 2308a8a.

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

[SparkQA]

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/38091/

[SparkQA]

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/38091/

[maropu]

cc: @wzhfy

[maropu]

I've checked the update and it seems fine. Could you add some tests based on the example in the PR description?

[tanelk]

I've checked the update and it seems fine. Could you add some tests based on the example in the PR description?

Ahh, good point. Adding this I also realised, that I had the inequality check in the wrong order - this greatly reduced the amount of plan changes.

[SparkQA]

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/38129/

[SparkQA]

Test build #133539 has finished for PR 30965 at commit 4b05711.

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

[SparkQA]

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/38129/

[SparkQA]

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/38130/

[SparkQA]

Kubernetes integration test status failure
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/38130/

[SparkQA]

Test build #133534 has finished for PR 30965 at commit 6dbb9fe.

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

[SparkQA]

Test build #133536 has finished for PR 30965 at commit 24f65b5.

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

[maropu]

Thanks for adding this. This fix looks fine to me. cc: @cloud-fan @HyukjinKwon

[cloud-fan]

retest this please

[LuciferYang]

@tanelk Can you help to run UTs in sql/catalyst and sql/core in Scala 2.13 with this pr manually ? thx ~

[SparkQA]

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/38210/

[SparkQA]

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/38210/

[SparkQA]

Test build #133621 has finished for PR 30965 at commit 4b05711.

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

[maropu]

Thanks! Merged to master/3.1.

[maropu]

@tanelk Can you help to run UTs in sql/catalyst and sql/core in Scala 2.13 with this pr manually ? thx ~

Could you check this? Have this fix resolved the previous issue, too?

[maropu]

@tanelk Could you open a PR to fix it for branch-3.0/2.4?

[tanelk]

I created the pulls for 3.0 and 2.4: #31042 and #31043

[tanelk]

@tanelk Can you help to run UTs in sql/catalyst and sql/core in Scala 2.13 with this pr manually ? thx ~

The sql/catalyst passed.
With sql/core I had some issues:
at random tests it would fail with the following exception:

*** RUN ABORTED ***
  java.net.BindException: Cannot assign requested address: Service 'sparkDriver' failed after 100 retries (on a random free port)!

I assume you want the results of *PlanStability*Suite - these all passed.

[maropu]

The sql/catalyst passed.
I assume you want the results of PlanStabilitySuite - these all passed.

Awesome!

[wzhfy]

@tanelk Hi, sorry to see this so late.

IIRC the reason to use a relative value for rowCount and size, is to normalize them to a similar scale while comparing cost. Otherwise, one (size) may overwhelm the other (rowCount), then the weight and cost function become meaningless.

To resolve the stability issue, can we have a betterPlan() function instead of current betterThan()? Inside that plan, we can fix the comparison order based on rowCount or size. And at caller side, we can use !existingPlan.get.eq(betterPlan(existingPlan.get, newJoinPlan, conf)) to decide whether to update the best plan so far.

    def betterPlan(existing: JoinPlan, newPlan: JoinPlan, conf: SQLConf): JoinPlan = {
      // To fix the comparison order, set the one with smaller cardinality as the baseline.
      val (baseline, toCompare) = if (existing.planCost.card <= newPlan.planCost.card) {
        (existing, newPlan)
      } else {
        (newPlan, existing)
      }

      if (toCompare.planCost.card == 0 || toCompare.planCost.size == 0) {
        return existing
      }

      val relativeRows = BigDecimal(baseline.planCost.card) / BigDecimal(toCompare.planCost.card)
      val relativeSize = BigDecimal(baseline.planCost.size) / BigDecimal(toCompare.planCost.size)
      val relativeCost = relativeRows * conf.joinReorderCardWeight +
        relativeSize * (1 - conf.joinReorderCardWeight)
      if (relativeCost == 1) {
        // If they have same cost, we don't update the best plan and return the existing one.
        existing
      } else if (relativeCost < 1) {
        baseline
      } else {
        toCompare
      }
    }

What do you think?

[cloud-fan]

I think what @wzhfy said makes sense, @tanelk do you have time to try this idea?

[tanelk]

@wzhfy and @cloud-fan

I'm not a fan of adding up the relative costs.

A simple example, where the weight is 0.5:
If this plans size (bytes) is 2x larger, then no matter how many times more rows does the other plan have, the other plan will allways be considered to be better - 0.5*2 + 0.5*0.00000000000001 > 1.
This basically the same situation, where one cost overwhelms the other.

Perhaps this would be a best of both worlds:
(this.card / other.card) ^ cardWeight * (this.size / other.size) ^ (1 - cardWeight) < 1.
In short - multiply the relative costs instead of adding them.

[cloud-fan]

I think q19 exposes a problem. Previously this BroadcastHashJoin is run before the SortMergeJoin, which reduces the input data of shuffle, because this BroadcastHashJoin has a filter on the right side and likely makes this join very selective.

@tanelk , if the idea from @wzhfy doesn't look good to you, can you try with some other ideas and see if we can fix this issue?

[tanelk]

I'll experiment with it a bit, but it might take a while.

[cloud-fan]

Thanks for looking into it!

[HyukjinKwon]

This actually caused a significant regression at q19 in TPC-DS benchmark (performed internally). Can we just revert it for now, and do it with actual performance numbers? I think that's safer and easier for everybody here. Seems like at least the plan change here was overlooked and the performance of this had to be clarified.

[maropu]

branch-2.4 has the same regression? This PR was back-ported into branch-2.4, too. cc: @viirya

[viirya]

+1 for @HyukjinKwon's suggestion. Especially for branch-2.4, I think it is better to avoid unexpected performance change.

[HyukjinKwon]

I discussed with @cloud-fan offline. Shall we merge #32014 into master, branch-3.1 and branch-3.0 to fix the regression, and revert this one from branch-2.4? Spark 2.4 release is very soon and might be best to stay safe, and technically this was more like an improvement.

[viirya]

Agree with it. I prefer to make stable for branch-2.4.

[cloud-fan]

yea let's revert it from 2.4 to be safe

[viirya]

Revert at #32020