[SPARK-36760][SQL] Add interface SupportsPushDownV2Filters

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

@@ -47,4 +47,9 @@ public int hashCode() {
 
   @Override
   public NamedReference[] references() { return EMPTY_REFERENCE; }
+
+  @Override
+  public org.apache.spark.sql.sources.Filter toV1() {
+    return new org.apache.spark.sql.sources.AlwaysFalse();
+  }
 }

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

@@ -47,4 +47,9 @@ public int hashCode() {
 
   @Override
   public NamedReference[] references() { return EMPTY_REFERENCE; }
+
+  @Override
+  public org.apache.spark.sql.sources.Filter toV1() {
+    return new org.apache.spark.sql.sources.AlwaysTrue();
+  }
 }

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

@@ -36,4 +36,9 @@ public And(Filter left, Filter right) {
   public String toString() {
     return String.format("(%s) AND (%s)", left.describe(), right.describe());
   }
+
+  @Override
+  public org.apache.spark.sql.sources.Filter toV1() {
+    return new org.apache.spark.sql.sources.And(left.toV1(), right.toV1());
+  }
 }

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

@@ -18,6 +18,7 @@
 package org.apache.spark.sql.connector.expressions.filter;
 
 import org.apache.spark.annotation.Evolving;
+import org.apache.spark.sql.catalyst.CatalystTypeConverters;
 import org.apache.spark.sql.connector.expressions.Literal;
 import org.apache.spark.sql.connector.expressions.NamedReference;
 
@@ -37,4 +38,10 @@ public EqualNullSafe(NamedReference column, Literal<?> value) {
 
   @Override
   public String toString() { return this.column.describe() + " <=> " + value.describe(); }
+
+  @Override
+  public org.apache.spark.sql.sources.Filter toV1() {
+    return new org.apache.spark.sql.sources.EqualNullSafe(
+      column.describe(), CatalystTypeConverters.convertToScala(value.value(), value.dataType()));
+  }
 }

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

@@ -18,6 +18,7 @@
 package org.apache.spark.sql.connector.expressions.filter;
 
 import org.apache.spark.annotation.Evolving;
+import org.apache.spark.sql.catalyst.CatalystTypeConverters;
 import org.apache.spark.sql.connector.expressions.Literal;
 import org.apache.spark.sql.connector.expressions.NamedReference;
 
@@ -36,4 +37,10 @@ public EqualTo(NamedReference column, Literal<?> value) {
 
   @Override
   public String toString() { return column.describe() + " = " + value.describe(); }
+
+  @Override
+  public org.apache.spark.sql.sources.Filter toV1() {
+    return new org.apache.spark.sql.sources.EqualTo(
+      (column).describe(), CatalystTypeConverters.convertToScala(value.value(), value.dataType()));
+  }
 }

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

@@ -17,6 +17,8 @@
 
 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;
@@ -27,7 +29,7 @@
  * @since 3.3.0
  */
 @Evolving
