srielau commented on code in PR #59716:
URL: https://github.com/apache/spark/pull/59716#discussion_r4396520860
##########
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:
Generalized `containsUnsupportedCharPad` to `Expression` and check it before
`pushAggregation`, `pushTopN`, and `pushDownJoin` compile. Derby in `JDBCSuite`
now declines a SUM(CASE WHEN CHAR pad = ...) aggregate and a CHAR-pad sort
order with no compile warning. H2 pushes `SUM(CASE WHEN flag = 'Y' THEN 1 ELSE
0 END)` and returns 1.
##########
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:
The `<=>` unit test now asserts the operator name, wrapper child, and padded
literal. H2 runs `flag <=> 'Y'` and `NOT (flag <=> 'Y')` with `checkPushedInfo`
(the expanded null-aware form) and `checkAnswer`; the NULL row comes back only
for the NOT query.
##########
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:
Split into two bullets that both name
`spark.sql.charVarchar.standardSemantics.enabled` is false. Equality/`IN` stays
on the four dialects; `IS [NOT] NULL` unwrap is every DSv2 source, including
every JDBC dialect.
##########
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:
Added `Postgres.getCatalystType(Types.ARRAY, "_char", 1, md) ==
Some(ArrayType(StringType))` next to the `_bpchar` assert.
--
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]
