dongjoon-hyun commented on code in PR #60025:
URL: https://github.com/apache/spark/pull/60025#discussion_r4449329587
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/statsEstimation/FilterEstimation.scala:
##########
@@ -170,7 +171,14 @@ case class FilterEstimation(
* It returns None if the condition is not supported.
*/
def calculateSingleCondition(condition: Expression, update: Boolean):
Option[Double] = {
- condition match {
+ // A predicate pushed below a read-side CHAR padding Project compares the
padding expression
+ // rather than the column. The column's statistics already describe padded
values (see
+ // `EstimationUtils.getAliasStats`), so estimate it as a predicate on the
column itself.
+ val unpadded = condition.mapChildren {
+ case CharVarcharUtils.ReadSidePadding(attr: Attribute) => attr
Review Comment:
**3. [Low] The unwrap also changes partition pruning and DSv2 estimates**
`FilterEstimation` is not only used for `Filter` nodes, so these callers now
estimate predicates on padded CHAR columns too:
- `PruneFileSourcePartitions` estimates the partition filters and saves the
result as the pruned table's `CatalogStatistics` (so does
`PruneHiveTablePartitions` when the partitions have no row counts). For a
partition column `p CHAR(5)` and `WHERE p = 'a'`, the pruned relation's
`rowCount` used to stay at the table's row count. It now becomes the row count
divided by the NDV, and the distinct count of `p` is saved as 1.
- `DataSourceV2Relation.estimateStatsWithFilters` estimates the pushed and
inferred filters against the statistics the connector reports.
These look like improvements, but the description and the tests don't
mention them. Could the description list them, and could a test cover at least
the partition pruning case?
##########
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.
+ */
+ private[sql] object ReadSidePadding {
Review Comment:
**2. [Medium] Standard semantics still lose the statistics**
`ReadSidePadding` only matches `readSidePadding`. With
`spark.sql.charVarchar.standardSemantics.enabled`, `addPaddingForScan` wraps
CHAR columns in `charTypeReadSideCheck` and VARCHAR columns in
`varcharTypeReadSideCheck` instead. Both still lose their statistics at the
first `Project`, and their filters are still estimated with selectivity 1.0.
For VARCHAR this is new in 4.4.0, because VARCHAR had no read-side wrapper
before that config.
The reasoning in `getAliasStats` holds there as well: `ANALYZE TABLE` reads
through the same checks, so the stats describe the values the checks return.
The PR description says these columns behave as before. Since the config is new
in 4.4.0, could this PR cover them too, or is there a JIRA for it?
If they are added, they need an extractor of their own.
`V2ExpressionBuilder` now uses `ReadSidePadding` to decide when to compile
`CHAR_READ_SIDE_PADDING`, and that must not match the checks, which trim and
fail on over-long values.
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/statsEstimation/FilterEstimationSuite.scala:
##########
@@ -963,6 +964,26 @@ 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,
Review Comment:
**4. [Low] `InSet` and `IsNotNull` are not covered**
These cases cover `=`, `IN` with two values and `IS NULL`. Two other
branches of `calculateSingleCondition` that a padded column reaches have no
test:
- `InSet`: `OptimizeIn` turns an `IN` list with more than
`spark.sql.optimizer.inSetConversionThreshold` (10) values into
`InSet(readSidePadding(c, 5), set)`. The `IN ('3', '4')` query in
`StatisticsCollectionSuite` stays an `In`.
- `IsNotNull`, which `InferFiltersFromConstraints` adds next to the equality.
Could two more cases be added here? For example:
```scala
InSet(padded, Set(UTF8String.fromString("A0"), UTF8String.fromString("A1")))
-> 2,
IsNotNull(padded) -> 10,
```
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/statsEstimation/FilterEstimation.scala:
##########
@@ -170,7 +171,14 @@ case class FilterEstimation(
* It returns None if the condition is not supported.
*/
def calculateSingleCondition(condition: Expression, update: Boolean):
Option[Double] = {
- condition match {
+ // A predicate pushed below a read-side CHAR padding Project compares the
padding expression
Review Comment:
**6. [Nit] The scaladoc still says one side must be a column**
The scaladoc of this method (L164-L165) still says that only binary
predicates with a column on one side and a literal on the other are supported.
Could the rule in this new comment move into the scaladoc, e.g. "A side may
also be the read-side CHAR padding of a column, which is estimated as the
column itself"?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/V2ExpressionBuilder.scala:
##########
@@ -484,12 +484,8 @@ class V2ExpressionBuilder(e: Expression, isPredicate:
Boolean = false) extends L
}
}
- private def isCharReadSidePaddingInvoke(invoke: StaticInvoke): Boolean = {
- invoke.scalarFunction.isEmpty &&
- invoke.staticObject == classOf[CharVarcharCodegenUtils] &&
- invoke.functionName == CharVarcharUtils.READ_SIDE_PADDING &&
- invoke.arguments.length == 2
- }
+ private def isCharReadSidePaddingInvoke(invoke: StaticInvoke): Boolean =
Review Comment:
**5. [Nit] The helper could go away**
This wrapper now only adds `.unapply(invoke).isDefined`, and its callers
still read `invoke.arguments.head` and `invoke.arguments(1)` by position. If
the extractor returned both arguments, the call sites could match it directly,
and `charPadLength` could become a `Literal(n: Int, _)` pattern:
```scala
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 &&
s.arguments.length == 2 =>
Some((s.arguments.head, s.arguments(1)))
case _ => None
}
}
```
For example, `case IsNull(CharVarcharUtils.ReadSidePadding(child, _))` here.
Feel free to skip this if you prefer to keep the change small.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/statsEstimation/EstimationUtils.scala:
##########
@@ -87,6 +88,11 @@ 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.
+ case alias @ Alias(CharVarcharUtils.ReadSidePadding(attr: Attribute), _)
Review Comment:
**1. [Medium] The pass-through assumes the stats were collected through the
padding**
This hands the column's `ColumnStat` to the padded alias whenever the column
has stats, but that is only exact when the stats were collected through the
same padding, and nothing records how they were collected:
- A table analyzed with `spark.sql.readSideCharPadding=false`, or before
3.4.0, over files that Spark did not write has stats of the unpadded values.
The PR description mentions this case.
- `ApplyCharTypePadding` also pads `DataSourceV2Relation`, whose column
stats come from the connector through `DataSourceV2Relation.transformV2Stats`.
Those describe the stored values, which `ANALYZE TABLE` never read.
The distinct count can then be too high, but only when stored values differ
just in trailing spaces. The lengths matter more. `getSizePerRow` counts a
string attribute as `avgLen + 12` bytes when it has stats and as `defaultSize`
(20) when it does not. A `CHAR(100)` column whose stored values average 3
characters is now counted as 15 bytes per row instead of 20, while each padded
value takes 112. The smaller `sizeInBytes` can make a large side look small
enough to broadcast.
Every non-null padded value has at least N characters, so could the
pass-through raise `avgLen` and `maxLen` to at least N? Assuming
`ReadSidePadding` also returns the length argument (see comment 5), something
like:
```scala
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)))
```
--
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]
