cloud-fan commented on code in PR #59716:
URL: https://github.com/apache/spark/pull/59716#discussion_r4395624810
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/jdbc/JDBCScanBuilder.scala:
##########
@@ -66,11 +67,25 @@ case class JDBCScanBuilder(
private var sortOrders: Array[String] = Array.empty[String]
+ private def containsUnsupportedCharPad(pred: Predicate): Boolean = {
+ if
(dialect.isSupportedFunction(CharVarcharUtils.V2_CHAR_READ_SIDE_PADDING)) {
+ false
+ } else {
+ def walk(e: Expression): Boolean = e match {
+ case g: GeneralScalarExpression
+ if g.name() == CharVarcharUtils.V2_CHAR_READ_SIDE_PADDING => true
+ case _ => Option(e.children()).exists(_.exists(walk))
+ }
+ walk(pred)
+ }
+ }
+
override def pushPredicates(predicates: Array[Predicate]): Array[Predicate]
= {
if (jdbcOptions.pushDownPredicate) {
- val (pushed, unSupported) =
predicates.partition(dialect.compileExpression(_).isDefined)
+ val (unsupportedCharPad, rest) =
predicates.partition(containsUnsupportedCharPad)
Review Comment:
**Nit (P3):** This decline only runs for filters. V2ExpressionBuilder also
translates a CHAR `=`, `<=>`, `<>`, or `IN` nested inside other expressions; a
`CASE WHEN` condition, for example, goes through `generateExpression(_, true)`.
`rewriteAggregate` substitutes the read-side padding alias into aggregate and
grouping expressions, so on a MySQL catalog table `t(flag CHAR(1), amt INT)`
```sql
SELECT SUM(CASE WHEN flag = 'Y' THEN amt END) FROM mysql.t
```
reaches `pushAggregation` as `SUM(CASE WHEN CHAR_READ_SIDE_PADDING(flag, 1,
' ') = 'Y' ...)`. `pushAggregation` (lines 113 and 116), `pushTopN` (line 334),
and `pushDownJoin` (line 222) call `dialect.compileExpression` directly.
`JDBCSQLBuilder.build` throws `_LEGACY_ERROR_TEMP_3177`, and
`compileExpression` logs `Error occurs while compiling V2 expression` with the
stack trace before Spark falls back to local evaluation. That is the same
planning-failure-looking log this decline removes for filters. It now shows up
on aggregate, top-N, and join planning for MySQL, DB2, Derby, Teradata,
Snowflake, Databricks, the base dialect, and third-party dialects. Results
don't change.
Generalizing `containsUnsupportedCharPad` from `Predicate` to `Expression`
and checking it before those three compile calls would keep them quiet too.
**Verification:**
- **Regression:** The scan builder of a dialect that does not opt in
declines an aggregation whose argument contains a CHAR comparison over
CHAR_READ_SIDE_PADDING, and a top-N sort order containing one, without logging
the compile warning.
- **Behavior:** On an opt-in dialect, an aggregate such as SUM(CASE WHEN
flag = 'Y' THEN 1 ELSE 0 END) over a CHAR column is pushed and returns the same
value as Spark-local evaluation.
- **Compatibility:** On dialects that do not opt in, CHAR comparisons in
filters still stay post-scan and IS [NOT] NULL is still pushed on the raw
column, and aggregates, top-N, and joins without the function push exactly as
before.
##########
sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCSuite.scala:
##########
@@ -1656,6 +1658,8 @@ class JDBCSuite extends SharedSparkSession {
Some(ArrayType(DecimalType.SYSTEM_DEFAULT)))
assert(Postgres.getCatalystType(java.sql.Types.ARRAY, "_bpchar", 64, md) ==
Some(ArrayType(CharType(64))))
+ assert(Postgres.getCatalystType(java.sql.Types.CHAR, "char", 1, md) ===
Some(StringType))
Review Comment:
**Nit (P3):** This pins the scalar internal `"char"` mapping, but the PR
also changes its array element path. `toCatalystType` now has `case "char" =>
Some(StringType)`, so a `"char"[]` column (JDBC type name `_char`) reads as
`ARRAY<STRING>` instead of `ARRAY<CHAR(n)>`. Nothing calls
`getCatalystType(Types.ARRAY, "_char", ...)`, and no docker suite declares a
`"char"[]` column. Re-merging the arm with `bpchar` would leave every test
green, and empty elements would read as `' '` again. An assertion next to the
`_bpchar` one, such as `Postgres.getCatalystType(java.sql.Types.ARRAY, "_char",
1, md) == Some(ArrayType(StringType))`, would pin it.
**Verification:**
- **Regression:** Reverting the internal "char" array element mapping to
CHAR(n) makes the PostgreSQL type-mapping test fail.
- **Compatibility:** The _bpchar array mapping and the scalar char and
bpchar mappings keep their current expectations.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2StrategySuite.scala:
##########
@@ -1882,6 +1883,115 @@ class DataSourceV2StrategySuite extends
SharedSparkSession {
}
}
+ test("SPARK-60479: read-side CHAR padding StaticInvoke translates") {
+ val md = new MetadataBuilder()
+ .putString(CharVarcharUtils.CHAR_VARCHAR_TYPE_STRING_METADATA_KEY,
"char(1)")
+ .build()
+ val col = AttributeReference("flag", StringType, metadata = md)()
+ val padded = CharVarcharUtils.addPaddingForScan(col)
+ val eq = EqualTo(padded, Literal("Y"))
+ val result = new V2ExpressionBuilder(eq, isPredicate = true).build()
+ assert(result.isDefined, "readSidePadding should translate to a V2
predicate")
+ result.get match {
+ case p: Predicate if p.name() == "=" =>
+ p.children()(0) match {
+ case g: GeneralScalarExpression
+ if g.name() == CharVarcharUtils.V2_CHAR_READ_SIDE_PADDING =>
+ assert(g.children()(0) === FieldReference("flag"))
+ assert(g.children()(1) === LiteralValue(1, IntegerType))
+ assert(g.children()(2) ===
+ LiteralValue(UTF8String.fromString(" "), StringType))
+ case other =>
+ fail(s"expected CHAR_READ_SIDE_PADDING, got ${other.describe()}")
+ }
+ assert(p.children()(1) === LiteralValue(UTF8String.fromString("Y"),
StringType))
+ case other =>
+ fail(s"expected '=' predicate, got ${other.describe()}")
+ }
+
+ val isNull = new V2ExpressionBuilder(IsNull(padded), isPredicate =
true).build()
+ assert(isNull.isDefined)
+ isNull.get match {
+ case p: Predicate if p.name() == "IS_NULL" =>
+ assert(p.children()(0) === FieldReference("flag"))
+ case other => fail(s"expected IS_NULL on the raw column, got
${other.describe()}")
+ }
+
+ val nseq = new V2ExpressionBuilder(EqualNullSafe(padded, Literal("Y")),
isPredicate = true)
+ .build()
+ assert(nseq.isDefined)
Review Comment:
**Nit (P3):** Unlike the `=`, `<>`, and `InSet` cases here, this only checks
that some predicate comes back. The JDBCV2Suite `flag <=> 'Y'` query doesn't
pin the operator either. A `WHERE` clause drops the NULL row whether the pushed
predicate is `<=>` or `=`, so the answer is the same. The difference shows up
under `NOT`: `NOT (flag <=> 'Y')` must return the NULL row, and JDBC keeps it
only while the operator stays `<=>`, which it renders with the NULL-aware `IS
NULL` form. If this arm emitted `=`, the remote would drop that row with
Spark's filter already removed, and every current assertion would still pass.
Asserting the `<=>` name, the wrapper child, and the padded literal, as the `=`
case does, plus an H2 `NOT (flag <=> 'Y')` query with `checkPushedInfo` and
`checkAnswer`, would pin it.
**Verification:**
- **Regression:** Translating EqualNullSafe over the padded CHAR column with
= instead of <=> fails both the translation-shape assertion and the end-to-end
NOT (flag <=> 'Y') answer.
- **Behavior:** The pushed NOT (flag <=> 'Y') filter leaves no post-scan
filter and returns the rows whose flag is not 'Y' together with the NULL row.
##########
docs/sql-migration-guide.md:
##########
@@ -40,6 +40,7 @@ license: |
- Since Spark 4.4, when an already-analyzed Data Source V2 query is refreshed
after a compatible schema change, connectors can return more data columns from
the current table schema in `Scan.readSchema()` than requested by
`SupportsPushDownRequiredColumns.pruneColumns`. Previously, this partial
pruning could fail planning because the scan reported columns absent from the
analyzed relation output.
- Since Spark 4.4, when an already-analyzed Data Source V2 query is refreshed,
if rebinding an expanded struct used as a map key causes distinct current keys
to collide under the analyzed schema, the query fails with `DUPLICATED_MAP_KEY`
by default instead of returning duplicate keys. With
`spark.sql.mapKeyDedupPolicy=LAST_WIN`, the last value is kept.
- Since Spark 4.4, the Oracle JDBC datasource maps bare `NUMBER` columns
(declared without explicit precision or scale) to `DecimalType(38, 18)` instead
of `DecimalType(38, 10)`. This uses `DecimalType.DEFAULT_SCALE` and preserves
up to 18 fractional digits instead of 10, reducing silent precision loss.
However, the integer range shrinks from 28 to 20 digits; bare `NUMBER` values
with more than 20 integer digits that previously read correctly will raise
`NUMERIC_VALUE_OUT_OF_RANGE`. The same change applies to Oracle `FLOAT`
columns, which JDBC reports as `NUMERIC` with `scale = -127`. Columns with
explicit `NUMBER(p,s)` are not affected. To restore the previous
`DecimalType(38, 10)` mapping, set
`spark.sql.legacy.oracle.numberMapping.enabled` to `true`.
+- Since Spark 4.4, JDBC Data Source V2 catalog tables on H2, PostgreSQL, SQL
Server, and Oracle push CHAR equality, inequality, null-safe equality,
`IN`/`NOT IN`, and `IS [NOT] NULL` again when `spark.sql.readSideCharPadding`
is true (the default). In 4.3 the default read-side padding wrap kept those
filters in Spark. Pushed predicates use the remote collation, so a
case-insensitive SQL Server collation can match a stored `'ABC'` with `flag =
'abc'` while a Spark-local comparison would not. The V1 path (`spark.read.jdbc`
and `USING jdbc`) still evaluates these CHAR filters in Spark. PostgreSQL's
internal `"char"` type (JDBC type name `char`) is mapped to `STRING` instead of
`CHAR(1)`, so Spark no longer blank-pads its empty value.
Review Comment:
**Nit (P3):** Two scope statements here don't match the code. First, line 32
of this section says later CHAR/VARCHAR bullets that don't name
`preserveCharVarcharTypeInfo` apply when
`spark.sql.charVarchar.standardSemantics.enabled` is true. This pushdown
happens only when that flag is false. Under standard semantics the scan gets
`charTypeReadSideCheck`, which isn't translated, and JDBCV2Suite asserts that
the equality stays in Spark. Read under that rule, this bullet describes the
wrong mode. Second, the `IS [NOT] NULL` unwrap in `V2ExpressionBuilder` has no
dialect gate. Every JDBC dialect and every DSv2 source gets the raw-column null
check, not only H2, PostgreSQL, SQL Server, and Oracle, and JDBCSuite asserts
that Derby pushes it. Naming the standard-semantics condition and moving `IS
[NOT] NULL` out of the four-dialect list would fix both.
**Verification:**
- **Inspection:** The bullet names the standard-semantics condition and
separates the IS [NOT] NULL change from the four-dialect equality and IN
change, consistent with the translation and JDBC tests.
--
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]
