dongjoon-hyun commented on code in PR #60025:
URL: https://github.com/apache/spark/pull/60025#discussion_r4464484994
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/statsEstimation/JoinEstimation.scala:
##########
@@ -201,7 +201,9 @@ case class JoinEstimation(join: Join) extends Logging {
if (ValueInterval.isIntersected(lInterval, rInterval)) {
val (newMin, newMax) = ValueInterval.intersect(lInterval, rInterval,
leftKey.dataType)
val (card, joinStat) = (leftKeyStat.histogram, rightKeyStat.histogram)
match {
- case (Some(l: Histogram), Some(r: Histogram)) =>
+ // The intersection has no bounds for string and binary keys, which
have no
+ // equi-height histograms even if a connector reported one.
+ case (Some(l: Histogram), Some(r: Histogram)) if newMin.isDefined &&
newMax.isDefined =>
Review Comment:
**14. [Low] The join guard tells string keys apart by their missing bounds**
This guard keeps string and binary keys away from `computeByHistogram`
because `ValueInterval.intersect` returns `(None, None)` for a
`DefaultValueInterval`. That holds today only because `ValueInterval` has no
string min/max yet; its scaladoc says "This version of Spark does not have
min/max for binary/string types". If that changes, the intersection gets
bounds, a string histogram passes this guard again, and `computeByHistogram`
fails on `newMin.get.toString.toDouble`.
`FilterEstimation.evaluateEquality` checks the type directly. Could the join
check it the same way, so both estimators state the same rule? For example
(with `BinaryType` and `StringType` imported from `org.apache.spark.sql.types`):
```scala
case (Some(l: Histogram), Some(r: Histogram))
if !leftKey.dataType.isInstanceOf[StringType] &&
leftKey.dataType != BinaryType =>
```
A shared helper, as in comment 16, would work too.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/statsEstimation/FilterEstimation.scala:
##########
@@ -366,7 +373,14 @@ case class FilterEstimation(
if (!valueInRange) return Some(0.0)
- val percent = if (colStat.histogram.isDefined) {
+ // Only numeric-like types have equi-height histograms: the histogram path
converts the
+ // literal to a double. A string or binary column uses its distinct count
even if a connector
+ // reported a histogram for it.
+ val histogram = attr.dataType match {
+ case _: StringType | BinaryType => None
+ case _ => colStat.histogram
+ }
+ val percent = if (histogram.isDefined) {
Review Comment:
**15. [Low] Histogram-only STRING stats now get 1/ndv instead of 1.0**
Before this change, a string column with a histogram but no min/max did not
fail here: `hasMinMaxStats` was false, so `percent` was `None` and the
predicate was estimated at 1.0. Only stats that also carry min and max reached
`computeEqualityPossibilityByHistogram` and the `MatchError`. Now both cases
use `1 / distinctCount`.
The new estimate is the better one, but it changes plans for STRING and
BINARY columns too, not only CHAR. The description says such a column "already
failed with a `MatchError`" without that condition. The stat in the new
`JoinEstimationSuite` test (a histogram, no min/max) is an example: `WHERE ls =
'x'` over it used to keep all 10 rows and now keeps 1.
Could the description say that the failure needed min and max as well, and
mention the estimate change for histogram-only stats in the user-facing change
section?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/statsEstimation/FilterEstimation.scala:
##########
@@ -366,7 +373,14 @@ case class FilterEstimation(
if (!valueInRange) return Some(0.0)
- val percent = if (colStat.histogram.isDefined) {
+ // Only numeric-like types have equi-height histograms: the histogram path
converts the
Review Comment:
**16. [Low] Could the histogram be dropped where it enters?**
The rule "string and binary columns have no usable histogram" is now
enforced in two consumers: by type here, and by bounds in `JoinEstimation`
(comment 14). The histogram itself comes from
`DataSourceV2Relation.transformV2Stats`, which copies a connector's histogram
for any column type. Dropping it there for types that
`EstimationUtils.toDouble` cannot convert would cover both estimators and any
later reader of `ColumnStat.histogram`, and the two guards could go.
If you'd rather keep the change in the estimators, a shared helper such as
`EstimationUtils.supportsHistogram(dataType)` would at least keep the two
checks from drifting apart.
##########
sql/core/src/test/scala/org/apache/spark/sql/StatisticsCollectionSuite.scala:
##########
@@ -202,6 +203,59 @@ class StatisticsCollectionSuite extends
StatisticsCollectionTestBase with Shared
}
}
+ test("SPARK-60810: CBO keeps CHAR column statistics through read-side
padding") {
+ withSQLConf(
+ SQLConf.CBO_ENABLED.key -> "true",
+ SQLConf.READ_SIDE_CHAR_PADDING.key -> "true") {
+ val tableName = "char_read_side_padding_stats"
+ withTable(tableName) {
+ sql(s"CREATE TABLE $tableName(c CHAR(5)) USING parquet")
+ sql(s"INSERT INTO $tableName SELECT CAST(id % 10 AS STRING) FROM
range(1000)")
+ sql(s"ANALYZE TABLE $tableName COMPUTE STATISTICS FOR COLUMNS c")
+
+ def estimatedRows(query: String)(
+ pick: PartialFunction[LogicalPlan, LogicalPlan]): Option[BigInt] =
{
+ val plan = sql(query).queryExecution.optimizedPlan
+ plan.collectFirst(pick).getOrElse(fail(s"no matching node
in\n${plan.treeString}"))
+ .stats.rowCount
+ }
+ // Above the padding Project, which carries the column's statistics.
+ assert(estimatedRows(s"SELECT c, count(*) FROM $tableName GROUP BY c")
{
+ case a: Aggregate => a
+ }.contains(BigInt(10)))
+ // Pushed below it, where the predicate compares the padding
expression itself.
+ assert(estimatedRows(s"SELECT * FROM $tableName WHERE c = '3'") {
+ case f: Filter => f
+ }.contains(BigInt(100)))
+ assert(estimatedRows(s"SELECT * FROM $tableName WHERE c IN ('3',
'4')") {
+ case f: Filter => f
+ }.contains(BigInt(200)))
+ }
+ }
+ }
+
+ test("SPARK-60810: partition pruning estimates a filter on a CHAR partition
column") {
Review Comment:
**19. [Low] The DSv2 estimate is not covered**
This test covers `PruneFileSourcePartitions`. The description also lists
`DataSourceV2Relation.estimateStatsWithFilters`, which estimates the pushed and
the inferred filters against the connector's stats and keeps the smaller
estimate (`estimateWithInferredCondition`). No test sends a predicate on a
padded CHAR column through that path, so it could stop seeing the column's
stats without any test failing.
`DataSourceV2RelationSuite` already builds relations with reported column
stats and `pushedFilters` / `inferredFilters` (for example "inferred filters
use the smaller estimate for fully pushed originals"). Could one case there use
a predicate on `readSidePadding` of a CHAR column?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/statsEstimation/FilterEstimation.scala:
##########
@@ -366,7 +373,14 @@ case class FilterEstimation(
if (!valueInRange) return Some(0.0)
- val percent = if (colStat.histogram.isDefined) {
+ // Only numeric-like types have equi-height histograms: the histogram path
converts the
+ // literal to a double. A string or binary column uses its distinct count
even if a connector
+ // reported a histogram for it.
+ val histogram = attr.dataType match {
Review Comment:
**20. [Nit] `histogram` is only used for `isDefined`**
The local `histogram` decides the branch, but
`computeEqualityPossibilityByHistogram` reads `colStat.histogram` again, so the
value that is checked and the value that is used are different expressions. A
Boolean would say what the local is for, for example:
```scala
val useHistogram = colStat.histogram.isDefined && (attr.dataType match {
case _: StringType | BinaryType => false
case _ => true
})
val percent = if (useHistogram) {
```
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/statsEstimation/EstimationUtils.scala:
##########
@@ -87,6 +88,23 @@ object EstimationUtils {
expressions.collect {
case alias @ Alias(attr: Attribute, _) if attributeStats.contains(attr)
=>
alias.toAttribute -> attributeStats(attr)
+ // Read-side CHAR padding wraps each CHAR column of a scan. With padding
on, ANALYZE TABLE
+ // reads the table through it as well, so the statistics already
describe padded values.
+ // Statistics collected otherwise (with padding off, or reported by a
connector) can
+ // describe shorter values, and every non-null padded value has at least
N characters.
+ // A distinct count of 0 means no non-null values, so there is nothing
to pad (ANALYZE
+ // then stores the type's default size as the lengths). The distinct
count is left as is.
+ // It is too high only when stored values differ just in trailing
spaces, which the
+ // padding makes equal.
+ case alias @ Alias(CharVarcharUtils.ReadSidePadding(attr: Attribute,
Literal(n: Int, _)), _)
+ if attributeStats.contains(attr) =>
+ val stat = attributeStats(attr)
+ val padded = if (stat.distinctCount.contains(BigInt(0))) {
Review Comment:
**17. [Nit] The floor still applies to all-null columns without a distinct
count**
The skip keys only on a distinct count of 0, which `ANALYZE TABLE` always
stores. A connector can report `nullCount` without `distinctCount`, and then an
all-null column still has its lengths raised to N. For a `CHAR(100)` column
reported with `avgLen = 0`, `getSizePerRow` then counts 112 bytes per row for a
column of nulls. `getAliasStats` already receives `rowCount`, so the skip could
cover that case too:
```scala
val allNull = stat.distinctCount.contains(BigInt(0)) ||
stat.nullCount.exists(_ >= rowCount)
val padded = if (allNull) {
```
This error is an overestimate, which is the safe direction, so feel free to
leave it.
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/statsEstimation/JoinEstimationSuite.scala:
##########
@@ -560,6 +560,20 @@ class JoinEstimationSuite extends StatsEstimationTestBase {
assert(join.stats == expectedStats)
}
+ test("SPARK-60810: string join keys with reported histograms use the
distinct count") {
Review Comment:
**18. [Low] The join test covers only STRING keys without min/max**
This test checks the guard for STRING keys whose stats are built directly.
Two cases are not covered:
- BINARY keys, which the comment in `JoinEstimation` names as well.
- CHAR keys above the padding `Project`, which is the path this PR opens.
`getAliasStats` now hands the column's stat, histogram included, to the alias,
and the join reads it from there.
Could the test also join two
`Project(Seq(Alias(CharVarcharUtils.addPaddingForScan(c), "c")()), child)`
plans whose child stats carry a histogram, and a pair of BINARY keys?
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
