dongjoon-hyun commented on code in PR #60025:
URL: https://github.com/apache/spark/pull/60025#discussion_r4463155132
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/statsEstimation/EstimationUtils.scala:
##########
@@ -87,6 +88,15 @@ 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.
+ case alias @ Alias(CharVarcharUtils.ReadSidePadding(attr: Attribute,
Literal(n: Int, _)), _)
+ if attributeStats.contains(attr) =>
+ val stat = attributeStats(attr)
+ alias.toAttribute -> stat.copy(
+ avgLen = stat.avgLen.map(_.max(n)), maxLen =
stat.maxLen.map(_.max(n)))
Review Comment:
**9. [Low] The length floor also applies to all-null columns**
The floor I suggested in comment 1 misses one case. For an all-null or empty
column, `ANALYZE TABLE` stores `defaultSize` (20 for `StringType`) as `avgLen`
and `maxLen`, from the `Coalesce(..., defaultSize)` in
`CommandUtils.statExprs`. That 20 is a placeholder, not the length of any
padded value, but the floor raises it to N. For an all-null `CHAR(100)` column,
`getSizePerRow` now counts 112 bytes per row instead of 32.
A distinct count of 0 marks this case, so the floor could skip it, for
example:
```scala
val stat = attributeStats(attr)
val padded = if (stat.distinctCount.contains(BigInt(0))) stat else stat.copy(
avgLen = stat.avgLen.map(_.max(n)), maxLen = stat.maxLen.map(_.max(n)))
alias.toAttribute -> padded
```
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/statsEstimation/FilterEstimation.scala:
##########
@@ -170,7 +173,11 @@ case class FilterEstimation(
* It returns None if the condition is not supported.
*/
def calculateSingleCondition(condition: Expression, update: Boolean):
Option[Double] = {
- condition match {
+ val unpadded = condition.mapChildren {
+ case CharVarcharUtils.ReadSidePadding(attr: Attribute, _) => attr
+ case other => other
+ }
+ unpadded match {
Review Comment:
**7. [Medium] An equality on a padded CHAR column can now reach `toDouble`
with a string**
Unwrapping the padding sends `c = 'x'` on a CHAR column to
`evaluateEquality`. If the column stat has a histogram and min/max, it
continues to `computeEqualityPossibilityByHistogram`, which calls
`EstimationUtils.toDouble(literal.value, StringType)`. `toDouble` only handles
numeric, date, timestamp and boolean types, so the optimizer fails with a
`scala.MatchError`.
`ANALYZE TABLE` never builds a histogram for strings, but
`DataSourceV2Relation.transformV2Stats` copies min, max and the histogram from
the connector for any type. A STRING column with such stats already fails the
same way, so the bug itself is older. Before this PR, though, a CHAR column
never got here, because the padded predicate was unsupported and estimated at
1.0.
Could `evaluateEquality` fall back to the distinct count for string and
binary types, as the `valueInRange` match above does? If you'd rather keep this
PR focused, a separate JIRA would work too, since STRING columns are affected
as well.
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/statsEstimation/FilterEstimationSuite.scala:
##########
@@ -963,6 +965,31 @@ class FilterEstimationSuite extends
StatsEstimationTestBase {
expectedRowCount = 10)
}
+ test("SPARK-60810: predicates on a read-side padded CHAR column use the
column's stats") {
+ val padded = CharVarcharUtils.addPaddingForScan(
+
DataTypeUtils.toAttribute(CharVarcharUtils.replaceCharVarcharWithStringInSchema(
+ StructType(Seq(StructField("cstring",
CharType(2))))).head).withExprId(attrString.exprId))
+ assert(padded match {
+ case CharVarcharUtils.ReadSidePadding(a: Attribute, _) => a.exprId ==
attrString.exprId
+ case _ => false
+ }, s"expected read-side padding over cstring, got $padded")
+ val child = childStatsTestPlan(Seq(attrString), tableRowCount = 10L)
+ // colStatString has 10 distinct values and no nulls over 10 rows.
+ Seq(
+ EqualTo(padded, Literal("A0")) -> 1,
+ EqualTo(Literal("A0"), padded) -> 1,
+ In(padded, Seq(Literal("A0"), Literal("A1"))) -> 2,
+ InSet(padded, Set(UTF8String.fromString("A0"),
UTF8String.fromString("A1"))) -> 2,
+ IsNull(padded) -> 0).foreach { case (condition, expectedRows) =>
+ val filter = Filter(condition, child)
+ assert(filter.stats.rowCount.contains(BigInt(expectedRows)), condition)
Review Comment:
**11. [Low] The cases check only the row count**
Each case asserts `rowCount` only. The other half of the change is that,
with `update`, the padded predicate's result is written to the stat of
`cstring` itself, which the padding `Project` above then passes to its alias.
If that update went to the wrong attribute or were skipped, every case here
would still pass. Could the equality case also check the output column stat, as
the `cstring = 'A2'` test does? For example:
```scala
validateEstimatedStats(
Filter(EqualTo(padded, Literal("A0")), child),
Seq(attrString -> ColumnStat(distinctCount = Some(1), min = None, max =
None,
nullCount = Some(0), avgLen = Some(2), maxLen = Some(2))),
expectedRowCount = 1)
```
A `Not(EqualTo(padded, Literal("A0")))` case would also cover the path
through `calculateFilterSelectivity` with `update = false`.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/statsEstimation/FilterEstimation.scala:
##########
@@ -170,7 +173,11 @@ case class FilterEstimation(
* It returns None if the condition is not supported.
*/
def calculateSingleCondition(condition: Expression, update: Boolean):
Option[Double] = {
- condition match {
+ val unpadded = condition.mapChildren {
Review Comment:
**13. [Nit] `mapChildren` also rebuilds predicates that are not supported**
`mapChildren` returns the same instance when no child changes, so predicates
without padding are not copied. When a child is padded, though, the predicate
is rebuilt before the match even if no case below supports it (for example
`Like`, `StartsWith`, or a UDF), and the copy is then dropped. This is cheap,
so feel free to ignore it. If you want to avoid it, the cases could match an
extractor for "an attribute or its read-side padding" instead.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/CharVarcharUtils.scala:
##########
@@ -391,6 +391,22 @@ object CharVarcharUtils extends Logging with
SparkCharVarcharUtils {
}.getOrElse(attr)
}
+ /**
+ * Matches the read-side CHAR padding that [[addPaddingForScan]] wraps a
column in (when
+ * standard semantics are off) and extracts the padded expression and the
length argument.
+ */
+ private[sql] object ReadSidePadding {
+ def unapply(e: Expression): Option[(Expression, Expression)] = e match {
+ case s: StaticInvoke
+ if s.scalarFunction.isEmpty &&
+ s.staticObject == classOf[CharVarcharCodegenUtils] &&
+ s.functionName == READ_SIDE_PADDING &&
Review Comment:
**12. [Nit] For SPARK-60830, a dedicated expression instead of another name
check?**
One idea for SPARK-60830: `ReadSidePadding` recognizes the scan wrapper by
the `StaticInvoke` function name, and as you noted, the checks need a matcher
of their own with different length handling. Each wrapper then needs another
name-based matcher, and every consumer has to know which one to use. A
dedicated expression for the scan wrappers, carrying the length and whether
values are padded or checked, might let these consumers match one type instead.
Not for this PR.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/statsEstimation/FilterEstimation.scala:
##########
@@ -161,7 +162,9 @@ case class FilterEstimation(
/**
* Returns a percentage of rows meeting a single condition in Filter node.
* Currently we only support binary predicates where one side is a column,
- * and the other is a literal.
+ * and the other is a literal. A side may also be the read-side CHAR padding
of a column, which
+ * is what a predicate pushed below the padding Project compares. It is
estimated as the column
+ * itself, as `EstimationUtils.getAliasStats` does above the padding Project.
Review Comment:
**8. [Low] With CBO, the pruned relation's size now comes from this
estimate**
Following up on comment 3, where I called the partition pruning change an
improvement: I missed how the result is used. `PruneFileSourcePartitions` saves
the estimated `rowCount` next to `sizeInBytes = prunedFileIndex.sizeInBytes`.
When CBO or plan stats are enabled, `CatalogStatistics.toPlanStats` replaces
that size with `rowCount * rowSize`. For a CHAR partition column, the pruned
relation's size now comes from the uniform `1 / ndv` estimate instead of the
size of the remaining files.
With skewed partitions, the error goes the unsafe way. Suppose a table has
10 partition values and `p = '3'` holds 90% of the rows. The row count used to
stay at the table's row count, an overestimate. It is now 10% of it, so
`sizeInBytes` is about 9x too small, which can make the relation look small
enough to broadcast.
STRING partition columns already behave this way, so I don't think this
blocks the PR. Could the description mention that, with CBO, this also changes
the pruned relation's size?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/statsEstimation/EstimationUtils.scala:
##########
@@ -87,6 +88,15 @@ 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.
Review Comment:
**10. [Nit] Could the comment mention the distinct count?**
Following up on your reply to comment 1: I agree that the distinct count can
only be too high when stored values differ just in trailing spaces. Since the
pass-through adjusts the lengths but not the distinct count, could the comment
say so? Before this PR such predicates were estimated at 1.0, so a too-high
distinct count now gives an underestimate. For example:
```scala
// 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.
// Their distinct count is left as is. It is too high only when stored
values differ
// just in trailing spaces, which the padding makes equal.
```
--
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]
