You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "peter-toth (via GitHub)" <gi...@apache.org> on 2026/03/20 10:48:19 UTC
[PR] [SPARK-56102][SQL] `UnionEstimation` code cleanup [spark]
peter-toth opened a new pull request, #54918:
URL: https://github.com/apache/spark/pull/54918
### What changes were proposed in this pull request?
Refactored `UnionEstimation` to simplify and consolidate the column stats merging logic:
- Stats merging (estimate): replaced the multi-pass layering approach (newMinMaxStats ++ withNullCount ++ withDistinctCount) with a single-pass merge over all attributes using AttributeMap
lookups, eliminating redundant concatenations.
- `computeMinMaxStats`: replaced zipWithIndex + index-based child/output lookups with zip(union.children) and zip(union.output). Collapsed the separate filter + map into a single flatMap with
a short-circuiting foldLeft, seeded from the first element to avoid inner Option[Any] in the accumulator. Used collect to combine the hasMinMaxStats guard and value extraction.
- `computeNullCountStats`: same structural simplification — flatMap + short-circuiting fold, eliminating the awkward head/tail split and double attributeStats lookups.
- `computeDistinctCountStats`: same structural simplification — flatMap + short-circuiting fold, replacing the separate filter and map passes. Replaced outputRows match with
outputRows.fold(...).
### Why are the changes needed?
The previous implementation had several readability and efficiency issues:
- attributeStats was looked up twice per attribute per child (once in the filter guard, once in the map body).
- Index-based access (zipWithIndex, unionOutput(outputIndex), union.children(childIndex)) made the data relationships implicit; direct zipping makes them explicit.
- The stats merging duplicated newMinMaxStats entries multiple times, relying on AttributeMap deduplication by position.
- Inner Option[Any] in the computeMinMaxStats accumulator was only needed as an uninitialized sentinel, which could be eliminated by seeding from the first element.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Existing unit tests in `UnionEstimationSuite` cover the changed logic. No new tests were added as this is a pure refactoring with no behavior change.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Sonnet 4.6
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56102][SQL] `UnionEstimation` code cleanup [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #54918:
URL: https://github.com/apache/spark/pull/54918#discussion_r2965521747
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/statsEstimation/UnionEstimation.scala:
##########
@@ -147,31 +66,49 @@ object UnionEstimation {
* (e.g. web_sales and catalog_sales reference the same date keys),
* so max is a reasonable approximation in the common case.
*/
- private def computeDistinctCountStats(
+ private def computeColumnStats(
union: Union,
outputRows: Option[BigInt]): Seq[(Attribute, ColumnStat)] = {
- val unionOutput = union.output
- val attrToComputeDistinctCount = union.children.map(_.output).transpose.zipWithIndex.filter {
- case (attrs, _) => attrs.zipWithIndex.forall {
- case (attr, childIndex) =>
- val attrStats = union.children(childIndex).stats.attributeStats
- attrStats.get(attr).isDefined && attrStats(attr).distinctCount.isDefined
+ // For each child, look up the ColumnStat for each of its output attributes.
+ // After transposing, maybeColStats(i) holds the ColumnStat (if available) contributed
+ // by the i-th child for the current output column.
+ union.output.zip(union.children.map(c => c.output.map(c.stats.attributeStats.get)).transpose)
+ .flatMap {
+ case (outputAttr, maybeColStats) =>
+ val maybeMinMax: Option[(Any, Any)] = if (isTypeSupported(outputAttr.dataType)) {
+ val statComparator = createStatComparator(outputAttr.dataType)
+ val initial = maybeColStats.head
+ .collect { case s if s.hasMinMaxStats => (s.min.get, s.max.get) }
+ maybeColStats.tail.foldLeft(initial) {
+ case (Some((minVal, maxVal)), Some(s)) if s.hasMinMaxStats =>
+ Some((
+ if (statComparator(s.min.get, minVal)) s.min.get else minVal,
+ if (statComparator(maxVal, s.max.get)) s.max.get else maxVal
+ ))
+ case _ => None
+ }
+ } else None
+
+ val maybeNullCount = maybeColStats.foldLeft(Option(BigInt(0))) {
+ case (Some(total), Some(s)) => s.nullCount.map(total + _)
+ case _ => None
+ }
+
+ val maybeDistinctCount = maybeColStats.foldLeft(Option(BigInt(0))) {
+ case (Some(currentMax), Some(s)) => s.distinctCount.map(currentMax.max)
+ case _ => None
+ }
+ // Cap distinctCount by outputRows if available
Review Comment:
Hmm, if an estimation in Spark estimated outputRows = x for a filter like `WHERE id < 5`, but kept distinctCount > x for any columns, then it would sound like a Spark bug to me. But yeah, the base stats might not be correct and this cap does not harm.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56102][SQL] `UnionEstimation` code cleanup [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on PR #54918:
URL: https://github.com/apache/spark/pull/54918#issuecomment-4098859489
Thank you for the review @LuciferYang, @dongjoon-hyun, @yaooqinn.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56102][SQL] `UnionEstimation` code cleanup [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #54918:
URL: https://github.com/apache/spark/pull/54918#discussion_r2965372895
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/statsEstimation/UnionEstimation.scala:
##########
@@ -147,31 +66,49 @@ object UnionEstimation {
* (e.g. web_sales and catalog_sales reference the same date keys),
* so max is a reasonable approximation in the common case.
*/
- private def computeDistinctCountStats(
+ private def computeColumnStats(
union: Union,
outputRows: Option[BigInt]): Seq[(Attribute, ColumnStat)] = {
- val unionOutput = union.output
- val attrToComputeDistinctCount = union.children.map(_.output).transpose.zipWithIndex.filter {
- case (attrs, _) => attrs.zipWithIndex.forall {
- case (attr, childIndex) =>
- val attrStats = union.children(childIndex).stats.attributeStats
- attrStats.get(attr).isDefined && attrStats(attr).distinctCount.isDefined
+ // For each child, look up the ColumnStat for each of its output attributes.
+ // After transposing, maybeColStats(i) holds the ColumnStat (if available) contributed
+ // by the i-th child for the current output column.
+ union.output.zip(union.children.map(c => c.output.map(c.stats.attributeStats.get)).transpose)
+ .flatMap {
+ case (outputAttr, maybeColStats) =>
+ val maybeMinMax: Option[(Any, Any)] = if (isTypeSupported(outputAttr.dataType)) {
+ val statComparator = createStatComparator(outputAttr.dataType)
+ val initial = maybeColStats.head
+ .collect { case s if s.hasMinMaxStats => (s.min.get, s.max.get) }
+ maybeColStats.tail.foldLeft(initial) {
+ case (Some((minVal, maxVal)), Some(s)) if s.hasMinMaxStats =>
+ Some((
+ if (statComparator(s.min.get, minVal)) s.min.get else minVal,
+ if (statComparator(maxVal, s.max.get)) s.max.get else maxVal
+ ))
+ case _ => None
+ }
+ } else None
+
+ val maybeNullCount = maybeColStats.foldLeft(Option(BigInt(0))) {
+ case (Some(total), Some(s)) => s.nullCount.map(total + _)
+ case _ => None
+ }
+
+ val maybeDistinctCount = maybeColStats.foldLeft(Option(BigInt(0))) {
+ case (Some(currentMax), Some(s)) => s.distinctCount.map(currentMax.max)
+ case _ => None
+ }
+ // Cap distinctCount by outputRows if available
Review Comment:
@LuciferYang , qq: why exactly did we introduce this `outputRows` cap in https://github.com/apache/spark/pull/54883? How can `distinctCount` be larger than `outputRows`?
I wonder if we need that cap on `nullCount` as well for some reasons?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56102][SQL] `UnionEstimation` code cleanup [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on PR #54918:
URL: https://github.com/apache/spark/pull/54918#issuecomment-4097236372
cc @LuciferYang
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56102][SQL] `UnionEstimation` code cleanup [spark]
Posted by "LuciferYang (via GitHub)" <gi...@apache.org>.
LuciferYang commented on code in PR #54918:
URL: https://github.com/apache/spark/pull/54918#discussion_r2965448066
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/statsEstimation/UnionEstimation.scala:
##########
@@ -147,31 +66,49 @@ object UnionEstimation {
* (e.g. web_sales and catalog_sales reference the same date keys),
* so max is a reasonable approximation in the common case.
*/
- private def computeDistinctCountStats(
+ private def computeColumnStats(
union: Union,
outputRows: Option[BigInt]): Seq[(Attribute, ColumnStat)] = {
- val unionOutput = union.output
- val attrToComputeDistinctCount = union.children.map(_.output).transpose.zipWithIndex.filter {
- case (attrs, _) => attrs.zipWithIndex.forall {
- case (attr, childIndex) =>
- val attrStats = union.children(childIndex).stats.attributeStats
- attrStats.get(attr).isDefined && attrStats(attr).distinctCount.isDefined
+ // For each child, look up the ColumnStat for each of its output attributes.
+ // After transposing, maybeColStats(i) holds the ColumnStat (if available) contributed
+ // by the i-th child for the current output column.
+ union.output.zip(union.children.map(c => c.output.map(c.stats.attributeStats.get)).transpose)
+ .flatMap {
+ case (outputAttr, maybeColStats) =>
+ val maybeMinMax: Option[(Any, Any)] = if (isTypeSupported(outputAttr.dataType)) {
+ val statComparator = createStatComparator(outputAttr.dataType)
+ val initial = maybeColStats.head
+ .collect { case s if s.hasMinMaxStats => (s.min.get, s.max.get) }
+ maybeColStats.tail.foldLeft(initial) {
+ case (Some((minVal, maxVal)), Some(s)) if s.hasMinMaxStats =>
+ Some((
+ if (statComparator(s.min.get, minVal)) s.min.get else minVal,
+ if (statComparator(maxVal, s.max.get)) s.max.get else maxVal
+ ))
+ case _ => None
+ }
+ } else None
+
+ val maybeNullCount = maybeColStats.foldLeft(Option(BigInt(0))) {
+ case (Some(total), Some(s)) => s.nullCount.map(total + _)
+ case _ => None
+ }
+
+ val maybeDistinctCount = maybeColStats.foldLeft(Option(BigInt(0))) {
+ case (Some(currentMax), Some(s)) => s.distinctCount.map(currentMax.max)
+ case _ => None
+ }
+ // Cap distinctCount by outputRows if available
Review Comment:
The cap is needed because CBO column statistics can be inherited from base tables (via `ANALYZE TABLE`) and may not reflect filters applied in subqueries. For example, a child plan like `SELECT * FROM big_table WHERE id < 5` might still carry `distinctCount = 1000000` from the base table stats if the filter's selectivity wasn't perfectly estimated. In that case, `max(dc_i)` across children could exceed the union's total `outputRows`, producing a nonsensical estimate. The `min(maxDc, outputRows)` cap corrects this.
For `nullCount`, theoretically the same issue could arise — the summed `nullCount` across children could exceed `outputRows` if any child carries stale stats. However, since `nullCount` is summed (not max'd), it naturally scales with the union's row count when stats are fresh, so in practice this is less likely to produce an obviously incorrect result. That said, adding `nullCount = nullCount.min(outputRows)` would be a defensively improvement if we want to be thorough.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56102][SQL] `UnionEstimation` code cleanup [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun closed pull request #54918: [SPARK-56102][SQL] `UnionEstimation` code cleanup
URL: https://github.com/apache/spark/pull/54918
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56102][SQL] `UnionEstimation` code cleanup [spark]
Posted by "LuciferYang (via GitHub)" <gi...@apache.org>.
LuciferYang commented on code in PR #54918:
URL: https://github.com/apache/spark/pull/54918#discussion_r2965448066
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/statsEstimation/UnionEstimation.scala:
##########
@@ -147,31 +66,49 @@ object UnionEstimation {
* (e.g. web_sales and catalog_sales reference the same date keys),
* so max is a reasonable approximation in the common case.
*/
- private def computeDistinctCountStats(
+ private def computeColumnStats(
union: Union,
outputRows: Option[BigInt]): Seq[(Attribute, ColumnStat)] = {
- val unionOutput = union.output
- val attrToComputeDistinctCount = union.children.map(_.output).transpose.zipWithIndex.filter {
- case (attrs, _) => attrs.zipWithIndex.forall {
- case (attr, childIndex) =>
- val attrStats = union.children(childIndex).stats.attributeStats
- attrStats.get(attr).isDefined && attrStats(attr).distinctCount.isDefined
+ // For each child, look up the ColumnStat for each of its output attributes.
+ // After transposing, maybeColStats(i) holds the ColumnStat (if available) contributed
+ // by the i-th child for the current output column.
+ union.output.zip(union.children.map(c => c.output.map(c.stats.attributeStats.get)).transpose)
+ .flatMap {
+ case (outputAttr, maybeColStats) =>
+ val maybeMinMax: Option[(Any, Any)] = if (isTypeSupported(outputAttr.dataType)) {
+ val statComparator = createStatComparator(outputAttr.dataType)
+ val initial = maybeColStats.head
+ .collect { case s if s.hasMinMaxStats => (s.min.get, s.max.get) }
+ maybeColStats.tail.foldLeft(initial) {
+ case (Some((minVal, maxVal)), Some(s)) if s.hasMinMaxStats =>
+ Some((
+ if (statComparator(s.min.get, minVal)) s.min.get else minVal,
+ if (statComparator(maxVal, s.max.get)) s.max.get else maxVal
+ ))
+ case _ => None
+ }
+ } else None
+
+ val maybeNullCount = maybeColStats.foldLeft(Option(BigInt(0))) {
+ case (Some(total), Some(s)) => s.nullCount.map(total + _)
+ case _ => None
+ }
+
+ val maybeDistinctCount = maybeColStats.foldLeft(Option(BigInt(0))) {
+ case (Some(currentMax), Some(s)) => s.distinctCount.map(currentMax.max)
+ case _ => None
+ }
+ // Cap distinctCount by outputRows if available
Review Comment:
The cap is needed because CBO column statistics can be inherited from base tables (via `ANALYZE TABLE`) and may not reflect filters applied in subqueries. For example, a child plan like `SELECT * FROM big_table WHERE id < 5` might still carry `distinctCount = 1000000` from the base table stats if the filter's selectivity wasn't perfectly estimated. In that case, `max(dc_i)` across children could exceed the union's total `outputRows`, producing a nonsensical estimate. The `min(maxDc, outputRows)` cap corrects this.
For `nullCount`, theoretically the same issue could arise — the summed `nullCount` across children could exceed `outputRows` if any child carries stale stats. However, since `nullCount` is summed (not max'd), it naturally scales with the union's row count when stats are fresh, so in practice this is less likely to produce an obviously incorrect result. That said, adding `nullCount = nullCount.min(outputRows)` would be a defensively correct improvement if we want to be thorough.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56102][SQL] `UnionEstimation` code cleanup [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on PR #54918:
URL: https://github.com/apache/spark/pull/54918#issuecomment-4097547535
cc @dongjoon-hyun , @yaooqinn as well
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56102][SQL] `UnionEstimation` code cleanup [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on PR #54918:
URL: https://github.com/apache/spark/pull/54918#issuecomment-4097337641
actually, let me try to make some more simplifications
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56102][SQL] `UnionEstimation` code cleanup [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #54918:
URL: https://github.com/apache/spark/pull/54918#discussion_r2965521747
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/statsEstimation/UnionEstimation.scala:
##########
@@ -147,31 +66,49 @@ object UnionEstimation {
* (e.g. web_sales and catalog_sales reference the same date keys),
* so max is a reasonable approximation in the common case.
*/
- private def computeDistinctCountStats(
+ private def computeColumnStats(
union: Union,
outputRows: Option[BigInt]): Seq[(Attribute, ColumnStat)] = {
- val unionOutput = union.output
- val attrToComputeDistinctCount = union.children.map(_.output).transpose.zipWithIndex.filter {
- case (attrs, _) => attrs.zipWithIndex.forall {
- case (attr, childIndex) =>
- val attrStats = union.children(childIndex).stats.attributeStats
- attrStats.get(attr).isDefined && attrStats(attr).distinctCount.isDefined
+ // For each child, look up the ColumnStat for each of its output attributes.
+ // After transposing, maybeColStats(i) holds the ColumnStat (if available) contributed
+ // by the i-th child for the current output column.
+ union.output.zip(union.children.map(c => c.output.map(c.stats.attributeStats.get)).transpose)
+ .flatMap {
+ case (outputAttr, maybeColStats) =>
+ val maybeMinMax: Option[(Any, Any)] = if (isTypeSupported(outputAttr.dataType)) {
+ val statComparator = createStatComparator(outputAttr.dataType)
+ val initial = maybeColStats.head
+ .collect { case s if s.hasMinMaxStats => (s.min.get, s.max.get) }
+ maybeColStats.tail.foldLeft(initial) {
+ case (Some((minVal, maxVal)), Some(s)) if s.hasMinMaxStats =>
+ Some((
+ if (statComparator(s.min.get, minVal)) s.min.get else minVal,
+ if (statComparator(maxVal, s.max.get)) s.max.get else maxVal
+ ))
+ case _ => None
+ }
+ } else None
+
+ val maybeNullCount = maybeColStats.foldLeft(Option(BigInt(0))) {
+ case (Some(total), Some(s)) => s.nullCount.map(total + _)
+ case _ => None
+ }
+
+ val maybeDistinctCount = maybeColStats.foldLeft(Option(BigInt(0))) {
+ case (Some(currentMax), Some(s)) => s.distinctCount.map(currentMax.max)
+ case _ => None
+ }
+ // Cap distinctCount by outputRows if available
Review Comment:
Hmm, if an estimation in Spark would estimate outputRows = x for a filter like `WHERE id < 5`, but keep distinctCount > x for any columns, then it would sound like a Spark bug to me. But yeah, the base stats might not be correct and this cap does not harm.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56102][SQL] `UnionEstimation` code cleanup [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #54918:
URL: https://github.com/apache/spark/pull/54918#discussion_r2965521747
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/statsEstimation/UnionEstimation.scala:
##########
@@ -147,31 +66,49 @@ object UnionEstimation {
* (e.g. web_sales and catalog_sales reference the same date keys),
* so max is a reasonable approximation in the common case.
*/
- private def computeDistinctCountStats(
+ private def computeColumnStats(
union: Union,
outputRows: Option[BigInt]): Seq[(Attribute, ColumnStat)] = {
- val unionOutput = union.output
- val attrToComputeDistinctCount = union.children.map(_.output).transpose.zipWithIndex.filter {
- case (attrs, _) => attrs.zipWithIndex.forall {
- case (attr, childIndex) =>
- val attrStats = union.children(childIndex).stats.attributeStats
- attrStats.get(attr).isDefined && attrStats(attr).distinctCount.isDefined
+ // For each child, look up the ColumnStat for each of its output attributes.
+ // After transposing, maybeColStats(i) holds the ColumnStat (if available) contributed
+ // by the i-th child for the current output column.
+ union.output.zip(union.children.map(c => c.output.map(c.stats.attributeStats.get)).transpose)
+ .flatMap {
+ case (outputAttr, maybeColStats) =>
+ val maybeMinMax: Option[(Any, Any)] = if (isTypeSupported(outputAttr.dataType)) {
+ val statComparator = createStatComparator(outputAttr.dataType)
+ val initial = maybeColStats.head
+ .collect { case s if s.hasMinMaxStats => (s.min.get, s.max.get) }
+ maybeColStats.tail.foldLeft(initial) {
+ case (Some((minVal, maxVal)), Some(s)) if s.hasMinMaxStats =>
+ Some((
+ if (statComparator(s.min.get, minVal)) s.min.get else minVal,
+ if (statComparator(maxVal, s.max.get)) s.max.get else maxVal
+ ))
+ case _ => None
+ }
+ } else None
+
+ val maybeNullCount = maybeColStats.foldLeft(Option(BigInt(0))) {
+ case (Some(total), Some(s)) => s.nullCount.map(total + _)
+ case _ => None
+ }
+
+ val maybeDistinctCount = maybeColStats.foldLeft(Option(BigInt(0))) {
+ case (Some(currentMax), Some(s)) => s.distinctCount.map(currentMax.max)
+ case _ => None
+ }
+ // Cap distinctCount by outputRows if available
Review Comment:
Hmm, if an estimation in Spark estimated outputRows = x for a filter like `WHERE id < 5`, but kept distinctCount > x for any columns, then it would sound like a Spark bug to me. But yeah, the base stats might not be correct and this cap does no harm.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56102][SQL] `UnionEstimation` code cleanup [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on PR #54918:
URL: https://github.com/apache/spark/pull/54918#issuecomment-4098369316
Merged to master. Thank you, @peter-toth and all. For the following part, we can proceed independently later.
> https://github.com/apache/spark/pull/54918#discussion_r2965521747
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56102][SQL] `UnionEstimation` code cleanup [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on PR #54918:
URL: https://github.com/apache/spark/pull/54918#issuecomment-4097410935
it looks better now
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56102][SQL] `UnionEstimation` code cleanup [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #54918:
URL: https://github.com/apache/spark/pull/54918#discussion_r2965372895
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/statsEstimation/UnionEstimation.scala:
##########
@@ -147,31 +66,49 @@ object UnionEstimation {
* (e.g. web_sales and catalog_sales reference the same date keys),
* so max is a reasonable approximation in the common case.
*/
- private def computeDistinctCountStats(
+ private def computeColumnStats(
union: Union,
outputRows: Option[BigInt]): Seq[(Attribute, ColumnStat)] = {
- val unionOutput = union.output
- val attrToComputeDistinctCount = union.children.map(_.output).transpose.zipWithIndex.filter {
- case (attrs, _) => attrs.zipWithIndex.forall {
- case (attr, childIndex) =>
- val attrStats = union.children(childIndex).stats.attributeStats
- attrStats.get(attr).isDefined && attrStats(attr).distinctCount.isDefined
+ // For each child, look up the ColumnStat for each of its output attributes.
+ // After transposing, maybeColStats(i) holds the ColumnStat (if available) contributed
+ // by the i-th child for the current output column.
+ union.output.zip(union.children.map(c => c.output.map(c.stats.attributeStats.get)).transpose)
+ .flatMap {
+ case (outputAttr, maybeColStats) =>
+ val maybeMinMax: Option[(Any, Any)] = if (isTypeSupported(outputAttr.dataType)) {
+ val statComparator = createStatComparator(outputAttr.dataType)
+ val initial = maybeColStats.head
+ .collect { case s if s.hasMinMaxStats => (s.min.get, s.max.get) }
+ maybeColStats.tail.foldLeft(initial) {
+ case (Some((minVal, maxVal)), Some(s)) if s.hasMinMaxStats =>
+ Some((
+ if (statComparator(s.min.get, minVal)) s.min.get else minVal,
+ if (statComparator(maxVal, s.max.get)) s.max.get else maxVal
+ ))
+ case _ => None
+ }
+ } else None
+
+ val maybeNullCount = maybeColStats.foldLeft(Option(BigInt(0))) {
+ case (Some(total), Some(s)) => s.nullCount.map(total + _)
+ case _ => None
+ }
+
+ val maybeDistinctCount = maybeColStats.foldLeft(Option(BigInt(0))) {
+ case (Some(currentMax), Some(s)) => s.distinctCount.map(currentMax.max)
+ case _ => None
+ }
+ // Cap distinctCount by outputRows if available
Review Comment:
@LuciferYang , qq: why exactly did we introduce this `outputRows` cap in https://github.com/apache/spark/pull/54883? How can distinctCount be larger than outputRows? I wonder if we need that cap on
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/statsEstimation/UnionEstimation.scala:
##########
@@ -147,31 +66,49 @@ object UnionEstimation {
* (e.g. web_sales and catalog_sales reference the same date keys),
* so max is a reasonable approximation in the common case.
*/
- private def computeDistinctCountStats(
+ private def computeColumnStats(
union: Union,
outputRows: Option[BigInt]): Seq[(Attribute, ColumnStat)] = {
- val unionOutput = union.output
- val attrToComputeDistinctCount = union.children.map(_.output).transpose.zipWithIndex.filter {
- case (attrs, _) => attrs.zipWithIndex.forall {
- case (attr, childIndex) =>
- val attrStats = union.children(childIndex).stats.attributeStats
- attrStats.get(attr).isDefined && attrStats(attr).distinctCount.isDefined
+ // For each child, look up the ColumnStat for each of its output attributes.
+ // After transposing, maybeColStats(i) holds the ColumnStat (if available) contributed
+ // by the i-th child for the current output column.
+ union.output.zip(union.children.map(c => c.output.map(c.stats.attributeStats.get)).transpose)
+ .flatMap {
+ case (outputAttr, maybeColStats) =>
+ val maybeMinMax: Option[(Any, Any)] = if (isTypeSupported(outputAttr.dataType)) {
+ val statComparator = createStatComparator(outputAttr.dataType)
+ val initial = maybeColStats.head
+ .collect { case s if s.hasMinMaxStats => (s.min.get, s.max.get) }
+ maybeColStats.tail.foldLeft(initial) {
+ case (Some((minVal, maxVal)), Some(s)) if s.hasMinMaxStats =>
+ Some((
+ if (statComparator(s.min.get, minVal)) s.min.get else minVal,
+ if (statComparator(maxVal, s.max.get)) s.max.get else maxVal
+ ))
+ case _ => None
+ }
+ } else None
+
+ val maybeNullCount = maybeColStats.foldLeft(Option(BigInt(0))) {
+ case (Some(total), Some(s)) => s.nullCount.map(total + _)
+ case _ => None
+ }
+
+ val maybeDistinctCount = maybeColStats.foldLeft(Option(BigInt(0))) {
+ case (Some(currentMax), Some(s)) => s.distinctCount.map(currentMax.max)
+ case _ => None
+ }
+ // Cap distinctCount by outputRows if available
Review Comment:
@LuciferYang , qq: why exactly did we introduce this `outputRows` cap in https://github.com/apache/spark/pull/54883? How can distinctCount be larger than outputRows?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56102][SQL] `UnionEstimation` code cleanup [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #54918:
URL: https://github.com/apache/spark/pull/54918#discussion_r2965372895
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/statsEstimation/UnionEstimation.scala:
##########
@@ -147,31 +66,49 @@ object UnionEstimation {
* (e.g. web_sales and catalog_sales reference the same date keys),
* so max is a reasonable approximation in the common case.
*/
- private def computeDistinctCountStats(
+ private def computeColumnStats(
union: Union,
outputRows: Option[BigInt]): Seq[(Attribute, ColumnStat)] = {
- val unionOutput = union.output
- val attrToComputeDistinctCount = union.children.map(_.output).transpose.zipWithIndex.filter {
- case (attrs, _) => attrs.zipWithIndex.forall {
- case (attr, childIndex) =>
- val attrStats = union.children(childIndex).stats.attributeStats
- attrStats.get(attr).isDefined && attrStats(attr).distinctCount.isDefined
+ // For each child, look up the ColumnStat for each of its output attributes.
+ // After transposing, maybeColStats(i) holds the ColumnStat (if available) contributed
+ // by the i-th child for the current output column.
+ union.output.zip(union.children.map(c => c.output.map(c.stats.attributeStats.get)).transpose)
+ .flatMap {
+ case (outputAttr, maybeColStats) =>
+ val maybeMinMax: Option[(Any, Any)] = if (isTypeSupported(outputAttr.dataType)) {
+ val statComparator = createStatComparator(outputAttr.dataType)
+ val initial = maybeColStats.head
+ .collect { case s if s.hasMinMaxStats => (s.min.get, s.max.get) }
+ maybeColStats.tail.foldLeft(initial) {
+ case (Some((minVal, maxVal)), Some(s)) if s.hasMinMaxStats =>
+ Some((
+ if (statComparator(s.min.get, minVal)) s.min.get else minVal,
+ if (statComparator(maxVal, s.max.get)) s.max.get else maxVal
+ ))
+ case _ => None
+ }
+ } else None
+
+ val maybeNullCount = maybeColStats.foldLeft(Option(BigInt(0))) {
+ case (Some(total), Some(s)) => s.nullCount.map(total + _)
+ case _ => None
+ }
+
+ val maybeDistinctCount = maybeColStats.foldLeft(Option(BigInt(0))) {
+ case (Some(currentMax), Some(s)) => s.distinctCount.map(currentMax.max)
+ case _ => None
+ }
+ // Cap distinctCount by outputRows if available
Review Comment:
@LuciferYang , qq: why exactly did we introduce this `outputRows` cap in https://github.com/apache/spark/pull/54883? How can `distinctCount` be larger than `outputRows`?
I wonder if we need that cap on `nullCount` as well for some reason?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org