-public abstract class Filter implements Expression {
+public abstract class Filter implements Expression, Serializable {
 
   protected static final NamedReference[] EMPTY_REFERENCE = new NamedReference[0];
 
@@ -38,4 +40,9 @@ public abstract class Filter implements Expression {
 
   @Override
   public String describe() { return this.toString(); }
+
+  /**
+   * Returns a V1 Filter.
+   */
+  public abstract org.apache.spark.sql.sources.Filter toV1();
 }

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

@@ -18,6 +18,7 @@
 package org.apache.spark.sql.connector.expressions.filter;
 
 import org.apache.spark.annotation.Evolving;
+import org.apache.spark.sql.catalyst.CatalystTypeConverters;
 import org.apache.spark.sql.connector.expressions.Literal;
 import org.apache.spark.sql.connector.expressions.NamedReference;
 
@@ -36,4 +37,11 @@ public GreaterThan(NamedReference column, Literal<?> value) {
 
   @Override
   public String toString() { return column.describe() + " > " + value.describe(); }
+
+  @Override
+  public org.apache.spark.sql.sources.Filter toV1() {
+    return new org.apache.spark.sql.sources.GreaterThan(
+      column.describe(), CatalystTypeConverters.convertToScala(value.value(), value.dataType()));
+  }
+
 }

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

@@ -18,6 +18,7 @@
 package org.apache.spark.sql.connector.expressions.filter;
 
 import org.apache.spark.annotation.Evolving;
+import org.apache.spark.sql.catalyst.CatalystTypeConverters;
 import org.apache.spark.sql.connector.expressions.Literal;
 import org.apache.spark.sql.connector.expressions.NamedReference;
 
@@ -36,4 +37,10 @@ public GreaterThanOrEqual(NamedReference column, Literal<?> value) {
 
   @Override
   public String toString() { return column.describe() + " >= " + value.describe(); }
+
+  @Override
+  public org.apache.spark.sql.sources.Filter toV1() {
+    return new org.apache.spark.sql.sources.GreaterThanOrEqual(
+      column.describe(), CatalystTypeConverters.convertToScala(value.value(), value.dataType()));
+  }
 }

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

@@ -22,6 +22,7 @@
 import java.util.stream.Collectors;
 
 import org.apache.spark.annotation.Evolving;
+import org.apache.spark.sql.catalyst.CatalystTypeConverters;
 import org.apache.spark.sql.connector.expressions.Literal;
 import org.apache.spark.sql.connector.expressions.NamedReference;
 
@@ -73,4 +74,15 @@ public String toString() {
 
   @Override
   public NamedReference[] references() { return new NamedReference[] { column }; }
+
+  @Override
+  public org.apache.spark.sql.sources.Filter toV1() {
+    Object[] array = new Object[values.length];
+    int index = 0;
+    for (Literal value: values) {
+      array[index] = CatalystTypeConverters.convertToScala(value.value(), value.dataType());
+      index++;
+    }
+    return new org.apache.spark.sql.sources.In(column.describe(), array);
+  }
 }

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

@@ -55,4 +55,9 @@ public int hashCode() {
 
   @Override
   public NamedReference[] references() { return new NamedReference[] { column }; }
+
+  @Override
+  public org.apache.spark.sql.sources.Filter toV1() {
+    return new org.apache.spark.sql.sources.IsNotNull(column.describe());
+  }
 }

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

@@ -55,4 +55,9 @@ public int hashCode() {
 
   @Override
   public NamedReference[] references() { return new NamedReference[] { column }; }
+
+  @Override
+  public org.apache.spark.sql.sources.Filter toV1() {
+    return new org.apache.spark.sql.sources.IsNull(column.describe());
+  }
 }

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

@@ -18,6 +18,7 @@
 package org.apache.spark.sql.connector.expressions.filter;
 
 import org.apache.spark.annotation.Evolving;
+import org.apache.spark.sql.catalyst.CatalystTypeConverters;
 import org.apache.spark.sql.connector.expressions.Literal;
 import org.apache.spark.sql.connector.expressions.NamedReference;
 
@@ -36,4 +37,10 @@ public LessThan(NamedReference column, Literal<?> value) {
 
   @Override
   public String toString() { return column.describe() + " < " + value.describe(); }
+
+  @Override
+  public org.apache.spark.sql.sources.Filter toV1() {
+    return new org.apache.spark.sql.sources.LessThan(
+      column.describe(), CatalystTypeConverters.convertToScala(value.value(), value.dataType()));
+  }
 }

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

@@ -18,6 +18,7 @@
 package org.apache.spark.sql.connector.expressions.filter;
 
 import org.apache.spark.annotation.Evolving;
+import org.apache.spark.sql.catalyst.CatalystTypeConverters;
 import org.apache.spark.sql.connector.expressions.Literal;
 import org.apache.spark.sql.connector.expressions.NamedReference;
 
@@ -36,4 +37,10 @@ public LessThanOrEqual(NamedReference column, Literal<?> value) {
 
   @Override
   public String toString() { return column.describe() + " <= " + value.describe(); }
+
+  @Override
+  public org.apache.spark.sql.sources.Filter toV1() {
+    return new org.apache.spark.sql.sources.LessThanOrEqual(
+      column.describe(), CatalystTypeConverters.convertToScala(value.value(), value.dataType()));
+  }
 }

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

@@ -53,4 +53,9 @@ public int hashCode() {
 
   @Override
   public NamedReference[] references() { return child.references(); }
+
+  @Override
+  public org.apache.spark.sql.sources.Filter toV1() {
+    return new org.apache.spark.sql.sources.Not(child.toV1());
+  }
 }

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

@@ -36,4 +36,9 @@ public Or(Filter left, Filter right) {
   public String toString() {
     return String.format("(%s) OR (%s)", left.describe(), right.describe());
   }
+
+  @Override
+  public org.apache.spark.sql.sources.Filter toV1() {
+    return new org.apache.spark.sql.sources.Or(left.toV1(), right.toV1());
+  }
 }

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

@@ -36,4 +36,9 @@ public StringContains(NamedReference column, UTF8String value) {
 
   @Override
   public String toString() { return "STRING_CONTAINS(" + column.describe() + ", " + value + ")"; }
+
+  @Override
+  public org.apache.spark.sql.sources.Filter toV1() {
+    return new org.apache.spark.sql.sources.StringContains(column.describe(), value.toString());
+  }
 }

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

@@ -36,4 +36,9 @@ public StringEndsWith(NamedReference column, UTF8String value) {
 
   @Override
   public String toString() { return "STRING_ENDS_WITH(" + column.describe() + ", " + value + ")"; }
+
+  @Override
+  public org.apache.spark.sql.sources.Filter toV1() {
+    return new org.apache.spark.sql.sources.StringEndsWith(column.describe(), value.toString());
+  }
 }

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

@@ -38,4 +38,9 @@ public StringStartsWith(NamedReference column, UTF8String value) {
   public String toString() {
     return "STRING_STARTS_WITH(" + column.describe() + ", " + value + ")";
   }
+
+  @Override
+  public org.apache.spark.sql.sources.Filter toV1() {
+    return new org.apache.spark.sql.sources.StringStartsWith(column.describe(), value.toString());
+  }
 }

sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala

@@ -35,6 +35,7 @@ import org.apache.spark.sql.connector.catalog._
 import org.apache.spark.sql.connector.catalog.CatalogV2Implicits._
 import org.apache.spark.sql.connector.catalog.functions.{BoundFunction, UnboundFunction}
 import org.apache.spark.sql.connector.expressions.{NamedReference, Transform}
+import org.apache.spark.sql.connector.expressions.filter.{Filter => V2Filter}
 import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.internal.SQLConf.{LEGACY_ALLOW_NEGATIVE_SCALE_OF_DECIMAL_ENABLED, LEGACY_CTE_PRECEDENCE_POLICY}
 import org.apache.spark.sql.sources.Filter
@@ -1136,6 +1137,11 @@ object QueryCompilationErrors {
       s"Fail to rebuild expression: missing key $filter in `translatedFilterToExpr`")
   }
 
+  def failedToRebuildExpressionError(filter: V2Filter): Throwable = {
+    new AnalysisException(
+      s"Fail to rebuild expression: missing key $filter in `translatedFilterToExpr`")
+  }
+
   def dataTypeUnsupportedByDataSourceError(format: String, field: StructField): Throwable = {
     new AnalysisException(
       s"$format data source does not support ${field.dataType.catalogString} data type.")
@@ -2392,4 +2398,8 @@ object QueryCompilationErrors {
       errorClass = "INVALID_JSON_SCHEMA_MAPTYPE",
       messageParameters = Array(schema.toString))
   }
+
+  def invalidDataTypeForFilterValue(value: Any): Throwable = {
+    new AnalysisException(s"Filter value $value has invalid data type")
+  }
 }