You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by we...@apache.org on 2022/03/22 07:49:05 UTC
[spark] branch branch-3.3 updated: [SPARK-38432][SQL] Refactor framework so as JDBC dialect could compile filter by self way
This is an automated email from the ASF dual-hosted git repository.
wenchen pushed a commit to branch branch-3.3
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-3.3 by this push:
new a14f3a5 [SPARK-38432][SQL] Refactor framework so as JDBC dialect could compile filter by self way
a14f3a5 is described below
commit a14f3a52e8f9d9e58f6fdcf53aebb0802f477436
Author: Jiaan Geng <be...@163.com>
AuthorDate: Tue Mar 22 15:45:51 2022 +0800
[SPARK-38432][SQL] Refactor framework so as JDBC dialect could compile filter by self way
### What changes were proposed in this pull request?
Currently, Spark DS V2 could push down filters into JDBC source. However, only the most basic form of filter is supported.
On the other hand, some JDBC source could not compile the filters by themselves way.
This PR reactor the framework so as JDBC dialect could compile expression by self way.
First, The framework translate catalyst expressions to DS V2 filters.
Second, The JDBC dialect could compile DS V2 filters to different SQL syntax.
### Why are the changes needed?
Make the framework be more common use.
### Does this PR introduce _any_ user-facing change?
'No'.
The feature is not released.
### How was this patch tested?
Exists tests.
Closes #35768 from beliefer/SPARK-38432_new.
Authored-by: Jiaan Geng <be...@163.com>
Signed-off-by: Wenchen Fan <we...@databricks.com>
(cherry picked from commit 53df45650af1e48e01e392caed6c1f83c2e9e9f1)
Signed-off-by: Wenchen Fan <we...@databricks.com>
---
project/MimaExcludes.scala | 2 +
.../sql/connector/expressions/Expression.java | 17 ++
.../expressions/GeneralScalarExpression.java | 98 ++------
.../spark/sql/connector/expressions/Literal.java | 3 +
.../sql/connector/expressions/NamedReference.java | 6 +
.../spark/sql/connector/expressions/SortOrder.java | 3 +
.../spark/sql/connector/expressions/Transform.java | 8 +-
.../sql/connector/expressions/aggregate/Avg.java | 3 +
.../sql/connector/expressions/aggregate/Count.java | 3 +
.../connector/expressions/aggregate/CountStar.java | 4 +
.../aggregate/GeneralAggregateFunc.java | 13 +-
.../sql/connector/expressions/aggregate/Max.java | 3 +
.../sql/connector/expressions/aggregate/Min.java | 3 +
.../sql/connector/expressions/aggregate/Sum.java | 3 +
.../connector/expressions/filter/AlwaysFalse.java | 30 +--
.../connector/expressions/filter/AlwaysTrue.java | 28 +--
.../sql/connector/expressions/filter/And.java | 14 +-
.../expressions/filter/BinaryComparison.java | 60 -----
.../connector/expressions/filter/BinaryFilter.java | 65 -----
.../expressions/filter/EqualNullSafe.java | 40 ---
.../sql/connector/expressions/filter/EqualTo.java | 39 ---
.../sql/connector/expressions/filter/Filter.java | 40 ---
.../connector/expressions/filter/GreaterThan.java | 39 ---
.../expressions/filter/GreaterThanOrEqual.java | 39 ---
.../spark/sql/connector/expressions/filter/In.java | 76 ------
.../connector/expressions/filter/IsNotNull.java | 58 -----
.../sql/connector/expressions/filter/IsNull.java | 58 -----
.../sql/connector/expressions/filter/LessThan.java | 39 ---
.../expressions/filter/LessThanOrEqual.java | 39 ---
.../sql/connector/expressions/filter/Not.java | 31 +--
.../spark/sql/connector/expressions/filter/Or.java | 14 +-
.../Predicate.java} | 126 +++-------
.../expressions/filter/StringContains.java | 39 ---
.../expressions/filter/StringEndsWith.java | 39 ---
.../expressions/filter/StringPredicate.java | 60 -----
.../expressions/filter/StringStartsWith.java | 41 ----
.../connector/read/SupportsPushDownV2Filters.java | 35 +--
.../sql/connector/util/V2ExpressionSQLBuilder.java | 100 ++++++--
.../org/apache/spark/sql/sources/filters.scala | 60 +++++
.../expressions/TransformExtractorSuite.scala | 1 -
.../sql/catalyst/util/V2ExpressionBuilder.scala | 113 +++++++--
.../spark/sql/execution/DataSourceScanExec.scala | 11 +-
.../execution/datasources/DataSourceStrategy.scala | 11 +-
.../sql/execution/datasources/jdbc/JDBCRDD.scala | 77 +-----
.../execution/datasources/jdbc/JDBCRelation.scala | 18 +-
.../datasources/v2/DataSourceV2Strategy.scala | 137 ++++-------
.../execution/datasources/v2/PushDownUtils.scala | 17 +-
.../datasources/v2/PushedDownOperators.scala | 4 +-
.../datasources/v2/V2ScanRelationPushDown.scala | 8 +-
.../execution/datasources/v2/jdbc/JDBCScan.scala | 9 +-
.../datasources/v2/jdbc/JDBCScanBuilder.scala | 20 +-
.../org/apache/spark/sql/jdbc/DB2Dialect.scala | 24 +-
.../org/apache/spark/sql/jdbc/DerbyDialect.scala | 16 +-
.../org/apache/spark/sql/jdbc/H2Dialect.scala | 28 +--
.../org/apache/spark/sql/jdbc/JdbcDialects.scala | 22 +-
.../apache/spark/sql/jdbc/MsSqlServerDialect.scala | 16 +-
.../org/apache/spark/sql/jdbc/MySQLDialect.scala | 16 +-
.../org/apache/spark/sql/jdbc/OracleDialect.scala | 28 +--
.../apache/spark/sql/jdbc/PostgresDialect.scala | 28 +--
.../apache/spark/sql/jdbc/TeradataDialect.scala | 28 +--
.../JavaAdvancedDataSourceV2WithV2Filter.java | 75 +++---
.../spark/sql/connector/DataSourceV2Suite.scala | 43 ++--
.../datasources/v2/DataSourceV2StrategySuite.scala | 10 +-
.../execution/datasources/v2/V2FiltersSuite.scala | 204 ----------------
.../datasources/v2/V2PredicateSuite.scala | 188 +++++++++++++++
.../org/apache/spark/sql/jdbc/JDBCSuite.scala | 19 +-
.../org/apache/spark/sql/jdbc/JDBCV2Suite.scala | 268 ++++++++++++++++-----
67 files changed, 1101 insertions(+), 1716 deletions(-)
diff --git a/project/MimaExcludes.scala b/project/MimaExcludes.scala
index b045d46..dc3b661 100644
--- a/project/MimaExcludes.scala
+++ b/project/MimaExcludes.scala
@@ -49,6 +49,8 @@ object MimaExcludes {
ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.spark.sql.util.ExecutionListenerManager.this"),
// [SPARK-37786][SQL] StreamingQueryListener support use SQLConf.get to get corresponding SessionState's SQLConf
ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.spark.sql.streaming.StreamingQueryManager.this"),
+ // [SPARK-38432][SQL] Reactor framework so as JDBC dialect could compile filter by self way
+ ProblemFilters.exclude[ReversedMissingMethodProblem]("org.apache.spark.sql.sources.Filter.toV2"),
// [SPARK-37600][BUILD] Upgrade to Hadoop 3.3.2
ProblemFilters.exclude[MissingClassProblem]("org.apache.hadoop.shaded.net.jpountz.lz4.LZ4Compressor"),
diff --git a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/Expression.java b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/Expression.java
index 9f6c097..76dfe73 100644
--- a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/Expression.java
+++ b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/Expression.java
@@ -17,6 +17,8 @@
package org.apache.spark.sql.connector.expressions;
+import java.util.Arrays;
+
import org.apache.spark.annotation.Evolving;
/**
@@ -26,8 +28,23 @@ import org.apache.spark.annotation.Evolving;
*/
@Evolving
public interface Expression {
+ Expression[] EMPTY_EXPRESSION = new Expression[0];
+
/**
* Format the expression as a human readable SQL-like string.
*/
default String describe() { return this.toString(); }
+
+ /**
+ * Returns an array of the children of this node. Children should not change.
+ */
+ Expression[] children();
+
+ /**
+ * List of fields or columns that are referenced by this expression.
+ */
+ default NamedReference[] references() {
+ return Arrays.stream(children()).map(e -> e.references())
+ .flatMap(Arrays::stream).distinct().toArray(NamedReference[]::new);
+ }
}
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 b3dd2cb..8952761 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
@@ -19,77 +19,19 @@ package org.apache.spark.sql.connector.expressions;
import java.io.Serializable;
import java.util.Arrays;
+import java.util.Objects;
import org.apache.spark.annotation.Evolving;
+import org.apache.spark.sql.connector.expressions.filter.Predicate;
import org.apache.spark.sql.connector.util.V2ExpressionSQLBuilder;
-// scalastyle:off line.size.limit
/**
* The general representation of SQL scalar expressions, which contains the upper-cased
- * expression name and all the children expressions.
+ * expression name and all the children expressions. Please also see {@link Predicate}
+ * for the supported predicate expressions.
* <p>
* The currently supported SQL scalar expressions:
* <ol>
- * <li>Name: <code>IS_NULL</code>
- * <ul>
- * <li>SQL semantic: <code>expr IS NULL</code></li>
- * <li>Since version: 3.3.0</li>
- * </ul>
- * </li>
- * <li>Name: <code>IS_NOT_NULL</code>
- * <ul>
- * <li>SQL semantic: <code>expr IS NOT NULL</code></li>
- * <li>Since version: 3.3.0</li>
- * </ul>
- * </li>
- * <li>Name: <code>=</code>
- * <ul>
- * <li>SQL semantic: <code>expr1 = expr2</code></li>
- * <li>Since version: 3.3.0</li>
- * </ul>
- * </li>
- * <li>Name: <code>!=</code>
- * <ul>
- * <li>SQL semantic: <code>expr1 != expr2</code></li>
- * <li>Since version: 3.3.0</li>
- * </ul>
- * </li>
- * <li>Name: <code><></code>
- * <ul>
- * <li>SQL semantic: <code>expr1 <> expr2</code></li>
- * <li>Since version: 3.3.0</li>
- * </ul>
- * </li>
- * <li>Name: <code><=></code>
- * <ul>
- * <li>SQL semantic: <code>expr1 <=> expr2</code></li>
- * <li>Since version: 3.3.0</li>
- * </ul>
- * </li>
- * <li>Name: <code><</code>
- * <ul>
- * <li>SQL semantic: <code>expr1 < expr2</code></li>
- * <li>Since version: 3.3.0</li>
- * </ul>
- * </li>
- * <li>Name: <code><=</code>
- * <ul>
- * <li>SQL semantic: <code>expr1 <= expr2</code></li>
- * <li>Since version: 3.3.0</li>
- * </ul>
- * </li>
- * <li>Name: <code>></code>
- * <ul>
- * <li>SQL semantic: <code>expr1 > expr2</code></li>
- * <li>Since version: 3.3.0</li>
- * </ul>
- * </li>
- * <li>Name: <code>>=</code>
- * <ul>
- * <li>SQL semantic: <code>expr1 >= expr2</code></li>
- * <li>Since version: 3.3.0</li>
- * </ul>
- * </li>
* <li>Name: <code>+</code>
* <ul>
* <li>SQL semantic: <code>expr1 + expr2</code></li>
@@ -138,24 +80,6 @@ import org.apache.spark.sql.connector.util.V2ExpressionSQLBuilder;
* <li>Since version: 3.3.0</li>
* </ul>
* </li>
- * <li>Name: <code>AND</code>
- * <ul>
- * <li>SQL semantic: <code>expr1 AND expr2</code></li>
- * <li>Since version: 3.3.0</li>
- * </ul>
- * </li>
- * <li>Name: <code>OR</code>
- * <ul>
- * <li>SQL semantic: <code>expr1 OR expr2</code></li>
- * <li>Since version: 3.3.0</li>
- * </ul>
- * </li>
- * <li>Name: <code>NOT</code>
- * <ul>
- * <li>SQL semantic: <code>NOT expr</code></li>
- * <li>Since version: 3.3.0</li>
- * </ul>
- * </li>
* <li>Name: <code>~</code>
* <ul>
* <li>SQL semantic: <code>~ expr</code></li>
@@ -176,7 +100,6 @@ import org.apache.spark.sql.connector.util.V2ExpressionSQLBuilder;
*
* @since 3.3.0
*/
-// scalastyle:on line.size.limit
@Evolving
public class GeneralScalarExpression implements Expression, Serializable {
private String name;
@@ -191,6 +114,19 @@ public class GeneralScalarExpression implements Expression, Serializable {
public Expression[] children() { return children; }
@Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+ GeneralScalarExpression that = (GeneralScalarExpression) o;
+ return Objects.equals(name, that.name) && Arrays.equals(children, that.children);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(name, children);
+ }
+
+ @Override
public String toString() {
V2ExpressionSQLBuilder builder = new V2ExpressionSQLBuilder();
try {
diff --git a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/Literal.java b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/Literal.java
index df9e58f..5e8aeaf 100644
--- a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/Literal.java
+++ b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/Literal.java
@@ -40,4 +40,7 @@ public interface Literal<T> extends Expression {
* Returns the SQL data type of the literal.
*/
DataType dataType();
+
+ @Override
+ default Expression[] children() { return EMPTY_EXPRESSION; }
}
diff --git a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/NamedReference.java b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/NamedReference.java
index 167432f..8c0f029 100644
--- a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/NamedReference.java
+++ b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/NamedReference.java
@@ -32,4 +32,10 @@ public interface NamedReference extends Expression {
* Each string in the returned array represents a field name.
*/
String[] fieldNames();
+
+ @Override
+ default Expression[] children() { return EMPTY_EXPRESSION; }
+
+ @Override
+ default NamedReference[] references() { return new NamedReference[]{ this }; }
}
diff --git a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/SortOrder.java b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/SortOrder.java
index 7225245..5140178 100644
--- a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/SortOrder.java
+++ b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/SortOrder.java
@@ -40,4 +40,7 @@ public interface SortOrder extends Expression {
* Returns the null ordering.
*/
NullOrdering nullOrdering();
+
+ @Override
+ default Expression[] children() { return new Expression[]{ expression() }; }
}
diff --git a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/Transform.java b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/Transform.java
index 2972058..e9ead7f 100644
--- a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/Transform.java
+++ b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/Transform.java
@@ -35,12 +35,10 @@ public interface Transform extends Expression {
String name();
/**
- * Returns all field references in the transform arguments.
- */
- NamedReference[] references();
-
- /**
* Returns the arguments passed to the transform function.
*/
Expression[] arguments();
+
+ @Override
+ default Expression[] children() { return arguments(); }
}
diff --git a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/aggregate/Avg.java b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/aggregate/Avg.java
index cc9d27a..d09e5f7 100644
--- a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/aggregate/Avg.java
+++ b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/aggregate/Avg.java
@@ -39,6 +39,9 @@ public final class Avg implements AggregateFunc {
public boolean isDistinct() { return isDistinct; }
@Override
+ public Expression[] children() { return new Expression[]{ input }; }
+
+ @Override
public String toString() {
if (isDistinct) {
return "AVG(DISTINCT " + input.describe() + ")";
diff --git a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/aggregate/Count.java b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/aggregate/Count.java
index 54c64b8..c840b29 100644
--- a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/aggregate/Count.java
+++ b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/aggregate/Count.java
@@ -39,6 +39,9 @@ public final class Count implements AggregateFunc {
public boolean isDistinct() { return isDistinct; }
@Override
+ public Expression[] children() { return new Expression[]{ input }; }
+
+ @Override
public String toString() {
if (isDistinct) {
return "COUNT(DISTINCT " + input.describe() + ")";
diff --git a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/aggregate/CountStar.java b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/aggregate/CountStar.java
index 1380119..ff8639c 100644
--- a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/aggregate/CountStar.java
+++ b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/aggregate/CountStar.java
@@ -18,6 +18,7 @@
package org.apache.spark.sql.connector.expressions.aggregate;
import org.apache.spark.annotation.Evolving;
+import org.apache.spark.sql.connector.expressions.Expression;
/**
* An aggregate function that returns the number of rows in a group.
@@ -31,5 +32,8 @@ public final class CountStar implements AggregateFunc {
}
@Override
+ public Expression[] children() { return EMPTY_EXPRESSION; }
+
+ @Override
public String toString() { return "COUNT(*)"; }
}
diff --git a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/aggregate/GeneralAggregateFunc.java b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/aggregate/GeneralAggregateFunc.java
index 0ff26c8..7016644 100644
--- a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/aggregate/GeneralAggregateFunc.java
+++ b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/aggregate/GeneralAggregateFunc.java
@@ -22,7 +22,6 @@ import java.util.stream.Collectors;
import org.apache.spark.annotation.Evolving;
import org.apache.spark.sql.connector.expressions.Expression;
-import org.apache.spark.sql.connector.expressions.NamedReference;
/**
* The general implementation of {@link AggregateFunc}, which contains the upper-cased function
@@ -46,21 +45,23 @@ import org.apache.spark.sql.connector.expressions.NamedReference;
public final class GeneralAggregateFunc implements AggregateFunc {
private final String name;
private final boolean isDistinct;
- private final NamedReference[] inputs;
+ private final Expression[] children;
public String name() { return name; }
public boolean isDistinct() { return isDistinct; }
- public NamedReference[] inputs() { return inputs; }
- public GeneralAggregateFunc(String name, boolean isDistinct, NamedReference[] inputs) {
+ public GeneralAggregateFunc(String name, boolean isDistinct, Expression[] children) {
this.name = name;
this.isDistinct = isDistinct;
- this.inputs = inputs;
+ this.children = children;
}
@Override
+ public Expression[] children() { return children; }
+
+ @Override
public String toString() {
- String inputsString = Arrays.stream(inputs)
+ String inputsString = Arrays.stream(children)
.map(Expression::describe)
.collect(Collectors.joining(", "));
if (isDistinct) {
diff --git a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/aggregate/Max.java b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/aggregate/Max.java
index 971aac2..089d2bd 100644
--- a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/aggregate/Max.java
+++ b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/aggregate/Max.java
@@ -34,5 +34,8 @@ public final class Max implements AggregateFunc {
public Expression column() { return input; }
@Override
+ public Expression[] children() { return new Expression[]{ input }; }
+
+ @Override
public String toString() { return "MAX(" + input.describe() + ")"; }
}
diff --git a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/aggregate/Min.java b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/aggregate/Min.java
index 8d0644b..253cdea 100644
--- a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/aggregate/Min.java
+++ b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/aggregate/Min.java
@@ -34,5 +34,8 @@ public final class Min implements AggregateFunc {
public Expression column() { return input; }
@Override
+ public Expression[] children() { return new Expression[]{ input }; }
+
+ @Override
public String toString() { return "MIN(" + input.describe() + ")"; }
}
diff --git a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/aggregate/Sum.java b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/aggregate/Sum.java
index 721ef31..4e01b92 100644
--- a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/aggregate/Sum.java
+++ b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/aggregate/Sum.java
@@ -39,6 +39,9 @@ public final class Sum implements AggregateFunc {
public boolean isDistinct() { return isDistinct; }
@Override
+ public Expression[] children() { return new Expression[]{ input }; }
+
+ @Override
public String toString() {
if (isDistinct) {
return "SUM(DISTINCT " + input.describe() + ")";
diff --git a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/filter/AlwaysFalse.java b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/filter/AlwaysFalse.java
index 72ed83f..accdd1a 100644
--- a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/filter/AlwaysFalse.java
+++ b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/filter/AlwaysFalse.java
@@ -17,34 +17,30 @@
package org.apache.spark.sql.connector.expressions.filter;
-import java.util.Objects;
-
import org.apache.spark.annotation.Evolving;
-import org.apache.spark.sql.connector.expressions.NamedReference;
+import org.apache.spark.sql.connector.expressions.Literal;
+import org.apache.spark.sql.types.DataType;
+import org.apache.spark.sql.types.DataTypes;
/**
- * A filter that always evaluates to {@code false}.
+ * A predicate that always evaluates to {@code false}.
*
* @since 3.3.0
*/
@Evolving
-public final class AlwaysFalse extends Filter {
+public final class AlwaysFalse extends Predicate implements Literal<Boolean> {
- @Override
- public boolean equals(Object o) {
- if (this == o) return true;
- if (o == null || getClass() != o.getClass()) return false;
- return true;
+ public AlwaysFalse() {
+ super("ALWAYS_FALSE", new Predicate[]{});
}
- @Override
- public int hashCode() {
- return Objects.hash();
+ public Boolean value() {
+ return false;
}
- @Override
- public String toString() { return "FALSE"; }
+ public DataType dataType() {
+ return DataTypes.BooleanType;
+ }
- @Override
- public NamedReference[] references() { return EMPTY_REFERENCE; }
+ public String toString() { return "FALSE"; }
}
diff --git a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/filter/AlwaysTrue.java b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/filter/AlwaysTrue.java
index b6d39c3..5a14f64 100644
--- a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/filter/AlwaysTrue.java
+++ b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/filter/AlwaysTrue.java
@@ -17,34 +17,30 @@
package org.apache.spark.sql.connector.expressions.filter;
-import java.util.Objects;
-
import org.apache.spark.annotation.Evolving;
-import org.apache.spark.sql.connector.expressions.NamedReference;
+import org.apache.spark.sql.connector.expressions.Literal;
+import org.apache.spark.sql.types.DataType;
+import org.apache.spark.sql.types.DataTypes;
/**
- * A filter that always evaluates to {@code true}.
+ * A predicate that always evaluates to {@code true}.
*
* @since 3.3.0
*/
@Evolving
-public final class AlwaysTrue extends Filter {
+public final class AlwaysTrue extends Predicate implements Literal<Boolean> {
+
+ public AlwaysTrue() {
+ super("ALWAYS_TRUE", new Predicate[]{});
+ }
- @Override
- public boolean equals(Object o) {
- if (this == o) return true;
- if (o == null || getClass() != o.getClass()) return false;
+ public Boolean value() {
return true;
}
- @Override
- public int hashCode() {
- return Objects.hash();
+ public DataType dataType() {
+ return DataTypes.BooleanType;
}
- @Override
public String toString() { return "TRUE"; }
-
- @Override
- public NamedReference[] references() { return EMPTY_REFERENCE; }
}
diff --git a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/filter/And.java b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/filter/And.java
index e0b8b13..179a4b3 100644
--- a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/filter/And.java
+++ b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/filter/And.java
@@ -20,20 +20,18 @@ package org.apache.spark.sql.connector.expressions.filter;
import org.apache.spark.annotation.Evolving;
/**
- * A filter that evaluates to {@code true} iff both {@code left} and {@code right} evaluate to
+ * A predicate that evaluates to {@code true} iff both {@code left} and {@code right} evaluate to
* {@code true}.
*
* @since 3.3.0
*/
@Evolving
-public final class And extends BinaryFilter {
+public final class And extends Predicate {
- public And(Filter left, Filter right) {
- super(left, right);
+ public And(Predicate left, Predicate right) {
+ super("AND", new Predicate[]{left, right});
}
- @Override
- public String toString() {
- return String.format("(%s) AND (%s)", left.describe(), right.describe());
- }
+ public Predicate left() { return (Predicate) children()[0]; }
+ public Predicate right() { return (Predicate) children()[1]; }
}
diff --git a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/filter/BinaryComparison.java b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/filter/BinaryComparison.java
deleted file mode 100644
index 0ae6e5a..0000000
--- a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/filter/BinaryComparison.java
+++ /dev/null
@@ -1,60 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.spark.sql.connector.expressions.filter;
-
-import java.util.Objects;
-
-import org.apache.spark.annotation.Evolving;
-import org.apache.spark.sql.connector.expressions.Literal;
-import org.apache.spark.sql.connector.expressions.NamedReference;
-
-/**
- * Base class for {@link EqualNullSafe}, {@link EqualTo}, {@link GreaterThan},
- * {@link GreaterThanOrEqual}, {@link LessThan}, {@link LessThanOrEqual}
- *
- * @since 3.3.0
- */
-@Evolving
-abstract class BinaryComparison extends Filter {
- protected final NamedReference column;
- protected final Literal<?> value;
-
- protected BinaryComparison(NamedReference column, Literal<?> value) {
- this.column = column;
- this.value = value;
- }
-
- public NamedReference column() { return column; }
- public Literal<?> value() { return value; }
-
- @Override
- public boolean equals(Object o) {
- if (this == o) return true;
- if (o == null || getClass() != o.getClass()) return false;
- BinaryComparison that = (BinaryComparison) o;
- return Objects.equals(column, that.column) && Objects.equals(value, that.value);
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(column, value);
- }
-
- @Override
- public NamedReference[] references() { return new NamedReference[] { column }; }
-}
diff --git a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/filter/BinaryFilter.java b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/filter/BinaryFilter.java
deleted file mode 100644
index ac4b9f2..0000000
--- a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/filter/BinaryFilter.java
+++ /dev/null
@@ -1,65 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.spark.sql.connector.expressions.filter;
-
-import java.util.Objects;
-
-import org.apache.spark.annotation.Evolving;
-import org.apache.spark.sql.connector.expressions.NamedReference;
-
-/**
- * Base class for {@link And}, {@link Or}
- *
- * @since 3.3.0
- */
-@Evolving
-abstract class BinaryFilter extends Filter {
- protected final Filter left;
- protected final Filter right;
-
- protected BinaryFilter(Filter left, Filter right) {
- this.left = left;
- this.right = right;
- }
-
- public Filter left() { return left; }
- public Filter right() { return right; }
-
- @Override
- public boolean equals(Object o) {
- if (this == o) return true;
- if (o == null || getClass() != o.getClass()) return false;
- BinaryFilter and = (BinaryFilter) o;
- return Objects.equals(left, and.left) && Objects.equals(right, and.right);
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(left, right);
- }
-
- @Override
- public NamedReference[] references() {
- NamedReference[] refLeft = left.references();
- NamedReference[] refRight = right.references();
- NamedReference[] arr = new NamedReference[refLeft.length + refRight.length];
- System.arraycopy(refLeft, 0, arr, 0, refLeft.length);
- System.arraycopy(refRight, 0, arr, refLeft.length, refRight.length);
- return arr;
- }
-}
diff --git a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/filter/EqualNullSafe.java b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/filter/EqualNullSafe.java
deleted file mode 100644
index 34b5291..0000000
--- a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/filter/EqualNullSafe.java
+++ /dev/null
@@ -1,40 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.spark.sql.connector.expressions.filter;
-
-import org.apache.spark.annotation.Evolving;
-import org.apache.spark.sql.connector.expressions.Literal;
-import org.apache.spark.sql.connector.expressions.NamedReference;
-
-/**
- * Performs equality comparison, similar to {@link EqualTo}. However, this differs from
- * {@link EqualTo} in that it returns {@code true} (rather than NULL) if both inputs are NULL,
- * and {@code false} (rather than NULL) if one of the input is NULL and the other is not NULL.
- *
- * @since 3.3.0
- */
-@Evolving
-public final class EqualNullSafe extends BinaryComparison {
-
- public EqualNullSafe(NamedReference column, Literal<?> value) {
- super(column, value);
- }
-
- @Override
- public String toString() { return this.column.describe() + " <=> " + value.describe(); }
-}
diff --git a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/filter/EqualTo.java b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/filter/EqualTo.java
deleted file mode 100644
index b9c4fe0..0000000
--- a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/filter/EqualTo.java
+++ /dev/null
@@ -1,39 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.spark.sql.connector.expressions.filter;
-
-import org.apache.spark.annotation.Evolving;
-import org.apache.spark.sql.connector.expressions.Literal;
-import org.apache.spark.sql.connector.expressions.NamedReference;
-
-/**
- * A filter that evaluates to {@code true} iff the {@code column} evaluates to a value
- * equal to {@code value}.
- *
- * @since 3.3.0
- */
-@Evolving
-public final class EqualTo extends BinaryComparison {
-
- public EqualTo(NamedReference column, Literal<?> value) {
- super(column, value);
- }
-
- @Override
- public String toString() { return column.describe() + " = " + value.describe(); }
-}
diff --git a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/filter/Filter.java b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/filter/Filter.java
deleted file mode 100644
index af87e76..0000000
--- a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/filter/Filter.java
+++ /dev/null
@@ -1,40 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.spark.sql.connector.expressions.filter;
-
-import java.io.Serializable;
-
-import org.apache.spark.annotation.Evolving;
-import org.apache.spark.sql.connector.expressions.Expression;
-import org.apache.spark.sql.connector.expressions.NamedReference;
-
-/**
- * Filter base class
- *
- * @since 3.3.0
- */
-@Evolving
-public abstract class Filter implements Expression, Serializable {
-
- protected static final NamedReference[] EMPTY_REFERENCE = new NamedReference[0];
-
- /**
- * Returns list of columns that are referenced by this filter.
- */
- public abstract NamedReference[] references();
-}
diff --git a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/filter/GreaterThan.java b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/filter/GreaterThan.java
deleted file mode 100644
index a3374f3..0000000
--- a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/filter/GreaterThan.java
+++ /dev/null
@@ -1,39 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.spark.sql.connector.expressions.filter;
-
-import org.apache.spark.annotation.Evolving;
-import org.apache.spark.sql.connector.expressions.Literal;
-import org.apache.spark.sql.connector.expressions.NamedReference;
-
-/**
- * A filter that evaluates to {@code true} iff the {@code column} evaluates to a value
- * greater than {@code value}.
- *
- * @since 3.3.0
- */
-@Evolving
-public final class GreaterThan extends BinaryComparison {
-
- public GreaterThan(NamedReference column, Literal<?> value) {
- super(column, value);
- }
-
- @Override
- public String toString() { return column.describe() + " > " + value.describe(); }
-}
diff --git a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/filter/GreaterThanOrEqual.java b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/filter/GreaterThanOrEqual.java
deleted file mode 100644
index 4ee9210..0000000
--- a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/filter/GreaterThanOrEqual.java
+++ /dev/null
@@ -1,39 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.spark.sql.connector.expressions.filter;
-
-import org.apache.spark.annotation.Evolving;
-import org.apache.spark.sql.connector.expressions.Literal;
-import org.apache.spark.sql.connector.expressions.NamedReference;
-
-/**
- * A filter that evaluates to {@code true} iff the {@code column} evaluates to a value
- * greater than or equal to {@code value}.
- *
- * @since 3.3.0
- */
-@Evolving
-public final class GreaterThanOrEqual extends BinaryComparison {
-
- public GreaterThanOrEqual(NamedReference column, Literal<?> value) {
- super(column, value);
- }
-
- @Override
- public String toString() { return column.describe() + " >= " + value.describe(); }
-}
diff --git a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/filter/In.java b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/filter/In.java
deleted file mode 100644
index 8d6490b..0000000
--- a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/filter/In.java
+++ /dev/null
@@ -1,76 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.spark.sql.connector.expressions.filter;
-
-import java.util.Arrays;
-import java.util.Objects;
-import java.util.stream.Collectors;
-
-import org.apache.spark.annotation.Evolving;
-import org.apache.spark.sql.connector.expressions.Literal;
-import org.apache.spark.sql.connector.expressions.NamedReference;
-
-/**
- * A filter that evaluates to {@code true} iff the {@code column} evaluates to one of the
- * {@code values} in the array.
- *
- * @since 3.3.0
- */
-@Evolving
-public final class In extends Filter {
- static final int MAX_LEN_TO_PRINT = 50;
- private final NamedReference column;
- private final Literal<?>[] values;
-
- public In(NamedReference column, Literal<?>[] values) {
- this.column = column;
- this.values = values;
- }
-
- public NamedReference column() { return column; }
- public Literal<?>[] values() { return values; }
-
- @Override
- public boolean equals(Object o) {
- if (this == o) return true;
- if (o == null || getClass() != o.getClass()) return false;
- In in = (In) o;
- return Objects.equals(column, in.column) && values.length == in.values.length
- && Arrays.asList(values).containsAll(Arrays.asList(in.values));
- }
-
- @Override
- public int hashCode() {
- int result = Objects.hash(column);
- result = 31 * result + Arrays.hashCode(values);
- return result;
- }
-
- @Override
- public String toString() {
- String res = Arrays.stream(values).limit((MAX_LEN_TO_PRINT)).map(Literal::describe)
- .collect(Collectors.joining(", "));
- if (values.length > MAX_LEN_TO_PRINT) {
- res += "...";
- }
- return column.describe() + " IN (" + res + ")";
- }
-
- @Override
- public NamedReference[] references() { return new NamedReference[] { column }; }
-}
diff --git a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/filter/IsNotNull.java b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/filter/IsNotNull.java
deleted file mode 100644
index 2cf000e..0000000
--- a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/filter/IsNotNull.java
+++ /dev/null
@@ -1,58 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.spark.sql.connector.expressions.filter;
-
-import java.util.Objects;
-
-import org.apache.spark.annotation.Evolving;
-import org.apache.spark.sql.connector.expressions.NamedReference;
-
-/**
- * A filter that evaluates to {@code true} iff the {@code column} evaluates to a non-null value.
- *
- * @since 3.3.0
- */
-@Evolving
-public final class IsNotNull extends Filter {
- private final NamedReference column;
-
- public IsNotNull(NamedReference column) {
- this.column = column;
- }
-
- public NamedReference column() { return column; }
-
- @Override
- public String toString() { return column.describe() + " IS NOT NULL"; }
-
- @Override
- public boolean equals(Object o) {
- if (this == o) return true;
- if (o == null || getClass() != o.getClass()) return false;
- IsNotNull isNotNull = (IsNotNull) o;
- return Objects.equals(column, isNotNull.column);
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(column);
- }
-
- @Override
- public NamedReference[] references() { return new NamedReference[] { column }; }
-}
diff --git a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/filter/IsNull.java b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/filter/IsNull.java
deleted file mode 100644
index 1cd497c..0000000
--- a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/filter/IsNull.java
+++ /dev/null
@@ -1,58 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.spark.sql.connector.expressions.filter;
-
-import java.util.Objects;
-
-import org.apache.spark.annotation.Evolving;
-import org.apache.spark.sql.connector.expressions.NamedReference;
-
-/**
- * A filter that evaluates to {@code true} iff the {@code column} evaluates to null.
- *
- * @since 3.3.0
- */
-@Evolving
-public final class IsNull extends Filter {
- private final NamedReference column;
-
- public IsNull(NamedReference column) {
- this.column = column;
- }
-
- public NamedReference column() { return column; }
-
- @Override
- public String toString() { return column.describe() + " IS NULL"; }
-
- @Override
- public boolean equals(Object o) {
- if (this == o) return true;
- if (o == null || getClass() != o.getClass()) return false;
- IsNull isNull = (IsNull) o;
- return Objects.equals(column, isNull.column);
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(column);
- }
-
- @Override
- public NamedReference[] references() { return new NamedReference[] { column }; }
-}
diff --git a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/filter/LessThan.java b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/filter/LessThan.java
deleted file mode 100644
index 9fa5cfb..0000000
--- a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/filter/LessThan.java
+++ /dev/null
@@ -1,39 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.spark.sql.connector.expressions.filter;
-
-import org.apache.spark.annotation.Evolving;
-import org.apache.spark.sql.connector.expressions.Literal;
-import org.apache.spark.sql.connector.expressions.NamedReference;
-
-/**
- * A filter that evaluates to {@code true} iff the {@code column} evaluates to a value
- * less than {@code value}.
- *
- * @since 3.3.0
- */
-@Evolving
-public final class LessThan extends BinaryComparison {
-
- public LessThan(NamedReference column, Literal<?> value) {
- super(column, value);
- }
-
- @Override
- public String toString() { return column.describe() + " < " + value.describe(); }
-}
diff --git a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/filter/LessThanOrEqual.java b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/filter/LessThanOrEqual.java
deleted file mode 100644
index a41b3c8..0000000
--- a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/filter/LessThanOrEqual.java
+++ /dev/null
@@ -1,39 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.spark.sql.connector.expressions.filter;
-
-import org.apache.spark.annotation.Evolving;
-import org.apache.spark.sql.connector.expressions.Literal;
-import org.apache.spark.sql.connector.expressions.NamedReference;
-
-/**
- * A filter that evaluates to {@code true} iff the {@code column} evaluates to a value
- * less than or equal to {@code value}.
- *
- * @since 3.3.0
- */
-@Evolving
-public final class LessThanOrEqual extends BinaryComparison {
-
- public LessThanOrEqual(NamedReference column, Literal<?> value) {
- super(column, value);
- }
-
- @Override
- public String toString() { return column.describe() + " <= " + value.describe(); }
-}
diff --git a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/filter/Not.java b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/filter/Not.java
index 69746f5..d65c9f0 100644
--- a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/filter/Not.java
+++ b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/filter/Not.java
@@ -17,40 +17,19 @@
package org.apache.spark.sql.connector.expressions.filter;
-import java.util.Objects;
-
import org.apache.spark.annotation.Evolving;
-import org.apache.spark.sql.connector.expressions.NamedReference;
/**
- * A filter that evaluates to {@code true} iff {@code child} is evaluated to {@code false}.
+ * A predicate that evaluates to {@code true} iff {@code child} is evaluated to {@code false}.
*
* @since 3.3.0
*/
@Evolving
-public final class Not extends Filter {
- private final Filter child;
-
- public Not(Filter child) { this.child = child; }
-
- public Filter child() { return child; }
+public final class Not extends Predicate {
- @Override
- public boolean equals(Object o) {
- if (this == o) return true;
- if (o == null || getClass() != o.getClass()) return false;
- Not not = (Not) o;
- return Objects.equals(child, not.child);
+ public Not(Predicate child) {
+ super("NOT", new Predicate[]{child});
}
- @Override
- public int hashCode() {
- return Objects.hash(child);
- }
-
- @Override
- public String toString() { return "NOT (" + child.describe() + ")"; }
-
- @Override
- public NamedReference[] references() { return child.references(); }
+ public Predicate child() { return (Predicate) children()[0]; }
}
diff --git a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/filter/Or.java b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/filter/Or.java
index baa33d8..7f1717c 100644
--- a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/filter/Or.java
+++ b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/filter/Or.java
@@ -20,20 +20,18 @@ package org.apache.spark.sql.connector.expressions.filter;
import org.apache.spark.annotation.Evolving;
/**
- * A filter that evaluates to {@code true} iff at least one of {@code left} or {@code right}
+ * A predicate that evaluates to {@code true} iff at least one of {@code left} or {@code right}
* evaluates to {@code true}.
*
* @since 3.3.0
*/
@Evolving
-public final class Or extends BinaryFilter {
+public final class Or extends Predicate {
- public Or(Filter left, Filter right) {
- super(left, right);
+ public Or(Predicate left, Predicate right) {
+ super("OR", new Predicate[]{left, right});
}
- @Override
- public String toString() {
- return String.format("(%s) OR (%s)", left.describe(), right.describe());
- }
+ public Predicate left() { return (Predicate) children()[0]; }
+ public Predicate right() { return (Predicate) children()[1]; }
}
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/filter/Predicate.java
similarity index 53%
copy from sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/GeneralScalarExpression.java
copy to sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/filter/Predicate.java
index b3dd2cb..e58cddc 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/filter/Predicate.java
@@ -15,20 +15,18 @@
* limitations under the License.
*/
-package org.apache.spark.sql.connector.expressions;
-
-import java.io.Serializable;
-import java.util.Arrays;
+package org.apache.spark.sql.connector.expressions.filter;
import org.apache.spark.annotation.Evolving;
-import org.apache.spark.sql.connector.util.V2ExpressionSQLBuilder;
+import org.apache.spark.sql.connector.expressions.Expression;
+import org.apache.spark.sql.connector.expressions.GeneralScalarExpression;
-// scalastyle:off line.size.limit
/**
- * The general representation of SQL scalar expressions, which contains the upper-cased
- * expression name and all the children expressions.
+ * The general representation of predicate expressions, which contains the upper-cased expression
+ * name and all the children expressions. You can also use these concrete subclasses for better
+ * type safety: {@link And}, {@link Or}, {@link Not}, {@link AlwaysTrue}, {@link AlwaysFalse}.
* <p>
- * The currently supported SQL scalar expressions:
+ * The currently supported predicate expressions:
* <ol>
* <li>Name: <code>IS_NULL</code>
* <ul>
@@ -42,99 +40,69 @@ import org.apache.spark.sql.connector.util.V2ExpressionSQLBuilder;
* <li>Since version: 3.3.0</li>
* </ul>
* </li>
- * <li>Name: <code>=</code>
- * <ul>
- * <li>SQL semantic: <code>expr1 = expr2</code></li>
- * <li>Since version: 3.3.0</li>
- * </ul>
- * </li>
- * <li>Name: <code>!=</code>
- * <ul>
- * <li>SQL semantic: <code>expr1 != expr2</code></li>
- * <li>Since version: 3.3.0</li>
- * </ul>
- * </li>
- * <li>Name: <code><></code>
- * <ul>
- * <li>SQL semantic: <code>expr1 <> expr2</code></li>
- * <li>Since version: 3.3.0</li>
- * </ul>
- * </li>
- * <li>Name: <code><=></code>
- * <ul>
- * <li>SQL semantic: <code>expr1 <=> expr2</code></li>
- * <li>Since version: 3.3.0</li>
- * </ul>
- * </li>
- * <li>Name: <code><</code>
- * <ul>
- * <li>SQL semantic: <code>expr1 < expr2</code></li>
- * <li>Since version: 3.3.0</li>
- * </ul>
- * </li>
- * <li>Name: <code><=</code>
+ * <li>Name: <code>STARTS_WITH</code>
* <ul>
- * <li>SQL semantic: <code>expr1 <= expr2</code></li>
+ * <li>SQL semantic: <code>expr1 LIKE 'expr2%'</code></li>
* <li>Since version: 3.3.0</li>
* </ul>
* </li>
- * <li>Name: <code>></code>
+ * <li>Name: <code>ENDS_WITH</code>
* <ul>
- * <li>SQL semantic: <code>expr1 > expr2</code></li>
+ * <li>SQL semantic: <code>expr1 LIKE '%expr2'</code></li>
* <li>Since version: 3.3.0</li>
* </ul>
* </li>
- * <li>Name: <code>>=</code>
+ * <li>Name: <code>CONTAINS</code>
* <ul>
- * <li>SQL semantic: <code>expr1 >= expr2</code></li>
+ * <li>SQL semantic: <code>expr1 LIKE '%expr2%'</code></li>
* <li>Since version: 3.3.0</li>
* </ul>
* </li>
- * <li>Name: <code>+</code>
+ * <li>Name: <code>IN</code>
* <ul>
- * <li>SQL semantic: <code>expr1 + expr2</code></li>
+ * <li>SQL semantic: <code>expr IN (expr1, expr2, ...)</code></li>
* <li>Since version: 3.3.0</li>
* </ul>
* </li>
- * <li>Name: <code>-</code>
+ * <li>Name: <code>=</code>
* <ul>
- * <li>SQL semantic: <code>expr1 - expr2</code> or <code>- expr</code></li>
+ * <li>SQL semantic: <code>expr1 = expr2</code></li>
* <li>Since version: 3.3.0</li>
* </ul>
* </li>
- * <li>Name: <code>*</code>
+ * <li>Name: <code><></code>
* <ul>
- * <li>SQL semantic: <code>expr1 * expr2</code></li>
+ * <li>SQL semantic: <code>expr1 <> expr2</code></li>
* <li>Since version: 3.3.0</li>
* </ul>
* </li>
- * <li>Name: <code>/</code>
+ * <li>Name: <code><=></code>
* <ul>
- * <li>SQL semantic: <code>expr1 / expr2</code></li>
+ * <li>SQL semantic: null-safe version of <code>expr1 = expr2</code></li>
* <li>Since version: 3.3.0</li>
* </ul>
* </li>
- * <li>Name: <code>%</code>
+ * <li>Name: <code><</code>
* <ul>
- * <li>SQL semantic: <code>expr1 % expr2</code></li>
+ * <li>SQL semantic: <code>expr1 < expr2</code></li>
* <li>Since version: 3.3.0</li>
* </ul>
* </li>
- * <li>Name: <code>&</code>
+ * <li>Name: <code><=</code>
* <ul>
- * <li>SQL semantic: <code>expr1 & expr2</code></li>
+ * <li>SQL semantic: <code>expr1 <= expr2</code></li>
* <li>Since version: 3.3.0</li>
* </ul>
* </li>
- * <li>Name: <code>|</code>
+ * <li>Name: <code>></code>
* <ul>
- * <li>SQL semantic: <code>expr1 | expr2</code></li>
+ * <li>SQL semantic: <code>expr1 > expr2</code></li>
* <li>Since version: 3.3.0</li>
* </ul>
* </li>
- * <li>Name: <code>^</code>
+ * <li>Name: <code>>=</code>
* <ul>
- * <li>SQL semantic: <code>expr1 ^ expr2</code></li>
+ * <li>SQL semantic: <code>expr1 >= expr2</code></li>
* <li>Since version: 3.3.0</li>
* </ul>
* </li>
@@ -156,48 +124,26 @@ import org.apache.spark.sql.connector.util.V2ExpressionSQLBuilder;
* <li>Since version: 3.3.0</li>
* </ul>
* </li>
- * <li>Name: <code>~</code>
+ * <li>Name: <code>ALWAYS_TRUE</code>
* <ul>
- * <li>SQL semantic: <code>~ expr</code></li>
+ * <li>SQL semantic: <code>TRUE</code></li>
* <li>Since version: 3.3.0</li>
* </ul>
* </li>
- * <li>Name: <code>CASE_WHEN</code>
+ * <li>Name: <code>ALWAYS_FALSE</code>
* <ul>
- * <li>SQL semantic:
- * <code>CASE WHEN expr1 THEN expr2 [WHEN expr3 THEN expr4]* [ELSE expr5] END</code>
- * </li>
+ * <li>SQL semantic: <code>FALSE</code></li>
* <li>Since version: 3.3.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.
*
* @since 3.3.0
*/
-// scalastyle:on line.size.limit
@Evolving
-public class GeneralScalarExpression implements Expression, Serializable {
- private String name;
- private Expression[] children;
-
- public GeneralScalarExpression(String name, Expression[] children) {
- this.name = name;
- this.children = children;
- }
-
- public String name() { return name; }
- public Expression[] children() { return children; }
+public class Predicate extends GeneralScalarExpression {
- @Override
- public String toString() {
- V2ExpressionSQLBuilder builder = new V2ExpressionSQLBuilder();
- try {
- return builder.build(this);
- } catch (Throwable e) {
- return name + "(" +
- Arrays.stream(children).map(child -> child.toString()).reduce((a,b) -> a + "," + b) + ")";
- }
+ public Predicate(String name, Expression[] children) {
+ super(name, children);
}
}
diff --git a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/filter/StringContains.java b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/filter/StringContains.java
deleted file mode 100644
index 9a01e4d..0000000
--- a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/filter/StringContains.java
+++ /dev/null
@@ -1,39 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.spark.sql.connector.expressions.filter;
-
-import org.apache.spark.annotation.Evolving;
-import org.apache.spark.sql.connector.expressions.NamedReference;
-import org.apache.spark.unsafe.types.UTF8String;
-
-/**
- * A filter that evaluates to {@code true} iff the {@code column} evaluates to
- * a string that contains {@code value}.
- *
- * @since 3.3.0
- */
-@Evolving
-public final class StringContains extends StringPredicate {
-
- public StringContains(NamedReference column, UTF8String value) {
- super(column, value);
- }
-
- @Override
- public String toString() { return "STRING_CONTAINS(" + column.describe() + ", " + value + ")"; }
-}
diff --git a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/filter/StringEndsWith.java b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/filter/StringEndsWith.java
deleted file mode 100644
index 11b8317..0000000
--- a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/filter/StringEndsWith.java
+++ /dev/null
@@ -1,39 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.spark.sql.connector.expressions.filter;
-
-import org.apache.spark.annotation.Evolving;
-import org.apache.spark.sql.connector.expressions.NamedReference;
-import org.apache.spark.unsafe.types.UTF8String;
-
-/**
- * A filter that evaluates to {@code true} iff the {@code column} evaluates to
- * a string that ends with {@code value}.
- *
- * @since 3.3.0
- */
-@Evolving
-public final class StringEndsWith extends StringPredicate {
-
- public StringEndsWith(NamedReference column, UTF8String value) {
- super(column, value);
- }
-
- @Override
- public String toString() { return "STRING_ENDS_WITH(" + column.describe() + ", " + value + ")"; }
-}
diff --git a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/filter/StringPredicate.java b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/filter/StringPredicate.java
deleted file mode 100644
index ffe5d5d..0000000
--- a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/filter/StringPredicate.java
+++ /dev/null
@@ -1,60 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.spark.sql.connector.expressions.filter;
-
-import java.util.Objects;
-
-import org.apache.spark.annotation.Evolving;
-import org.apache.spark.sql.connector.expressions.NamedReference;
-import org.apache.spark.unsafe.types.UTF8String;
-
-/**
- * Base class for {@link StringContains}, {@link StringStartsWith},
- * {@link StringEndsWith}
- *
- * @since 3.3.0
- */
-@Evolving
-abstract class StringPredicate extends Filter {
- protected final NamedReference column;
- protected final UTF8String value;
-
- protected StringPredicate(NamedReference column, UTF8String value) {
- this.column = column;
- this.value = value;
- }
-
- public NamedReference column() { return column; }
- public UTF8String value() { return value; }
-
- @Override
- public boolean equals(Object o) {
- if (this == o) return true;
- if (o == null || getClass() != o.getClass()) return false;
- StringPredicate that = (StringPredicate) o;
- return Objects.equals(column, that.column) && Objects.equals(value, that.value);
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(column, value);
- }
-
- @Override
- public NamedReference[] references() { return new NamedReference[] { column }; }
-}
diff --git a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/filter/StringStartsWith.java b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/filter/StringStartsWith.java
deleted file mode 100644
index 38a5de1..0000000
--- a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/filter/StringStartsWith.java
+++ /dev/null
@@ -1,41 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.spark.sql.connector.expressions.filter;
-
-import org.apache.spark.annotation.Evolving;
-import org.apache.spark.sql.connector.expressions.NamedReference;
-import org.apache.spark.unsafe.types.UTF8String;
-
-/**
- * A filter that evaluates to {@code true} iff the {@code column} evaluates to
- * a string that starts with {@code value}.
- *
- * @since 3.3.0
- */
-@Evolving
-public final class StringStartsWith extends StringPredicate {
-
- public StringStartsWith(NamedReference column, UTF8String value) {
- super(column, value);
- }
-
- @Override
- public String toString() {
- return "STRING_STARTS_WITH(" + column.describe() + ", " + value + ")";
- }
-}
diff --git a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/read/SupportsPushDownV2Filters.java b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/read/SupportsPushDownV2Filters.java
index e1a5401..1fec939 100644
--- a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/read/SupportsPushDownV2Filters.java
+++ b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/read/SupportsPushDownV2Filters.java
@@ -18,11 +18,11 @@
package org.apache.spark.sql.connector.read;
import org.apache.spark.annotation.Evolving;
-import org.apache.spark.sql.connector.expressions.filter.Filter;
+import org.apache.spark.sql.connector.expressions.filter.Predicate;
/**
* A mix-in interface for {@link ScanBuilder}. Data sources can implement this interface to
- * push down V2 {@link Filter} to the data source and reduce the size of the data to be read.
+ * push down V2 {@link Predicate} to the data source and reduce the size of the data to be read.
* Please Note that this interface is preferred over {@link SupportsPushDownFilters}, which uses
* V1 {@link org.apache.spark.sql.sources.Filter} and is less efficient due to the
* internal -> external data conversion.
@@ -33,28 +33,31 @@ import org.apache.spark.sql.connector.expressions.filter.Filter;
public interface SupportsPushDownV2Filters extends ScanBuilder {
/**
- * Pushes down filters, and returns filters that need to be evaluated after scanning.
+ * Pushes down predicates, and returns predicates that need to be evaluated after scanning.
* <p>
- * Rows should be returned from the data source if and only if all of the filters match. That is,
- * filters must be interpreted as ANDed together.
+ * Rows should be returned from the data source if and only if all of the predicates match.
+ * That is, predicates must be interpreted as ANDed together.
*/
- Filter[] pushFilters(Filter[] filters);
+ Predicate[] pushPredicates(Predicate[] predicates);
/**
- * Returns the filters that are pushed to the data source via {@link #pushFilters(Filter[])}.
+ * Returns the predicates that are pushed to the data source via
+ * {@link #pushPredicates(Predicate[])}.
* <p>
- * There are 3 kinds of filters:
+ * There are 3 kinds of predicates:
* <ol>
- * <li>pushable filters which don't need to be evaluated again after scanning.</li>
- * <li>pushable filters which still need to be evaluated after scanning, e.g. parquet row
- * group filter.</li>
- * <li>non-pushable filters.</li>
+ * <li>pushable predicates which don't need to be evaluated again after scanning.</li>
+ * <li>pushable predicates which still need to be evaluated after scanning, e.g. parquet row
+ * group predicate.</li>
+ * <li>non-pushable predicates.</li>
* </ol>
* <p>
- * Both case 1 and 2 should be considered as pushed filters and should be returned by this method.
+ * Both case 1 and 2 should be considered as pushed predicates and should be returned
+ * by this method.
* <p>
- * It's possible that there is no filters in the query and {@link #pushFilters(Filter[])}
- * is never called, empty array should be returned for this case.
+ * It's possible that there is no predicates in the query and
+ * {@link #pushPredicates(Predicate[])} is never called,
+ * empty array should be returned for this case.
*/
- Filter[] pushedFilters();
+ Predicate[] pushedPredicates();
}
diff --git a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/util/V2ExpressionSQLBuilder.java b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/util/V2ExpressionSQLBuilder.java
index 0af0d88..91dae74 100644
--- a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/util/V2ExpressionSQLBuilder.java
+++ b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/util/V2ExpressionSQLBuilder.java
@@ -17,39 +17,53 @@
package org.apache.spark.sql.connector.util;
-import java.util.ArrayList;
+import java.util.Arrays;
import java.util.List;
+import java.util.stream.Collectors;
import org.apache.spark.sql.connector.expressions.Expression;
-import org.apache.spark.sql.connector.expressions.FieldReference;
+import org.apache.spark.sql.connector.expressions.NamedReference;
import org.apache.spark.sql.connector.expressions.GeneralScalarExpression;
-import org.apache.spark.sql.connector.expressions.LiteralValue;
+import org.apache.spark.sql.connector.expressions.Literal;
/**
* The builder to generate SQL from V2 expressions.
*/
public class V2ExpressionSQLBuilder {
+
public String build(Expression expr) {
- if (expr instanceof LiteralValue) {
- return visitLiteral((LiteralValue) expr);
- } else if (expr instanceof FieldReference) {
- return visitFieldReference((FieldReference) expr);
+ if (expr instanceof Literal) {
+ return visitLiteral((Literal) expr);
+ } else if (expr instanceof NamedReference) {
+ return visitNamedReference((NamedReference) expr);
} else if (expr instanceof GeneralScalarExpression) {
GeneralScalarExpression e = (GeneralScalarExpression) expr;
String name = e.name();
switch (name) {
+ case "IN": {
+ List<String> children =
+ Arrays.stream(e.children()).map(c -> build(c)).collect(Collectors.toList());
+ return visitIn(children.get(0), children.subList(1, children.size()));
+ }
case "IS_NULL":
return visitIsNull(build(e.children()[0]));
case "IS_NOT_NULL":
return visitIsNotNull(build(e.children()[0]));
+ case "STARTS_WITH":
+ return visitStartsWith(build(e.children()[0]), build(e.children()[1]));
+ case "ENDS_WITH":
+ return visitEndsWith(build(e.children()[0]), build(e.children()[1]));
+ case "CONTAINS":
+ return visitContains(build(e.children()[0]), build(e.children()[1]));
case "=":
- case "!=":
+ case "<>":
case "<=>":
case "<":
case "<=":
case ">":
case ">=":
- return visitBinaryComparison(name, build(e.children()[0]), build(e.children()[1]));
+ return visitBinaryComparison(
+ name, inputToSQL(e.children()[0]), inputToSQL(e.children()[1]));
case "+":
case "*":
case "/":
@@ -57,12 +71,14 @@ public class V2ExpressionSQLBuilder {
case "&":
case "|":
case "^":
- return visitBinaryArithmetic(name, build(e.children()[0]), build(e.children()[1]));
+ return visitBinaryArithmetic(
+ name, inputToSQL(e.children()[0]), inputToSQL(e.children()[1]));
case "-":
if (e.children().length == 1) {
return visitUnaryArithmetic(name, build(e.children()[0]));
} else {
- return visitBinaryArithmetic(name, build(e.children()[0]), build(e.children()[1]));
+ return visitBinaryArithmetic(
+ name, inputToSQL(e.children()[0]), inputToSQL(e.children()[1]));
}
case "AND":
return visitAnd(name, build(e.children()[0]), build(e.children()[1]));
@@ -72,12 +88,11 @@ public class V2ExpressionSQLBuilder {
return visitNot(build(e.children()[0]));
case "~":
return visitUnaryArithmetic(name, build(e.children()[0]));
- case "CASE_WHEN":
- List<String> children = new ArrayList<>();
- for (Expression child : e.children()) {
- children.add(build(child));
- }
+ case "CASE_WHEN": {
+ List<String> children =
+ Arrays.stream(e.children()).map(c -> build(c)).collect(Collectors.toList());
return visitCaseWhen(children.toArray(new String[e.children().length]));
+ }
// TODO supports other expressions
default:
return visitUnexpectedExpr(expr);
@@ -87,12 +102,19 @@ public class V2ExpressionSQLBuilder {
}
}
- protected String visitLiteral(LiteralValue literalValue) {
- return literalValue.toString();
+ protected String visitLiteral(Literal literal) {
+ return literal.toString();
}
- protected String visitFieldReference(FieldReference fieldRef) {
- return fieldRef.toString();
+ protected String visitNamedReference(NamedReference namedRef) {
+ return namedRef.toString();
+ }
+
+ protected String visitIn(String v, List<String> list) {
+ if (list.isEmpty()) {
+ return "CASE WHEN " + v + " IS NULL THEN NULL ELSE FALSE END";
+ }
+ return v + " IN (" + list.stream().collect(Collectors.joining(", ")) + ")";
}
protected String visitIsNull(String v) {
@@ -103,12 +125,46 @@ public class V2ExpressionSQLBuilder {
return v + " IS NOT NULL";
}
+ protected String visitStartsWith(String l, String r) {
+ // Remove quotes at the beginning and end.
+ // e.g. converts "'str'" to "str".
+ String value = r.substring(1, r.length() - 1);
+ return l + " LIKE '" + value + "%'";
+ }
+
+ protected String visitEndsWith(String l, String r) {
+ // Remove quotes at the beginning and end.
+ // e.g. converts "'str'" to "str".
+ String value = r.substring(1, r.length() - 1);
+ return l + " LIKE '%" + value + "'";
+ }
+
+ protected String visitContains(String l, String r) {
+ // Remove quotes at the beginning and end.
+ // e.g. converts "'str'" to "str".
+ String value = r.substring(1, r.length() - 1);
+ return l + " LIKE '%" + value + "%'";
+ }
+
+ private String inputToSQL(Expression input) {
+ if (input.children().length > 1) {
+ return "(" + build(input) + ")";
+ } else {
+ return build(input);
+ }
+ }
+
protected String visitBinaryComparison(String name, String l, String r) {
- return "(" + l + ") " + name + " (" + r + ")";
+ switch (name) {
+ case "<=>":
+ return "(" + l + " = " + r + ") OR (" + l + " IS NULL AND " + r + " IS NULL)";
+ default:
+ return l + " " + name + " " + r;
+ }
}
protected String visitBinaryArithmetic(String name, String l, String r) {
- return "(" + l + ") " + name + " (" + r + ")";
+ return l + " " + name + " " + r;
}
protected String visitAnd(String name, String l, String r) {
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/sources/filters.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/sources/filters.scala
index 174dd08..9954821 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/sources/filters.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/sources/filters.scala
@@ -18,7 +18,12 @@
package org.apache.spark.sql.sources
import org.apache.spark.annotation.{Evolving, Stable}
+import org.apache.spark.sql.catalyst.expressions.Literal
import org.apache.spark.sql.connector.catalog.CatalogV2Implicits.parseColumnPath
+import org.apache.spark.sql.connector.expressions.{FieldReference, LiteralValue}
+import org.apache.spark.sql.connector.expressions.filter.{AlwaysFalse => V2AlwaysFalse, AlwaysTrue => V2AlwaysTrue, Predicate}
+import org.apache.spark.sql.types.StringType
+import org.apache.spark.unsafe.types.UTF8String
////////////////////////////////////////////////////////////////////////////////////////////////////
// This file defines all the filters that we can push down to the data sources.
@@ -64,6 +69,11 @@ sealed abstract class Filter {
private[sql] def containsNestedColumn: Boolean = {
this.v2references.exists(_.length > 1)
}
+
+ /**
+ * Converts V1 filter to V2 filter
+ */
+ private[sql] def toV2: Predicate
}
/**
@@ -78,6 +88,11 @@ sealed abstract class Filter {
@Stable
case class EqualTo(attribute: String, value: Any) extends Filter {
override def references: Array[String] = Array(attribute) ++ findReferences(value)
+ override def toV2: Predicate = {
+ val literal = Literal(value)
+ new Predicate("=",
+ Array(FieldReference(attribute), LiteralValue(literal.value, literal.dataType)))
+ }
}
/**
@@ -93,6 +108,11 @@ case class EqualTo(attribute: String, value: Any) extends Filter {
@Stable
case class EqualNullSafe(attribute: String, value: Any) extends Filter {
override def references: Array[String] = Array(attribute) ++ findReferences(value)
+ override def toV2: Predicate = {
+ val literal = Literal(value)
+ new Predicate("<=>",
+ Array(FieldReference(attribute), LiteralValue(literal.value, literal.dataType)))
+ }
}
/**
@@ -107,6 +127,11 @@ case class EqualNullSafe(attribute: String, value: Any) extends Filter {
@Stable
case class GreaterThan(attribute: String, value: Any) extends Filter {
override def references: Array[String] = Array(attribute) ++ findReferences(value)
+ override def toV2: Predicate = {
+ val literal = Literal(value)
+ new Predicate(">",
+ Array(FieldReference(attribute), LiteralValue(literal.value, literal.dataType)))
+ }
}
/**
@@ -121,6 +146,11 @@ case class GreaterThan(attribute: String, value: Any) extends Filter {
@Stable
case class GreaterThanOrEqual(attribute: String, value: Any) extends Filter {
override def references: Array[String] = Array(attribute) ++ findReferences(value)
+ override def toV2: Predicate = {
+ val literal = Literal(value)
+ new Predicate(">=",
+ Array(FieldReference(attribute), LiteralValue(literal.value, literal.dataType)))
+ }
}
/**
@@ -135,6 +165,11 @@ case class GreaterThanOrEqual(attribute: String, value: Any) extends Filter {
@Stable
case class LessThan(attribute: String, value: Any) extends Filter {
override def references: Array[String] = Array(attribute) ++ findReferences(value)
+ override def toV2: Predicate = {
+ val literal = Literal(value)
+ new Predicate("<",
+ Array(FieldReference(attribute), LiteralValue(literal.value, literal.dataType)))
+ }
}
/**
@@ -149,6 +184,11 @@ case class LessThan(attribute: String, value: Any) extends Filter {
@Stable
case class LessThanOrEqual(attribute: String, value: Any) extends Filter {
override def references: Array[String] = Array(attribute) ++ findReferences(value)
+ override def toV2: Predicate = {
+ val literal = Literal(value)
+ new Predicate("<=",
+ Array(FieldReference(attribute), LiteralValue(literal.value, literal.dataType)))
+ }
}
/**
@@ -185,6 +225,13 @@ case class In(attribute: String, values: Array[Any]) extends Filter {
}
override def references: Array[String] = Array(attribute) ++ values.flatMap(findReferences)
+ override def toV2: Predicate = {
+ val literals = values.map { value =>
+ val literal = Literal(value)
+ LiteralValue(literal.value, literal.dataType)
+ }
+ new Predicate("IN", FieldReference(attribute) +: literals)
+ }
}
/**
@@ -198,6 +245,7 @@ case class In(attribute: String, values: Array[Any]) extends Filter {
@Stable
case class IsNull(attribute: String) extends Filter {
override def references: Array[String] = Array(attribute)
+ override def toV2: Predicate = new Predicate("IS_NULL", Array(FieldReference(attribute)))
}
/**
@@ -211,6 +259,7 @@ case class IsNull(attribute: String) extends Filter {
@Stable
case class IsNotNull(attribute: String) extends Filter {
override def references: Array[String] = Array(attribute)
+ override def toV2: Predicate = new Predicate("IS_NOT_NULL", Array(FieldReference(attribute)))
}
/**
@@ -221,6 +270,7 @@ case class IsNotNull(attribute: String) extends Filter {
@Stable
case class And(left: Filter, right: Filter) extends Filter {
override def references: Array[String] = left.references ++ right.references
+ override def toV2: Predicate = new Predicate("AND", Seq(left, right).map(_.toV2).toArray)
}
/**
@@ -231,6 +281,7 @@ case class And(left: Filter, right: Filter) extends Filter {
@Stable
case class Or(left: Filter, right: Filter) extends Filter {
override def references: Array[String] = left.references ++ right.references
+ override def toV2: Predicate = new Predicate("OR", Seq(left, right).map(_.toV2).toArray)
}
/**
@@ -241,6 +292,7 @@ case class Or(left: Filter, right: Filter) extends Filter {
@Stable
case class Not(child: Filter) extends Filter {
override def references: Array[String] = child.references
+ override def toV2: Predicate = new Predicate("NOT", Array(child.toV2))
}
/**
@@ -255,6 +307,8 @@ case class Not(child: Filter) extends Filter {
@Stable
case class StringStartsWith(attribute: String, value: String) extends Filter {
override def references: Array[String] = Array(attribute)
+ override def toV2: Predicate = new Predicate("STARTS_WITH",
+ Array(FieldReference(attribute), LiteralValue(UTF8String.fromString(value), StringType)))
}
/**
@@ -269,6 +323,8 @@ case class StringStartsWith(attribute: String, value: String) extends Filter {
@Stable
case class StringEndsWith(attribute: String, value: String) extends Filter {
override def references: Array[String] = Array(attribute)
+ override def toV2: Predicate = new Predicate("ENDS_WITH",
+ Array(FieldReference(attribute), LiteralValue(UTF8String.fromString(value), StringType)))
}
/**
@@ -283,6 +339,8 @@ case class StringEndsWith(attribute: String, value: String) extends Filter {
@Stable
case class StringContains(attribute: String, value: String) extends Filter {
override def references: Array[String] = Array(attribute)
+ override def toV2: Predicate = new Predicate("CONTAINS",
+ Array(FieldReference(attribute), LiteralValue(UTF8String.fromString(value), StringType)))
}
/**
@@ -293,6 +351,7 @@ case class StringContains(attribute: String, value: String) extends Filter {
@Evolving
case class AlwaysTrue() extends Filter {
override def references: Array[String] = Array.empty
+ override def toV2: Predicate = new V2AlwaysTrue()
}
@Evolving
@@ -307,6 +366,7 @@ object AlwaysTrue extends AlwaysTrue {
@Evolving
case class AlwaysFalse() extends Filter {
override def references: Array[String] = Array.empty
+ override def toV2: Predicate = new V2AlwaysFalse()
}
@Evolving
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/expressions/TransformExtractorSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/expressions/TransformExtractorSuite.scala
index 54ab1df..62cae3c8 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/expressions/TransformExtractorSuite.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/expressions/TransformExtractorSuite.scala
@@ -45,7 +45,6 @@ class TransformExtractorSuite extends SparkFunSuite {
*/
private def transform(func: String, ref: NamedReference): Transform = new Transform {
override def name: String = func
- override def references: Array[NamedReference] = Array(ref)
override def arguments: Array[Expression] = Array(ref)
override def toString: String = ref.describe
}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/catalyst/util/V2ExpressionBuilder.scala b/sql/core/src/main/scala/org/apache/spark/sql/catalyst/util/V2ExpressionBuilder.scala
index 1e36169..5c8e6a6 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/catalyst/util/V2ExpressionBuilder.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/catalyst/util/V2ExpressionBuilder.scala
@@ -17,18 +17,23 @@
package org.apache.spark.sql.catalyst.util
-import org.apache.spark.sql.catalyst.expressions.{Add, And, Attribute, BinaryComparison, BinaryOperator, BitwiseAnd, BitwiseNot, BitwiseOr, BitwiseXor, CaseWhen, Divide, EqualTo, Expression, IsNotNull, IsNull, Literal, Multiply, Not, Or, Remainder, Subtract, UnaryMinus}
+import org.apache.spark.sql.catalyst.expressions.{Add, And, BinaryComparison, BinaryOperator, BitwiseAnd, BitwiseNot, BitwiseOr, BitwiseXor, CaseWhen, Contains, Divide, EndsWith, EqualTo, Expression, In, InSet, IsNotNull, IsNull, Literal, Multiply, Not, Or, Predicate, Remainder, StartsWith, StringPredicate, Subtract, UnaryMinus}
import org.apache.spark.sql.connector.expressions.{Expression => V2Expression, FieldReference, GeneralScalarExpression, LiteralValue}
+import org.apache.spark.sql.connector.expressions.filter.{AlwaysFalse, AlwaysTrue, And => V2And, Not => V2Not, Or => V2Or, Predicate => V2Predicate}
+import org.apache.spark.sql.execution.datasources.PushableColumn
+import org.apache.spark.sql.types.BooleanType
/**
* The builder to generate V2 expressions from catalyst expressions.
*/
-class V2ExpressionBuilder(e: Expression) {
+class V2ExpressionBuilder(
+ e: Expression, nestedPredicatePushdownEnabled: Boolean = false, isPredicate: Boolean = false) {
- def build(): Option[V2Expression] = generateExpression(e)
+ val pushableColumn = PushableColumn(nestedPredicatePushdownEnabled)
+
+ def build(): Option[V2Expression] = generateExpression(e, isPredicate)
private def canTranslate(b: BinaryOperator) = b match {
- case _: And | _: Or => true
case _: BinaryComparison => true
case _: BitwiseAnd | _: BitwiseOr | _: BitwiseXor => true
case add: Add => add.failOnError
@@ -39,18 +44,83 @@ class V2ExpressionBuilder(e: Expression) {
case _ => false
}
- private def generateExpression(expr: Expression): Option[V2Expression] = expr match {
+ private def generateExpression(
+ expr: Expression, isPredicate: Boolean = false): Option[V2Expression] = expr match {
+ case Literal(true, BooleanType) => Some(new AlwaysTrue())
+ case Literal(false, BooleanType) => Some(new AlwaysFalse())
case Literal(value, dataType) => Some(LiteralValue(value, dataType))
- case attr: Attribute => Some(FieldReference.column(attr.name))
+ case col @ pushableColumn(name) if nestedPredicatePushdownEnabled =>
+ if (isPredicate && col.dataType.isInstanceOf[BooleanType]) {
+ Some(new V2Predicate("=", Array(FieldReference(name), LiteralValue(true, BooleanType))))
+ } else {
+ Some(FieldReference(name))
+ }
+ case pushableColumn(name) if !nestedPredicatePushdownEnabled =>
+ Some(FieldReference.column(name))
+ case in @ InSet(child, hset) =>
+ generateExpression(child).map { v =>
+ val children =
+ (v +: hset.toSeq.map(elem => LiteralValue(elem, in.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
+ }
case IsNull(col) => generateExpression(col)
- .map(c => new GeneralScalarExpression("IS_NULL", Array[V2Expression](c)))
+ .map(c => new V2Predicate("IS_NULL", Array[V2Expression](c)))
case IsNotNull(col) => generateExpression(col)
- .map(c => new GeneralScalarExpression("IS_NOT_NULL", Array[V2Expression](c)))
- case b: BinaryOperator if canTranslate(b) =>
- val left = generateExpression(b.left)
- val right = generateExpression(b.right)
+ .map(c => new V2Predicate("IS_NOT_NULL", Array[V2Expression](c)))
+ case p: StringPredicate =>
+ val left = generateExpression(p.left)
+ val right = generateExpression(p.right)
if (left.isDefined && right.isDefined) {
- Some(new GeneralScalarExpression(b.sqlOperator, Array[V2Expression](left.get, right.get)))
+ val name = p match {
+ case _: StartsWith => "STARTS_WITH"
+ case _: EndsWith => "ENDS_WITH"
+ case _: Contains => "CONTAINS"
+ }
+ Some(new V2Predicate(name, Array[V2Expression](left.get, right.get)))
+ } else {
+ None
+ }
+ case and: And =>
+ val l = generateExpression(and.left, true)
+ val r = generateExpression(and.right, true)
+ if (l.isDefined && r.isDefined) {
+ assert(l.get.isInstanceOf[V2Predicate] && r.get.isInstanceOf[V2Predicate])
+ Some(new V2And(l.get.asInstanceOf[V2Predicate], r.get.asInstanceOf[V2Predicate]))
+ } else {
+ None
+ }
+ case or: Or =>
+ val l = generateExpression(or.left, true)
+ val r = generateExpression(or.right, true)
+ if (l.isDefined && r.isDefined) {
+ assert(l.get.isInstanceOf[V2Predicate] && r.get.isInstanceOf[V2Predicate])
+ Some(new V2Or(l.get.asInstanceOf[V2Predicate], r.get.asInstanceOf[V2Predicate]))
+ } else {
+ 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 _: 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
}
@@ -58,32 +128,35 @@ class V2ExpressionBuilder(e: Expression) {
val left = generateExpression(eq.left)
val right = generateExpression(eq.right)
if (left.isDefined && right.isDefined) {
- Some(new GeneralScalarExpression("!=", Array[V2Expression](left.get, right.get)))
+ Some(new V2Predicate("<>", Array[V2Expression](left.get, right.get)))
} else {
None
}
- case Not(child) => generateExpression(child)
- .map(v => new GeneralScalarExpression("NOT", Array[V2Expression](v)))
+ case Not(child) => generateExpression(child, true) // NOT expects predicate
+ .map { v =>
+ assert(v.isInstanceOf[V2Predicate])
+ new V2Not(v.asInstanceOf[V2Predicate])
+ }
case UnaryMinus(child, true) => generateExpression(child)
.map(v => new GeneralScalarExpression("-", Array[V2Expression](v)))
case BitwiseNot(child) => generateExpression(child)
.map(v => new GeneralScalarExpression("~", Array[V2Expression](v)))
case CaseWhen(branches, elseValue) =>
- val conditions = branches.map(_._1).flatMap(generateExpression)
- val values = branches.map(_._2).flatMap(generateExpression)
+ val conditions = branches.map(_._1).flatMap(generateExpression(_, true))
+ val values = branches.map(_._2).flatMap(generateExpression(_, true))
if (conditions.length == branches.length && values.length == branches.length) {
val branchExpressions = conditions.zip(values).flatMap { case (c, v) =>
Seq[V2Expression](c, v)
}
if (elseValue.isDefined) {
- elseValue.flatMap(generateExpression).map { v =>
+ elseValue.flatMap(generateExpression(_)).map { v =>
val children = (branchExpressions :+ v).toArray[V2Expression]
// The children looks like [condition1, value1, ..., conditionN, valueN, elseValue]
- new GeneralScalarExpression("CASE_WHEN", children)
+ new V2Predicate("CASE_WHEN", children)
}
} else {
// The children looks like [condition1, value1, ..., conditionN, valueN]
- Some(new GeneralScalarExpression("CASE_WHEN", branchExpressions.toArray[V2Expression]))
+ Some(new V2Predicate("CASE_WHEN", branchExpressions.toArray[V2Expression]))
}
} else {
None
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/DataSourceScanExec.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/DataSourceScanExec.scala
index 1e2fa41..e6de7d0 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/DataSourceScanExec.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/DataSourceScanExec.scala
@@ -153,9 +153,14 @@ case class RowDataSourceScanExec(
pushedDownOperators.limit.map(value => "PushedLimit" -> s"LIMIT $value")
}
- Map(
- "ReadSchema" -> requiredSchema.catalogString,
- "PushedFilters" -> seqToString(markedFilters.toSeq)) ++
+ val pushedFilters = if (pushedDownOperators.pushedPredicates.nonEmpty) {
+ seqToString(pushedDownOperators.pushedPredicates.map(_.describe()))
+ } else {
+ seqToString(markedFilters.toSeq)
+ }
+
+ Map("ReadSchema" -> requiredSchema.catalogString,
+ "PushedFilters" -> pushedFilters) ++
pushedDownOperators.aggregation.fold(Map[String, String]()) { v =>
Map("PushedAggregates" -> seqToString(v.aggregateExpressions.map(_.describe())),
"PushedGroupByColumns" -> seqToString(v.groupByColumns.map(_.describe())))} ++
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSourceStrategy.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSourceStrategy.scala
index 4e5014c..1b14884 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSourceStrategy.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSourceStrategy.scala
@@ -340,7 +340,7 @@ object DataSourceStrategy
l.output.toStructType,
Set.empty,
Set.empty,
- PushedDownOperators(None, None, None, Seq.empty),
+ PushedDownOperators(None, None, None, Seq.empty, Seq.empty),
toCatalystRDD(l, baseRelation.buildScan()),
baseRelation,
None) :: Nil
@@ -414,7 +414,7 @@ object DataSourceStrategy
requestedColumns.toStructType,
pushedFilters.toSet,
handledFilters,
- PushedDownOperators(None, None, None, Seq.empty),
+ PushedDownOperators(None, None, None, Seq.empty, Seq.empty),
scanBuilder(requestedColumns, candidatePredicates, pushedFilters),
relation.relation,
relation.catalogTable.map(_.identifier))
@@ -437,7 +437,7 @@ object DataSourceStrategy
requestedColumns.toStructType,
pushedFilters.toSet,
handledFilters,
- PushedDownOperators(None, None, None, Seq.empty),
+ PushedDownOperators(None, None, None, Seq.empty, Seq.empty),
scanBuilder(requestedColumns, candidatePredicates, pushedFilters),
relation.relation,
relation.catalogTable.map(_.identifier))
@@ -864,8 +864,5 @@ object PushableColumnWithoutNestedColumn extends PushableColumnBase {
* Get the expression of DS V2 to represent catalyst expression that can be pushed down.
*/
object PushableExpression {
- def unapply(e: Expression): Option[V2Expression] = e match {
- case PushableColumnWithoutNestedColumn(name) => Some(FieldReference.column(name))
- case _ => new V2ExpressionBuilder(e).build()
- }
+ def unapply(e: Expression): Option[V2Expression] = new V2ExpressionBuilder(e).build()
}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JDBCRDD.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JDBCRDD.scala
index 3cd2e03..b30b460 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JDBCRDD.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JDBCRDD.scala
@@ -24,13 +24,11 @@ import scala.util.control.NonFatal
import org.apache.spark.{InterruptibleIterator, Partition, SparkContext, TaskContext}
import org.apache.spark.internal.Logging
import org.apache.spark.rdd.RDD
-import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.connector.expressions.SortOrder
-import org.apache.spark.sql.errors.QueryCompilationErrors
+import org.apache.spark.sql.connector.expressions.filter.Predicate
import org.apache.spark.sql.execution.datasources.v2.TableSampleInfo
import org.apache.spark.sql.jdbc.{JdbcDialect, JdbcDialects}
-import org.apache.spark.sql.sources._
import org.apache.spark.sql.types._
import org.apache.spark.util.CompletionIterator
@@ -95,67 +93,12 @@ object JDBCRDD extends Logging {
}
/**
- * Turns a single Filter into a String representing a SQL expression.
- * Returns None for an unhandled filter.
- */
- def compileFilter(f: Filter, dialect: JdbcDialect): Option[String] = {
-
- def quote(colName: String): String = {
- val nameParts = SparkSession.active.sessionState.sqlParser.parseMultipartIdentifier(colName)
- if(nameParts.length > 1) {
- throw QueryCompilationErrors.commandNotSupportNestedColumnError("Filter push down", colName)
- }
- dialect.quoteIdentifier(nameParts.head)
- }
-
- Option(f match {
- case EqualTo(attr, value) => s"${quote(attr)} = ${dialect.compileValue(value)}"
- case EqualNullSafe(attr, value) =>
- val col = quote(attr)
- s"(NOT ($col != ${dialect.compileValue(value)} OR $col IS NULL OR " +
- s"${dialect.compileValue(value)} IS NULL) OR " +
- s"($col IS NULL AND ${dialect.compileValue(value)} IS NULL))"
- case LessThan(attr, value) => s"${quote(attr)} < ${dialect.compileValue(value)}"
- case GreaterThan(attr, value) => s"${quote(attr)} > ${dialect.compileValue(value)}"
- case LessThanOrEqual(attr, value) => s"${quote(attr)} <= ${dialect.compileValue(value)}"
- case GreaterThanOrEqual(attr, value) => s"${quote(attr)} >= ${dialect.compileValue(value)}"
- case IsNull(attr) => s"${quote(attr)} IS NULL"
- case IsNotNull(attr) => s"${quote(attr)} IS NOT NULL"
- case StringStartsWith(attr, value) => s"${quote(attr)} LIKE '${value}%'"
- case StringEndsWith(attr, value) => s"${quote(attr)} LIKE '%${value}'"
- case StringContains(attr, value) => s"${quote(attr)} LIKE '%${value}%'"
- case In(attr, value) if value.isEmpty =>
- s"CASE WHEN ${quote(attr)} IS NULL THEN NULL ELSE FALSE END"
- case In(attr, value) => s"${quote(attr)} IN (${dialect.compileValue(value)})"
- case Not(f) => compileFilter(f, dialect).map(p => s"(NOT ($p))").getOrElse(null)
- case Or(f1, f2) =>
- // We can't compile Or filter unless both sub-filters are compiled successfully.
- // It applies too for the following And filter.
- // If we can make sure compileFilter supports all filters, we can remove this check.
- val or = Seq(f1, f2).flatMap(compileFilter(_, dialect))
- if (or.size == 2) {
- or.map(p => s"($p)").mkString(" OR ")
- } else {
- null
- }
- case And(f1, f2) =>
- val and = Seq(f1, f2).flatMap(compileFilter(_, dialect))
- if (and.size == 2) {
- and.map(p => s"($p)").mkString(" AND ")
- } else {
- null
- }
- case _ => null
- })
- }
-
- /**
* Build and return JDBCRDD from the given information.
*
* @param sc - Your SparkContext.
* @param schema - The Catalyst schema of the underlying database table.
* @param requiredColumns - The names of the columns or aggregate columns to SELECT.
- * @param filters - The filters to include in all WHERE clauses.
+ * @param predicates - The predicates to include in all WHERE clauses.
* @param parts - An array of JDBCPartitions specifying partition ids and
* per-partition WHERE clauses.
* @param options - JDBC options that contains url, table and other information.
@@ -164,7 +107,7 @@ object JDBCRDD extends Logging {
* @param sample - The pushed down tableSample.
* @param limit - The pushed down limit. If the value is 0, it means no limit or limit
* is not pushed down.
- * @param sortValues - The sort values cooperates with limit to realize top N.
+ * @param sortOrders - The sort orders cooperates with limit to realize top N.
*
* @return An RDD representing "SELECT requiredColumns FROM fqTable".
*/
@@ -173,7 +116,7 @@ object JDBCRDD extends Logging {
sc: SparkContext,
schema: StructType,
requiredColumns: Array[String],
- filters: Array[Filter],
+ predicates: Array[Predicate],
parts: Array[Partition],
options: JDBCOptions,
outputSchema: Option[StructType] = None,
@@ -194,7 +137,7 @@ object JDBCRDD extends Logging {
dialect.createConnectionFactory(options),
outputSchema.getOrElse(pruneSchema(schema, requiredColumns)),
quotedColumns,
- filters,
+ predicates,
parts,
url,
options,
@@ -216,7 +159,7 @@ private[jdbc] class JDBCRDD(
getConnection: Int => Connection,
schema: StructType,
columns: Array[String],
- filters: Array[Filter],
+ predicates: Array[Predicate],
partitions: Array[Partition],
url: String,
options: JDBCOptions,
@@ -239,10 +182,10 @@ private[jdbc] class JDBCRDD(
/**
* `filters`, but as a WHERE clause suitable for injection into a SQL query.
*/
- private val filterWhereClause: String =
- filters
- .flatMap(JDBCRDD.compileFilter(_, JdbcDialects.get(url)))
- .map(p => s"($p)").mkString(" AND ")
+ private val filterWhereClause: String = {
+ val dialect = JdbcDialects.get(url)
+ predicates.flatMap(dialect.compileExpression(_)).map(p => s"($p)").mkString(" AND ")
+ }
/**
* A WHERE clause representing both `filters`, if any, and the current partition.
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JDBCRelation.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JDBCRelation.scala
index ecb2073..0f1a1b6 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JDBCRelation.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JDBCRelation.scala
@@ -28,6 +28,7 @@ import org.apache.spark.sql.catalyst.analysis._
import org.apache.spark.sql.catalyst.util.{DateFormatter, DateTimeUtils, TimestampFormatter}
import org.apache.spark.sql.catalyst.util.DateTimeUtils.{getZoneId, stringToDate, stringToTimestamp}
import org.apache.spark.sql.connector.expressions.SortOrder
+import org.apache.spark.sql.connector.expressions.filter.Predicate
import org.apache.spark.sql.errors.QueryCompilationErrors
import org.apache.spark.sql.execution.datasources.v2.TableSampleInfo
import org.apache.spark.sql.internal.SQLConf
@@ -270,10 +271,11 @@ private[sql] case class JDBCRelation(
override val needConversion: Boolean = false
- // Check if JDBCRDD.compileFilter can accept input filters
+ // Check if JdbcDialect can compile input filters
override def unhandledFilters(filters: Array[Filter]): Array[Filter] = {
if (jdbcOptions.pushDownPredicate) {
- filters.filter(JDBCRDD.compileFilter(_, JdbcDialects.get(jdbcOptions.url)).isEmpty)
+ val dialect = JdbcDialects.get(jdbcOptions.url)
+ filters.filter(f => dialect.compileExpression(f.toV2).isEmpty)
} else {
filters
}
@@ -281,17 +283,17 @@ private[sql] case class JDBCRelation(
override def buildScan(requiredColumns: Array[String], filters: Array[Filter]): RDD[Row] = {
// When pushDownPredicate is false, all Filters that need to be pushed down should be ignored
- val pushedFilters = if (jdbcOptions.pushDownPredicate) {
- filters
+ val pushedPredicates = if (jdbcOptions.pushDownPredicate) {
+ filters.map(_.toV2)
} else {
- Array.empty[Filter]
+ Array.empty[Predicate]
}
// Rely on a type erasure hack to pass RDD[InternalRow] back as RDD[Row]
JDBCRDD.scanTable(
sparkSession.sparkContext,
schema,
requiredColumns,
- pushedFilters,
+ pushedPredicates,
parts,
jdbcOptions).asInstanceOf[RDD[Row]]
}
@@ -299,7 +301,7 @@ private[sql] case class JDBCRelation(
def buildScan(
requiredColumns: Array[String],
finalSchema: StructType,
- filters: Array[Filter],
+ predicates: Array[Predicate],
groupByColumns: Option[Array[String]],
tableSample: Option[TableSampleInfo],
limit: Int,
@@ -309,7 +311,7 @@ private[sql] case class JDBCRelation(
sparkSession.sparkContext,
schema,
requiredColumns,
- filters,
+ predicates,
parts,
jdbcOptions,
Some(finalSchema),
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala
index e544a7c..c6d271b 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala
@@ -24,27 +24,25 @@ import org.apache.spark.sql.{SparkSession, Strategy}
import org.apache.spark.sql.catalyst.analysis.{ResolvedDBObjectName, ResolvedNamespace, ResolvedPartitionSpec, ResolvedTable}
import org.apache.spark.sql.catalyst.catalog.CatalogUtils
import org.apache.spark.sql.catalyst.expressions
-import org.apache.spark.sql.catalyst.expressions.{And, Attribute, DynamicPruning, EmptyRow, Expression, Literal, NamedExpression, PredicateHelper, SubqueryExpression}
+import org.apache.spark.sql.catalyst.expressions.{And, Attribute, DynamicPruning, Expression, NamedExpression, Not, Or, PredicateHelper, SubqueryExpression}
import org.apache.spark.sql.catalyst.planning.PhysicalOperation
import org.apache.spark.sql.catalyst.plans.logical._
-import org.apache.spark.sql.catalyst.util.toPrettySQL
+import org.apache.spark.sql.catalyst.util.{toPrettySQL, V2ExpressionBuilder}
import org.apache.spark.sql.connector.catalog.{Identifier, StagingTableCatalog, SupportsNamespaces, SupportsPartitionManagement, SupportsWrite, Table, TableCapability, TableCatalog}
import org.apache.spark.sql.connector.catalog.index.SupportsIndex
-import org.apache.spark.sql.connector.expressions.{FieldReference, Literal => V2Literal, LiteralValue}
-import org.apache.spark.sql.connector.expressions.filter.{AlwaysFalse => V2AlwaysFalse, AlwaysTrue => V2AlwaysTrue, And => V2And, EqualNullSafe => V2EqualNullSafe, EqualTo => V2EqualTo, Filter => V2Filter, GreaterThan => V2GreaterThan, GreaterThanOrEqual => V2GreaterThanOrEqual, In => V2In, IsNotNull => V2IsNotNull, IsNull => V2IsNull, LessThan => V2LessThan, LessThanOrEqual => V2LessThanOrEqual, Not => V2Not, Or => V2Or, StringContains => V2StringContains, StringEndsWith => V2StringEnds [...]
+import org.apache.spark.sql.connector.expressions.{FieldReference}
+import org.apache.spark.sql.connector.expressions.filter.{And => V2And, Not => V2Not, Or => V2Or, Predicate}
import org.apache.spark.sql.connector.read.LocalScan
import org.apache.spark.sql.connector.read.streaming.{ContinuousStream, MicroBatchStream}
import org.apache.spark.sql.connector.write.V1Write
import org.apache.spark.sql.errors.{QueryCompilationErrors, QueryExecutionErrors}
import org.apache.spark.sql.execution.{FilterExec, LeafExecNode, LocalTableScanExec, ProjectExec, RowDataSourceScanExec, SparkPlan}
-import org.apache.spark.sql.execution.datasources.{DataSourceStrategy, PushableColumn, PushableColumnBase}
+import org.apache.spark.sql.execution.datasources.DataSourceStrategy
import org.apache.spark.sql.execution.streaming.continuous.{WriteToContinuousDataSource, WriteToContinuousDataSourceExec}
import org.apache.spark.sql.internal.StaticSQLConf.WAREHOUSE_PATH
import org.apache.spark.sql.sources.{BaseRelation, TableScan}
-import org.apache.spark.sql.types.{BooleanType, StringType}
import org.apache.spark.sql.util.CaseInsensitiveStringMap
import org.apache.spark.storage.StorageLevel
-import org.apache.spark.unsafe.types.UTF8String
class DataSourceV2Strategy(session: SparkSession) extends Strategy with PredicateHelper {
@@ -473,74 +471,12 @@ private[sql] object DataSourceV2Strategy {
private def translateLeafNodeFilterV2(
predicate: Expression,
- pushableColumn: PushableColumnBase): Option[V2Filter] = predicate match {
- case expressions.EqualTo(pushableColumn(name), Literal(v, t)) =>
- Some(new V2EqualTo(FieldReference(name), LiteralValue(v, t)))
- case expressions.EqualTo(Literal(v, t), pushableColumn(name)) =>
- Some(new V2EqualTo(FieldReference(name), LiteralValue(v, t)))
-
- case expressions.EqualNullSafe(pushableColumn(name), Literal(v, t)) =>
- Some(new V2EqualNullSafe(FieldReference(name), LiteralValue(v, t)))
- case expressions.EqualNullSafe(Literal(v, t), pushableColumn(name)) =>
- Some(new V2EqualNullSafe(FieldReference(name), LiteralValue(v, t)))
-
- case expressions.GreaterThan(pushableColumn(name), Literal(v, t)) =>
- Some(new V2GreaterThan(FieldReference(name), LiteralValue(v, t)))
- case expressions.GreaterThan(Literal(v, t), pushableColumn(name)) =>
- Some(new V2LessThan(FieldReference(name), LiteralValue(v, t)))
-
- case expressions.LessThan(pushableColumn(name), Literal(v, t)) =>
- Some(new V2LessThan(FieldReference(name), LiteralValue(v, t)))
- case expressions.LessThan(Literal(v, t), pushableColumn(name)) =>
- Some(new V2GreaterThan(FieldReference(name), LiteralValue(v, t)))
-
- case expressions.GreaterThanOrEqual(pushableColumn(name), Literal(v, t)) =>
- Some(new V2GreaterThanOrEqual(FieldReference(name), LiteralValue(v, t)))
- case expressions.GreaterThanOrEqual(Literal(v, t), pushableColumn(name)) =>
- Some(new V2LessThanOrEqual(FieldReference(name), LiteralValue(v, t)))
-
- case expressions.LessThanOrEqual(pushableColumn(name), Literal(v, t)) =>
- Some(new V2LessThanOrEqual(FieldReference(name), LiteralValue(v, t)))
- case expressions.LessThanOrEqual(Literal(v, t), pushableColumn(name)) =>
- Some(new V2GreaterThanOrEqual(FieldReference(name), LiteralValue(v, t)))
-
- case in @ expressions.InSet(pushableColumn(name), set) =>
- val values: Array[V2Literal[_]] =
- set.toSeq.map(elem => LiteralValue(elem, in.dataType)).toArray
- Some(new V2In(FieldReference(name), values))
-
- // 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 @ expressions.In(pushableColumn(name), list) if list.forall(_.isInstanceOf[Literal]) =>
- val hSet = list.map(_.eval(EmptyRow))
- Some(new V2In(FieldReference(name),
- hSet.toArray.map(LiteralValue(_, in.value.dataType))))
-
- case expressions.IsNull(pushableColumn(name)) =>
- Some(new V2IsNull(FieldReference(name)))
- case expressions.IsNotNull(pushableColumn(name)) =>
- Some(new V2IsNotNull(FieldReference(name)))
-
- case expressions.StartsWith(pushableColumn(name), Literal(v: UTF8String, StringType)) =>
- Some(new V2StringStartsWith(FieldReference(name), v))
-
- case expressions.EndsWith(pushableColumn(name), Literal(v: UTF8String, StringType)) =>
- Some(new V2StringEndsWith(FieldReference(name), v))
-
- case expressions.Contains(pushableColumn(name), Literal(v: UTF8String, StringType)) =>
- Some(new V2StringContains(FieldReference(name), v))
-
- case expressions.Literal(true, BooleanType) =>
- Some(new V2AlwaysTrue)
-
- case expressions.Literal(false, BooleanType) =>
- Some(new V2AlwaysFalse)
-
- case e @ pushableColumn(name) if e.dataType.isInstanceOf[BooleanType] =>
- Some(new V2EqualTo(FieldReference(name), LiteralValue(true, BooleanType)))
-
- case _ => None
+ supportNestedPredicatePushdown: Boolean): Option[Predicate] = {
+ val pushablePredicate = PushablePredicate(supportNestedPredicatePushdown)
+ predicate match {
+ case pushablePredicate(expr) => Some(expr)
+ case _ => None
+ }
}
/**
@@ -550,7 +486,7 @@ private[sql] object DataSourceV2Strategy {
*/
protected[sql] def translateFilterV2(
predicate: Expression,
- supportNestedPredicatePushdown: Boolean): Option[V2Filter] = {
+ supportNestedPredicatePushdown: Boolean): Option[Predicate] = {
translateFilterV2WithMapping(predicate, None, supportNestedPredicatePushdown)
}
@@ -565,11 +501,11 @@ private[sql] object DataSourceV2Strategy {
*/
protected[sql] def translateFilterV2WithMapping(
predicate: Expression,
- translatedFilterToExpr: Option[mutable.HashMap[V2Filter, Expression]],
+ translatedFilterToExpr: Option[mutable.HashMap[Predicate, Expression]],
nestedPredicatePushdownEnabled: Boolean)
- : Option[V2Filter] = {
+ : Option[Predicate] = {
predicate match {
- case expressions.And(left, right) =>
+ case And(left, right) =>
// See SPARK-12218 for detailed discussion
// It is not safe to just convert one side if we do not understand the
// other side. Here is an example used to explain the reason.
@@ -586,7 +522,7 @@ private[sql] object DataSourceV2Strategy {
right, translatedFilterToExpr, nestedPredicatePushdownEnabled)
} yield new V2And(leftFilter, rightFilter)
- case expressions.Or(left, right) =>
+ case Or(left, right) =>
for {
leftFilter <- translateFilterV2WithMapping(
left, translatedFilterToExpr, nestedPredicatePushdownEnabled)
@@ -594,13 +530,12 @@ private[sql] object DataSourceV2Strategy {
right, translatedFilterToExpr, nestedPredicatePushdownEnabled)
} yield new V2Or(leftFilter, rightFilter)
- case expressions.Not(child) =>
+ case Not(child) =>
translateFilterV2WithMapping(child, translatedFilterToExpr, nestedPredicatePushdownEnabled)
.map(new V2Not(_))
case other =>
- val filter = translateLeafNodeFilterV2(
- other, PushableColumn(nestedPredicatePushdownEnabled))
+ val filter = translateLeafNodeFilterV2(other, nestedPredicatePushdownEnabled)
if (filter.isDefined && translatedFilterToExpr.isDefined) {
translatedFilterToExpr.get(filter.get) = predicate
}
@@ -609,20 +544,34 @@ private[sql] object DataSourceV2Strategy {
}
protected[sql] def rebuildExpressionFromFilter(
- filter: V2Filter,
- translatedFilterToExpr: mutable.HashMap[V2Filter, Expression]): Expression = {
- filter match {
+ predicate: Predicate,
+ translatedFilterToExpr: mutable.HashMap[Predicate, Expression]): Expression = {
+ predicate match {
case and: V2And =>
- expressions.And(rebuildExpressionFromFilter(and.left, translatedFilterToExpr),
- rebuildExpressionFromFilter(and.right, translatedFilterToExpr))
+ expressions.And(
+ rebuildExpressionFromFilter(and.left(), translatedFilterToExpr),
+ rebuildExpressionFromFilter(and.right(), translatedFilterToExpr))
case or: V2Or =>
- expressions.Or(rebuildExpressionFromFilter(or.left, translatedFilterToExpr),
- rebuildExpressionFromFilter(or.right, translatedFilterToExpr))
+ expressions.Or(
+ rebuildExpressionFromFilter(or.left(), translatedFilterToExpr),
+ rebuildExpressionFromFilter(or.right(), translatedFilterToExpr))
case not: V2Not =>
- expressions.Not(rebuildExpressionFromFilter(not.child, translatedFilterToExpr))
- case other =>
- translatedFilterToExpr.getOrElse(other,
- throw new IllegalStateException("Failed to rebuild Expression for filter: " + filter))
+ expressions.Not(rebuildExpressionFromFilter(not.child(), translatedFilterToExpr))
+ case _ =>
+ translatedFilterToExpr.getOrElse(predicate,
+ throw new IllegalStateException("Failed to rebuild Expression for filter: " + predicate))
}
}
}
+
+/**
+ * Get the expression of DS V2 to represent catalyst predicate that can be pushed down.
+ */
+case class PushablePredicate(nestedPredicatePushdownEnabled: Boolean) {
+
+ def unapply(e: Expression): Option[Predicate] =
+ new V2ExpressionBuilder(e, nestedPredicatePushdownEnabled, true).build().map { v =>
+ assert(v.isInstanceOf[Predicate])
+ v.asInstanceOf[Predicate]
+ }
+}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownUtils.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownUtils.scala
index 9953658..1149bff 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownUtils.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownUtils.scala
@@ -22,7 +22,7 @@ import scala.collection.mutable
import org.apache.spark.sql.catalyst.expressions.{AttributeReference, AttributeSet, Expression, NamedExpression, PredicateHelper, SchemaPruning}
import org.apache.spark.sql.catalyst.util.CharVarcharUtils
import org.apache.spark.sql.connector.expressions.SortOrder
-import org.apache.spark.sql.connector.expressions.filter.{Filter => V2Filter}
+import org.apache.spark.sql.connector.expressions.filter.Predicate
import org.apache.spark.sql.connector.read.{Scan, ScanBuilder, SupportsPushDownFilters, SupportsPushDownLimit, SupportsPushDownRequiredColumns, SupportsPushDownTableSample, SupportsPushDownTopN, SupportsPushDownV2Filters}
import org.apache.spark.sql.execution.datasources.DataSourceStrategy
import org.apache.spark.sql.internal.SQLConf
@@ -35,9 +35,8 @@ object PushDownUtils extends PredicateHelper {
*
* @return pushed filter and post-scan filters.
*/
- def pushFilters(
- scanBuilder: ScanBuilder,
- filters: Seq[Expression]): (Either[Seq[sources.Filter], Seq[V2Filter]], Seq[Expression]) = {
+ def pushFilters(scanBuilder: ScanBuilder, filters: Seq[Expression])
+ : (Either[Seq[sources.Filter], Seq[Predicate]], Seq[Expression]) = {
scanBuilder match {
case r: SupportsPushDownFilters =>
// A map from translated data source leaf node filters to original catalyst filter
@@ -73,8 +72,8 @@ object PushDownUtils extends PredicateHelper {
// expressions. For a `And`/`Or` predicate, it is possible that the predicate is partially
// pushed down. This map can be used to construct a catalyst filter expression from the
// input filter, or a superset(partial push down filter) of the input filter.
- val translatedFilterToExpr = mutable.HashMap.empty[V2Filter, Expression]
- val translatedFilters = mutable.ArrayBuffer.empty[V2Filter]
+ val translatedFilterToExpr = mutable.HashMap.empty[Predicate, Expression]
+ val translatedFilters = mutable.ArrayBuffer.empty[Predicate]
// Catalyst filter expression that can't be translated to data source filters.
val untranslatableExprs = mutable.ArrayBuffer.empty[Expression]
@@ -92,10 +91,10 @@ object PushDownUtils extends PredicateHelper {
// Data source filters that need to be evaluated again after scanning. which means
// the data source cannot guarantee the rows returned can pass these filters.
// As a result we must return it so Spark can plan an extra filter operator.
- val postScanFilters = r.pushFilters(translatedFilters.toArray).map { filter =>
- DataSourceV2Strategy.rebuildExpressionFromFilter(filter, translatedFilterToExpr)
+ val postScanFilters = r.pushPredicates(translatedFilters.toArray).map { predicate =>
+ DataSourceV2Strategy.rebuildExpressionFromFilter(predicate, translatedFilterToExpr)
}
- (Right(r.pushedFilters), (untranslatableExprs ++ postScanFilters).toSeq)
+ (Right(r.pushedPredicates), (untranslatableExprs ++ postScanFilters).toSeq)
case f: FileScanBuilder =>
val postScanFilters = f.pushFilters(filters)
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushedDownOperators.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushedDownOperators.scala
index 20ced9c..a95b459 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushedDownOperators.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushedDownOperators.scala
@@ -19,6 +19,7 @@ package org.apache.spark.sql.execution.datasources.v2
import org.apache.spark.sql.connector.expressions.SortOrder
import org.apache.spark.sql.connector.expressions.aggregate.Aggregation
+import org.apache.spark.sql.connector.expressions.filter.Predicate
/**
* Pushed down operators
@@ -27,6 +28,7 @@ case class PushedDownOperators(
aggregation: Option[Aggregation],
sample: Option[TableSampleInfo],
limit: Option[Int],
- sortValues: Seq[SortOrder]) {
+ sortValues: Seq[SortOrder],
+ pushedPredicates: Seq[Predicate]) {
assert((limit.isEmpty && sortValues.isEmpty) || limit.isDefined)
}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala
index b4bd027..44cdff1 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala
@@ -27,6 +27,7 @@ import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, Filter, LeafNode,
import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.connector.expressions.SortOrder
import org.apache.spark.sql.connector.expressions.aggregate.{Aggregation, Avg, Count, GeneralAggregateFunc, Sum}
+import org.apache.spark.sql.connector.expressions.filter.Predicate
import org.apache.spark.sql.connector.read.{Scan, ScanBuilder, SupportsPushDownAggregates, SupportsPushDownFilters, V1Scan}
import org.apache.spark.sql.execution.datasources.DataSourceStrategy
import org.apache.spark.sql.sources
@@ -72,6 +73,7 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper {
val pushedFiltersStr = if (pushedFilters.isLeft) {
pushedFilters.left.get.mkString(", ")
} else {
+ sHolder.pushedPredicates = pushedFilters.right.get
pushedFilters.right.get.mkString(", ")
}
@@ -405,8 +407,8 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper {
f.pushedFilters()
case _ => Array.empty[sources.Filter]
}
- val pushedDownOperators = PushedDownOperators(aggregation,
- sHolder.pushedSample, sHolder.pushedLimit, sHolder.sortOrders)
+ val pushedDownOperators = PushedDownOperators(aggregation, sHolder.pushedSample,
+ sHolder.pushedLimit, sHolder.sortOrders, sHolder.pushedPredicates)
V1ScanWrapper(v1, pushedFilters, pushedDownOperators)
case _ => scan
}
@@ -422,6 +424,8 @@ case class ScanBuilderHolder(
var sortOrders: Seq[SortOrder] = Seq.empty[SortOrder]
var pushedSample: Option[TableSampleInfo] = None
+
+ var pushedPredicates: Seq[Predicate] = Seq.empty[Predicate]
}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/jdbc/JDBCScan.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/jdbc/JDBCScan.scala
index 87ec9f4..f68f78d 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/jdbc/JDBCScan.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/jdbc/JDBCScan.scala
@@ -19,16 +19,17 @@ package org.apache.spark.sql.execution.datasources.v2.jdbc
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{Row, SQLContext}
import org.apache.spark.sql.connector.expressions.SortOrder
+import org.apache.spark.sql.connector.expressions.filter.Predicate
import org.apache.spark.sql.connector.read.V1Scan
import org.apache.spark.sql.execution.datasources.jdbc.JDBCRelation
import org.apache.spark.sql.execution.datasources.v2.TableSampleInfo
-import org.apache.spark.sql.sources.{BaseRelation, Filter, TableScan}
+import org.apache.spark.sql.sources.{BaseRelation, TableScan}
import org.apache.spark.sql.types.StructType
case class JDBCScan(
relation: JDBCRelation,
prunedSchema: StructType,
- pushedFilters: Array[Filter],
+ pushedPredicates: Array[Predicate],
pushedAggregateColumn: Array[String] = Array(),
groupByColumns: Option[Array[String]],
tableSample: Option[TableSampleInfo],
@@ -48,7 +49,7 @@ case class JDBCScan(
} else {
pushedAggregateColumn
}
- relation.buildScan(columnList, prunedSchema, pushedFilters, groupByColumns, tableSample,
+ relation.buildScan(columnList, prunedSchema, pushedPredicates, groupByColumns, tableSample,
pushedLimit, sortOrders)
}
}.asInstanceOf[T]
@@ -63,7 +64,7 @@ case class JDBCScan(
("[]", "[]")
}
super.description() + ", prunedSchema: " + seqToString(prunedSchema) +
- ", PushedFilters: " + seqToString(pushedFilters) +
+ ", PushedPredicates: " + seqToString(pushedPredicates) +
", PushedAggregates: " + aggString + ", PushedGroupBy: " + groupByString
}
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 61bf729..475f563 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,12 +22,12 @@ import org.apache.spark.internal.Logging
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.connector.expressions.SortOrder
import org.apache.spark.sql.connector.expressions.aggregate.Aggregation
-import org.apache.spark.sql.connector.read.{Scan, ScanBuilder, SupportsPushDownAggregates, SupportsPushDownFilters, SupportsPushDownLimit, SupportsPushDownRequiredColumns, SupportsPushDownTableSample, SupportsPushDownTopN}
+import org.apache.spark.sql.connector.expressions.filter.Predicate
+import org.apache.spark.sql.connector.read.{Scan, ScanBuilder, SupportsPushDownAggregates, SupportsPushDownLimit, SupportsPushDownRequiredColumns, SupportsPushDownTableSample, SupportsPushDownTopN, SupportsPushDownV2Filters}
import org.apache.spark.sql.execution.datasources.PartitioningUtils
import org.apache.spark.sql.execution.datasources.jdbc.{JDBCOptions, JDBCRDD, JDBCRelation}
import org.apache.spark.sql.execution.datasources.v2.TableSampleInfo
import org.apache.spark.sql.jdbc.JdbcDialects
-import org.apache.spark.sql.sources.Filter
import org.apache.spark.sql.types.StructType
case class JDBCScanBuilder(
@@ -35,7 +35,7 @@ case class JDBCScanBuilder(
schema: StructType,
jdbcOptions: JDBCOptions)
extends ScanBuilder
- with SupportsPushDownFilters
+ with SupportsPushDownV2Filters
with SupportsPushDownRequiredColumns
with SupportsPushDownAggregates
with SupportsPushDownLimit
@@ -45,7 +45,7 @@ case class JDBCScanBuilder(
private val isCaseSensitive = session.sessionState.conf.caseSensitiveAnalysis
- private var pushedFilter = Array.empty[Filter]
+ private var pushedPredicate = Array.empty[Predicate]
private var finalSchema = schema
@@ -55,18 +55,18 @@ case class JDBCScanBuilder(
private var sortOrders: Array[SortOrder] = Array.empty[SortOrder]
- override def pushFilters(filters: Array[Filter]): Array[Filter] = {
+ override def pushPredicates(predicates: Array[Predicate]): Array[Predicate] = {
if (jdbcOptions.pushDownPredicate) {
val dialect = JdbcDialects.get(jdbcOptions.url)
- val (pushed, unSupported) = filters.partition(JDBCRDD.compileFilter(_, dialect).isDefined)
- this.pushedFilter = pushed
+ val (pushed, unSupported) = predicates.partition(dialect.compileExpression(_).isDefined)
+ this.pushedPredicate = pushed
unSupported
} else {
- filters
+ predicates
}
}
- override def pushedFilters(): Array[Filter] = pushedFilter
+ override def pushedPredicates(): Array[Predicate] = pushedPredicate
private var pushedAggregateList: Array[String] = Array()
@@ -170,7 +170,7 @@ case class JDBCScanBuilder(
// "DEPT","NAME",MAX("SALARY"),MIN("BONUS"), instead of getting column names from
// prunedSchema and quote them (will become "MAX(SALARY)", "MIN(BONUS)" and can't
// be used in sql string.
- JDBCScan(JDBCRelation(schema, parts, jdbcOptions)(session), finalSchema, pushedFilter,
+ JDBCScan(JDBCRelation(schema, parts, jdbcOptions)(session), finalSchema, pushedPredicate,
pushedAggregateList, pushedGroupByCols, tableSample, pushedLimit, sortOrders)
}
}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/jdbc/DB2Dialect.scala b/sql/core/src/main/scala/org/apache/spark/sql/jdbc/DB2Dialect.scala
index 6af5cc0..35293b3 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/jdbc/DB2Dialect.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/jdbc/DB2Dialect.scala
@@ -35,27 +35,27 @@ private object DB2Dialect extends JdbcDialect {
super.compileAggregate(aggFunction).orElse(
aggFunction match {
case f: GeneralAggregateFunc if f.name() == "VAR_POP" =>
- assert(f.inputs().length == 1)
+ assert(f.children().length == 1)
val distinct = if (f.isDistinct) "DISTINCT " else ""
- Some(s"VARIANCE($distinct${f.inputs().head})")
+ Some(s"VARIANCE($distinct${f.children().head})")
case f: GeneralAggregateFunc if f.name() == "VAR_SAMP" =>
- assert(f.inputs().length == 1)
+ assert(f.children().length == 1)
val distinct = if (f.isDistinct) "DISTINCT " else ""
- Some(s"VARIANCE_SAMP($distinct${f.inputs().head})")
+ Some(s"VARIANCE_SAMP($distinct${f.children().head})")
case f: GeneralAggregateFunc if f.name() == "STDDEV_POP" =>
- assert(f.inputs().length == 1)
+ assert(f.children().length == 1)
val distinct = if (f.isDistinct) "DISTINCT " else ""
- Some(s"STDDEV($distinct${f.inputs().head})")
+ Some(s"STDDEV($distinct${f.children().head})")
case f: GeneralAggregateFunc if f.name() == "STDDEV_SAMP" =>
- assert(f.inputs().length == 1)
+ assert(f.children().length == 1)
val distinct = if (f.isDistinct) "DISTINCT " else ""
- Some(s"STDDEV_SAMP($distinct${f.inputs().head})")
+ Some(s"STDDEV_SAMP($distinct${f.children().head})")
case f: GeneralAggregateFunc if f.name() == "COVAR_POP" && f.isDistinct == false =>
- assert(f.inputs().length == 2)
- Some(s"COVARIANCE(${f.inputs().head}, ${f.inputs().last})")
+ assert(f.children().length == 2)
+ Some(s"COVARIANCE(${f.children().head}, ${f.children().last})")
case f: GeneralAggregateFunc if f.name() == "COVAR_SAMP" && f.isDistinct == false =>
- assert(f.inputs().length == 2)
- Some(s"COVARIANCE_SAMP(${f.inputs().head}, ${f.inputs().last})")
+ assert(f.children().length == 2)
+ Some(s"COVARIANCE_SAMP(${f.children().head}, ${f.children().last})")
case _ => None
}
)
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/jdbc/DerbyDialect.scala b/sql/core/src/main/scala/org/apache/spark/sql/jdbc/DerbyDialect.scala
index bf838b8..36c3c6b 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/jdbc/DerbyDialect.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/jdbc/DerbyDialect.scala
@@ -35,17 +35,17 @@ private object DerbyDialect extends JdbcDialect {
super.compileAggregate(aggFunction).orElse(
aggFunction match {
case f: GeneralAggregateFunc if f.name() == "VAR_POP" && f.isDistinct == false =>
- assert(f.inputs().length == 1)
- Some(s"VAR_POP(${f.inputs().head})")
+ assert(f.children().length == 1)
+ Some(s"VAR_POP(${f.children().head})")
case f: GeneralAggregateFunc if f.name() == "VAR_SAMP" && f.isDistinct == false =>
- assert(f.inputs().length == 1)
- Some(s"VAR_SAMP(${f.inputs().head})")
+ assert(f.children().length == 1)
+ Some(s"VAR_SAMP(${f.children().head})")
case f: GeneralAggregateFunc if f.name() == "STDDEV_POP" && f.isDistinct == false =>
- assert(f.inputs().length == 1)
- Some(s"STDDEV_POP(${f.inputs().head})")
+ assert(f.children().length == 1)
+ Some(s"STDDEV_POP(${f.children().head})")
case f: GeneralAggregateFunc if f.name() == "STDDEV_SAMP" && f.isDistinct == false =>
- assert(f.inputs().length == 1)
- Some(s"STDDEV_SAMP(${f.inputs().head})")
+ assert(f.children().length == 1)
+ Some(s"STDDEV_SAMP(${f.children().head})")
case _ => None
}
)
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 7bd51f8..643376c 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
@@ -32,33 +32,33 @@ private object H2Dialect extends JdbcDialect {
super.compileAggregate(aggFunction).orElse(
aggFunction match {
case f: GeneralAggregateFunc if f.name() == "VAR_POP" =>
- assert(f.inputs().length == 1)
+ assert(f.children().length == 1)
val distinct = if (f.isDistinct) "DISTINCT " else ""
- Some(s"VAR_POP($distinct${f.inputs().head})")
+ Some(s"VAR_POP($distinct${f.children().head})")
case f: GeneralAggregateFunc if f.name() == "VAR_SAMP" =>
- assert(f.inputs().length == 1)
+ assert(f.children().length == 1)
val distinct = if (f.isDistinct) "DISTINCT " else ""
- Some(s"VAR_SAMP($distinct${f.inputs().head})")
+ Some(s"VAR_SAMP($distinct${f.children().head})")
case f: GeneralAggregateFunc if f.name() == "STDDEV_POP" =>
- assert(f.inputs().length == 1)
+ assert(f.children().length == 1)
val distinct = if (f.isDistinct) "DISTINCT " else ""
- Some(s"STDDEV_POP($distinct${f.inputs().head})")
+ Some(s"STDDEV_POP($distinct${f.children().head})")
case f: GeneralAggregateFunc if f.name() == "STDDEV_SAMP" =>
- assert(f.inputs().length == 1)
+ assert(f.children().length == 1)
val distinct = if (f.isDistinct) "DISTINCT " else ""
- Some(s"STDDEV_SAMP($distinct${f.inputs().head})")
+ Some(s"STDDEV_SAMP($distinct${f.children().head})")
case f: GeneralAggregateFunc if f.name() == "COVAR_POP" =>
- assert(f.inputs().length == 2)
+ assert(f.children().length == 2)
val distinct = if (f.isDistinct) "DISTINCT " else ""
- Some(s"COVAR_POP($distinct${f.inputs().head}, ${f.inputs().last})")
+ Some(s"COVAR_POP($distinct${f.children().head}, ${f.children().last})")
case f: GeneralAggregateFunc if f.name() == "COVAR_SAMP" =>
- assert(f.inputs().length == 2)
+ assert(f.children().length == 2)
val distinct = if (f.isDistinct) "DISTINCT " else ""
- Some(s"COVAR_SAMP($distinct${f.inputs().head}, ${f.inputs().last})")
+ Some(s"COVAR_SAMP($distinct${f.children().head}, ${f.children().last})")
case f: GeneralAggregateFunc if f.name() == "CORR" =>
- assert(f.inputs().length == 2)
+ assert(f.children().length == 2)
val distinct = if (f.isDistinct) "DISTINCT " else ""
- Some(s"CORR($distinct${f.inputs().head}, ${f.inputs().last})")
+ Some(s"CORR($distinct${f.children().head}, ${f.children().last})")
case _ => None
}
)
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 c9dcbb2..4b28de2 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
@@ -22,17 +22,19 @@ import java.time.{Instant, LocalDate}
import java.util
import scala.collection.mutable.ArrayBuilder
+import scala.util.control.NonFatal
import org.apache.commons.lang3.StringUtils
import org.apache.spark.annotation.{DeveloperApi, Since}
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.util.{DateFormatter, DateTimeUtils, TimestampFormatter}
import org.apache.spark.sql.connector.catalog.TableChange
import org.apache.spark.sql.connector.catalog.TableChange._
import org.apache.spark.sql.connector.catalog.index.TableIndex
-import org.apache.spark.sql.connector.expressions.{Expression, FieldReference, NamedReference}
+import org.apache.spark.sql.connector.expressions.{Expression, Literal, NamedReference}
import org.apache.spark.sql.connector.expressions.aggregate.{AggregateFunc, Avg, Count, CountStar, Max, Min, Sum}
import org.apache.spark.sql.connector.util.V2ExpressionSQLBuilder
import org.apache.spark.sql.errors.QueryCompilationErrors
@@ -220,12 +222,18 @@ abstract class JdbcDialect extends Serializable with Logging{
}
class JDBCSQLBuilder extends V2ExpressionSQLBuilder {
- override def visitFieldReference(fieldRef: FieldReference): String = {
- if (fieldRef.fieldNames().length != 1) {
+ override def visitLiteral(literal: Literal[_]): String = {
+ compileValue(
+ CatalystTypeConverters.convertToScala(literal.value(), literal.dataType())).toString
+ }
+
+ override def visitNamedReference(namedRef: NamedReference): String = {
+ if (namedRef.fieldNames().length > 1) {
throw new IllegalArgumentException(
- "FieldReference with field name has multiple or zero parts unsupported: " + fieldRef);
+ QueryCompilationErrors.commandNotSupportNestedColumnError(
+ "Filter push down", namedRef.toString).getMessage);
}
- quoteIdentifier(fieldRef.fieldNames.head)
+ quoteIdentifier(namedRef.fieldNames.head)
}
}
@@ -240,7 +248,9 @@ abstract class JdbcDialect extends Serializable with Logging{
try {
Some(jdbcSQLBuilder.build(expr))
} catch {
- case _: IllegalArgumentException => None
+ case NonFatal(e) =>
+ logWarning("Error occurs while compiling V2 expression", e)
+ None
}
}
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 841f1c8..8d2fbec 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
@@ -47,21 +47,21 @@ private object MsSqlServerDialect extends JdbcDialect {
super.compileAggregate(aggFunction).orElse(
aggFunction match {
case f: GeneralAggregateFunc if f.name() == "VAR_POP" =>
- assert(f.inputs().length == 1)
+ assert(f.children().length == 1)
val distinct = if (f.isDistinct) "DISTINCT " else ""
- Some(s"VARP($distinct${f.inputs().head})")
+ Some(s"VARP($distinct${f.children().head})")
case f: GeneralAggregateFunc if f.name() == "VAR_SAMP" =>
- assert(f.inputs().length == 1)
+ assert(f.children().length == 1)
val distinct = if (f.isDistinct) "DISTINCT " else ""
- Some(s"VAR($distinct${f.inputs().head})")
+ Some(s"VAR($distinct${f.children().head})")
case f: GeneralAggregateFunc if f.name() == "STDDEV_POP" =>
- assert(f.inputs().length == 1)
+ assert(f.children().length == 1)
val distinct = if (f.isDistinct) "DISTINCT " else ""
- Some(s"STDEVP($distinct${f.inputs().head})")
+ Some(s"STDEVP($distinct${f.children().head})")
case f: GeneralAggregateFunc if f.name() == "STDDEV_SAMP" =>
- assert(f.inputs().length == 1)
+ assert(f.children().length == 1)
val distinct = if (f.isDistinct) "DISTINCT " else ""
- Some(s"STDEV($distinct${f.inputs().head})")
+ Some(s"STDEV($distinct${f.children().head})")
case _ => None
}
)
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/jdbc/MySQLDialect.scala b/sql/core/src/main/scala/org/apache/spark/sql/jdbc/MySQLDialect.scala
index d73721d..24f9bac 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/jdbc/MySQLDialect.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/jdbc/MySQLDialect.scala
@@ -43,17 +43,17 @@ private case object MySQLDialect extends JdbcDialect with SQLConfHelper {
super.compileAggregate(aggFunction).orElse(
aggFunction match {
case f: GeneralAggregateFunc if f.name() == "VAR_POP" && f.isDistinct == false =>
- assert(f.inputs().length == 1)
- Some(s"VAR_POP(${f.inputs().head})")
+ assert(f.children().length == 1)
+ Some(s"VAR_POP(${f.children().head})")
case f: GeneralAggregateFunc if f.name() == "VAR_SAMP" && f.isDistinct == false =>
- assert(f.inputs().length == 1)
- Some(s"VAR_SAMP(${f.inputs().head})")
+ assert(f.children().length == 1)
+ Some(s"VAR_SAMP(${f.children().head})")
case f: GeneralAggregateFunc if f.name() == "STDDEV_POP" && f.isDistinct == false =>
- assert(f.inputs().length == 1)
- Some(s"STDDEV_POP(${f.inputs().head})")
+ assert(f.children().length == 1)
+ Some(s"STDDEV_POP(${f.children().head})")
case f: GeneralAggregateFunc if f.name() == "STDDEV_SAMP" && f.isDistinct == false =>
- assert(f.inputs().length == 1)
- Some(s"STDDEV_SAMP(${f.inputs().head})")
+ assert(f.children().length == 1)
+ Some(s"STDDEV_SAMP(${f.children().head})")
case _ => None
}
)
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 71db7e9..40333c1 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
@@ -41,26 +41,26 @@ private case object OracleDialect extends JdbcDialect {
super.compileAggregate(aggFunction).orElse(
aggFunction match {
case f: GeneralAggregateFunc if f.name() == "VAR_POP" && f.isDistinct == false =>
- assert(f.inputs().length == 1)
- Some(s"VAR_POP(${f.inputs().head})")
+ assert(f.children().length == 1)
+ Some(s"VAR_POP(${f.children().head})")
case f: GeneralAggregateFunc if f.name() == "VAR_SAMP" && f.isDistinct == false =>
- assert(f.inputs().length == 1)
- Some(s"VAR_SAMP(${f.inputs().head})")
+ assert(f.children().length == 1)
+ Some(s"VAR_SAMP(${f.children().head})")
case f: GeneralAggregateFunc if f.name() == "STDDEV_POP" && f.isDistinct == false =>
- assert(f.inputs().length == 1)
- Some(s"STDDEV_POP(${f.inputs().head})")
+ assert(f.children().length == 1)
+ Some(s"STDDEV_POP(${f.children().head})")
case f: GeneralAggregateFunc if f.name() == "STDDEV_SAMP" && f.isDistinct == false =>
- assert(f.inputs().length == 1)
- Some(s"STDDEV_SAMP(${f.inputs().head})")
+ assert(f.children().length == 1)
+ Some(s"STDDEV_SAMP(${f.children().head})")
case f: GeneralAggregateFunc if f.name() == "COVAR_POP" && f.isDistinct == false =>
- assert(f.inputs().length == 2)
- Some(s"COVAR_POP(${f.inputs().head}, ${f.inputs().last})")
+ assert(f.children().length == 2)
+ Some(s"COVAR_POP(${f.children().head}, ${f.children().last})")
case f: GeneralAggregateFunc if f.name() == "COVAR_SAMP" && f.isDistinct == false =>
- assert(f.inputs().length == 2)
- Some(s"COVAR_SAMP(${f.inputs().head}, ${f.inputs().last})")
+ assert(f.children().length == 2)
+ Some(s"COVAR_SAMP(${f.children().head}, ${f.children().last})")
case f: GeneralAggregateFunc if f.name() == "CORR" && f.isDistinct == false =>
- assert(f.inputs().length == 2)
- Some(s"CORR(${f.inputs().head}, ${f.inputs().last})")
+ assert(f.children().length == 2)
+ Some(s"CORR(${f.children().head}, ${f.children().last})")
case _ => None
}
)
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 e2023d1..a668d66 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
@@ -41,33 +41,33 @@ private object PostgresDialect extends JdbcDialect with SQLConfHelper {
super.compileAggregate(aggFunction).orElse(
aggFunction match {
case f: GeneralAggregateFunc if f.name() == "VAR_POP" =>
- assert(f.inputs().length == 1)
+ assert(f.children().length == 1)
val distinct = if (f.isDistinct) "DISTINCT " else ""
- Some(s"VAR_POP($distinct${f.inputs().head})")
+ Some(s"VAR_POP($distinct${f.children().head})")
case f: GeneralAggregateFunc if f.name() == "VAR_SAMP" =>
- assert(f.inputs().length == 1)
+ assert(f.children().length == 1)
val distinct = if (f.isDistinct) "DISTINCT " else ""
- Some(s"VAR_SAMP($distinct${f.inputs().head})")
+ Some(s"VAR_SAMP($distinct${f.children().head})")
case f: GeneralAggregateFunc if f.name() == "STDDEV_POP" =>
- assert(f.inputs().length == 1)
+ assert(f.children().length == 1)
val distinct = if (f.isDistinct) "DISTINCT " else ""
- Some(s"STDDEV_POP($distinct${f.inputs().head})")
+ Some(s"STDDEV_POP($distinct${f.children().head})")
case f: GeneralAggregateFunc if f.name() == "STDDEV_SAMP" =>
- assert(f.inputs().length == 1)
+ assert(f.children().length == 1)
val distinct = if (f.isDistinct) "DISTINCT " else ""
- Some(s"STDDEV_SAMP($distinct${f.inputs().head})")
+ Some(s"STDDEV_SAMP($distinct${f.children().head})")
case f: GeneralAggregateFunc if f.name() == "COVAR_POP" =>
- assert(f.inputs().length == 2)
+ assert(f.children().length == 2)
val distinct = if (f.isDistinct) "DISTINCT " else ""
- Some(s"COVAR_POP($distinct${f.inputs().head}, ${f.inputs().last})")
+ Some(s"COVAR_POP($distinct${f.children().head}, ${f.children().last})")
case f: GeneralAggregateFunc if f.name() == "COVAR_SAMP" =>
- assert(f.inputs().length == 2)
+ assert(f.children().length == 2)
val distinct = if (f.isDistinct) "DISTINCT " else ""
- Some(s"COVAR_SAMP($distinct${f.inputs().head}, ${f.inputs().last})")
+ Some(s"COVAR_SAMP($distinct${f.children().head}, ${f.children().last})")
case f: GeneralAggregateFunc if f.name() == "CORR" =>
- assert(f.inputs().length == 2)
+ assert(f.children().length == 2)
val distinct = if (f.isDistinct) "DISTINCT " else ""
- Some(s"CORR($distinct${f.inputs().head}, ${f.inputs().last})")
+ Some(s"CORR($distinct${f.children().head}, ${f.children().last})")
case _ => None
}
)
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/jdbc/TeradataDialect.scala b/sql/core/src/main/scala/org/apache/spark/sql/jdbc/TeradataDialect.scala
index 13e16d2..79fb710 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/jdbc/TeradataDialect.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/jdbc/TeradataDialect.scala
@@ -35,30 +35,30 @@ private case object TeradataDialect extends JdbcDialect {
super.compileAggregate(aggFunction).orElse(
aggFunction match {
case f: GeneralAggregateFunc if f.name() == "VAR_POP" =>
- assert(f.inputs().length == 1)
+ assert(f.children().length == 1)
val distinct = if (f.isDistinct) "DISTINCT " else ""
- Some(s"VAR_POP($distinct${f.inputs().head})")
+ Some(s"VAR_POP($distinct${f.children().head})")
case f: GeneralAggregateFunc if f.name() == "VAR_SAMP" =>
- assert(f.inputs().length == 1)
+ assert(f.children().length == 1)
val distinct = if (f.isDistinct) "DISTINCT " else ""
- Some(s"VAR_SAMP($distinct${f.inputs().head})")
+ Some(s"VAR_SAMP($distinct${f.children().head})")
case f: GeneralAggregateFunc if f.name() == "STDDEV_POP" =>
- assert(f.inputs().length == 1)
+ assert(f.children().length == 1)
val distinct = if (f.isDistinct) "DISTINCT " else ""
- Some(s"STDDEV_POP($distinct${f.inputs().head})")
+ Some(s"STDDEV_POP($distinct${f.children().head})")
case f: GeneralAggregateFunc if f.name() == "STDDEV_SAMP" =>
- assert(f.inputs().length == 1)
+ assert(f.children().length == 1)
val distinct = if (f.isDistinct) "DISTINCT " else ""
- Some(s"STDDEV_SAMP($distinct${f.inputs().head})")
+ Some(s"STDDEV_SAMP($distinct${f.children().head})")
case f: GeneralAggregateFunc if f.name() == "COVAR_POP" && f.isDistinct == false =>
- assert(f.inputs().length == 2)
- Some(s"COVAR_POP(${f.inputs().head}, ${f.inputs().last})")
+ assert(f.children().length == 2)
+ Some(s"COVAR_POP(${f.children().head}, ${f.children().last})")
case f: GeneralAggregateFunc if f.name() == "COVAR_SAMP" && f.isDistinct == false =>
- assert(f.inputs().length == 2)
- Some(s"COVAR_SAMP(${f.inputs().head}, ${f.inputs().last})")
+ assert(f.children().length == 2)
+ Some(s"COVAR_SAMP(${f.children().head}, ${f.children().last})")
case f: GeneralAggregateFunc if f.name() == "CORR" && f.isDistinct == false =>
- assert(f.inputs().length == 2)
- Some(s"CORR(${f.inputs().head}, ${f.inputs().last})")
+ assert(f.children().length == 2)
+ Some(s"CORR(${f.children().head}, ${f.children().last})")
case _ => None
}
)
diff --git a/sql/core/src/test/java/test/org/apache/spark/sql/connector/JavaAdvancedDataSourceV2WithV2Filter.java b/sql/core/src/test/java/test/org/apache/spark/sql/connector/JavaAdvancedDataSourceV2WithV2Filter.java
index b92206c..ec532da 100644
--- a/sql/core/src/test/java/test/org/apache/spark/sql/connector/JavaAdvancedDataSourceV2WithV2Filter.java
+++ b/sql/core/src/test/java/test/org/apache/spark/sql/connector/JavaAdvancedDataSourceV2WithV2Filter.java
@@ -17,21 +17,23 @@
package test.org.apache.spark.sql.connector;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
import org.apache.spark.sql.catalyst.InternalRow;
import org.apache.spark.sql.catalyst.expressions.GenericInternalRow;
import org.apache.spark.sql.connector.TestingV2Source;
import org.apache.spark.sql.connector.catalog.Table;
-import org.apache.spark.sql.connector.expressions.filter.Filter;
+import org.apache.spark.sql.connector.expressions.FieldReference;
+import org.apache.spark.sql.connector.expressions.Literal;
+import org.apache.spark.sql.connector.expressions.LiteralValue;
+import org.apache.spark.sql.connector.expressions.filter.Predicate;
import org.apache.spark.sql.connector.read.*;
-import org.apache.spark.sql.connector.expressions.filter.GreaterThan;
import org.apache.spark.sql.types.StructType;
import org.apache.spark.sql.util.CaseInsensitiveStringMap;
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.List;
-
public class JavaAdvancedDataSourceV2WithV2Filter implements TestingV2Source {
@Override
@@ -48,7 +50,7 @@ public class JavaAdvancedDataSourceV2WithV2Filter implements TestingV2Source {
SupportsPushDownV2Filters, SupportsPushDownRequiredColumns {
private StructType requiredSchema = TestingV2Source.schema();
- private Filter[] filters = new Filter[0];
+ private Predicate[] predicates = new Predicate[0];
@Override
public void pruneColumns(StructType requiredSchema) {
@@ -61,32 +63,38 @@ public class JavaAdvancedDataSourceV2WithV2Filter implements TestingV2Source {
}
@Override
- public Filter[] pushFilters(Filter[] filters) {
- Filter[] supported = Arrays.stream(filters).filter(f -> {
- if (f instanceof GreaterThan) {
- GreaterThan gt = (GreaterThan) f;
- return gt.column().describe().equals("i") && gt.value().value() instanceof Integer;
+ public Predicate[] pushPredicates(Predicate[] predicates) {
+ Predicate[] supported = Arrays.stream(predicates).filter(f -> {
+ if (f.name().equals(">")) {
+ assert(f.children()[0] instanceof FieldReference);
+ FieldReference column = (FieldReference) f.children()[0];
+ assert(f.children()[1] instanceof LiteralValue);
+ Literal value = (Literal) f.children()[1];
+ return column.describe().equals("i") && value.value() instanceof Integer;
} else {
return false;
}
- }).toArray(Filter[]::new);
-
- Filter[] unsupported = Arrays.stream(filters).filter(f -> {
- if (f instanceof GreaterThan) {
- GreaterThan gt = (GreaterThan) f;
- return !gt.column().describe().equals("i") || !(gt.value().value() instanceof Integer);
+ }).toArray(Predicate[]::new);
+
+ Predicate[] unsupported = Arrays.stream(predicates).filter(f -> {
+ if (f.name().equals(">")) {
+ assert(f.children()[0] instanceof FieldReference);
+ FieldReference column = (FieldReference) f.children()[0];
+ assert(f.children()[1] instanceof LiteralValue);
+ Literal value = (LiteralValue) f.children()[1];
+ return !column.describe().equals("i") || !(value.value() instanceof Integer);
} else {
return true;
}
- }).toArray(Filter[]::new);
+ }).toArray(Predicate[]::new);
- this.filters = supported;
+ this.predicates = supported;
return unsupported;
}
@Override
- public Filter[] pushedFilters() {
- return filters;
+ public Predicate[] pushedPredicates() {
+ return predicates;
}
@Override
@@ -96,18 +104,18 @@ public class JavaAdvancedDataSourceV2WithV2Filter implements TestingV2Source {
@Override
public Batch toBatch() {
- return new AdvancedBatchWithV2Filter(requiredSchema, filters);
+ return new AdvancedBatchWithV2Filter(requiredSchema, predicates);
}
}
public static class AdvancedBatchWithV2Filter implements Batch {
// Exposed for testing.
public StructType requiredSchema;
- public Filter[] filters;
+ public Predicate[] predicates;
- AdvancedBatchWithV2Filter(StructType requiredSchema, Filter[] filters) {
+ AdvancedBatchWithV2Filter(StructType requiredSchema, Predicate[] predicates) {
this.requiredSchema = requiredSchema;
- this.filters = filters;
+ this.predicates = predicates;
}
@Override
@@ -115,11 +123,14 @@ public class JavaAdvancedDataSourceV2WithV2Filter implements TestingV2Source {
List<InputPartition> res = new ArrayList<>();
Integer lowerBound = null;
- for (Filter filter : filters) {
- if (filter instanceof GreaterThan) {
- GreaterThan f = (GreaterThan) filter;
- if ("i".equals(f.column().describe()) && f.value().value() instanceof Integer) {
- lowerBound = (Integer) f.value().value();
+ for (Predicate predicate : predicates) {
+ if (predicate.name().equals(">")) {
+ assert(predicate.children()[0] instanceof FieldReference);
+ FieldReference column = (FieldReference) predicate.children()[0];
+ assert(predicate.children()[1] instanceof LiteralValue);
+ Literal value = (Literal) predicate.children()[1];
+ if ("i".equals(column.describe()) && value.value() instanceof Integer) {
+ lowerBound = (Integer) value.value();
break;
}
}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2Suite.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2Suite.scala
index 8f37e42..b7533c6 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2Suite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2Suite.scala
@@ -27,8 +27,8 @@ import org.apache.spark.sql.{AnalysisException, DataFrame, QueryTest, Row}
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.connector.catalog.{SupportsRead, Table, TableCapability, TableProvider}
import org.apache.spark.sql.connector.catalog.TableCapability._
-import org.apache.spark.sql.connector.expressions.Transform
-import org.apache.spark.sql.connector.expressions.filter.{Filter => V2Filter, GreaterThan => V2GreaterThan}
+import org.apache.spark.sql.connector.expressions.{Literal, Transform}
+import org.apache.spark.sql.connector.expressions.filter.Predicate
import org.apache.spark.sql.connector.read._
import org.apache.spark.sql.connector.read.partitioning.{ClusteredDistribution, Distribution, Partitioning}
import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
@@ -155,11 +155,11 @@ class DataSourceV2Suite extends QueryTest with SharedSparkSession with AdaptiveS
checkAnswer(q1, (0 until 10).map(i => Row(-i)))
if (cls == classOf[AdvancedDataSourceV2WithV2Filter]) {
val batch = getBatchWithV2Filter(q1)
- assert(batch.filters.isEmpty)
+ assert(batch.predicates.isEmpty)
assert(batch.requiredSchema.fieldNames === Seq("j"))
} else {
val batch = getJavaBatchWithV2Filter(q1)
- assert(batch.filters.isEmpty)
+ assert(batch.predicates.isEmpty)
assert(batch.requiredSchema.fieldNames === Seq("j"))
}
@@ -167,11 +167,11 @@ class DataSourceV2Suite extends QueryTest with SharedSparkSession with AdaptiveS
checkAnswer(q2, (4 until 10).map(i => Row(i, -i)))
if (cls == classOf[AdvancedDataSourceV2WithV2Filter]) {
val batch = getBatchWithV2Filter(q2)
- assert(batch.filters.flatMap(_.references.map(_.describe)).toSet == Set("i"))
+ assert(batch.predicates.flatMap(_.references.map(_.describe)).toSet == Set("i"))
assert(batch.requiredSchema.fieldNames === Seq("i", "j"))
} else {
val batch = getJavaBatchWithV2Filter(q2)
- assert(batch.filters.flatMap(_.references.map(_.describe)).toSet == Set("i"))
+ assert(batch.predicates.flatMap(_.references.map(_.describe)).toSet == Set("i"))
assert(batch.requiredSchema.fieldNames === Seq("i", "j"))
}
@@ -179,11 +179,11 @@ class DataSourceV2Suite extends QueryTest with SharedSparkSession with AdaptiveS
checkAnswer(q3, (7 until 10).map(i => Row(i)))
if (cls == classOf[AdvancedDataSourceV2WithV2Filter]) {
val batch = getBatchWithV2Filter(q3)
- assert(batch.filters.flatMap(_.references.map(_.describe)).toSet == Set("i"))
+ assert(batch.predicates.flatMap(_.references.map(_.describe)).toSet == Set("i"))
assert(batch.requiredSchema.fieldNames === Seq("i"))
} else {
val batch = getJavaBatchWithV2Filter(q3)
- assert(batch.filters.flatMap(_.references.map(_.describe)).toSet == Set("i"))
+ assert(batch.predicates.flatMap(_.references.map(_.describe)).toSet == Set("i"))
assert(batch.requiredSchema.fieldNames === Seq("i"))
}
@@ -192,12 +192,12 @@ class DataSourceV2Suite extends QueryTest with SharedSparkSession with AdaptiveS
if (cls == classOf[AdvancedDataSourceV2WithV2Filter]) {
val batch = getBatchWithV2Filter(q4)
// Symbol("j") < 10 is not supported by the testing data source.
- assert(batch.filters.isEmpty)
+ assert(batch.predicates.isEmpty)
assert(batch.requiredSchema.fieldNames === Seq("j"))
} else {
val batch = getJavaBatchWithV2Filter(q4)
// Symbol("j") < 10 is not supported by the testing data source.
- assert(batch.filters.isEmpty)
+ assert(batch.predicates.isEmpty)
assert(batch.requiredSchema.fieldNames === Seq("j"))
}
}
@@ -704,7 +704,7 @@ class AdvancedScanBuilderWithV2Filter extends ScanBuilder
with Scan with SupportsPushDownV2Filters with SupportsPushDownRequiredColumns {
var requiredSchema = TestingV2Source.schema
- var filters = Array.empty[V2Filter]
+ var predicates = Array.empty[Predicate]
override def pruneColumns(requiredSchema: StructType): Unit = {
this.requiredSchema = requiredSchema
@@ -712,29 +712,32 @@ class AdvancedScanBuilderWithV2Filter extends ScanBuilder
override def readSchema(): StructType = requiredSchema
- override def pushFilters(filters: Array[V2Filter]): Array[V2Filter] = {
- val (supported, unsupported) = filters.partition {
- case _: V2GreaterThan => true
+ override def pushPredicates(predicates: Array[Predicate]): Array[Predicate] = {
+ val (supported, unsupported) = predicates.partition {
+ case p: Predicate if p.name() == ">" => true
case _ => false
}
- this.filters = supported
+ this.predicates = supported
unsupported
}
- override def pushedFilters(): Array[V2Filter] = filters
+ override def pushedPredicates(): Array[Predicate] = predicates
override def build(): Scan = this
- override def toBatch: Batch = new AdvancedBatchWithV2Filter(filters, requiredSchema)
+ override def toBatch: Batch = new AdvancedBatchWithV2Filter(predicates, requiredSchema)
}
class AdvancedBatchWithV2Filter(
- val filters: Array[V2Filter],
+ val predicates: Array[Predicate],
val requiredSchema: StructType) extends Batch {
override def planInputPartitions(): Array[InputPartition] = {
- val lowerBound = filters.collectFirst {
- case gt: V2GreaterThan => gt.value
+ val lowerBound = predicates.collectFirst {
+ case p: Predicate if p.name().equals(">") =>
+ val value = p.children()(1)
+ assert(value.isInstanceOf[Literal[_]])
+ value.asInstanceOf[Literal[_]]
}
val res = scala.collection.mutable.ArrayBuffer.empty[InputPartition]
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 0fb6fc5..6296da4 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
@@ -21,21 +21,21 @@ import org.apache.spark.sql.catalyst.dsl.expressions._
import org.apache.spark.sql.catalyst.expressions.Expression
import org.apache.spark.sql.catalyst.plans.PlanTest
import org.apache.spark.sql.connector.expressions.{FieldReference, LiteralValue}
-import org.apache.spark.sql.connector.expressions.filter.{EqualTo => V2EqualTo, Filter => V2Filter}
+import org.apache.spark.sql.connector.expressions.filter.Predicate
import org.apache.spark.sql.test.SharedSparkSession
import org.apache.spark.sql.types.BooleanType
class DataSourceV2StrategySuite extends PlanTest with SharedSparkSession {
test("SPARK-36644: Push down boolean column filter") {
testTranslateFilter(Symbol("col").boolean,
- Some(new V2EqualTo(FieldReference("col"), LiteralValue(true, BooleanType))))
+ Some(new Predicate("=", Array(FieldReference("col"), LiteralValue(true, BooleanType)))))
}
/**
- * Translate the given Catalyst [[Expression]] into data source [[V2Filter]]
- * then verify against the given [[V2Filter]].
+ * Translate the given Catalyst [[Expression]] into data source V2 [[Predicate]]
+ * then verify against the given [[Predicate]].
*/
- def testTranslateFilter(catalystFilter: Expression, result: Option[V2Filter]): Unit = {
+ def testTranslateFilter(catalystFilter: Expression, result: Option[Predicate]): Unit = {
assertResult(result) {
DataSourceV2Strategy.translateFilterV2(catalystFilter, true)
}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/V2FiltersSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/V2FiltersSuite.scala
deleted file mode 100644
index b457211..0000000
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/V2FiltersSuite.scala
+++ /dev/null
@@ -1,204 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.spark.sql.execution.datasources.v2
-
-import org.apache.spark.SparkFunSuite
-import org.apache.spark.sql.connector.expressions.{FieldReference, Literal, LiteralValue}
-import org.apache.spark.sql.connector.expressions.filter._
-import org.apache.spark.sql.execution.datasources.v2.FiltersV2Suite.ref
-import org.apache.spark.sql.types.IntegerType
-import org.apache.spark.unsafe.types.UTF8String
-
-class FiltersV2Suite extends SparkFunSuite {
-
- test("nested columns") {
- val filter1 = new EqualTo(ref("a", "B"), LiteralValue(1, IntegerType))
- assert(filter1.references.map(_.describe()).toSeq == Seq("a.B"))
- assert(filter1.describe.equals("a.B = 1"))
-
- val filter2 = new EqualTo(ref("a", "b.c"), LiteralValue(1, IntegerType))
- assert(filter2.references.map(_.describe()).toSeq == Seq("a.`b.c`"))
- assert(filter2.describe.equals("a.`b.c` = 1"))
-
- val filter3 = new EqualTo(ref("`a`.b", "c"), LiteralValue(1, IntegerType))
- assert(filter3.references.map(_.describe()).toSeq == Seq("```a``.b`.c"))
- assert(filter3.describe.equals("```a``.b`.c = 1"))
- }
-
- test("AlwaysTrue") {
- val filter1 = new AlwaysTrue
- val filter2 = new AlwaysTrue
- assert(filter1.equals(filter2))
- assert(filter1.references.map(_.describe()).length == 0)
- assert(filter1.describe.equals("TRUE"))
- }
-
- test("AlwaysFalse") {
- val filter1 = new AlwaysFalse
- val filter2 = new AlwaysFalse
- assert(filter1.equals(filter2))
- assert(filter1.references.map(_.describe()).length == 0)
- assert(filter1.describe.equals("FALSE"))
- }
-
- test("EqualTo") {
- val filter1 = new EqualTo(ref("a"), LiteralValue(1, IntegerType))
- val filter2 = new EqualTo(ref("a"), LiteralValue(1, IntegerType))
- assert(filter1.equals(filter2))
- assert(filter1.references.map(_.describe()).toSeq == Seq("a"))
- assert(filter1.describe.equals("a = 1"))
- }
-
- test("EqualNullSafe") {
- val filter1 = new EqualNullSafe(ref("a"), LiteralValue(1, IntegerType))
- val filter2 = new EqualNullSafe(ref("a"), LiteralValue(1, IntegerType))
- assert(filter1.equals(filter2))
- assert(filter1.references.map(_.describe()).toSeq == Seq("a"))
- assert(filter1.describe.equals("a <=> 1"))
- }
-
- test("GreaterThan") {
- val filter1 = new GreaterThan(ref("a"), LiteralValue(1, IntegerType))
- val filter2 = new GreaterThan(ref("a"), LiteralValue(1, IntegerType))
- assert(filter1.equals(filter2))
- assert(filter1.references.map(_.describe()).toSeq == Seq("a"))
- assert(filter1.describe.equals("a > 1"))
- }
-
- test("GreaterThanOrEqual") {
- val filter1 = new GreaterThanOrEqual(ref("a"), LiteralValue(1, IntegerType))
- val filter2 = new GreaterThanOrEqual(ref("a"), LiteralValue(1, IntegerType))
- assert(filter1.equals(filter2))
- assert(filter1.references.map(_.describe()).toSeq == Seq("a"))
- assert(filter1.describe.equals("a >= 1"))
- }
-
- test("LessThan") {
- val filter1 = new LessThan(ref("a"), LiteralValue(1, IntegerType))
- val filter2 = new LessThan(ref("a"), LiteralValue(1, IntegerType))
- assert(filter1.equals(filter2))
- assert(filter1.references.map(_.describe()).toSeq == Seq("a"))
- assert(filter1.describe.equals("a < 1"))
- }
-
- test("LessThanOrEqual") {
- val filter1 = new LessThanOrEqual(ref("a"), LiteralValue(1, IntegerType))
- val filter2 = new LessThanOrEqual(ref("a"), LiteralValue(1, IntegerType))
- assert(filter1.equals(filter2))
- assert(filter1.references.map(_.describe()).toSeq == Seq("a"))
- assert(filter1.describe.equals("a <= 1"))
- }
-
- test("In") {
- val filter1 = new In(ref("a"),
- Array(LiteralValue(1, IntegerType), LiteralValue(2, IntegerType),
- LiteralValue(3, IntegerType), LiteralValue(4, IntegerType)))
- val filter2 = new In(ref("a"),
- Array(LiteralValue(4, IntegerType), LiteralValue(2, IntegerType),
- LiteralValue(3, IntegerType), LiteralValue(1, IntegerType)))
- assert(filter1.equals(filter2))
- assert(filter1.references.map(_.describe()).toSeq == Seq("a"))
- assert(filter1.describe.equals("a IN (1, 2, 3, 4)"))
- val values: Array[Literal[_]] = new Array[Literal[_]](1000)
- for (i <- 0 until 1000) {
- values(i) = LiteralValue(i, IntegerType)
- }
- val filter3 = new In(ref("a"), values)
- var expected = "a IN ("
- for (i <- 0 until 50) {
- expected += i + ", "
- }
- expected = expected.dropRight(2) // remove the last ", "
- expected += "...)"
- assert(filter3.describe.equals(expected))
- }
-
- test("IsNull") {
- val filter1 = new IsNull(ref("a"))
- val filter2 = new IsNull(ref("a"))
- assert(filter1.equals(filter2))
- assert(filter1.references.map(_.describe()).toSeq == Seq("a"))
- assert(filter1.describe.equals("a IS NULL"))
- }
-
- test("IsNotNull") {
- val filter1 = new IsNotNull(ref("a"))
- val filter2 = new IsNotNull(ref("a"))
- assert(filter1.equals(filter2))
- assert(filter1.references.map(_.describe()).toSeq == Seq("a"))
- assert(filter1.describe.equals("a IS NOT NULL"))
- }
-
- test("Not") {
- val filter1 = new Not(new LessThan(ref("a"), LiteralValue(1, IntegerType)))
- val filter2 = new Not(new LessThan(ref("a"), LiteralValue(1, IntegerType)))
- assert(filter1.equals(filter2))
- assert(filter1.references.map(_.describe()).toSeq == Seq("a"))
- assert(filter1.describe.equals("NOT (a < 1)"))
- }
-
- test("And") {
- val filter1 = new And(new EqualTo(ref("a"), LiteralValue(1, IntegerType)),
- new EqualTo(ref("b"), LiteralValue(1, IntegerType)))
- val filter2 = new And(new EqualTo(ref("a"), LiteralValue(1, IntegerType)),
- new EqualTo(ref("b"), LiteralValue(1, IntegerType)))
- assert(filter1.equals(filter2))
- assert(filter1.references.map(_.describe()).toSeq == Seq("a", "b"))
- assert(filter1.describe.equals("(a = 1) AND (b = 1)"))
- }
-
- test("Or") {
- val filter1 = new Or(new EqualTo(ref("a"), LiteralValue(1, IntegerType)),
- new EqualTo(ref("b"), LiteralValue(1, IntegerType)))
- val filter2 = new Or(new EqualTo(ref("a"), LiteralValue(1, IntegerType)),
- new EqualTo(ref("b"), LiteralValue(1, IntegerType)))
- assert(filter1.equals(filter2))
- assert(filter1.references.map(_.describe()).toSeq == Seq("a", "b"))
- assert(filter1.describe.equals("(a = 1) OR (b = 1)"))
- }
-
- test("StringStartsWith") {
- val filter1 = new StringStartsWith(ref("a"), UTF8String.fromString("str"))
- val filter2 = new StringStartsWith(ref("a"), UTF8String.fromString("str"))
- assert(filter1.equals(filter2))
- assert(filter1.references.map(_.describe()).toSeq == Seq("a"))
- assert(filter1.describe.equals("STRING_STARTS_WITH(a, str)"))
- }
-
- test("StringEndsWith") {
- val filter1 = new StringEndsWith(ref("a"), UTF8String.fromString("str"))
- val filter2 = new StringEndsWith(ref("a"), UTF8String.fromString("str"))
- assert(filter1.equals(filter2))
- assert(filter1.references.map(_.describe()).toSeq == Seq("a"))
- assert(filter1.describe.equals("STRING_ENDS_WITH(a, str)"))
- }
-
- test("StringContains") {
- val filter1 = new StringContains(ref("a"), UTF8String.fromString("str"))
- val filter2 = new StringContains(ref("a"), UTF8String.fromString("str"))
- assert(filter1.equals(filter2))
- assert(filter1.references.map(_.describe()).toSeq == Seq("a"))
- assert(filter1.describe.equals("STRING_CONTAINS(a, str)"))
- }
-}
-
-object FiltersV2Suite {
- private[sql] def ref(parts: String*): FieldReference = {
- new FieldReference(parts)
- }
-}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/V2PredicateSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/V2PredicateSuite.scala
new file mode 100644
index 0000000..2d6e6fc
--- /dev/null
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/V2PredicateSuite.scala
@@ -0,0 +1,188 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.datasources.v2
+
+import org.apache.spark.SparkFunSuite
+import org.apache.spark.sql.connector.expressions.{Expression, FieldReference, Literal, LiteralValue}
+import org.apache.spark.sql.connector.expressions.filter._
+import org.apache.spark.sql.execution.datasources.v2.V2PredicateSuite.ref
+import org.apache.spark.sql.types.{IntegerType, StringType}
+import org.apache.spark.unsafe.types.UTF8String
+
+class V2PredicateSuite extends SparkFunSuite {
+
+ test("nested columns") {
+ val predicate1 =
+ new Predicate("=", Array[Expression](ref("a", "B"), LiteralValue(1, IntegerType)))
+ assert(predicate1.references.map(_.describe()).toSeq == Seq("a.B"))
+ assert(predicate1.describe.equals("a.B = 1"))
+
+ val predicate2 =
+ new Predicate("=", Array[Expression](ref("a", "b.c"), LiteralValue(1, IntegerType)))
+ assert(predicate2.references.map(_.describe()).toSeq == Seq("a.`b.c`"))
+ assert(predicate2.describe.equals("a.`b.c` = 1"))
+
+ val predicate3 =
+ new Predicate("=", Array[Expression](ref("`a`.b", "c"), LiteralValue(1, IntegerType)))
+ assert(predicate3.references.map(_.describe()).toSeq == Seq("```a``.b`.c"))
+ assert(predicate3.describe.equals("```a``.b`.c = 1"))
+ }
+
+ test("AlwaysTrue") {
+ val predicate1 = new AlwaysTrue
+ val predicate2 = new AlwaysTrue
+ assert(predicate1.equals(predicate2))
+ assert(predicate1.references.map(_.describe()).length == 0)
+ assert(predicate1.describe.equals("TRUE"))
+ }
+
+ test("AlwaysFalse") {
+ val predicate1 = new AlwaysFalse
+ val predicate2 = new AlwaysFalse
+ assert(predicate1.equals(predicate2))
+ assert(predicate1.references.map(_.describe()).length == 0)
+ assert(predicate1.describe.equals("FALSE"))
+ }
+
+ test("EqualTo") {
+ val predicate1 = new Predicate("=", Array[Expression](ref("a"), LiteralValue(1, IntegerType)))
+ val predicate2 = new Predicate("=", Array[Expression](ref("a"), LiteralValue(1, IntegerType)))
+ assert(predicate1.equals(predicate2))
+ assert(predicate1.references.map(_.describe()).toSeq == Seq("a"))
+ assert(predicate1.describe.equals("a = 1"))
+ }
+
+ test("EqualNullSafe") {
+ val predicate1 = new Predicate("<=>", Array[Expression](ref("a"), LiteralValue(1, IntegerType)))
+ val predicate2 = new Predicate("<=>", Array[Expression](ref("a"), LiteralValue(1, IntegerType)))
+ assert(predicate1.equals(predicate2))
+ assert(predicate1.references.map(_.describe()).toSeq == Seq("a"))
+ assert(predicate1.describe.equals("(a = 1) OR (a IS NULL AND 1 IS NULL)"))
+ }
+
+ test("In") {
+ val predicate1 = new Predicate("IN",
+ Array(ref("a"), LiteralValue(1, IntegerType), LiteralValue(2, IntegerType),
+ LiteralValue(3, IntegerType), LiteralValue(4, IntegerType)))
+ val predicate2 = new Predicate("IN",
+ Array(ref("a"), LiteralValue(4, IntegerType), LiteralValue(2, IntegerType),
+ LiteralValue(3, IntegerType), LiteralValue(1, IntegerType)))
+ assert(!predicate1.equals(predicate2))
+ assert(predicate1.references.map(_.describe()).toSeq == Seq("a"))
+ assert(predicate1.describe.equals("a IN (1, 2, 3, 4)"))
+ val values: Array[Literal[_]] = new Array[Literal[_]](1000)
+ var expected = "a IN ("
+ for (i <- 0 until 1000) {
+ values(i) = LiteralValue(i, IntegerType)
+ expected += i + ", "
+ }
+ val predicate3 = new Predicate("IN", (ref("a") +: values).toArray[Expression])
+ expected = expected.dropRight(2) // remove the last ", "
+ expected += ")"
+ assert(predicate3.describe.equals(expected))
+ }
+
+ test("IsNull") {
+ val predicate1 = new Predicate("IS_NULL", Array[Expression](ref("a")))
+ val predicate2 = new Predicate("IS_NULL", Array[Expression](ref("a")))
+ assert(predicate1.equals(predicate2))
+ assert(predicate1.references.map(_.describe()).toSeq == Seq("a"))
+ assert(predicate1.describe.equals("a IS NULL"))
+ }
+
+ test("IsNotNull") {
+ val predicate1 = new Predicate("IS_NOT_NULL", Array[Expression](ref("a")))
+ val predicate2 = new Predicate("IS_NOT_NULL", Array[Expression](ref("a")))
+ assert(predicate1.equals(predicate2))
+ assert(predicate1.references.map(_.describe()).toSeq == Seq("a"))
+ assert(predicate1.describe.equals("a IS NOT NULL"))
+ }
+
+ test("Not") {
+ val predicate1 = new Not(
+ new Predicate("<", Array[Expression](ref("a"), LiteralValue(1, IntegerType))))
+ val predicate2 = new Not(
+ new Predicate("<", Array[Expression](ref("a"), LiteralValue(1, IntegerType))))
+ assert(predicate1.equals(predicate2))
+ assert(predicate1.references.map(_.describe()).toSeq == Seq("a"))
+ assert(predicate1.describe.equals("NOT (a < 1)"))
+ }
+
+ test("And") {
+ val predicate1 = new And(
+ new Predicate("=", Array[Expression](ref("a"), LiteralValue(1, IntegerType))),
+ new Predicate("=", Array[Expression](ref("b"), LiteralValue(1, IntegerType))))
+ val predicate2 = new And(
+ new Predicate("=", Array[Expression](ref("a"), LiteralValue(1, IntegerType))),
+ new Predicate("=", Array[Expression](ref("b"), LiteralValue(1, IntegerType))))
+ assert(predicate1.equals(predicate2))
+ assert(predicate1.references.map(_.describe()).toSeq == Seq("a", "b"))
+ assert(predicate1.describe.equals("(a = 1) AND (b = 1)"))
+ }
+
+ test("Or") {
+ val predicate1 = new Or(
+ new Predicate("=", Array[Expression](ref("a"), LiteralValue(1, IntegerType))),
+ new Predicate("=", Array[Expression](ref("b"), LiteralValue(1, IntegerType))))
+ val predicate2 = new Or(
+ new Predicate("=", Array[Expression](ref("a"), LiteralValue(1, IntegerType))),
+ new Predicate("=", Array[Expression](ref("b"), LiteralValue(1, IntegerType))))
+ assert(predicate1.equals(predicate2))
+ assert(predicate1.references.map(_.describe()).toSeq == Seq("a", "b"))
+ assert(predicate1.describe.equals("(a = 1) OR (b = 1)"))
+ }
+
+ test("StringStartsWith") {
+ val literal = LiteralValue(UTF8String.fromString("str"), StringType)
+ val predicate1 = new Predicate("STARTS_WITH",
+ Array[Expression](ref("a"), literal))
+ val predicate2 = new Predicate("STARTS_WITH",
+ Array[Expression](ref("a"), literal))
+ assert(predicate1.equals(predicate2))
+ assert(predicate1.references.map(_.describe()).toSeq == Seq("a"))
+ assert(predicate1.describe.equals("a LIKE 'str%'"))
+ }
+
+ test("StringEndsWith") {
+ val literal = LiteralValue(UTF8String.fromString("str"), StringType)
+ val predicate1 = new Predicate("ENDS_WITH",
+ Array[Expression](ref("a"), literal))
+ val predicate2 = new Predicate("ENDS_WITH",
+ Array[Expression](ref("a"), literal))
+ assert(predicate1.equals(predicate2))
+ assert(predicate1.references.map(_.describe()).toSeq == Seq("a"))
+ assert(predicate1.describe.equals("a LIKE '%str'"))
+ }
+
+ test("StringContains") {
+ val literal = LiteralValue(UTF8String.fromString("str"), StringType)
+ val predicate1 = new Predicate("CONTAINS",
+ Array[Expression](ref("a"), literal))
+ val predicate2 = new Predicate("CONTAINS",
+ Array[Expression](ref("a"), literal))
+ assert(predicate1.equals(predicate2))
+ assert(predicate1.references.map(_.describe()).toSeq == Seq("a"))
+ assert(predicate1.describe.equals("a LIKE '%str%'"))
+ }
+}
+
+object V2PredicateSuite {
+ private[sql] def ref(parts: String*): FieldReference = {
+ new FieldReference(parts)
+ }
+}
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 d32e958..b1f5fd0 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
@@ -37,7 +37,7 @@ import org.apache.spark.sql.catalyst.util.{CaseInsensitiveMap, DateTimeTestUtils
import org.apache.spark.sql.execution.{DataSourceScanExec, ExtendedMode}
import org.apache.spark.sql.execution.command.{ExplainCommand, ShowCreateTableCommand}
import org.apache.spark.sql.execution.datasources.LogicalRelation
-import org.apache.spark.sql.execution.datasources.jdbc.{JDBCOptions, JDBCPartition, JDBCRDD, JDBCRelation, JdbcUtils}
+import org.apache.spark.sql.execution.datasources.jdbc.{JDBCOptions, JDBCPartition, JDBCRelation, JdbcUtils}
import org.apache.spark.sql.execution.metric.InputOutputMetricsHelper
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.sources._
@@ -773,12 +773,12 @@ class JDBCSuite extends QueryTest
}
test("compile filters") {
- val compileFilter = PrivateMethod[Option[String]](Symbol("compileFilter"))
def doCompileFilter(f: Filter): String =
- JDBCRDD invokePrivate compileFilter(f, JdbcDialects.get("jdbc:")) getOrElse("")
+ JdbcDialects.get("jdbc:").compileExpression(f.toV2).getOrElse("")
+
Seq(("col0", "col1"), ("`col0`", "`col1`")).foreach { case(col0, col1) =>
assert(doCompileFilter(EqualTo(col0, 3)) === """"col0" = 3""")
- assert(doCompileFilter(Not(EqualTo(col1, "abc"))) === """(NOT ("col1" = 'abc'))""")
+ assert(doCompileFilter(Not(EqualTo(col1, "abc"))) === """NOT ("col1" = 'abc')""")
assert(doCompileFilter(And(EqualTo(col0, 0), EqualTo(col1, "def")))
=== """("col0" = 0) AND ("col1" = 'def')""")
assert(doCompileFilter(Or(EqualTo(col0, 2), EqualTo(col1, "ghi")))
@@ -795,17 +795,14 @@ class JDBCSuite extends QueryTest
assert(doCompileFilter(In(col1, Array.empty)) ===
"""CASE WHEN "col1" IS NULL THEN NULL ELSE FALSE END""")
assert(doCompileFilter(Not(In(col1, Array("mno", "pqr"))))
- === """(NOT ("col1" IN ('mno', 'pqr')))""")
+ === """NOT ("col1" IN ('mno', 'pqr'))""")
assert(doCompileFilter(IsNull(col1)) === """"col1" IS NULL""")
assert(doCompileFilter(IsNotNull(col1)) === """"col1" IS NOT NULL""")
assert(doCompileFilter(And(EqualNullSafe(col0, "abc"), EqualTo(col1, "def")))
- === """((NOT ("col0" != 'abc' OR "col0" IS NULL OR 'abc' IS NULL) """
- + """OR ("col0" IS NULL AND 'abc' IS NULL))) AND ("col1" = 'def')""")
+ === """(("col0" = 'abc') OR ("col0" IS NULL AND 'abc' IS NULL))"""
+ + """ AND ("col1" = 'def')""")
}
- val e = intercept[AnalysisException] {
- doCompileFilter(EqualTo("col0.nested", 3))
- }.getMessage
- assert(e.contains("Filter push down does not support nested column: col0.nested"))
+ assert(doCompileFilter(EqualTo("col0.nested", 3)).isEmpty)
}
test("Dialect unregister") {
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 17bd7f7..d50a055 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
@@ -27,7 +27,7 @@ import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, Filter, Sort}
import org.apache.spark.sql.connector.expressions.{FieldReference, NullOrdering, SortDirection, SortValue}
import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2ScanRelation, V1ScanWrapper}
import org.apache.spark.sql.execution.datasources.v2.jdbc.JDBCTableCatalog
-import org.apache.spark.sql.functions.{avg, count, count_distinct, lit, sum, udf}
+import org.apache.spark.sql.functions.{avg, count, count_distinct, lit, not, sum, udf, when}
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.test.SharedSparkSession
import org.apache.spark.util.Utils
@@ -70,17 +70,17 @@ class JDBCV2Suite extends QueryTest with SharedSparkSession with ExplainSuiteHel
conn.prepareStatement("INSERT INTO \"test\".\"people\" VALUES ('mary', 2)").executeUpdate()
conn.prepareStatement(
"CREATE TABLE \"test\".\"employee\" (dept INTEGER, name TEXT(32), salary NUMERIC(20, 2)," +
- " bonus DOUBLE)").executeUpdate()
- conn.prepareStatement("INSERT INTO \"test\".\"employee\" VALUES (1, 'amy', 10000, 1000)")
- .executeUpdate()
- conn.prepareStatement("INSERT INTO \"test\".\"employee\" VALUES (2, 'alex', 12000, 1200)")
- .executeUpdate()
- conn.prepareStatement("INSERT INTO \"test\".\"employee\" VALUES (1, 'cathy', 9000, 1200)")
- .executeUpdate()
- conn.prepareStatement("INSERT INTO \"test\".\"employee\" VALUES (2, 'david', 10000, 1300)")
- .executeUpdate()
- conn.prepareStatement("INSERT INTO \"test\".\"employee\" VALUES (6, 'jen', 12000, 1200)")
- .executeUpdate()
+ " bonus DOUBLE, is_manager BOOLEAN)").executeUpdate()
+ conn.prepareStatement(
+ "INSERT INTO \"test\".\"employee\" VALUES (1, 'amy', 10000, 1000, true)").executeUpdate()
+ conn.prepareStatement(
+ "INSERT INTO \"test\".\"employee\" VALUES (2, 'alex', 12000, 1200, false)").executeUpdate()
+ conn.prepareStatement(
+ "INSERT INTO \"test\".\"employee\" VALUES (1, 'cathy', 9000, 1200, false)").executeUpdate()
+ conn.prepareStatement(
+ "INSERT INTO \"test\".\"employee\" VALUES (2, 'david', 10000, 1300, true)").executeUpdate()
+ conn.prepareStatement(
+ "INSERT INTO \"test\".\"employee\" VALUES (6, 'jen', 12000, 1200, true)").executeUpdate()
conn.prepareStatement(
"CREATE TABLE \"test\".\"dept\" (\"dept id\" INTEGER NOT NULL)").executeUpdate()
conn.prepareStatement("INSERT INTO \"test\".\"dept\" VALUES (1)").executeUpdate()
@@ -131,7 +131,7 @@ class JDBCV2Suite extends QueryTest with SharedSparkSession with ExplainSuiteHel
val df1 = spark.read.table("h2.test.employee")
.where($"dept" === 1).limit(1)
checkPushedLimit(df1, Some(1))
- checkAnswer(df1, Seq(Row(1, "amy", 10000.00, 1000.0)))
+ checkAnswer(df1, Seq(Row(1, "amy", 10000.00, 1000.0, true)))
val df2 = spark.read
.option("partitionColumn", "dept")
@@ -142,7 +142,7 @@ class JDBCV2Suite extends QueryTest with SharedSparkSession with ExplainSuiteHel
.filter($"dept" > 1)
.limit(1)
checkPushedLimit(df2, Some(1))
- checkAnswer(df2, Seq(Row(2, "alex", 12000.00, 1200.0)))
+ checkAnswer(df2, Seq(Row(2, "alex", 12000.00, 1200.0, false)))
val df3 = sql("SELECT name FROM h2.test.employee WHERE dept > 1 LIMIT 1")
checkSchemaNames(df3, Seq("NAME"))
@@ -191,12 +191,12 @@ class JDBCV2Suite extends QueryTest with SharedSparkSession with ExplainSuiteHel
.sort("salary")
.limit(1)
checkPushedLimit(df1, Some(1), createSortValues())
- checkAnswer(df1, Seq(Row(1, "cathy", 9000.00, 1200.0)))
+ checkAnswer(df1, Seq(Row(1, "cathy", 9000.00, 1200.0, false)))
val df2 = spark.read.table("h2.test.employee")
.where($"dept" === 1).orderBy($"salary").limit(1)
checkPushedLimit(df2, Some(1), createSortValues())
- checkAnswer(df2, Seq(Row(1, "cathy", 9000.00, 1200.0)))
+ checkAnswer(df2, Seq(Row(1, "cathy", 9000.00, 1200.0, false)))
val df3 = spark.read
.option("partitionColumn", "dept")
@@ -209,7 +209,7 @@ class JDBCV2Suite extends QueryTest with SharedSparkSession with ExplainSuiteHel
.limit(1)
checkPushedLimit(
df3, Some(1), createSortValues(SortDirection.DESCENDING, NullOrdering.NULLS_LAST))
- checkAnswer(df3, Seq(Row(2, "alex", 12000.00, 1200.0)))
+ checkAnswer(df3, Seq(Row(2, "alex", 12000.00, 1200.0, false)))
val df4 =
sql("SELECT name FROM h2.test.employee WHERE dept > 1 ORDER BY salary NULLS LAST LIMIT 1")
@@ -220,7 +220,8 @@ class JDBCV2Suite extends QueryTest with SharedSparkSession with ExplainSuiteHel
val df5 = spark.read.table("h2.test.employee")
.where($"dept" === 1).orderBy($"salary")
checkPushedLimit(df5, None)
- checkAnswer(df5, Seq(Row(1, "cathy", 9000.00, 1200.0), Row(1, "amy", 10000.00, 1000.0)))
+ checkAnswer(df5,
+ Seq(Row(1, "cathy", 9000.00, 1200.0, false), Row(1, "amy", 10000.00, 1000.0, true)))
val df6 = spark.read
.table("h2.test.employee")
@@ -247,7 +248,7 @@ class JDBCV2Suite extends QueryTest with SharedSparkSession with ExplainSuiteHel
.sort(sub($"NAME"))
.limit(1)
checkPushedLimit(df8)
- checkAnswer(df8, Seq(Row(2, "alex", 12000.00, 1200.0)))
+ checkAnswer(df8, Seq(Row(2, "alex", 12000.00, 1200.0, false)))
}
private def createSortValues(
@@ -264,11 +265,165 @@ class JDBCV2Suite extends QueryTest with SharedSparkSession with ExplainSuiteHel
df.queryExecution.optimizedPlan.collect {
case _: DataSourceV2ScanRelation =>
val expected_plan_fragment =
- "PushedFilters: [IsNotNull(ID), GreaterThan(ID,1)]"
+ "PushedFilters: [ID IS NOT NULL, ID > 1]"
checkKeywordsExistsInExplain(df, expected_plan_fragment)
}
checkAnswer(df, Row("mary", 2))
+
+ val df2 = spark.table("h2.test.employee").filter($"name".isin("amy", "cathy"))
+
+ checkFiltersRemoved(df2)
+
+ df2.queryExecution.optimizedPlan.collect {
+ case _: DataSourceV2ScanRelation =>
+ val expected_plan_fragment =
+ "PushedFilters: [NAME IN ('amy', 'cathy')]"
+ checkKeywordsExistsInExplain(df2, expected_plan_fragment)
+ }
+
+ checkAnswer(df2, Seq(Row(1, "amy", 10000, 1000, true), Row(1, "cathy", 9000, 1200, false)))
+
+ val df3 = spark.table("h2.test.employee").filter($"name".startsWith("a"))
+
+ checkFiltersRemoved(df3)
+
+ df3.queryExecution.optimizedPlan.collect {
+ case _: DataSourceV2ScanRelation =>
+ val expected_plan_fragment =
+ "PushedFilters: [NAME IS NOT NULL, NAME LIKE 'a%']"
+ checkKeywordsExistsInExplain(df3, expected_plan_fragment)
+ }
+
+ checkAnswer(df3, Seq(Row(1, "amy", 10000, 1000, true), Row(2, "alex", 12000, 1200, false)))
+
+ val df4 = spark.table("h2.test.employee").filter($"is_manager")
+
+ checkFiltersRemoved(df4)
+
+ df4.queryExecution.optimizedPlan.collect {
+ case _: DataSourceV2ScanRelation =>
+ val expected_plan_fragment =
+ "PushedFilters: [IS_MANAGER IS NOT NULL, IS_MANAGER = true]"
+ checkKeywordsExistsInExplain(df4, expected_plan_fragment)
+ }
+
+ checkAnswer(df4, Seq(Row(1, "amy", 10000, 1000, true), Row(2, "david", 10000, 1300, true),
+ Row(6, "jen", 12000, 1200, true)))
+
+ val df5 = spark.table("h2.test.employee").filter($"is_manager".and($"salary" > 10000))
+
+ checkFiltersRemoved(df5)
+
+ df5.queryExecution.optimizedPlan.collect {
+ case _: DataSourceV2ScanRelation =>
+ val expected_plan_fragment =
+ "PushedFilters: [IS_MANAGER IS NOT NULL, SALARY IS NOT NULL, " +
+ "IS_MANAGER = true, SALARY > 10000.00]"
+ checkKeywordsExistsInExplain(df5, expected_plan_fragment)
+ }
+
+ checkAnswer(df5, Seq(Row(6, "jen", 12000, 1200, true)))
+
+ val df6 = spark.table("h2.test.employee").filter($"is_manager".or($"salary" > 10000))
+
+ checkFiltersRemoved(df6)
+
+ df6.queryExecution.optimizedPlan.collect {
+ case _: DataSourceV2ScanRelation =>
+ val expected_plan_fragment =
+ "PushedFilters: [(IS_MANAGER = true) OR (SALARY > 10000.00)], "
+ checkKeywordsExistsInExplain(df6, expected_plan_fragment)
+ }
+
+ checkAnswer(df6, Seq(Row(1, "amy", 10000, 1000, true), Row(2, "alex", 12000, 1200, false),
+ Row(2, "david", 10000, 1300, true), Row(6, "jen", 12000, 1200, true)))
+
+ val df7 = spark.table("h2.test.employee").filter(not($"is_manager") === true)
+
+ checkFiltersRemoved(df7)
+
+ df7.queryExecution.optimizedPlan.collect {
+ case _: DataSourceV2ScanRelation =>
+ val expected_plan_fragment =
+ "PushedFilters: [IS_MANAGER IS NOT NULL, NOT (IS_MANAGER = true)], "
+ checkKeywordsExistsInExplain(df7, expected_plan_fragment)
+ }
+
+ checkAnswer(df7, Seq(Row(1, "cathy", 9000, 1200, false), Row(2, "alex", 12000, 1200, false)))
+
+ val df8 = spark.table("h2.test.employee").filter($"is_manager" === true)
+
+ checkFiltersRemoved(df8)
+
+ df8.queryExecution.optimizedPlan.collect {
+ case _: DataSourceV2ScanRelation =>
+ val expected_plan_fragment =
+ "PushedFilters: [IS_MANAGER IS NOT NULL, IS_MANAGER = true], "
+ checkKeywordsExistsInExplain(df8, expected_plan_fragment)
+ }
+
+ checkAnswer(df8, Seq(Row(1, "amy", 10000, 1000, true),
+ Row(2, "david", 10000, 1300, true), Row(6, "jen", 12000, 1200, true)))
+
+ val df9 = spark.table("h2.test.employee")
+ .filter(when($"dept" > 1, true).when($"is_manager", false).otherwise($"dept" > 3))
+
+ checkFiltersRemoved(df9)
+
+ df9.queryExecution.optimizedPlan.collect {
+ case _: DataSourceV2ScanRelation =>
+ val expected_plan_fragment =
+ "PushedFilters: [CASE WHEN DEPT > 1 THEN TRUE WHEN IS_MANAGER = true THEN FALSE" +
+ " ELSE DEPT > 3 END], "
+ checkKeywordsExistsInExplain(df9, expected_plan_fragment)
+ }
+
+ checkAnswer(df9, Seq(Row(2, "alex", 12000, 1200, false),
+ Row(2, "david", 10000, 1300, true), Row(6, "jen", 12000, 1200, true)))
+ }
+
+ test("scan with complex filter push-down") {
+ Seq(false, true).foreach { ansiMode =>
+ withSQLConf(SQLConf.ANSI_ENABLED.key -> ansiMode.toString) {
+ val df = spark.table("h2.test.people").filter($"id" + 1 > 1)
+
+ checkFiltersRemoved(df, ansiMode)
+
+ df.queryExecution.optimizedPlan.collect {
+ case _: DataSourceV2ScanRelation =>
+ val expected_plan_fragment = if (ansiMode) {
+ "PushedFilters: [ID IS NOT NULL, (ID + 1) > 1]"
+ } else {
+ "PushedFilters: [ID IS NOT NULL]"
+ }
+ checkKeywordsExistsInExplain(df, expected_plan_fragment)
+ }
+
+ checkAnswer(df, Seq(Row("fred", 1), Row("mary", 2)))
+
+ val df2 = sql("""
+ |SELECT * FROM h2.test.employee
+ |WHERE (CASE WHEN SALARY > 10000 THEN BONUS ELSE BONUS + 200 END) > 1200
+ |""".stripMargin)
+
+ checkFiltersRemoved(df2, ansiMode)
+
+ df2.queryExecution.optimizedPlan.collect {
+ case _: DataSourceV2ScanRelation =>
+ val expected_plan_fragment = if (ansiMode) {
+ "PushedFilters: [(CASE WHEN SALARY > 10000.00 THEN BONUS" +
+ " ELSE BONUS + 200.0 END) > 1200.0]"
+ } else {
+ "PushedFilters: []"
+ }
+ checkKeywordsExistsInExplain(df2, expected_plan_fragment)
+ }
+
+ checkAnswer(df2,
+ Seq(Row(1, "cathy", 9000, 1200, false), Row(2, "david", 10000, 1300, true)))
+ }
+ }
}
test("scan with column pruning") {
@@ -412,18 +567,22 @@ class JDBCV2Suite extends QueryTest with SharedSparkSession with ExplainSuiteHel
case _: DataSourceV2ScanRelation =>
val expected_plan_fragment =
"PushedAggregates: [MAX(SALARY), AVG(BONUS)], " +
- "PushedFilters: [IsNotNull(DEPT), GreaterThan(DEPT,0)], " +
- "PushedGroupByColumns: [DEPT]"
+ "PushedFilters: [DEPT IS NOT NULL, DEPT > 0], " +
+ "PushedGroupByColumns: [DEPT], "
checkKeywordsExistsInExplain(df, expected_plan_fragment)
}
checkAnswer(df, Seq(Row(10000, 1100.0), Row(12000, 1250.0), Row(12000, 1200.0)))
}
- private def checkFiltersRemoved(df: DataFrame): Unit = {
+ private def checkFiltersRemoved(df: DataFrame, removed: Boolean = true): Unit = {
val filters = df.queryExecution.optimizedPlan.collect {
case f: Filter => f
}
- assert(filters.isEmpty)
+ if (removed) {
+ assert(filters.isEmpty)
+ } else {
+ assert(filters.nonEmpty)
+ }
}
test("scan with aggregate push-down: MAX AVG with filter without group by") {
@@ -434,8 +593,8 @@ class JDBCV2Suite extends QueryTest with SharedSparkSession with ExplainSuiteHel
case _: DataSourceV2ScanRelation =>
val expected_plan_fragment =
"PushedAggregates: [MAX(ID), AVG(ID)], " +
- "PushedFilters: [IsNotNull(ID), GreaterThan(ID,0)], " +
- "PushedGroupByColumns: []"
+ "PushedFilters: [ID IS NOT NULL, ID > 0], " +
+ "PushedGroupByColumns: [], "
checkKeywordsExistsInExplain(df, expected_plan_fragment)
}
checkAnswer(df, Seq(Row(2, 1.5)))
@@ -550,7 +709,7 @@ class JDBCV2Suite extends QueryTest with SharedSparkSession with ExplainSuiteHel
val expected_plan_fragment =
"PushedAggregates: [SUM(SALARY)], " +
"PushedFilters: [], " +
- "PushedGroupByColumns: [DEPT]"
+ "PushedGroupByColumns: [DEPT], "
checkKeywordsExistsInExplain(df, expected_plan_fragment)
}
checkAnswer(df, Seq(Row(19000), Row(22000), Row(12000)))
@@ -564,7 +723,7 @@ class JDBCV2Suite extends QueryTest with SharedSparkSession with ExplainSuiteHel
val expected_plan_fragment =
"PushedAggregates: [SUM(DISTINCT SALARY)], " +
"PushedFilters: [], " +
- "PushedGroupByColumns: [DEPT]"
+ "PushedGroupByColumns: [DEPT], "
checkKeywordsExistsInExplain(df, expected_plan_fragment)
}
checkAnswer(df, Seq(Row(19000), Row(22000), Row(12000)))
@@ -579,8 +738,8 @@ class JDBCV2Suite extends QueryTest with SharedSparkSession with ExplainSuiteHel
case _: DataSourceV2ScanRelation =>
val expected_plan_fragment =
"PushedAggregates: [MAX(SALARY), MIN(BONUS)], " +
- "PushedFilters: [IsNotNull(DEPT), GreaterThan(DEPT,0)], " +
- "PushedGroupByColumns: [DEPT, NAME]"
+ "PushedFilters: [DEPT IS NOT NULL, DEPT > 0], " +
+ "PushedGroupByColumns: [DEPT, NAME], "
checkKeywordsExistsInExplain(df, expected_plan_fragment)
}
checkAnswer(df, Seq(Row(9000, 1200), Row(12000, 1200), Row(10000, 1300),
@@ -599,8 +758,8 @@ class JDBCV2Suite extends QueryTest with SharedSparkSession with ExplainSuiteHel
case _: DataSourceV2ScanRelation =>
val expected_plan_fragment =
"PushedAggregates: [MAX(SALARY)], " +
- "PushedFilters: [IsNotNull(DEPT), GreaterThan(DEPT,0)], " +
- "PushedGroupByColumns: [DEPT, NAME]"
+ "PushedFilters: [DEPT IS NOT NULL, DEPT > 0], " +
+ "PushedGroupByColumns: [DEPT, NAME], "
checkKeywordsExistsInExplain(df1, expected_plan_fragment)
}
checkAnswer(df1, Seq(Row("1#amy", 10000), Row("1#cathy", 9000), Row("2#alex", 12000),
@@ -617,8 +776,8 @@ class JDBCV2Suite extends QueryTest with SharedSparkSession with ExplainSuiteHel
case _: DataSourceV2ScanRelation =>
val expected_plan_fragment =
"PushedAggregates: [MAX(SALARY), MIN(BONUS)], " +
- "PushedFilters: [IsNotNull(DEPT), GreaterThan(DEPT,0)], " +
- "PushedGroupByColumns: [DEPT, NAME]"
+ "PushedFilters: [DEPT IS NOT NULL, DEPT > 0], " +
+ "PushedGroupByColumns: [DEPT, NAME], "
checkKeywordsExistsInExplain(df2, expected_plan_fragment)
}
checkAnswer(df2, Seq(Row("1#amy", 11000), Row("1#cathy", 10200), Row("2#alex", 13200),
@@ -634,7 +793,7 @@ class JDBCV2Suite extends QueryTest with SharedSparkSession with ExplainSuiteHel
df3.queryExecution.optimizedPlan.collect {
case _: DataSourceV2ScanRelation =>
val expected_plan_fragment =
- "PushedFilters: [IsNotNull(DEPT), GreaterThan(DEPT,0)], "
+ "PushedFilters: [DEPT IS NOT NULL, DEPT > 0], "
checkKeywordsExistsInExplain(df3, expected_plan_fragment)
}
checkAnswer(df3, Seq(Row("1#amy", 11000), Row("1#cathy", 10200), Row("2#alex", 13200),
@@ -653,8 +812,8 @@ class JDBCV2Suite extends QueryTest with SharedSparkSession with ExplainSuiteHel
case _: DataSourceV2ScanRelation =>
val expected_plan_fragment =
"PushedAggregates: [MAX(SALARY), MIN(BONUS)], " +
- "PushedFilters: [IsNotNull(DEPT), GreaterThan(DEPT,0)], " +
- "PushedGroupByColumns: [DEPT]"
+ "PushedFilters: [DEPT IS NOT NULL, DEPT > 0], " +
+ "PushedGroupByColumns: [DEPT], "
checkKeywordsExistsInExplain(df, expected_plan_fragment)
}
checkAnswer(df, Seq(Row(12000, 1200), Row(12000, 1200)))
@@ -670,7 +829,7 @@ class JDBCV2Suite extends QueryTest with SharedSparkSession with ExplainSuiteHel
val expected_plan_fragment =
"PushedAggregates: [MIN(SALARY)], " +
"PushedFilters: [], " +
- "PushedGroupByColumns: [DEPT]"
+ "PushedGroupByColumns: [DEPT], "
checkKeywordsExistsInExplain(df, expected_plan_fragment)
}
checkAnswer(df, Seq(Row(1, 9000), Row(2, 10000), Row(6, 12000)))
@@ -693,8 +852,8 @@ class JDBCV2Suite extends QueryTest with SharedSparkSession with ExplainSuiteHel
case _: DataSourceV2ScanRelation =>
val expected_plan_fragment =
"PushedAggregates: [SUM(SALARY)], " +
- "PushedFilters: [IsNotNull(DEPT), GreaterThan(DEPT,0)], " +
- "PushedGroupByColumns: [DEPT]"
+ "PushedFilters: [DEPT IS NOT NULL, DEPT > 0], " +
+ "PushedGroupByColumns: [DEPT], "
checkKeywordsExistsInExplain(query, expected_plan_fragment)
}
checkAnswer(query, Seq(Row(6, 12000), Row(1, 19000), Row(2, 22000)))
@@ -736,8 +895,8 @@ class JDBCV2Suite extends QueryTest with SharedSparkSession with ExplainSuiteHel
case _: DataSourceV2ScanRelation =>
val expected_plan_fragment =
"PushedAggregates: [VAR_POP(BONUS), VAR_SAMP(BONUS)], " +
- "PushedFilters: [IsNotNull(DEPT), GreaterThan(DEPT,0)], " +
- "PushedGroupByColumns: [DEPT]"
+ "PushedFilters: [DEPT IS NOT NULL, DEPT > 0], " +
+ "PushedGroupByColumns: [DEPT], "
checkKeywordsExistsInExplain(df, expected_plan_fragment)
}
checkAnswer(df, Seq(Row(10000d, 20000d), Row(2500d, 5000d), Row(0d, null)))
@@ -752,8 +911,8 @@ class JDBCV2Suite extends QueryTest with SharedSparkSession with ExplainSuiteHel
case _: DataSourceV2ScanRelation =>
val expected_plan_fragment =
"PushedAggregates: [STDDEV_POP(BONUS), STDDEV_SAMP(BONUS)], " +
- "PushedFilters: [IsNotNull(DEPT), GreaterThan(DEPT,0)], " +
- "PushedGroupByColumns: [DEPT]"
+ "PushedFilters: [DEPT IS NOT NULL, DEPT > 0], " +
+ "PushedGroupByColumns: [DEPT], "
checkKeywordsExistsInExplain(df, expected_plan_fragment)
}
checkAnswer(df, Seq(Row(100d, 141.4213562373095d), Row(50d, 70.71067811865476d), Row(0d, null)))
@@ -768,8 +927,8 @@ class JDBCV2Suite extends QueryTest with SharedSparkSession with ExplainSuiteHel
case _: DataSourceV2ScanRelation =>
val expected_plan_fragment =
"PushedAggregates: [COVAR_POP(BONUS, BONUS), COVAR_SAMP(BONUS, BONUS)], " +
- "PushedFilters: [IsNotNull(DEPT), GreaterThan(DEPT,0)], " +
- "PushedGroupByColumns: [DEPT]"
+ "PushedFilters: [DEPT IS NOT NULL, DEPT > 0], " +
+ "PushedGroupByColumns: [DEPT], "
checkKeywordsExistsInExplain(df, expected_plan_fragment)
}
checkAnswer(df, Seq(Row(10000d, 20000d), Row(2500d, 5000d), Row(0d, null)))
@@ -784,15 +943,15 @@ class JDBCV2Suite extends QueryTest with SharedSparkSession with ExplainSuiteHel
case _: DataSourceV2ScanRelation =>
val expected_plan_fragment =
"PushedAggregates: [CORR(BONUS, BONUS)], " +
- "PushedFilters: [IsNotNull(DEPT), GreaterThan(DEPT,0)], " +
- "PushedGroupByColumns: [DEPT]"
+ "PushedFilters: [DEPT IS NOT NULL, DEPT > 0], " +
+ "PushedGroupByColumns: [DEPT], "
checkKeywordsExistsInExplain(df, expected_plan_fragment)
}
checkAnswer(df, Seq(Row(1d), Row(1d), Row(null)))
}
test("scan with aggregate push-down: aggregate over alias NOT push down") {
- val cols = Seq("a", "b", "c", "d")
+ val cols = Seq("a", "b", "c", "d", "e")
val df1 = sql("select * from h2.test.employee").toDF(cols: _*)
val df2 = df1.groupBy().sum("c")
checkAggregateRemoved(df2, false)
@@ -848,10 +1007,10 @@ class JDBCV2Suite extends QueryTest with SharedSparkSession with ExplainSuiteHel
df.queryExecution.optimizedPlan.collect {
case _: DataSourceV2ScanRelation =>
val expected_plan_fragment =
- "PushedAggregates: [COUNT(CASE WHEN ((SALARY) > (8000.00)) AND ((SALARY) < (10000.00))" +
- " THEN SALARY ELSE 0.00 END), C..., " +
+ "PushedAggregates: [COUNT(CASE WHEN (SALARY > 8000.00) AND (SALARY < 10000.00)" +
+ " THEN SALARY ELSE 0.00 END), COUNT(CAS..., " +
"PushedFilters: [], " +
- "PushedGroupByColumns: [DEPT]"
+ "PushedGroupByColumns: [DEPT], "
checkKeywordsExistsInExplain(df, expected_plan_fragment)
}
checkAnswer(df, Seq(Row(1, 1, 1, 1, 1, 0d, 12000d, 0d, 12000d, 12000d, 0d, 0d, 2, 0d),
@@ -865,8 +1024,9 @@ class JDBCV2Suite extends QueryTest with SharedSparkSession with ExplainSuiteHel
val df = sql("SELECT SUM(2147483647 + DEPT) FROM h2.test.employee")
checkAggregateRemoved(df, ansiMode)
val expected_plan_fragment = if (ansiMode) {
- "PushedAggregates: [SUM((2147483647) + (DEPT))], " +
- "PushedFilters: [], PushedGroupByColumns: []"
+ "PushedAggregates: [SUM(2147483647 + DEPT)], " +
+ "PushedFilters: [], " +
+ "PushedGroupByColumns: []"
} else {
"PushedFilters: []"
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org