This is an automated email from the ASF dual-hosted git repository.
cloud-fan pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 9d01b6320d10 [SPARK-60479][SQL] Push down CHAR filters when read-side
padding is enabled
9d01b6320d10 is described below
commit 9d01b6320d109168b14904e5f8bf7439f88508cb
Author: Serge Rielau <[email protected]>
AuthorDate: Wed Oct 28 16:02:59 2026 +0000
[SPARK-60479][SQL] Push down CHAR filters when read-side padding is enabled
### What changes were proposed in this pull request?
When `spark.sql.readSideCharPadding` is true (the default), CHAR scan
columns are wrapped in `StaticInvoke(CharVarcharCodegenUtils.readSidePadding,
col, n)`. DSv2 translation did not understand that invoke, so JDBC CHAR filters
that used to push as `flag = 'Y'` stayed local.
This PR restores that pushdown for JDBC **catalog (Data Source V2)**
tables, for the predicates that were sargable before the wrap. It does not
restore V1 `spark.read.jdbc` / `USING jdbc`. It maps `readSidePadding` to a
dedicated V2 function `CHAR_READ_SIDE_PADDING` (not SQL `RPAD`), and only emits
it for:
- `=`, `<=>`, `<>` against a foldable string whose length is the CHAR width
- `IN` / `NOT IN` of such strings
- `IS NULL` / `IS NOT NULL` (unwrapped to the raw column, on every DSv2
source)
JDBC dialects that opt in compile the function as the column. Spark still
pads the literal, so the remote predicate is `flag = 'Y '`. Ranges, `LIKE`,
nested `rpad`/`lpad`, and column-to-column compares stay local. Aggregates,
`GROUP BY`, and `ORDER BY` on the CHAR column itself stay local; on the four
opt-in dialects, an aggregate, sort, or join expression that contains a CHAR
equality or `IN` against a literal, such as `SUM(CASE WHEN flag = 'Y' THEN 1
ELSE 0 END)`, can be pushed. [...]
The generic `V2ExpressionSQLBuilder` does not allowlist the name. Base
`JdbcDialect.isSupportedFunction` stays false. Opt-in is H2, Postgres, SQL
Server, and Oracle. Returning true for `CHAR_READ_SIDE_PADDING` means the
remote compares the raw CHAR column with a blank-padded (PAD SPACE) literal the
same way Spark compares the padded value, and stored values never exceed that
length.
PostgreSQL's internal `"char"` type (JDBC type name `char`) is mapped to
`STRING` so it is not padded and is not compiled this way. Oracle `IN` lists
longer than 1000 values are split into `OR`-ed chunks.
### Why are the changes needed?
Read-side CHAR padding is on by default and wraps every DSv2 CHAR scan
column. That wrapper blocked V2 filter translation for JDBC CHAR equality. The
code before that wrap already pushed a native column predicate; remote CHAR
comparison already blank-pads. Compiling the wrapper as `RPAD` (or as the
column in every expression context) changes results on some dialects. This PR
only undoes the equality/IN/null regression on the DSv2 catalog path.
### Does this PR introduce _any_ user-facing change?
Yes. JDBC **catalog (DSv2)** queries that filter CHAR columns with equality
or `IN` can push those predicates again on H2, Postgres, SQL Server, and Oracle
when standard CHAR/VARCHAR semantics are off. That restores the
pre-readSidePadding JDBC path: the remote compares the stored CHAR value with
Spark's padded literal, including the remote collation. That is the same as
pushed VARCHAR equality. A case- or accent-insensitive remote collation can
therefore match rows that Spark-local b [...]
### How was this patch tested?
- `DataSourceV2StrategySuite`: equality, `<=>`, `<>`, `IN`, and `InSet`
translate to `CHAR_READ_SIDE_PADDING` (wrapper child and padded literal); `IS
NULL` unwraps to the raw column; `IN` with a NULL literal, over-long literals,
ranges, CHAR vs column, nested `rpad`, other classes, and
`charTypeReadSideCheck` do not translate
- `JDBCSuite`: H2/Postgres/SQL Server/Oracle compile
`CHAR_READ_SIDE_PADDING` as the quoted column; MySQL/DB2/Derby do not; Oracle
still does not compile SQL `RPAD`; Oracle IN of 2 values is unchanged, IN of
1001 values is OR-ed chunks of 1000; Postgres `char` and `_char` arrays map to
`STRING` / `ARRAY<STRING>` and `bpchar(n)` still defers; non-opt-in Derby skips
CHAR pad compile for filters, aggregates, and top-N with no warning and still
pushes `IS NULL`
- `JDBCV2Suite`: H2 `CHAR(5)` equality, `<>`, `IN`/`NOT IN`, `InSet`
(threshold 2), `IS NULL`, `OR`, `<=>`, and `NOT (flag <=> 'Y')` are pushed;
`SUM(CASE WHEN flag = 'Y' THEN 1 ELSE 0 END)` is pushed; long literals, range,
`LIKE`, CHAR vs VARCHAR, nested `rpad`, `COUNT`, `GROUP BY`, and top-N on the
CHAR column stay local; padding-false and standard-semantics paths
- Docker suites (not run here): SQL Server and Postgres CHAR equality plus
NULL and a non-matching `YES` row; Postgres internal `"char"`; Oracle CHAR
equality including CJK on `CHAR(4 CHAR)` and multibyte on `CHAR(4 BYTE)`
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Cursor Grok 4.6
Closes #59716 from srielau/SPARK-60479.
Authored-by: Serge Rielau <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
(cherry picked from commit 4913af9161d50cdcb5f15b3450604e431879033e)
Signed-off-by: Wenchen Fan <[email protected]>
---
.../sql/jdbc/v2/MsSqlServerIntegrationSuite.scala | 27 ++++
.../spark/sql/jdbc/v2/OracleIntegrationSuite.scala | 60 ++++++++
.../sql/jdbc/v2/PostgresIntegrationSuite.scala | 46 ++++++
docs/sql-migration-guide.md | 2 +
.../expressions/GeneralScalarExpression.java | 16 +++
.../spark/sql/catalyst/util/CharVarcharUtils.scala | 24 +++-
.../sql/catalyst/util/V2ExpressionBuilder.scala | 160 ++++++++++++++++-----
.../datasources/v2/jdbc/JDBCScanBuilder.scala | 30 +++-
.../org/apache/spark/sql/jdbc/H2Dialect.scala | 4 +-
.../org/apache/spark/sql/jdbc/JdbcDialects.scala | 26 +++-
.../apache/spark/sql/jdbc/MsSqlServerDialect.scala | 4 +-
.../org/apache/spark/sql/jdbc/OracleDialect.scala | 28 +++-
.../apache/spark/sql/jdbc/PostgresDialect.scala | 10 +-
.../datasources/v2/DataSourceV2StrategySuite.scala | 129 ++++++++++++++++-
.../org/apache/spark/sql/jdbc/JDBCSuite.scala | 91 +++++++++++-
.../org/apache/spark/sql/jdbc/JDBCV2Suite.scala | 135 +++++++++++++++++
16 files changed, 742 insertions(+), 50 deletions(-)
diff --git
a/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/v2/MsSqlServerIntegrationSuite.scala
b/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/v2/MsSqlServerIntegrationSuite.scala
index 85f23efcde44..1bc534d242c7 100644
---
a/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/v2/MsSqlServerIntegrationSuite.scala
+++
b/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/v2/MsSqlServerIntegrationSuite.scala
@@ -294,4 +294,31 @@ class MsSqlServerIntegrationSuite extends
DockerJDBCIntegrationV2Suite with V2JD
assert(rows.length == 1)
assert(rows(0).getString(0) === "amy")
}
+
+ test("SPARK-60479: CHAR read-side padding with NULL") {
+ val table = s"$catalogName.char_pad"
+ withTable(table) {
+ sql(s"CREATE TABLE $table (id INT, flag CHAR(5))")
+ sql(s"INSERT INTO $table VALUES (1, 'Y'), (2, NULL), (3, 'YES')")
+ val eq = sql(s"SELECT id FROM $table WHERE flag = 'Y'")
+ checkFilterPushed(eq)
+ val eqRows = eq.collect()
+ assert(eqRows.length === 1)
+ assert(eqRows(0).getInt(0) === 1)
+ val isNull = sql(s"SELECT id FROM $table WHERE flag IS NULL")
+ checkFilterPushed(isNull)
+ val nullRows = isNull.collect()
+ assert(nullRows.length === 1)
+ assert(nullRows(0).getInt(0) === 2)
+ val notIn = sql(s"SELECT id FROM $table WHERE flag NOT IN ('Y', 'N')")
+ checkFilterPushed(notIn)
+ val notInRows = notIn.collect().map(_.getInt(0)).toSet
+ assert(notInRows === Set(3))
+ val cnt = sql(s"SELECT COUNT(flag) FROM $table")
+ assert(cnt.collect()(0).getLong(0) === 2)
+ val grp = sql(s"SELECT flag IS NULL, COUNT(*) FROM $table GROUP BY flag")
+ val grpRows = grp.collect().map(r => (r.getBoolean(0),
r.getLong(1))).toSet
+ assert(grpRows === Set((false, 1L), (true, 1L)))
+ }
+ }
}
diff --git
a/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/v2/OracleIntegrationSuite.scala
b/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/v2/OracleIntegrationSuite.scala
index 207f259964cf..7920084527e3 100644
---
a/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/v2/OracleIntegrationSuite.scala
+++
b/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/v2/OracleIntegrationSuite.scala
@@ -428,4 +428,64 @@ class OracleIntegrationSuite extends
DockerJDBCIntegrationV2Suite with V2JDBCTes
assert(rows11(0).getString(0) === "amy")
assert(rows11(1).getString(0) === "alex")
}
+
+ test("SPARK-60479: CHAR equality is pushed, including CJK") {
+ val tableName = caseConvert("char_pad")
+ val table = s"$catalogName.$tableName"
+ withTable(table) {
+ // CHAR(n CHAR) so CJK occupies one length unit, matching Spark CHAR
padding.
+ Using.resource(getConnection()) { conn =>
+ conn.prepareStatement(
+ s"CREATE TABLE $tableName (id NUMBER, flag CHAR(4
CHAR))").executeUpdate()
+ }
+ // scalastyle:off nonascii
+ sql(s"INSERT INTO $table VALUES (1, 'Y'), (2, NULL), (3, '日本')")
+ // scalastyle:on nonascii
+ val eq = sql(s"SELECT id FROM $table WHERE flag = 'Y'")
+ checkFilterPushed(eq)
+ val eqRows = eq.collect()
+ assert(eqRows.length === 1)
+ assert(eqRows(0).getDecimal(0).intValueExact() === 1)
+ // scalastyle:off nonascii
+ val cjk = sql(s"SELECT id FROM $table WHERE flag = '日本'")
+ // scalastyle:on nonascii
+ checkFilterPushed(cjk)
+ val cjkRows = cjk.collect()
+ assert(cjkRows.length === 1)
+ assert(cjkRows(0).getDecimal(0).intValueExact() === 3)
+ val isNull = sql(s"SELECT id FROM $table WHERE flag IS NULL")
+ checkFilterPushed(isNull)
+ val nullRows = isNull.collect()
+ assert(nullRows.length === 1)
+ assert(nullRows(0).getDecimal(0).intValueExact() === 2)
+ }
+ }
+
+ test("SPARK-60479: CHAR(n BYTE) multibyte equality is pushed") {
+ val tableName = caseConvert("char_byte")
+ val table = s"$catalogName.$tableName"
+ withTable(table) {
+ Using.resource(getConnection()) { conn =>
+ conn.prepareStatement(
+ s"CREATE TABLE $tableName (id NUMBER, flag CHAR(4
BYTE))").executeUpdate()
+ // Insert via JDBC. A Spark INSERT would pad the 2-byte character to
+ // CHAR width in characters (5 bytes) and Oracle would raise ORA-12899.
+ // scalastyle:off nonascii
+ conn.prepareStatement(
+ s"INSERT INTO $tableName VALUES (1, 'ä'), (2, 'äb'), (3,
'Y')").executeUpdate()
+ // scalastyle:on nonascii
+ }
+ // scalastyle:off nonascii
+ val eq = sql(s"SELECT id FROM $table WHERE flag = 'ä'")
+ checkFilterPushed(eq)
+ assert(eq.collect().map(_.getDecimal(0).intValueExact()).toSet ===
Set(1))
+ val eqAb = sql(s"SELECT id FROM $table WHERE flag = 'äb'")
+ checkFilterPushed(eqAb)
+ assert(eqAb.collect().map(_.getDecimal(0).intValueExact()).toSet ===
Set(2))
+ // scalastyle:on nonascii
+ val eqY = sql(s"SELECT id FROM $table WHERE flag = 'Y'")
+ checkFilterPushed(eqY)
+ assert(eqY.collect().map(_.getDecimal(0).intValueExact()).toSet ===
Set(3))
+ }
+ }
}
diff --git
a/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/v2/PostgresIntegrationSuite.scala
b/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/v2/PostgresIntegrationSuite.scala
index 9b349b0484ff..311002da4d89 100644
---
a/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/v2/PostgresIntegrationSuite.scala
+++
b/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/v2/PostgresIntegrationSuite.scala
@@ -19,6 +19,8 @@ package org.apache.spark.sql.jdbc.v2
import java.sql.Connection
+import scala.util.Using
+
import org.apache.spark.{SparkConf, SparkSQLException}
import org.apache.spark.sql.AnalysisException
import org.apache.spark.sql.catalyst.analysis.TableAlreadyExistsException
@@ -416,4 +418,48 @@ class PostgresIntegrationSuite extends
DockerJDBCIntegrationV2Suite with V2JDBCT
assert(rows.length === 1)
assert(rows(0).getString(0) === "amy")
}
+
+ test("SPARK-60479: CHAR read-side padding with NULL") {
+ val table = s"$catalogName.char_pad"
+ withTable(table) {
+ sql(s"CREATE TABLE $table (id INT, flag CHAR(5))")
+ sql(s"INSERT INTO $table VALUES (1, 'Y'), (2, NULL), (3, 'YES')")
+ val eq = sql(s"SELECT id FROM $table WHERE flag = 'Y'")
+ checkFilterPushed(eq)
+ val eqRows = eq.collect()
+ assert(eqRows.length === 1)
+ assert(eqRows(0).getInt(0) === 1)
+ val isNull = sql(s"SELECT id FROM $table WHERE flag IS NULL")
+ checkFilterPushed(isNull)
+ val nullRows = isNull.collect()
+ assert(nullRows.length === 1)
+ assert(nullRows(0).getInt(0) === 2)
+ val notIn = sql(s"SELECT id FROM $table WHERE flag NOT IN ('Y', 'N')")
+ checkFilterPushed(notIn)
+ val notInRows = notIn.collect().map(_.getInt(0)).toSet
+ assert(notInRows === Set(3))
+ val cnt = sql(s"SELECT COUNT(flag) FROM $table")
+ assert(cnt.collect()(0).getLong(0) === 2)
+ }
+ }
+
+ test("SPARK-60479: Postgres internal char maps to STRING") {
+ val tableName = "char_internal"
+ val table = s"$catalogName.$tableName"
+ withTable(table) {
+ Using.resource(getConnection()) { conn =>
+ conn.prepareStatement(
+ s"""CREATE TABLE $tableName (id INT, c "char")""").executeUpdate()
+ conn.prepareStatement(
+ s"INSERT INTO $tableName VALUES (1, ''), (2, 'a'), (3, '
')").executeUpdate()
+ }
+ assert(spark.table(table).schema("c").dataType === StringType)
+ val eqSpace = sql(s"SELECT id FROM $table WHERE c = ' '")
+ assert(eqSpace.collect().map(_.getInt(0)).toSet === Set(3))
+ val eqEmpty = sql(s"SELECT id FROM $table WHERE c = ''")
+ assert(eqEmpty.collect().map(_.getInt(0)).toSet === Set(1))
+ val neqSpace = sql(s"SELECT id FROM $table WHERE c <> ' '")
+ assert(neqSpace.collect().map(_.getInt(0)).toSet === Set(1, 2))
+ }
+ }
}
diff --git a/docs/sql-migration-guide.md b/docs/sql-migration-guide.md
index 647151f9ba57..ee590cfb4f10 100644
--- a/docs/sql-migration-guide.md
+++ b/docs/sql-migration-guide.md
@@ -38,6 +38,8 @@ 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 [...]
+- Since Spark 4.4, when `spark.sql.charVarchar.standardSemantics.enabled` is
false and `spark.sql.readSideCharPadding` is true (the default), JDBC Data
Source V2 catalog tables on H2, PostgreSQL, SQL Server, and Oracle push CHAR
equality, inequality, null-safe equality, and `IN`/`NOT IN`. 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 [...]
+- Since Spark 4.4, when `spark.sql.charVarchar.standardSemantics.enabled` is
false and `spark.sql.readSideCharPadding` is true (the default), `IS [NOT]
NULL` over a CHAR scan column is unwrapped to the raw column for every Data
Source V2 source, including every JDBC dialect. Previously the padding wrap
kept that null check in Spark.
- Since Spark 4.4, the optimizer rule `SimplifyCaseConversionExpressions` no
longer simplifies mixed case conversions `LOWER(UPPER(str))` to `LOWER(str)` or
`UPPER(LOWER(str))` to `UPPER(str)`. This preserves standard Unicode case
mapping semantics for non-ASCII characters such as Latin small letter dotless i
(`ı` U+0131) and micro sign (`µ` U+00B5).
- Since Spark 4.4, the Oracle JDBC datasource maps Oracle DATE and TIMESTAMP
to TimestampNTZType unconditionally, ignoring the JDBC read option
`preferTimestampNTZ` (DATE only when the driver default
`oracle.jdbc.mapDateToTimestamp=true` surfaces it as TIMESTAMP; otherwise DATE
stays DateType). In 4.3 and earlier these were read as TimestampType, or as
TimestampNTZType only when `preferTimestampNTZ=true`. This faithfully
represents these zoneless Oracle types. Unaffected: Oracle DATE rea [...]
- Since Spark 4.4, a nondeterministic common expression that a conditional
branch reads more than once is evaluated once, and every read sees that one
value. `BETWEEN` and `NULLIF` read their input twice, so an expression such as
`CASE WHEN c THEN monotonically_increasing_id() BETWEEN 3 AND 5 END` previously
gave each read its own value and could return a different result. Queries whose
results depended on the earlier behaviour change, and an optimized or physical
plan may now show a `wi [...]
diff --git
a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/GeneralScalarExpression.java
b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/GeneralScalarExpression.java
index dc8f0426d77c..2f5748fe3499 100644
---
a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/GeneralScalarExpression.java
+++
b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/GeneralScalarExpression.java
@@ -428,6 +428,22 @@ import
org.apache.spark.sql.internal.connector.ExpressionWithToString;
* <li>Since version: 4.0.0</li>
* </ul>
* </li>
+ * <li>Name: <code>CHAR_READ_SIDE_PADDING</code>
+ * <ul>
+ * <li>SQL semantic: <code>CHAR_READ_SIDE_PADDING(str, len, pad)</code>.
+ * <code>len</code> is an integer literal; <code>pad</code> is always a
+ * single space. Right-pads <code>str</code> with spaces to
+ * <code>len</code> characters; values of <code>len</code> or more
+ * characters are returned unchanged. Spark only emits this in
+ * <code>=</code>, <code><=></code>, <code><></code>, and
+ * <code>IN</code> against a string of length <code>len</code>.
+ * A source may compile it as the raw value only if it compares that
+ * value with a <code>len</code>-character string using blank-padded
+ * (PAD SPACE) semantics, and its stored values never exceed
+ * <code>len</code> characters.</li>
+ * <li>Since version: 4.4.0</li>
+ * </ul>
+ * </li>
* </ol>
* Note: SQL semantic conforms ANSI standard, so some expressions are not
supported when ANSI off,
* including: add, subtract, multiply, divide, remainder, pmod.
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/CharVarcharUtils.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/CharVarcharUtils.scala
index 8c5450882673..0871f0a0b8a2 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/CharVarcharUtils.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/CharVarcharUtils.scala
@@ -34,6 +34,24 @@ object CharVarcharUtils extends Logging with
SparkCharVarcharUtils {
// visible for testing
private[sql] val CHAR_VARCHAR_TYPE_STRING_METADATA_KEY =
"__CHAR_VARCHAR_TYPE_STRING"
+ /** JVM method name used by [[addPaddingForScan]] when read-side CHAR
padding is on. */
+ private[sql] val READ_SIDE_PADDING = "readSidePadding"
+
+ /** JVM method name used by [[addPaddingForScan]] under standard
CHAR/VARCHAR semantics. */
+ private[sql] val CHAR_TYPE_READ_SIDE_CHECK = "charTypeReadSideCheck"
+
+ private[sql] val VARCHAR_TYPE_READ_SIDE_CHECK = "varcharTypeReadSideCheck"
+
+ /**
+ * V2 function name for CHAR read-side padding. Distinct from SQL `RPAD` so
connectors that
+ * already allowlist `RPAD` for explicit `rpad()` do not inherit CHAR-scan
rewrite. Emitted
+ * only for equality, null-safe equality, inequality, or IN against a string
of the CHAR
+ * length. JDBC may compile it as the column only when the remote compares
CHAR
+ * blank-padded (see `JdbcDialect.isSupportedFunction`) and stored values
never exceed
+ * that length.
+ */
+ private[sql] val V2_CHAR_READ_SIDE_PADDING = "CHAR_READ_SIDE_PADDING"
+
/**
* Like [[hasCharVarchar]], but also inspects [[UserDefinedType]] storage
types.
* Python/Arrow host checks must match PySpark `_has_type`, which unwraps
UDTs.
@@ -364,11 +382,11 @@ object CharVarcharUtils extends Logging with
SparkCharVarcharUtils {
processStringForCharVarchar(
attr,
dt,
- charFuncName = Some("charTypeReadSideCheck"),
- varcharFuncName = Some("varcharTypeReadSideCheck"))
+ charFuncName = Some(CHAR_TYPE_READ_SIDE_CHECK),
+ varcharFuncName = Some(VARCHAR_TYPE_READ_SIDE_CHECK))
} else {
processStringForCharVarchar(
- attr, dt, charFuncName = Some("readSidePadding"), varcharFuncName =
None)
+ attr, dt, charFuncName = Some(READ_SIDE_PADDING), varcharFuncName =
None)
}
}.getOrElse(attr)
}
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/V2ExpressionBuilder.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/V2ExpressionBuilder.scala
index db9f1b66c5da..ecc75ec3ae3a 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/V2ExpressionBuilder.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/V2ExpressionBuilder.scala
@@ -31,6 +31,7 @@ import
org.apache.spark.sql.connector.expressions.filter.{AlwaysFalse, AlwaysTru
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types.{AnyTimestampNanoType, BooleanType,
DataType, IntegerType,
StringType}
+import org.apache.spark.unsafe.types.UTF8String
/**
* The builder to generate V2 expressions from catalyst expressions.
@@ -126,24 +127,34 @@ class V2ExpressionBuilder(e: Expression, isPredicate:
Boolean = false) extends L
Some(ref)
}
case InSet(child, hset) =>
- generateExpression(child).map { v =>
- val children =
- (v +: hset.toSeq.map(elem => LiteralValue(elem,
child.dataType))).toArray[V2Expression]
- new V2Predicate("IN", children)
+ generateCharPadInSet(child, hset).orElse {
+ generateExpression(child).map { v =>
+ val children =
+ (v +: hset.toSeq.map(elem => LiteralValue(elem,
child.dataType))).toArray[V2Expression]
+ new V2Predicate("IN", children)
+ }
}
// Because we only convert In to InSet in Optimizer when there are more
than certain
// items. So it is possible we still get an In expression here that needs
to be pushed
// down.
case In(value, list) =>
- val v = generateExpression(value)
- val listExpressions = list.flatMap(generateExpression(_))
- if (v.isDefined && list.length == listExpressions.length) {
- val children = (v.get +: listExpressions).toArray[V2Expression]
- // The children looks like [expr, value1, ..., valueN]
- Some(new V2Predicate("IN", children))
- } else {
- None
+ generateCharPadIn(value, list).orElse {
+ val v = generateExpression(value)
+ val listExpressions = list.flatMap(generateExpression(_))
+ if (v.isDefined && list.length == listExpressions.length) {
+ val children = (v.get +: listExpressions).toArray[V2Expression]
+ // The children looks like [expr, value1, ..., valueN]
+ Some(new V2Predicate("IN", children))
+ } else {
+ None
+ }
}
+ case IsNull(invoke: StaticInvoke) if isCharReadSidePaddingInvoke(invoke) =>
+ generateExpression(invoke.arguments.head)
+ .map(c => new V2Predicate("IS_NULL", Array[V2Expression](c)))
+ case IsNotNull(invoke: StaticInvoke) if
isCharReadSidePaddingInvoke(invoke) =>
+ generateExpression(invoke.arguments.head)
+ .map(c => new V2Predicate("IS_NOT_NULL", Array[V2Expression](c)))
case IsNull(col) => generateExpression(col)
.map(c => new V2Predicate("IS_NULL", Array[V2Expression](c)))
case IsNotNull(col) => generateExpression(col)
@@ -222,29 +233,38 @@ class V2ExpressionBuilder(e: Expression, isPredicate:
Boolean = false) extends L
None
}
case b: BinaryOperator if canTranslate(b) =>
- val l = generateExpression(b.left)
- val r = generateExpression(b.right)
- if (l.isDefined && r.isDefined) {
- b match {
- case _: BinaryComparison if l.get.isInstanceOf[LiteralValue[_]] &&
- r.get.isInstanceOf[FieldReference] =>
- Some(new V2Predicate(flipComparisonOperatorName(b.sqlOperator),
- Array[V2Expression](r.get, l.get)))
- case _: Predicate =>
- Some(new V2Predicate(b.sqlOperator, Array[V2Expression](l.get,
r.get)))
- case _ =>
- Some(new GeneralScalarExpression(b.sqlOperator,
Array[V2Expression](l.get, r.get)))
+ val charPadCmp = b match {
+ case _: EqualTo | _: EqualNullSafe =>
+ generateCharPadLiteralCmp(b.sqlOperator, b.left, b.right)
+ case _ => None
+ }
+ charPadCmp.orElse {
+ val l = generateExpression(b.left)
+ val r = generateExpression(b.right)
+ if (l.isDefined && r.isDefined) {
+ b match {
+ case _: BinaryComparison if l.get.isInstanceOf[LiteralValue[_]] &&
+ r.get.isInstanceOf[FieldReference] =>
+ Some(new V2Predicate(flipComparisonOperatorName(b.sqlOperator),
+ Array[V2Expression](r.get, l.get)))
+ case _: Predicate =>
+ Some(new V2Predicate(b.sqlOperator, Array[V2Expression](l.get,
r.get)))
+ case _ =>
+ Some(new GeneralScalarExpression(b.sqlOperator,
Array[V2Expression](l.get, r.get)))
+ }
+ } else {
+ None
}
- } else {
- None
}
case Not(eq: EqualTo) =>
- val left = generateExpression(eq.left)
- val right = generateExpression(eq.right)
- if (left.isDefined && right.isDefined) {
- Some(new V2Predicate("<>", Array[V2Expression](left.get, right.get)))
- } else {
- None
+ generateCharPadLiteralCmp("<>", eq.left, eq.right).orElse {
+ val left = generateExpression(eq.left)
+ val right = generateExpression(eq.right)
+ if (left.isDefined && right.isDefined) {
+ Some(new V2Predicate("<>", Array[V2Expression](left.get, right.get)))
+ } else {
+ None
+ }
}
case Not(child) => generatePredicate(child).map(new V2Not(_))
case UnaryMinus(_, true) => generateExpressionWithName("-", expr,
isPredicate)
@@ -464,6 +484,82 @@ 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
+ }
+
+ // Identity compile is only valid vs a literal Spark already padded to the
CHAR length.
+ private def charPadLength(invoke: StaticInvoke): Option[Int] =
invoke.arguments(1) match {
+ case Literal(n: Int, _) => Some(n)
+ case _ => None
+ }
+
+ private def isFoldableStringOfLen(expr: Expression, n: Int): Boolean = expr
match {
+ case Literal(s: UTF8String, _) => s.numChars() == n
+ case _ => false
+ }
+
+ private def generateCharPadColumn(invoke: StaticInvoke):
Option[V2Expression] = {
+ generateExpressionWithNameByChildren(
+ CharVarcharUtils.V2_CHAR_READ_SIDE_PADDING,
+ Seq(invoke.arguments.head, invoke.arguments(1), Literal.create(" ",
StringType)),
+ invoke.dataType,
+ isPredicate = false)
+ }
+
+ private def generateCharPadLiteralCmp(
+ name: String, left: Expression, right: Expression): Option[V2Expression]
= {
+ def paddedVsLit(pad: Expression, lit: Expression): Option[V2Expression] =
pad match {
+ case invoke: StaticInvoke if isCharReadSidePaddingInvoke(invoke) =>
+ charPadLength(invoke).filter(isFoldableStringOfLen(lit, _)).flatMap {
_ =>
+ generateCharPadColumn(invoke).flatMap { col =>
+ generateExpression(lit).map { l =>
+ new V2Predicate(name, Array[V2Expression](col, l))
+ }
+ }
+ }
+ case _ => None
+ }
+ paddedVsLit(left, right).orElse(paddedVsLit(right, left))
+ }
+
+ private def generateCharPadIn(
+ value: Expression, list: Seq[Expression]): Option[V2Expression] = value
match {
+ case invoke: StaticInvoke if isCharReadSidePaddingInvoke(invoke) =>
+ charPadLength(invoke).filter(n => list.nonEmpty &&
list.forall(isFoldableStringOfLen(_, n)))
+ .flatMap { _ =>
+ generateCharPadColumn(invoke).flatMap { col =>
+ val lits = list.flatMap(generateExpression(_))
+ if (lits.length == list.length) {
+ Some(new V2Predicate("IN", (col +: lits).toArray[V2Expression]))
+ } else {
+ None
+ }
+ }
+ }
+ case _ => None
+ }
+
+ private def generateCharPadInSet(
+ child: Expression, hset: Set[Any]): Option[V2Expression] = child match {
+ case invoke: StaticInvoke if isCharReadSidePaddingInvoke(invoke) =>
+ charPadLength(invoke).filter { n =>
+ hset.nonEmpty && hset.forall {
+ case s: UTF8String => s.numChars() == n
+ case _ => false
+ }
+ }.flatMap { _ =>
+ generateCharPadColumn(invoke).map { col =>
+ val lits = hset.toSeq.map(elem => LiteralValue(elem, child.dataType))
+ new V2Predicate("IN", (col +: lits).toArray[V2Expression])
+ }
+ }
+ case _ => None
+ }
+
private def generateExpressionWithName(
v2ExpressionName: String,
expr: Expression,
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/jdbc/JDBCScanBuilder.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/jdbc/JDBCScanBuilder.scala
index e81f25c656c0..ef663e034295 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/jdbc/JDBCScanBuilder.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/jdbc/JDBCScanBuilder.scala
@@ -22,7 +22,8 @@ import org.apache.spark.internal.Logging
import org.apache.spark.internal.LogKeys.{JOIN_CONDITION, JOIN_TYPE, SCHEMA}
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.catalyst.expressions.V2ExpressionUtils
-import org.apache.spark.sql.connector.expressions.{FieldReference, SortOrder}
+import org.apache.spark.sql.catalyst.util.CharVarcharUtils
+import org.apache.spark.sql.connector.expressions.{Expression, FieldReference,
GeneralScalarExpression, SortOrder}
import org.apache.spark.sql.connector.expressions.aggregate.Aggregation
import org.apache.spark.sql.connector.expressions.filter.Predicate
import org.apache.spark.sql.connector.join.JoinType
@@ -66,11 +67,25 @@ case class JDBCScanBuilder(
private var sortOrders: Array[String] = Array.empty[String]
+ private def containsUnsupportedCharPad(expr: Expression): 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(expr)
+ }
+ }
+
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)
+ val (pushed, unSupported) =
rest.partition(dialect.compileExpression(_).isDefined)
this.pushedPredicate = pushed
- unSupported
+ unSupported ++ unsupportedCharPad
} else {
predicates
}
@@ -94,6 +109,10 @@ case class JDBCScanBuilder(
override def pushAggregation(aggregation: Aggregation): Boolean = {
if (!jdbcOptions.pushDownAggregate) return false
+ if (aggregation.aggregateExpressions.exists(containsUnsupportedCharPad) ||
+ aggregation.groupByExpressions.exists(containsUnsupportedCharPad)) {
+ return false
+ }
val compiledAggs =
aggregation.aggregateExpressions.flatMap(dialect.compileExpression)
if (compiledAggs.length != aggregation.aggregateExpressions.length) return
false
@@ -204,6 +223,10 @@ case class JDBCScanBuilder(
return false
}
+ if (containsUnsupportedCharPad(condition)) {
+ return false
+ }
+
val compiledCondition = dialect.compileExpression(condition)
if (!compiledCondition.isDefined) {
logError(log"Failed to push down join to JDBC due to unsupported join
condition " +
@@ -316,6 +339,7 @@ case class JDBCScanBuilder(
override def pushTopN(orders: Array[SortOrder], limit: Int): Boolean = {
if (jdbcOptions.pushDownLimit) {
+ if (orders.exists(containsUnsupportedCharPad)) return false
val compiledOrders = orders.flatMap(dialect.compileExpression(_))
if (orders.length != compiledOrders.length) return false
pushedLimit = limit
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/jdbc/H2Dialect.scala
b/sql/core/src/main/scala/org/apache/spark/sql/jdbc/H2Dialect.scala
index 11fe9506e09c..b9a04203ee84 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/jdbc/H2Dialect.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/jdbc/H2Dialect.scala
@@ -33,6 +33,7 @@ import org.apache.spark.sql.catalyst.analysis.{
NoSuchNamespaceException,
NoSuchTableException,
TableAlreadyExistsException}
+import org.apache.spark.sql.catalyst.util.CharVarcharUtils
import org.apache.spark.sql.connector.catalog.Identifier
import org.apache.spark.sql.connector.catalog.functions.UnboundFunction
import org.apache.spark.sql.connector.catalog.index.TableIndex
@@ -57,7 +58,8 @@ private[sql] case class H2Dialect() extends JdbcDialect with
NoLegacyJDBCError {
"POWER", "SQRT", "FLOOR", "CEIL", "ROUND", "SIN", "SINH", "COS", "COSH",
"TAN",
"TANH", "COT", "ASIN", "ACOS", "ATAN", "ATAN2", "DEGREES", "RADIANS",
"SIGN",
"PI", "SUBSTRING", "UPPER", "LOWER", "TRANSLATE", "TRIM", "MD5", "SHA1",
"SHA2",
- "BIT_LENGTH", "CHAR_LENGTH", "CONCAT", "RPAD", "LPAD")
+ "BIT_LENGTH", "CHAR_LENGTH", "CONCAT", "RPAD", "LPAD",
+ CharVarcharUtils.V2_CHAR_READ_SIDE_PADDING)
override def isSupportedFunction(funcName: String): Boolean =
supportedFunctions.contains(funcName)
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/jdbc/JdbcDialects.scala
b/sql/core/src/main/scala/org/apache/spark/sql/jdbc/JdbcDialects.scala
index 7d57fa6a5ed6..c2ae86ea9d4f 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/jdbc/JdbcDialects.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/jdbc/JdbcDialects.scala
@@ -34,14 +34,16 @@ import org.apache.spark.internal.Logging
import org.apache.spark.sql.AnalysisException
import org.apache.spark.sql.catalyst.CatalystTypeConverters
import org.apache.spark.sql.catalyst.plans.logical.SampleMethod
-import org.apache.spark.sql.catalyst.util.{DateFormatter, DateTimeUtils,
TimestampFormatter}
+import org.apache.spark.sql.catalyst.util.{
+ CharVarcharUtils, DateFormatter, DateTimeUtils, TimestampFormatter}
import
org.apache.spark.sql.catalyst.util.DateTimeUtils.{localDateTimeToMicros,
toJavaTimestampNoRebase}
import org.apache.spark.sql.catalyst.util.IntervalUtils.{fromDayTimeString,
fromYearMonthString, getDuration}
import org.apache.spark.sql.connector.catalog.{Identifier, TableChange}
import org.apache.spark.sql.connector.catalog.TableChange._
import org.apache.spark.sql.connector.catalog.functions.UnboundFunction
import org.apache.spark.sql.connector.catalog.index.TableIndex
-import org.apache.spark.sql.connector.expressions.{Expression, Literal,
NamedReference}
+import org.apache.spark.sql.connector.expressions.{Expression,
GeneralScalarExpression, Literal,
+ NamedReference}
import org.apache.spark.sql.connector.expressions.aggregate.AggregateFunc
import org.apache.spark.sql.connector.expressions.filter.{AlwaysFalse,
AlwaysTrue, Predicate}
import org.apache.spark.sql.connector.util.V2ExpressionSQLBuilder
@@ -417,6 +419,16 @@ abstract class JdbcDialect extends Serializable with
Logging {
override def build(expr: Expression): String = expr match {
case _: AlwaysTrue => "(1 = 1)"
case _: AlwaysFalse => "(1 = 0)"
+ case e: GeneralScalarExpression
+ if e.name() == CharVarcharUtils.V2_CHAR_READ_SIDE_PADDING =>
+ if (!isSupportedFunction(CharVarcharUtils.V2_CHAR_READ_SIDE_PADDING)) {
+ throw new SparkUnsupportedOperationException(
+ errorClass = "_LEGACY_ERROR_TEMP_3177",
+ messageParameters = Map(
+ "class" -> this.getClass.getSimpleName,
+ "funcName" -> e.name()))
+ }
+ inputToSQL(e.children()(0))
case _ => super.build(expr)
}
@@ -534,6 +546,16 @@ abstract class JdbcDialect extends Serializable with
Logging {
/**
* Returns whether the database supports function.
+ *
+ * For most names this means the remote can execute that SQL function.
+ * CHAR_READ_SIDE_PADDING is different: Spark only emits it as a CHAR scan
+ * column in equality, null-safe equality, inequality, or IN against a
+ * literal already padded to the CHAR width. Returning true for that name
+ * means JDBC may compile it as the raw column, so the remote must compare
+ * that column with a blank-padded (PAD SPACE) CHAR literal the same way
+ * Spark compares the padded value. Dialects that strip trailing spaces or
+ * use a NO PAD collation must not opt in.
+ *
* @param funcName Upper-cased function name
* @return True if the database supports function.
*/
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/jdbc/MsSqlServerDialect.scala
b/sql/core/src/main/scala/org/apache/spark/sql/jdbc/MsSqlServerDialect.scala
index b94e1a43eff7..302d3616a68a 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/jdbc/MsSqlServerDialect.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/jdbc/MsSqlServerDialect.scala
@@ -24,6 +24,7 @@ import scala.util.control.NonFatal
import org.apache.spark.SparkThrowable
import org.apache.spark.sql.catalyst.analysis.NonEmptyNamespaceException
+import org.apache.spark.sql.catalyst.util.CharVarcharUtils
import org.apache.spark.sql.connector.catalog.Identifier
import org.apache.spark.sql.connector.expressions.{Expression, NullOrdering,
SortDirection}
import org.apache.spark.sql.connector.expressions.filter.Predicate
@@ -58,7 +59,8 @@ private case class MsSqlServerDialect() extends JdbcDialect
with NoLegacyJDBCErr
// scalastyle:on line.size.limit
private val supportedAggregateFunctions = Set("MAX", "MIN", "SUM", "COUNT",
"AVG",
"VAR_POP", "VAR_SAMP", "STDDEV_POP", "STDDEV_SAMP")
- private val supportedStringFunctions = Set("RPAD", "LPAD")
+ private val supportedStringFunctions =
+ Set("RPAD", "LPAD", CharVarcharUtils.V2_CHAR_READ_SIDE_PADDING)
private val supportedFunctions = supportedAggregateFunctions ++
supportedStringFunctions
override def isSupportedFunction(funcName: String): Boolean =
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/jdbc/OracleDialect.scala
b/sql/core/src/main/scala/org/apache/spark/sql/jdbc/OracleDialect.scala
index 7b1fb9667397..d9147f7a7c95 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/jdbc/OracleDialect.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/jdbc/OracleDialect.scala
@@ -26,6 +26,7 @@ import scala.util.control.NonFatal
import org.apache.spark.{SparkThrowable, SparkUnsupportedOperationException}
import org.apache.spark.sql.catalyst.SQLConfHelper
+import org.apache.spark.sql.catalyst.util.CharVarcharUtils
import org.apache.spark.sql.connector.expressions.{Expression, Extract,
Literal}
import org.apache.spark.sql.errors.QueryCompilationErrors
import org.apache.spark.sql.execution.datasources.jdbc.{JDBCOptions, JdbcUtils}
@@ -46,7 +47,9 @@ 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", CharVarcharUtils.V2_CHAR_READ_SIDE_PADDING)
override def isSupportedFunction(funcName: String): Boolean =
supportedFunctions.contains(funcName)
@@ -118,6 +121,29 @@ private case class OracleDialect() extends JdbcDialect
with SQLConfHelper with N
super.visitAggregateFunction(funcName, isDistinct, inputs)
}
+ override def visitIn(v: String, list: java.util.List[String]): String = {
+ if (list.size() <= 1000) {
+ super.visitIn(v, list)
+ } else {
+ val chunks = new java.util.ArrayList[String]()
+ var i = 0
+ while (i < list.size()) {
+ val end = Math.min(i + 1000, list.size())
+ val ins = new java.lang.StringBuilder(v).append(" IN (")
+ var j = i
+ while (j < end) {
+ if (j > i) ins.append(", ")
+ ins.append(list.get(j))
+ j += 1
+ }
+ ins.append(")")
+ chunks.add(ins.toString)
+ i = end
+ }
+ chunks.toArray.mkString("(", " OR ", ")")
+ }
+ }
+
override def visitBinaryComparison(name: String, le: Expression, re:
Expression): String = {
(le, re) match {
case (lit: Literal[_], _) if lit.dataType == BinaryType =>
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/jdbc/PostgresDialect.scala
b/sql/core/src/main/scala/org/apache/spark/sql/jdbc/PostgresDialect.scala
index c78d9be492ab..95dcc06f3cdc 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/jdbc/PostgresDialect.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/jdbc/PostgresDialect.scala
@@ -29,6 +29,7 @@ import org.apache.spark.SparkThrowable
import org.apache.spark.internal.LogKeys.COLUMN_NAME
import org.apache.spark.sql.catalyst.SQLConfHelper
import org.apache.spark.sql.catalyst.analysis.{IndexAlreadyExistsException,
NonEmptyNamespaceException, NoSuchIndexException}
+import org.apache.spark.sql.catalyst.util.CharVarcharUtils
import org.apache.spark.sql.connector.catalog.Identifier
import org.apache.spark.sql.connector.expressions.{Expression, NamedReference}
import org.apache.spark.sql.errors.QueryCompilationErrors
@@ -47,7 +48,8 @@ private case class PostgresDialect()
private val supportedAggregateFunctions = Set("MAX", "MIN", "SUM", "COUNT",
"AVG",
"VAR_POP", "VAR_SAMP", "STDDEV_POP", "STDDEV_SAMP", "COVAR_POP",
"COVAR_SAMP", "CORR",
"REGR_INTERCEPT", "REGR_R2", "REGR_SLOPE", "REGR_SXY")
- private val supportedStringFunctions = Set("RPAD", "LPAD")
+ private val supportedStringFunctions =
+ Set("RPAD", "LPAD", CharVarcharUtils.V2_CHAR_READ_SIDE_PADDING)
private val supportedFunctions = supportedAggregateFunctions ++
supportedStringFunctions
override def isSupportedFunction(funcName: String): Boolean =
@@ -78,6 +80,9 @@ private case class PostgresDialect()
// timetz represents time with time zone, currently it maps to
Types.TIME.
// We need to change to Types.TIME_WITH_TIMEZONE if the upstream
changes.
Some(TimestampType)
+ case Types.CHAR if "char".equalsIgnoreCase(typeName) =>
+ // PostgreSQL internal "char" (single byte) compares exactly, not PAD
SPACE.
+ Some(StringType)
case Types.CHAR if "bpchar".equalsIgnoreCase(typeName) && size ==
Int.MaxValue =>
// bpchar with unspecified length same as text in postgres with
blank-trimmed
Some(StringType)
@@ -116,7 +121,8 @@ private case class PostgresDialect()
case "float8" => Some(DoubleType)
case "varchar" => Some(VarcharType(precision))
case "bpchar" if precision == Int.MaxValue => Some(StringType)
- case "char" | "bpchar" => Some(CharType(precision))
+ case "char" => Some(StringType)
+ case "bpchar" => Some(CharType(precision))
case "text" | "cidr" | "inet" | "json" | "jsonb" | "uuid" |
"xml" | "tsvector" | "tsquery" | "macaddr" | "macaddr8" |
"txid_snapshot" | "point" |
"line" | "lseg" | "box" | "path" | "polygon" | "circle" | "pg_lsn" |
"varbit" |
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2StrategySuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2StrategySuite.scala
index 533ac9b593cc..bdc1ab374ac1 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2StrategySuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2StrategySuite.scala
@@ -27,11 +27,12 @@ import
org.apache.spark.sql.catalyst.analysis.UnresolvedAttribute
import org.apache.spark.sql.catalyst.dsl.expressions._
import org.apache.spark.sql.catalyst.expressions._
import
org.apache.spark.sql.catalyst.expressions.aggregate.{AggregateExpression,
AggregateFunction, Complete, Count, Percentile, PercentileDisc}
+import org.apache.spark.sql.catalyst.expressions.objects.StaticInvoke
import org.apache.spark.sql.catalyst.expressions.variant.VariantGet
import org.apache.spark.sql.catalyst.optimizer.{BooleanSimplification,
CombineFilters, ConstantFolding}
import org.apache.spark.sql.catalyst.plans.Inner
import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, Filter, Join,
JoinHint, Limit, LogicalPlan, Offset, OneRowRelation, Project, Sort}
-import org.apache.spark.sql.catalyst.util.V2ExpressionBuilder
+import org.apache.spark.sql.catalyst.util.{CharVarcharCodegenUtils,
CharVarcharUtils, V2ExpressionBuilder}
import org.apache.spark.sql.connector.catalog.{SupportsRead, Table,
TableCapability}
import org.apache.spark.sql.connector.expressions.{Cast => V2Cast, Expression
=> V2Expression, FieldReference, GeneralScalarExpression, LiteralValue,
SortOrder => V2SortOrder, VariantGet => V2VariantGet}
import org.apache.spark.sql.connector.expressions.aggregate.Aggregation
@@ -43,7 +44,7 @@ import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.internal.connector.SupportsPushDownCatalystFilters
import org.apache.spark.sql.sources.{BaseRelation, TableScan}
import org.apache.spark.sql.test.SharedSparkSession
-import org.apache.spark.sql.types.{ArrayType, BooleanType, DoubleType,
FloatType, IntegerType, LongType, StringType, StructField, StructType,
TimestampNTZNanosType, TimestampType, TimeType, VariantType}
+import org.apache.spark.sql.types.{ArrayType, BooleanType, DoubleType,
FloatType, IntegerType, LongType, MetadataBuilder, StringType, StructField,
StructType, TimestampNTZNanosType, TimestampType, TimeType, VariantType}
import org.apache.spark.sql.util.CaseInsensitiveStringMap
import org.apache.spark.udf.worker.UDFWorkerSpecification
import org.apache.spark.unsafe.types.UTF8String
@@ -1919,6 +1920,130 @@ 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)
+ nseq.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 in = new V2ExpressionBuilder(In(padded, Seq(Literal("Y"))),
isPredicate = true).build()
+ assert(in.isDefined)
+ val neq = new V2ExpressionBuilder(Not(EqualTo(padded, Literal("Y"))),
isPredicate = true)
+ .build()
+ assert(neq.isDefined)
+ neq.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 inSet = new V2ExpressionBuilder(
+ InSet(padded, Set(UTF8String.fromString("Y"))), isPredicate =
true).build()
+ assert(inSet.isDefined)
+ inSet.get match {
+ case p: Predicate if p.name() == "IN" =>
+ assert(p.children().length === 2)
+ 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 IN predicate, got ${other.describe()}")
+ }
+ val inNull = new V2ExpressionBuilder(
+ In(padded, Seq(Literal("Y"), Literal(null, StringType))), isPredicate =
true).build()
+ assert(inNull.isEmpty, "IN list with NULL should stay local")
+ val litLeft = new V2ExpressionBuilder(EqualTo(Literal("Y"), padded),
isPredicate = true)
+ .build()
+ assert(litLeft.isDefined, "literal on the left of equality should still
translate")
+ assert(new V2ExpressionBuilder(EqualTo(padded, Literal("YY")), isPredicate
= true)
+ .build().isEmpty, "over-long literal should stay local")
+ assert(new V2ExpressionBuilder(GreaterThan(padded, Literal("A")),
isPredicate = true)
+ .build().isEmpty, "range comparison should stay local")
+ val other = AttributeReference("other", StringType)()
+ assert(new V2ExpressionBuilder(EqualTo(padded, other), isPredicate =
true).build().isEmpty,
+ "CHAR vs column should stay local")
+ val nestedRpad = StringRPad(padded, Literal(3), Literal("x"))
+ assert(new V2ExpressionBuilder(EqualTo(nestedRpad, Literal("Yxx")),
isPredicate = true)
+ .build().isEmpty, "rpad of padded CHAR should stay local")
+
+ def paddingInvoke(cls: Class[_], functionName: String): StaticInvoke = {
+ StaticInvoke(cls, StringType, functionName, col :: Literal(1) :: Nil,
returnNullable = false)
+ }
+ val otherClass = paddingInvoke(classOf[String],
CharVarcharUtils.READ_SIDE_PADDING)
+ assert(new V2ExpressionBuilder(EqualTo(otherClass, Literal("Y")),
isPredicate = true)
+ .build().isEmpty)
+ Seq(CharVarcharUtils.CHAR_TYPE_READ_SIDE_CHECK,
+ CharVarcharUtils.VARCHAR_TYPE_READ_SIDE_CHECK).foreach { name =>
+ assert(new V2ExpressionBuilder(
+ EqualTo(paddingInvoke(classOf[CharVarcharCodegenUtils], name),
Literal("Y")),
+ isPredicate = true).build().isEmpty,
+ s"$name should not translate")
+ }
+ }
+
test("SPARK-59301: translate runtime IN filter") {
attrInts.foreach { case (attr, name) =>
checkRuntimeFilter(runtimeFilter(attr, Array[Any](1, 2)), name,
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCSuite.scala
index baf29fe2c740..5efe9f10ad0b 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCSuite.scala
@@ -27,6 +27,7 @@ import java.util.concurrent.{CountDownLatch, Executors,
TimeUnit}
import scala.jdk.CollectionConverters._
import scala.util.Random
+import org.apache.logging.log4j.Level
import org.mockito.ArgumentCaptor
import org.mockito.ArgumentMatchers._
import org.mockito.Mockito._
@@ -42,18 +43,21 @@ import
org.apache.spark.sql.catalyst.parser.CatalystSqlParser
import org.apache.spark.sql.catalyst.plans.logical.ShowCreateTable
import org.apache.spark.sql.catalyst.util.{CaseInsensitiveMap,
CharVarcharUtils, DateTimeTestUtils, DateTimeUtils}
import org.apache.spark.sql.connector.catalog.Identifier
-import org.apache.spark.sql.connector.expressions.{Cast => V2Cast, Expression
=> V2Expression, FieldReference, GeneralScalarExpression, LiteralValue}
+import org.apache.spark.sql.connector.expressions.{Cast => V2Cast, Expression
=> V2Expression, FieldReference, GeneralScalarExpression, LiteralValue,
NullOrdering, SortDirection, SortValue}
+import org.apache.spark.sql.connector.expressions.aggregate.{Aggregation, Sum}
import org.apache.spark.sql.connector.expressions.filter.{AlwaysFalse,
AlwaysTrue, Predicate}
import org.apache.spark.sql.execution.{DataSourceScanExec, ExtendedMode,
ProjectExec}
import org.apache.spark.sql.execution.command.{ExplainCommand,
ShowCreateTableCommand}
import org.apache.spark.sql.execution.datasources.{LogicalRelation,
LogicalRelationWithTable}
import org.apache.spark.sql.execution.datasources.jdbc.{JDBCOptions,
JDBCPartition, JDBCRelation, JdbcUtils}
+import org.apache.spark.sql.execution.datasources.v2.jdbc.JDBCScanBuilder
import org.apache.spark.sql.execution.metric.InputOutputMetricsHelper
import org.apache.spark.sql.functions.{lit, percentile_approx}
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.sources._
import org.apache.spark.sql.test.SharedSparkSession
import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.types.UTF8String
import org.apache.spark.util.Utils
class JDBCSuite extends SharedSparkSession {
@@ -1432,7 +1436,6 @@ class JDBCSuite extends SharedSparkSession {
// LIKE-family operators render with a trailing ESCAPE clause and must be
delimited too.
// LiteralValue for StringType must use UTF8String (Spark's internal
string type).
- import org.apache.spark.unsafe.types.UTF8String
for ((op, pattern) <- Seq(
"STARTS_WITH" -> "abc%", "ENDS_WITH" -> "%abc", "CONTAINS" ->
"%abc%")) {
val like = new Predicate(op,
@@ -1656,6 +1659,10 @@ 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))
+ assert(Postgres.getCatalystType(java.sql.Types.CHAR, "bpchar", 5, md) ===
None)
+ assert(Postgres.getCatalystType(java.sql.Types.ARRAY, "_char", 1, md) ==
+ Some(ArrayType(StringType)))
assert(Postgres.getJDBCType(FloatType).map(_.databaseTypeDefinition).get
== "FLOAT4")
assert(Postgres.getJDBCType(DoubleType).map(_.databaseTypeDefinition).get
== "FLOAT8")
assert(Postgres.getJDBCType(ByteType).map(_.databaseTypeDefinition).get ==
"SMALLINT")
@@ -2350,13 +2357,91 @@ class JDBCSuite extends SharedSparkSession {
assert(getJdbcType(oracleDialect, TimestampNTZType) == "TIMESTAMP")
}
+ test("SPARK-60479: dialects compile CHAR_READ_SIDE_PADDING") {
+ val pad = new
GeneralScalarExpression(CharVarcharUtils.V2_CHAR_READ_SIDE_PADDING,
+ Array[V2Expression](
+ FieldReference("flag"),
+ LiteralValue(5, IntegerType),
+ LiteralValue(UTF8String.fromString(" "), StringType)))
+ val eq = new Predicate("=",
+ Array[V2Expression](pad, LiteralValue(UTF8String.fromString("Y "),
StringType)))
+ val rpad = new GeneralScalarExpression("RPAD",
+ Array[V2Expression](
+ FieldReference("flag"),
+ LiteralValue(5, IntegerType),
+ LiteralValue(UTF8String.fromString(" "), StringType)))
+ val rpadEq = new Predicate("=",
+ Array[V2Expression](rpad, LiteralValue(UTF8String.fromString("Y "),
StringType)))
+
+ Seq(
+ "jdbc:h2:mem:testdb0",
+ "jdbc:postgresql://127.0.0.1/db",
+ "jdbc:sqlserver://127.0.0.1/db",
+ "jdbc:oracle://127.0.0.1/db").foreach { url =>
+ assert(
+ JdbcDialects.get(url).compileExpression(eq).get === """("flag") = 'Y
'""", url)
+ }
+ Seq(
+ "jdbc:mysql://127.0.0.1/db",
+ "jdbc:db2://127.0.0.1/db",
+ "jdbc:derby:memory:db").foreach { url =>
+ assert(JdbcDialects.get(url).compileExpression(eq).isEmpty, url)
+ }
+
+ val h2 = JdbcDialects.get("jdbc:h2:mem:testdb0")
+ assert(h2.compileExpression(rpadEq).get === """(RPAD("flag", 5, ' ')) = 'Y
'""")
+
+ val oracle = JdbcDialects.get("jdbc:oracle://127.0.0.1/db")
+ assert(oracle.compileExpression(rpadEq).isEmpty)
+
+ val y = LiteralValue(UTF8String.fromString("Y "), StringType)
+ val n = LiteralValue(UTF8String.fromString("N "), StringType)
+ val inSmall = new Predicate("IN", Array[V2Expression](pad, y, n))
+ assert(oracle.compileExpression(inSmall).get === """"flag" IN ('Y ', 'N
')""")
+ val notIn = new Predicate("NOT", Array[V2Expression](inSmall))
+ assert(oracle.compileExpression(notIn).get === """NOT ("flag" IN ('Y ',
'N '))""")
+
+ val many = (1 to 1001).map { i =>
+ LiteralValue(UTF8String.fromString(f"$i%05d"), StringType)
+ }
+ val inLarge = new Predicate("IN", (pad +: many).toArray[V2Expression])
+ val largeSql = oracle.compileExpression(inLarge).get
+ assert(largeSql.startsWith("(") && largeSql.endsWith(")"))
+ assert(largeSql.contains(" OR "))
+ assert(" IN \\(".r.findAllMatchIn(largeSql).length === 2)
+ many.foreach { lit =>
+ assert(largeSql.contains(lit.value.toString))
+ }
+
+ val leftoverEq = new Predicate("=",
+ Array[V2Expression](pad, LiteralValue(UTF8String.fromString("Y "),
StringType)))
+ val isNull = new Predicate("IS_NULL",
Array[V2Expression](FieldReference("flag")))
+ val caseWhen = new GeneralScalarExpression("CASE_WHEN",
+ Array[V2Expression](leftoverEq, LiteralValue(1, IntegerType)))
+ val agg = new Aggregation(Array(new Sum(caseWhen, false)), Array.empty)
+ val sort = SortValue(leftoverEq, SortDirection.ASCENDING,
NullOrdering.NULLS_LAST)
+ val options = new JDBCOptions("jdbc:derby:memory:db", "t",
Map.empty[String, String])
+ val builder = JDBCScanBuilder(spark, new StructType().add("flag",
StringType), options)
+ val logAppender = new LogAppender("compile V2")
+ withLogAppender(logAppender, level = Some(Level.WARN)) {
+ val leftover = builder.pushPredicates(Array(leftoverEq, isNull))
+ assert(leftover.exists(_.name() == "="))
+ assert(!leftover.exists(_.name() == "IS_NULL"))
+ assert(builder.pushedPredicates().exists(_.name() == "IS_NULL"))
+ assert(!builder.pushAggregation(agg))
+ assert(!builder.pushTopN(Array(sort), 1))
+ }
+ assert(!logAppender.loggingEvents.exists { ev =>
+ ev.getMessage.getFormattedMessage.contains("Error occurs while compiling
V2 expression")
+ })
+ }
+
test("Oracle TRUNC pushdown should map Spark format strings to Oracle
format") {
val oracleDialect = JdbcDialects.get("jdbc:oracle://127.0.0.1/db")
val dateRef = FieldReference("d")
// LiteralValue for StringType must use UTF8String (Spark's internal
string type)
// to match what V2ExpressionBuilder produces in the real pushdown path.
- import org.apache.spark.unsafe.types.UTF8String
def truncExpr(fmt: String): GeneralScalarExpression = new
GeneralScalarExpression("TRUNC",
Array[V2Expression](dateRef, LiteralValue(UTF8String.fromString(fmt),
StringType)))
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCV2Suite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCV2Suite.scala
index e2145b691ca8..49a354e412d2 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCV2Suite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCV2Suite.scala
@@ -2017,6 +2017,141 @@ 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))")
+ // scalastyle:off nonascii
+ sql(s"INSERT INTO $table VALUES (1, 'Y'), (2, 'N'), (3, 'YES'), (4,
NULL), (5, 'ä')")
+
+ val eq = sql(s"SELECT id FROM $table WHERE flag = 'Y'")
+ checkFiltersRemoved(eq)
+ checkPushedInfo(eq,
+ "PushedFilters: [flag IS NOT NULL, (CHAR_READ_SIDE_PADDING(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,
+ "PushedFilters: [CHAR_READ_SIDE_PADDING(flag, 5, ' ') IN ('Y ', 'N
')]")
+ checkAnswer(in, Seq(Row(1), Row(2)))
+
+ withSQLConf(SQLConf.OPTIMIZER_INSET_CONVERSION_THRESHOLD.key -> "2") {
+ val inSetQ = sql(s"SELECT id FROM $table WHERE flag IN ('Y', 'N',
'X')")
+ checkFiltersRemoved(inSetQ)
+ checkPushedInfo(inSetQ,
+ "CHAR_READ_SIDE_PADDING(flag, 5, ' ') IN",
+ "'Y '",
+ "'N '",
+ "'X '")
+ checkAnswer(inSetQ, Seq(Row(1), Row(2)))
+ }
+
+ val isNull = sql(s"SELECT id FROM $table WHERE flag IS NULL")
+ checkFiltersRemoved(isNull)
+ checkPushedInfo(isNull, "PushedFilters: [flag IS NULL]")
+ checkAnswer(isNull, Seq(Row(4)))
+
+ val notIn = sql(s"SELECT id FROM $table WHERE flag NOT IN ('Y', 'N')")
+ checkFiltersRemoved(notIn)
+ checkPushedInfo(notIn,
+ "PushedFilters: [NOT (CHAR_READ_SIDE_PADDING(flag, 5, ' ') IN ('Y
', 'N '))]")
+ checkAnswer(notIn, Seq(Row(3), Row(5)))
+
+ val orPred = sql(s"SELECT id FROM $table WHERE flag = 'Y' OR flag = 'N'")
+ checkFiltersRemoved(orPred)
+ checkAnswer(orPred, Seq(Row(1), Row(2)))
+
+ val nseq = sql(s"SELECT id FROM $table WHERE flag <=> 'Y'")
+ checkFiltersRemoved(nseq)
+ checkPushedInfo(nseq,
+ "CHAR_READ_SIDE_PADDING(flag, 5, ' ')) = 'Y '",
+ "IS NULL AND 'Y ' IS NULL")
+ checkAnswer(nseq, Seq(Row(1)))
+
+ val notNseq = sql(s"SELECT id FROM $table WHERE NOT (flag <=> 'Y')")
+ checkFiltersRemoved(notNseq)
+ checkPushedInfo(notNseq,
+ "NOT ((((CHAR_READ_SIDE_PADDING(flag, 5, ' ')) IS NOT NULL",
+ "IS NULL AND 'Y ' IS NULL")
+ checkAnswer(notNseq, Seq(Row(2), Row(3), Row(4), Row(5)))
+
+ val neq = sql(s"SELECT id FROM $table WHERE flag <> 'Y'")
+ checkFiltersRemoved(neq)
+ checkPushedInfo(neq,
+ "PushedFilters: [flag IS NOT NULL, " +
+ "NOT ((CHAR_READ_SIDE_PADDING(flag, 5, ' ')) = 'Y ')]")
+ checkAnswer(neq, Seq(Row(2), Row(3), Row(5)))
+
+ val longLit = sql(s"SELECT id FROM $table WHERE flag = 'YYYYYY'")
+ checkFiltersRemoved(longLit, removed = false)
+ checkAnswer(longLit, Seq.empty)
+
+ val range = sql(s"SELECT id FROM $table WHERE flag > 'A'")
+ checkFiltersRemoved(range, removed = false)
+ checkAnswer(range, Seq(Row(1), Row(2), Row(3), Row(5)))
+
+ val likeFam = sql(s"SELECT id FROM $table WHERE flag LIKE 'Y%'")
+ checkFiltersRemoved(likeFam, removed = false)
+ checkAnswer(likeFam, Seq(Row(1), Row(3)))
+
+ val likeEq = sql(s"SELECT id FROM $table WHERE flag LIKE 'Y'")
+ checkFiltersRemoved(likeEq, removed = false)
+ checkAnswer(likeEq, Seq.empty)
+
+ val likeEnd = sql(s"SELECT id FROM $table WHERE flag LIKE '%Y'")
+ checkFiltersRemoved(likeEnd, removed = false)
+ checkAnswer(likeEnd, Seq.empty)
+
+ val vsCols = "h2.test.char_vs_vc"
+ withTable(vsCols) {
+ sql(s"CREATE TABLE $vsCols (flag CHAR(5), v VARCHAR(5))")
+ sql(s"INSERT INTO $vsCols VALUES ('Y', 'Y')")
+ val vsVarchar = sql(s"SELECT v FROM $vsCols WHERE flag = v")
+ checkFiltersRemoved(vsVarchar, removed = false)
+ checkAnswer(vsVarchar, Seq.empty)
+ }
+
+ val rpadNest = sql(s"SELECT id FROM $table WHERE rpad(flag, 10, 'x') =
'Y xxxxx'")
+ checkFiltersRemoved(rpadNest, removed = false)
+ checkAnswer(rpadNest, Seq(Row(1)))
+
+ val cnt = sql(s"SELECT COUNT(flag) FROM $table")
+ checkAggregateRemoved(cnt, removed = false)
+ checkAnswer(cnt, Seq(Row(4)))
+
+ val caseSum = sql(
+ s"SELECT SUM(CASE WHEN flag = 'Y' THEN 1 ELSE 0 END) FROM $table")
+ checkAggregateRemoved(caseSum)
+ checkAnswer(caseSum, Seq(Row(1)))
+
+ val grp = sql(s"SELECT flag, COUNT(*) FROM $table GROUP BY flag")
+ checkAggregateRemoved(grp, removed = false)
+ checkAnswer(grp, Seq(
+ Row("Y ", 1), Row("N ", 1), Row("YES ", 1), Row(null, 1),
Row("ä ", 1)))
+ // scalastyle:on nonascii
+
+ val topN = sql(s"SELECT id FROM $table WHERE flag IS NOT NULL ORDER BY
flag LIMIT 1")
+ checkSortRemoved(topN, removed = false)
+ checkAnswer(topN, Seq(Row(2)))
+
+ withSQLConf(SQLConf.READ_SIDE_CHAR_PADDING.key -> "false") {
+ val padFalse = sql(s"SELECT id FROM $table WHERE flag = 'Y'")
+ checkFiltersRemoved(padFalse)
+ checkPushedInfo(padFalse,
+ "PushedFilters: [flag IS NOT NULL, (RPAD(flag, 5, ' ')) = 'Y ']")
+ checkAnswer(padFalse, Seq(Row(1)))
+ }
+
+ withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true") {
+ val std = sql(s"SELECT id FROM $table WHERE flag = CAST('Y' AS
CHAR(5))")
+ checkFiltersRemoved(std, removed = false)
+ checkPushedInfo(std, "PushedFilters: [flag IS NOT NULL]")
+ checkAnswer(std, Seq(Row(1)))
+ }
+ }
+ }
+
test("scan with aggregate push-down: MAX AVG with filter and group by") {
val df = sql("SELECT MAX(SaLaRY), AVG(BONUS) FROM h2.test.employee WHERE
dept > 0" +
" GROUP BY DePt")
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
