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

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

@@ -20,6 +20,7 @@
 import java.util.Objects;
 
 import org.apache.spark.annotation.Evolving;
+import org.apache.spark.sql.connector.expressions.Expression;
 import org.apache.spark.sql.connector.expressions.NamedReference;
 
 /**
@@ -46,5 +47,5 @@ public int hashCode() {
   public String toString() { return "FALSE"; }
 
   @Override
-  public NamedReference[] references() { return EMPTY_REFERENCE; }
+  public Expression[] references() { return EMPTY_EXPRESSION; }
 }

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

@@ -20,6 +20,7 @@
 import java.util.Objects;
 
 import org.apache.spark.annotation.Evolving;
+import org.apache.spark.sql.connector.expressions.Expression;
 import org.apache.spark.sql.connector.expressions.NamedReference;
 
 /**
@@ -46,5 +47,5 @@ public int hashCode() {
   public String toString() { return "TRUE"; }
 
   @Override
-  public NamedReference[] references() { return EMPTY_REFERENCE; }
+  public Expression[] references() { return EMPTY_EXPRESSION; }
 }

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

@@ -20,8 +20,7 @@
 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;
+import org.apache.spark.sql.connector.expressions.Expression;
 
 /**
  * Base class for {@link EqualNullSafe}, {@link EqualTo}, {@link GreaterThan},
@@ -31,16 +30,16 @@
  */
 @Evolving
 abstract class BinaryComparison extends Filter {
-  protected final NamedReference column;
-  protected final Literal<?> value;
+  protected final Expression column;
+  protected final Expression value;
 
-  protected BinaryComparison(NamedReference column, Literal<?> value) {
+  protected BinaryComparison(Expression column, Expression value) {
     this.column = column;
     this.value = value;
   }
 
-  public NamedReference column() { return column; }
-  public Literal<?> value() { return value; }
+  public Expression column() { return column; }
+  public Expression value() { return value; }
 
   @Override
   public boolean equals(Object o) {
@@ -56,5 +55,5 @@ public int hashCode() {
   }
 
   @Override
-  public NamedReference[] references() { return new NamedReference[] { column }; }
+  public Expression[] references() { return new Expression[] { column }; }
 }

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

@@ -18,8 +18,7 @@
 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;
+import org.apache.spark.sql.connector.expressions.Expression;
 
 /**
  * Performs equality comparison, similar to {@link EqualTo}. However, this differs from
@@ -31,7 +30,7 @@
 @Evolving
 public final class EqualNullSafe extends BinaryComparison {
 
-  public EqualNullSafe(NamedReference column, Literal<?> value) {
+  public EqualNullSafe(Expression column, Expression value) {
     super(column, value);
   }
 

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

@@ -18,8 +18,7 @@
 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;
+import org.apache.spark.sql.connector.expressions.Expression;
 
 /**
  * A filter that evaluates to {@code true} iff the {@code column} evaluates to a value
@@ -30,7 +29,7 @@
 @Evolving
 public final class EqualTo extends BinaryComparison {
 
-  public EqualTo(NamedReference column, Literal<?> value) {
+  public EqualTo(Expression column, Expression value) {
     super(column, value);
   }
 

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

@@ -21,7 +21,6 @@
 
 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
@@ -31,10 +30,10 @@
 @Evolving
 public abstract class Filter implements Expression, Serializable {
 
-  protected static final NamedReference[] EMPTY_REFERENCE = new NamedReference[0];
+  protected static final Expression[] EMPTY_EXPRESSION = new Expression[0];
 
   /**
    * Returns list of columns that are referenced by this filter.
    */
-  public abstract NamedReference[] references();
+  public abstract Expression[] references();
 }

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

@@ -18,8 +18,7 @@
 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;
+import org.apache.spark.sql.connector.expressions.Expression;
 
 /**
  * A filter that evaluates to {@code true} iff the {@code column} evaluates to a value
@@ -30,7 +29,7 @@
 @Evolving
 public final class GreaterThan extends BinaryComparison {
 
-  public GreaterThan(NamedReference column, Literal<?> value) {
+  public GreaterThan(Expression column, Expression value) {
     super(column, value);
   }
 

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

@@ -18,8 +18,7 @@
 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;
+import org.apache.spark.sql.connector.expressions.Expression;
 
 /**
  * A filter that evaluates to {@code true} iff the {@code column} evaluates to a value
@@ -30,7 +29,7 @@
 @Evolving
 public final class GreaterThanOrEqual extends BinaryComparison {
 
-  public GreaterThanOrEqual(NamedReference column, Literal<?> value) {
+  public GreaterThanOrEqual(Expression column, Expression value) {
     super(column, value);
   }
 

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

@@ -22,8 +22,8 @@
 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.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
@@ -34,15 +34,15 @@
 @Evolving
 public final class In extends Filter {
   static final int MAX_LEN_TO_PRINT = 50;
-  private final NamedReference column;
+  private final Expression column;
   private final Literal<?>[] values;
 
-  public In(NamedReference column, Literal<?>[] values) {
+  public In(Expression column, Literal<?>[] values) {
     this.column = column;
     this.values = values;
   }
 
-  public NamedReference column() { return column; }
+  public Expression column() { return column; }
   public Literal<?>[] values() { return values; }
 
   @Override
@@ -72,5 +72,5 @@ public String toString() {
   }
 
   @Override
-  public NamedReference[] references() { return new NamedReference[] { column }; }
+  public Expression[] references() { return new Expression[] { column }; }
 }

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

@@ -20,7 +20,7 @@
 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.Expression;
 
 /**
  * A filter that evaluates to {@code true} iff the {@code column} evaluates to a non-null value.
@@ -29,13 +29,13 @@
  */
 @Evolving
 public final class IsNotNull extends Filter {
-  private final NamedReference column;
+  private final Expression column;
 
-  public IsNotNull(NamedReference column) {
+  public IsNotNull(Expression column) {
     this.column = column;
   }
 
-  public NamedReference column() { return column; }
+  public Expression column() { return column; }
 
   @Override
   public String toString() { return column.describe() + " IS NOT NULL"; }
@@ -54,5 +54,5 @@ public int hashCode() {
   }
 
   @Override
-  public NamedReference[] references() { return new NamedReference[] { column }; }
+  public Expression[] references() { return new Expression[] { column }; }
 }

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

@@ -20,7 +20,7 @@
 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.Expression;
 
 /**
  * A filter that evaluates to {@code true} iff the {@code column} evaluates to null.
@@ -29,13 +29,13 @@
  */
 @Evolving
 public final class IsNull extends Filter {
-  private final NamedReference column;
+  private final Expression column;
 
-  public IsNull(NamedReference column) {
+  public IsNull(Expression column) {
     this.column = column;
   }
 
-  public NamedReference column() { return column; }
+  public Expression column() { return column; }
 
   @Override
   public String toString() { return column.describe() + " IS NULL"; }
@@ -54,5 +54,5 @@ public int hashCode() {
   }
 
   @Override
-  public NamedReference[] references() { return new NamedReference[] { column }; }
+  public Expression[] references() { return new Expression[] { column }; }
 }

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

@@ -18,8 +18,7 @@
 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;
+import org.apache.spark.sql.connector.expressions.Expression;
 
 /**
  * A filter that evaluates to {@code true} iff the {@code column} evaluates to a value
@@ -30,7 +29,7 @@
 @Evolving
 public final class LessThan extends BinaryComparison {
 
-  public LessThan(NamedReference column, Literal<?> value) {
+  public LessThan(Expression column, Expression value) {
     super(column, value);
   }
 

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

@@ -18,8 +18,7 @@
 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;
+import org.apache.spark.sql.connector.expressions.Expression;
 
 /**
  * A filter that evaluates to {@code true} iff the {@code column} evaluates to a value
@@ -30,7 +29,7 @@
 @Evolving
 public final class LessThanOrEqual extends BinaryComparison {
 
-  public LessThanOrEqual(NamedReference column, Literal<?> value) {
+  public LessThanOrEqual(Expression column, Expression value) {
     super(column, value);
   }
 

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

@@ -18,7 +18,7 @@
 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.sql.connector.expressions.Expression;
 import org.apache.spark.unsafe.types.UTF8String;
 
 /**
@@ -30,7 +30,7 @@
 @Evolving
 public final class StringContains extends StringPredicate {
 
-  public StringContains(NamedReference column, UTF8String value) {
+  public StringContains(Expression column, UTF8String value) {
     super(column, value);
   }
 

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

@@ -18,7 +18,7 @@
 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.sql.connector.expressions.Expression;
 import org.apache.spark.unsafe.types.UTF8String;
 
 /**
@@ -30,7 +30,7 @@
 @Evolving
 public final class StringEndsWith extends StringPredicate {
 
-  public StringEndsWith(NamedReference column, UTF8String value) {
+  public StringEndsWith(Expression column, UTF8String value) {
     super(column, value);
   }
 

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

@@ -20,7 +20,7 @@
 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.Expression;
 import org.apache.spark.unsafe.types.UTF8String;
 
 /**
@@ -31,15 +31,15 @@
  */
 @Evolving
 abstract class StringPredicate extends Filter {
-  protected final NamedReference column;
+  protected final Expression column;
   protected final UTF8String value;
 
-  protected StringPredicate(NamedReference column, UTF8String value) {
+  protected StringPredicate(Expression column, UTF8String value) {
     this.column = column;
     this.value = value;
   }
 
-  public NamedReference column() { return column; }
+  public Expression column() { return column; }
   public UTF8String value() { return value; }
 
   @Override
@@ -56,5 +56,5 @@ public int hashCode() {
   }
 
   @Override
-  public NamedReference[] references() { return new NamedReference[] { column }; }
+  public Expression[] references() { return new Expression[] { column }; }
 }

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

@@ -18,7 +18,7 @@
 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.sql.connector.expressions.Expression;
 import org.apache.spark.unsafe.types.UTF8String;
 
 /**
@@ -30,7 +30,7 @@
 @Evolving
 public final class StringStartsWith extends StringPredicate {
 
-  public StringStartsWith(NamedReference column, UTF8String value) {
+  public StringStartsWith(Expression column, UTF8String value) {
     super(column, value);
   }