[SPARK-38432][SQL] Refactor framework so as JDBC dialect could compile filter by self way

project/MimaExcludes.scala

@@ -53,6 +53,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"),

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 @@
  */
 @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);
+  }
 }

sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/GeneralScalarExpression.java

@@ -19,77 +19,19 @@
 
 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>&lt;&gt;</code>
- *   <ul>
- *    <li>SQL semantic: <code>expr1 &lt;&gt; expr2</code></li>
- *    <li>Since version: 3.3.0</li>
- *   </ul>
- *  </li>
- *  <li>Name: <code>&lt;=&gt;</code>
- *   <ul>
- *    <li>SQL semantic: <code>expr1 &lt;=&gt; expr2</code></li>
- *    <li>Since version: 3.3.0</li>
- *   </ul>
- *  </li>
- *  <li>Name: <code>&lt;</code>
- *   <ul>
- *    <li>SQL semantic: <code>expr1 &lt; expr2</code></li>
- *    <li>Since version: 3.3.0</li>
- *   </ul>
- *  </li>
- *  <li>Name: <code>&lt;=</code>
- *   <ul>
- *    <li>SQL semantic: <code>expr1 &lt;= expr2</code></li>
- *    <li>Since version: 3.3.0</li>
- *   </ul>
- *  </li>
- *  <li>Name: <code>&gt;</code>
- *   <ul>
- *    <li>SQL semantic: <code>expr1 &gt; expr2</code></li>
- *    <li>Since version: 3.3.0</li>
- *   </ul>
- *  </li>
- *  <li>Name: <code>&gt;=</code>
- *   <ul>
- *    <li>SQL semantic: <code>expr1 &gt;= 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 @@
  *    <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 @@
  *
  * @since 3.3.0
  */
-// scalastyle:on line.size.limit
 @Evolving
 public class GeneralScalarExpression implements Expression, Serializable {
   private String name;
@@ -190,6 +113,19 @@ public GeneralScalarExpression(String name, Expression[] children) {
   public String name() { return name; }
   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();

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; }
 }

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 }; }
 }

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() }; }
 }

sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/Transform.java

@@ -34,13 +34,11 @@ 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(); }
 }

sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/aggregate/Avg.java

@@ -38,6 +38,9 @@ public Avg(Expression column, boolean isDistinct) {
   public Expression column() { return input; }
   public boolean isDistinct() { return isDistinct; }
 
+  @Override
+  public Expression[] children() { return new Expression[]{ input }; }
+
   @Override
   public String toString() {
     if (isDistinct) {

sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/aggregate/Count.java

@@ -38,6 +38,9 @@ public Count(Expression column, boolean isDistinct) {
   public Expression column() { return input; }
   public boolean isDistinct() { return isDistinct; }
 
+  @Override
+  public Expression[] children() { return new Expression[]{ input }; }
+
   @Override
   public String toString() {
     if (isDistinct) {

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.
@@ -30,6 +31,9 @@ public final class CountStar implements AggregateFunc {
   public CountStar() {
   }
 
+  @Override
+  public Expression[] children() { return EMPTY_EXPRESSION; }
+
   @Override
   public String toString() { return "COUNT(*)"; }
 }

sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/aggregate/GeneralAggregateFunc.java

@@ -22,7 +22,6 @@
 
 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 @@
 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) {

sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/aggregate/Max.java

@@ -33,6 +33,9 @@ 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() + ")"; }
 }

sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/aggregate/Min.java

@@ -33,6 +33,9 @@ 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() + ")"; }
 }

sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/aggregate/Sum.java

@@ -38,6 +38,9 @@ public Sum(Expression column, boolean isDistinct) {
   public Expression column() { return input; }
   public boolean isDistinct() { return isDistinct; }
 
+  @Override
+  public Expression[] children() { return new Expression[]{ input }; }
+
   @Override
   public String toString() {
     if (isDistinct) {

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"; }
 }