You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@spark.apache.org by "Ke Jia (Jira)" <ji...@apache.org> on 2026/04/15 13:07:00 UTC
[jira] [Updated] (SPARK-56485) Incorrect RowCount Estimation in CBO leads to unintended BroadcastHashJoin
[ https://issues.apache.org/jira/browse/SPARK-56485?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
Ke Jia updated SPARK-56485:
---------------------------
Description:
When running TPC-DS Q4 on the database 1TB with CBO enabled and {{}}
{code:java}
spark.sql.autoBroadcastJoinThreshold{code}
{{ set to }}
{code:java}
10MB{code}
{{{}{}}}, the query fails with a SparkException.
{code:java}
Py4JJavaError: An error occurred while calling o596.collectToPython.
: org.apache.spark.SparkException: Cannot broadcast the table that is larger than 8.0 GiB: 10.7 GiB
at org.apache.gluten.backendsapi.velox.VeloxSparkPlanExecApi.createBroadcastRelation(VeloxSparkPlanExecApi.scala:850)
at org.apache.spark.sql.execution.ColumnarBroadcastExchangeExec.$anonfun$relationFuture$2(ColumnarBroadcastExchangeExec.scala:79)
at org.apache.gluten.utils.Arm$.withResource(Arm.scala:25)
at org.apache.gluten.metrics.GlutenTimeMetric$.millis(GlutenTimeMetric.scala:37)
at org.apache.spark.sql.execution.ColumnarBroadcastExchangeExec.$anonfun$relationFuture$1(ColumnarBroadcastExchangeExec.scala:66)
at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withThreadLocalCaptured$2(SQLExecution.scala:230)
at org.apache.spark.JobArtifactSet$.withActiveJobArtifactState(JobArtifactSet.scala:94)
at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withThreadLocalCaptured$1(SQLExecution.scala:225)
at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
at java.base/java.lang.Thread.run(Thread.java:840){code}
The error indicates that Spark attempted to broadcast a table significantly larger than the 8GB limit (actual size ~10.7 GiB). Interestingly, the query runs successfully if CBO is disabled.
The issue stems from a bug in Spark's CBO Filter Estimation. When processing equality filters (e.g., d_year = 2001), the CBO incorrectly estimates the rowCount as 0.
Because the estimated row count is 0, the optimizer concludes that the join result will be very small and chooses a BroadcastHashJoin . However, the actual data size is roughly 10.7 GiB, which exceeds Spark's hard limit for broadcasting.
was:
When running TPC-DS Q4 on the database {{wxd_icebergtpcdsdb1000g}} with CBO enabled and {{spark.sql.autoBroadcastJoinThreshold }}set to {{{}10MB{}}}, the query fails with a SparkException.
{{Py4JJavaError: An error occurred while calling o596.collectToPython.
: org.apache.spark.SparkException: Cannot broadcast the table that is larger than 8.0 GiB: 10.7 GiB
at org.apache.gluten.backendsapi.velox.VeloxSparkPlanExecApi.createBroadcastRelation(VeloxSparkPlanExecApi.scala:850)
at org.apache.spark.sql.execution.ColumnarBroadcastExchangeExec.$anonfun$relationFuture$2(ColumnarBroadcastExchangeExec.scala:79)
at org.apache.gluten.utils.Arm$.withResource(Arm.scala:25)
at org.apache.gluten.metrics.GlutenTimeMetric$.millis(GlutenTimeMetric.scala:37)
at org.apache.spark.sql.execution.ColumnarBroadcastExchangeExec.$anonfun$relationFuture$1(ColumnarBroadcastExchangeExec.scala:66)
at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withThreadLocalCaptured$2(SQLExecution.scala:230)
at org.apache.spark.JobArtifactSet$.withActiveJobArtifactState(JobArtifactSet.scala:94)
at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withThreadLocalCaptured$1(SQLExecution.scala:225)
at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
at java.base/java.lang.Thread.run(Thread.java:840)}}
The error indicates that Spark attempted to broadcast a table significantly larger than the 8GB limit (actual size ~10.7 GiB). Interestingly, the query runs successfully if CBO is disabled.
The issue stems from a bug in Spark's CBO Filter Estimation. When processing equality filters (e.g., d_year = 2001), the CBO incorrectly estimates the rowCount as 0.
Because the estimated row count is 0, the optimizer concludes that the join result will be very small and chooses a BroadcastHashJoin . However, the actual data size is roughly 10.7 GiB, which exceeds Spark's hard limit for broadcasting.
The root of this miscalculation is in [FilterEstimation.scala|https://www.google.com/search?q=%5Bhttps://github.com/apache/spark/blob/15ffa544ca53cd9f8a25baaf11fa0171dac7c85f/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/statsEstimation/FilterEstimation.scala%23L347%5D(https://github.com/apache/spark/blob/15ffa544ca53cd9f8a25baaf11fa0171dac7c85f/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/statsEstimation/FilterEstimation.scala%23L347)]. Under normal circumstances, Spark should use column statistics to estimate selectivity. In this case, although column information is present, most of the statistical fields (min, max, distinct count, etc.) are null/None, except for versioning information.
{{Filter (isnotnull(d_year#7208) AND (d_year#7208 = 2001)), Statistics(sizeInBytes=1.0 B, rowCount=0)
+- RelationV2[d_date_sk#7202, d_date_id#7203, d_date#7204, d_month_seq#7205, d_week_seq#7206, d_quarter_seq#7207, d_year#7208, d_dow#7209, d_moy#7210, d_dom#7211, d_qoy#7212, d_fy_year#7213, d_fy_quarter_seq#7214, d_fy_week_seq#7215, d_day_name#7216, d_quarter_name#7217, d_holiday#7218, d_weekend#7219, d_following_holiday#7220, d_first_dom#7221, d_last_dom#7222, d_same_day_ly#7223, d_same_day_lq#7224, d_current_day#7225, ... 4 more fields] spark_catalog.wxd_icebergtpcdsdb1000g.date_dim, Statistics(sizeInBytes=20.1 MiB, rowCount=7.30E+4)}}
As a result, Spark fails to enter the correct evaluation logic in [lines 310-313|https://www.google.com/search?q=https://github.com/apache/spark/blob/15ffa544ca53cd9f8a25baaf11fa0171dac7c85f/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/statsEstimation/FilterEstimation.scala%23L310-L313] and defaults to an incorrect estimation.
{{// Example of the empty stats being passed:
d_year#20690 -> ColumnStat(None,None,None,None,None,None,None,2)}}
> Incorrect RowCount Estimation in CBO leads to unintended BroadcastHashJoin
> --------------------------------------------------------------------------
>
> Key: SPARK-56485
> URL: https://issues.apache.org/jira/browse/SPARK-56485
> Project: Spark
> Issue Type: Bug
> Components: SQL
> Affects Versions: 4.1.1
> Reporter: Ke Jia
> Priority: Major
>
> When running TPC-DS Q4 on the database 1TB with CBO enabled and {{}}
> {code:java}
> spark.sql.autoBroadcastJoinThreshold{code}
> {{ set to }}
> {code:java}
> 10MB{code}
> {{{}{}}}, the query fails with a SparkException.
>
> {code:java}
> Py4JJavaError: An error occurred while calling o596.collectToPython.
> : org.apache.spark.SparkException: Cannot broadcast the table that is larger than 8.0 GiB: 10.7 GiB
> at org.apache.gluten.backendsapi.velox.VeloxSparkPlanExecApi.createBroadcastRelation(VeloxSparkPlanExecApi.scala:850)
> at org.apache.spark.sql.execution.ColumnarBroadcastExchangeExec.$anonfun$relationFuture$2(ColumnarBroadcastExchangeExec.scala:79)
> at org.apache.gluten.utils.Arm$.withResource(Arm.scala:25)
> at org.apache.gluten.metrics.GlutenTimeMetric$.millis(GlutenTimeMetric.scala:37)
> at org.apache.spark.sql.execution.ColumnarBroadcastExchangeExec.$anonfun$relationFuture$1(ColumnarBroadcastExchangeExec.scala:66)
> at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withThreadLocalCaptured$2(SQLExecution.scala:230)
> at org.apache.spark.JobArtifactSet$.withActiveJobArtifactState(JobArtifactSet.scala:94)
> at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withThreadLocalCaptured$1(SQLExecution.scala:225)
> at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
> at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
> at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
> at java.base/java.lang.Thread.run(Thread.java:840){code}
>
> The error indicates that Spark attempted to broadcast a table significantly larger than the 8GB limit (actual size ~10.7 GiB). Interestingly, the query runs successfully if CBO is disabled.
> The issue stems from a bug in Spark's CBO Filter Estimation. When processing equality filters (e.g., d_year = 2001), the CBO incorrectly estimates the rowCount as 0.
> Because the estimated row count is 0, the optimizer concludes that the join result will be very small and chooses a BroadcastHashJoin . However, the actual data size is roughly 10.7 GiB, which exceeds Spark's hard limit for broadcasting.
>
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@spark.apache.org
For additional commands, e-mail: issues-help@spark.apache.org