dtenedor commented on code in PR #59716:
URL: https://github.com/apache/spark/pull/59716#discussion_r4352511490
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/V2ExpressionBuilder.scala:
##########
@@ -377,6 +377,17 @@ class V2ExpressionBuilder(e: Expression, isPredicate:
Boolean = false) extends L
case _ =>
None
}
+ // Scan CHAR padding is a JVM invoke; V2 only has RPAD. This matches local
eval when the
+ // stored length is <= n (pad-or-keep then equals truncating RPAD).
Over-long values would
+ // be truncated remotely. charTypeReadSideCheck is left untranslated so
length errors stay
Review Comment:
`readSidePadding` is
[documented](https://github.com/apache/spark/blob/f96ffb6921640b983f4744b47625ddafb2f43f06/sql/catalyst/src/main/java/org/apache/spark/sql/catalyst/util/CharVarcharCodegenUtils.java#L96-L98)
as the variant "which tolerates over-long values", but `RPAD` truncates them.
That's fine for JDBC, where the remote `CHAR(n)` enforces the length. This
builder serves every DSv2 source, though. A connector that accepts `RPAD` but
declares `CHAR(n)` without enforcing it would return extra rows, because
`RPAD('YESNO', 3, ' ')` equals `'YES'`.
Could we either translate exactly, e.g. `CASE WHEN CHAR_LENGTH(c) < n THEN
RPAD(c, n, ' ') ELSE c END` (which needs `CHAR_LENGTH` support), or make
"stored length <= n" an explicit part of the dialect/connector opt-in instead
of a code comment?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/V2ExpressionBuilder.scala:
##########
@@ -377,6 +377,17 @@ class V2ExpressionBuilder(e: Expression, isPredicate:
Boolean = false) extends L
case _ =>
None
}
+ // Scan CHAR padding is a JVM invoke; V2 only has RPAD. This matches local
eval when the
+ // stored length is <= n (pad-or-keep then equals truncating RPAD).
Over-long values would
+ // be truncated remotely. charTypeReadSideCheck is left untranslated so
length errors stay
+ // local under spark.sql.charVarchar.standardSemantics.enabled.
+ case StaticInvoke(staticObject, dataType, "readSidePadding", arguments, _,
_, _, _, None)
+ if staticObject == classOf[CharVarcharCodegenUtils] &&
arguments.length == 2 =>
+ generateExpressionWithNameByChildren(
+ "RPAD",
+ Seq(arguments.head, arguments(1), Literal.create(" ",
arguments.head.dataType)),
Review Comment:
Two small things on this case:
- The pad literal takes the column's type. With
`spark.sql.preserveCharVarcharTypeInfo=true` that type is `CharType(n)`, so
`Literal.create` goes through `CharConverter` and pads the literal to `n`
spaces. A `CharType` `LiteralValue` then reaches connectors (`RPAD(flag, 5, '
')`). It pads the same way, but a plain string literal would be cleaner.
- Matching on the string `"readSidePadding"` duplicates the name used in
`CharVarcharUtils.addPaddingForScan`. A shared constant would keep a rename
from silently disabling this translation.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/V2ExpressionBuilder.scala:
##########
@@ -377,6 +377,17 @@ class V2ExpressionBuilder(e: Expression, isPredicate:
Boolean = false) extends L
case _ =>
None
}
+ // Scan CHAR padding is a JVM invoke; V2 only has RPAD. This matches local
eval when the
+ // stored length is <= n (pad-or-keep then equals truncating RPAD).
Over-long values would
+ // be truncated remotely. charTypeReadSideCheck is left untranslated so
length errors stay
+ // local under spark.sql.charVarchar.standardSemantics.enabled.
+ case StaticInvoke(staticObject, dataType, "readSidePadding", arguments, _,
_, _, _, None)
+ if staticObject == classOf[CharVarcharCodegenUtils] &&
arguments.length == 2 =>
+ generateExpressionWithNameByChildren(
+ "RPAD",
Review Comment:
This is blocking for me: on SQL Server it turns NULL CHAR values into spaces.
`MsSqlServerDialect` already allows `RPAD` and [emulates
it](https://github.com/apache/spark/blob/f96ffb6921640b983f4744b47625ddafb2f43f06/sql/core/src/main/scala/org/apache/spark/sql/jdbc/MsSqlServerDialect.scala#L92-L99)
as `LEFT(CONCAT(str, REPLICATE(pad, len)), len)`. SQL Server's `CONCAT`
[treats NULL as an empty
string](https://learn.microsoft.com/en-us/sql/t-sql/functions/concat-transact-sql),
so a NULL input comes back as `len` spaces. Until now that only affected
explicit `rpad()` calls. With this case it applies to every CHAR column under
the default config:
```sql
-- SQL Server: t(id INT, flag CHAR(5)) containing (2, NULL)
SELECT id FROM t WHERE flag IS NULL -- pushed as LEFT(CONCAT(...),
5) IS NULL: no rows
SELECT id FROM t WHERE flag NOT IN ('Y', 'N') -- returns the NULL row
SELECT COUNT(flag) FROM t -- aggregate pushdown counts
the NULL row
SELECT flag, COUNT(*) FROM t GROUP BY flag -- the NULL group comes back
as ' '
```
The new tests can't see this for two reasons. H2's `RPAD` handles NULL
correctly. And for `flag = 'Y'`, the optimizer infers and pushes `flag IS NOT
NULL` too, because `StaticInvoke.nullIntolerant` [is
`propagateNull`](https://github.com/apache/spark/blob/f96ffb6921640b983f4744b47625ddafb2f43f06/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/objects/objects.scala#L342).
`NullDownPropagation` [only
unwraps](https://github.com/apache/spark/blob/f96ffb6921640b983f4744b47625ddafb2f43f06/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/expressions.scala#L1145-L1150)
`Not` and comparisons, and `In` isn't null-intolerant, so
`isnull(readSidePadding(flag, 5))` and `NOT IN` reach this builder with no
guard.
Suggestions:
1. Make the SQL Server `RPAD`/`LPAD` emulation NULL-preserving, e.g. `CASE
WHEN str IS NULL OR len IS NULL OR pad IS NULL THEN NULL ELSE LEFT(CONCAT(...),
len) END`.
2. `readSidePadding` preserves NULLs, so `IsNull`/`IsNotNull` over it could
translate straight to `c IS [NOT] NULL`. That's simpler and doesn't depend on
any dialect's `RPAD` NULL handling.
3. Add a `MsSqlServerIntegrationSuite` test with NULLs covering `IS NULL`,
`NOT IN`, `COUNT`, and `GROUP BY`.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2StrategySuite.scala:
##########
@@ -1882,6 +1883,42 @@ class DataSourceV2StrategySuite extends
SharedSparkSession {
}
}
+ test("SPARK-60479: read-side CHAR padding StaticInvoke translates to RPAD") {
+ val col = AttributeReference("flag", StringType)()
+ def paddingInvoke(functionName: String): StaticInvoke = {
Review Comment:
Could you add a negative case for a `readSidePadding` invoke on a different
class, to cover the `staticObject` guard? Also, this helper builds the invoke
from the method-name string, so a rename in `CharVarcharUtils` would leave the
test green. Building the expression with `CharVarcharUtils.addPaddingForScan`
on an attribute that carries `CHAR(1)` metadata would avoid that.
##########
sql/core/src/main/scala/org/apache/spark/sql/jdbc/OracleDialect.scala:
##########
@@ -46,7 +46,7 @@ private case class OracleDialect() extends JdbcDialect with
SQLConfHelper with N
// scalastyle:on line.size.limit
private val supportedAggregateFunctions =
Set("MAX", "MIN", "SUM", "COUNT", "AVG") ++
distinctUnsupportedAggregateFunctions
- private val supportedFunctions = supportedAggregateFunctions ++ Set("TRUNC")
+ private val supportedFunctions = supportedAggregateFunctions ++ Set("TRUNC",
"RPAD")
Review Comment:
I'd leave Oracle out of this PR, or gate it, until there's an Oracle
integration test. Oracle's `RPAD` doesn't count characters the way Spark does.
Oracle
[defines](https://docs.oracle.com/en/database/oracle/oracle-database/19/sqlrf/RPAD.html)
`n` as "the total length of the return value as it is displayed on your
terminal screen", and East Asian full-width characters count as two columns.
Spark's `readSidePadding` counts code points. Some consequences:
- **Missing rows.** A `CHAR(4 CHAR)` value `'日本'` is stored as `'日本 '`.
Spark pads both the column and the literal to `'日本 '`, so `flag = '日本'`
matches. Oracle's `RPAD(flag, 4, ' ')` keeps only what fits in 4 display
columns, which is `'日本'`. That result is a VARCHAR2, so the comparison is
non-padded and the row is dropped.
- **Extra rows.** Range predicates can return rows the query should exclude,
since `'日本'` sorts below `'日本人 '`.
- **Merged groups.** With aggregate pushdown, `'日本人'` and `'日本語'` both
truncate to `'日本'` and collapse into one group.
Allowing `RPAD` here also enables pushdown of explicit `rpad()` calls, where
Oracle's `'' = NULL` rule diverges. `rpad(c, 0, ' ')` is `''` in Spark but NULL
in Oracle, and `rpad(c, 5, '')` returns `c` unpadded in Spark but NULL in
Oracle.
I haven't run this against Oracle myself, so an integration test would
settle it. The v2 `OracleIntegrationSuite` already has a `char(3)` table that
could be extended with multibyte and CJK data.
##########
sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCSuite.scala:
##########
@@ -2350,13 +2350,25 @@ class JDBCSuite extends SharedSparkSession {
assert(getJdbcType(oracleDialect, TimestampNTZType) == "TIMESTAMP")
}
+ test("SPARK-60479: Oracle dialect compiles RPAD for CHAR filter pushdown") {
+ val oracleDialect = JdbcDialects.get("jdbc:oracle://127.0.0.1/db")
+ val rpad = new GeneralScalarExpression("RPAD",
+ Array[V2Expression](
+ FieldReference("PROD_VERSION_CURRENT"),
Review Comment:
nit: `PROD_VERSION_CURRENT` looks like it came from a real schema. A neutral
name like `flag` would match the other tests.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/V2ExpressionBuilder.scala:
##########
@@ -377,6 +377,17 @@ class V2ExpressionBuilder(e: Expression, isPredicate:
Boolean = false) extends L
case _ =>
None
}
+ // Scan CHAR padding is a JVM invoke; V2 only has RPAD. This matches local
eval when the
+ // stored length is <= n (pad-or-keep then equals truncating RPAD).
Over-long values would
+ // be truncated remotely. charTypeReadSideCheck is left untranslated so
length errors stay
+ // local under spark.sql.charVarchar.standardSemantics.enabled.
+ case StaticInvoke(staticObject, dataType, "readSidePadding", arguments, _,
_, _, _, None)
+ if staticObject == classOf[CharVarcharCodegenUtils] &&
arguments.length == 2 =>
Review Comment:
Because this lives in the shared translator, it reaches further than Oracle
filters:
- Every dialect that already allows `RPAD` now gets it on the default CHAR
read path, without any dialect change. In OSS that's H2, Postgres, and SQL
Server (emulated). Postgres looks right to me, since `bpchar` drops its padding
when cast to `text` and `rpad` counts characters. But it's untested, and SQL
Server is broken for NULLs (see the comment on the `"RPAD"` line).
-
[`translateAggregation`](https://github.com/apache/spark/blob/f96ffb6921640b983f4744b47625ddafb2f43f06/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSourceStrategy.scala#L837)
and
[`translateSortOrders`](https://github.com/apache/spark/blob/f96ffb6921640b983f4744b47625ddafb2f43f06/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSourceStrategy.scala#L856)
use this builder on alias-substituted expressions. So `GROUP BY char_col`,
`MIN`/`MAX`/`COUNT(char_col)`, and `ORDER BY char_col LIMIT k` are now pushed
as well (JDBC enables aggregate and limit pushdown by default). The tests only
cover filters, and `checkAggregateRemoved`/`checkSortRemoved` in `JDBCV2Suite`
would make these easy to add.
Dialects allowed `RPAD` to support explicit `rpad()` calls, not to rewrite
every CHAR read. I'd prefer making this opt-in per dialect, through a dedicated
V2 function name or a `JdbcDialect` capability. Then each dialect can compile
the padding in whatever form is correct for it, ideally one that doesn't wrap
the column in a function, since that makes the predicate non-sargable on large
remote tables. At minimum, an internal SQLConf to turn this off would help,
since it changes the pushed SQL for existing queries.
##########
sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCV2Suite.scala:
##########
@@ -2017,6 +2017,40 @@ class JDBCV2Suite extends SharedSparkSession with
ExplainSuiteHelper {
checkAnswer(df10, Seq(Row(6, "jen", 12000, 1200, true)))
}
+ test("SPARK-60479: CHAR equality is pushed down with read-side padding") {
+ val table = "h2.test.char_pad"
+ withTable(table) {
+ sql(s"CREATE TABLE $table (id INT, flag CHAR(5))")
+ sql(s"INSERT INTO $table VALUES (1, 'Y'), (2, 'N'), (3, 'YES')")
Review Comment:
Could the table also have a NULL row and some non-ASCII values? Then this
could cover `IS NULL`, `NOT IN`, `OR`, and `<=>`, a literal longer than 5
(which yields nested `RPAD(RPAD(...))`), and a range and a `LIKE`-family
predicate. H2 won't reproduce the SQL Server or Oracle behavior, but these
cases would pin the pushed shapes, and the same cases can run in the docker
suites.
##########
sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCV2Suite.scala:
##########
@@ -2017,6 +2017,40 @@ class JDBCV2Suite extends SharedSparkSession with
ExplainSuiteHelper {
checkAnswer(df10, Seq(Row(6, "jen", 12000, 1200, true)))
}
+ test("SPARK-60479: CHAR equality is pushed down with read-side padding") {
+ val table = "h2.test.char_pad"
+ withTable(table) {
+ sql(s"CREATE TABLE $table (id INT, flag CHAR(5))")
+ sql(s"INSERT INTO $table VALUES (1, 'Y'), (2, 'N'), (3, 'YES')")
+
+ val eq = sql(s"SELECT id FROM $table WHERE flag = 'Y'")
+ checkFiltersRemoved(eq)
+ checkPushedInfo(eq, "flag IS NOT NULL", "(RPAD(flag, 5, ' ')) = 'Y '")
+ checkAnswer(eq, Seq(Row(1)))
+
+ val in = sql(s"SELECT id FROM $table WHERE flag IN ('Y', 'N')")
+ checkFiltersRemoved(in)
+ checkPushedInfo(in, "RPAD(flag, 5, ' ') IN ('Y ', 'N ')")
Review Comment:
`checkPushedInfo` is a contains-check, so this doesn't pin whether `flag IS
NOT NULL` is also pushed. For `=` it is, through constraint inference, and that
guard is what hides dialect differences in `RPAD`'s NULL handling. `IN` and
`NOT IN` don't get it, so asserting the full `PushedFilters` here would make
that explicit.
--
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]
