Add v2 filter

sql/catalyst/src/main/java/org/apache/spark/sql/catalog/v2/expressions/NamedReference.java

@@ -18,6 +18,7 @@
 package org.apache.spark.sql.catalog.v2.expressions;
 
 import org.apache.spark.annotation.Experimental;
+import java.util.Arrays;
 
 /**
  * Represents a field or column reference in the public logical expression API.
@@ -30,4 +31,18 @@ public interface NamedReference extends Expression {
    * Each string in the returned array represents a field name.
    */
   String[] fieldNames();
+
+  /**
+   * Returns the full name of the field. For nested column, `.` character will be used as delimiter.
+   * If the name contains `.` character, it will be quoted to distinguish from the delimiter.
+   */
+  default String name() {
+    return Arrays.stream(fieldNames()).reduce((acc, name) -> {
+      if (!name.contains("`") && name.contains(".")) {
+        return acc + "." + "`" + name + "`";
+      } else {
+        return acc + "." + name;
+      }
+    }).get();
+  }
 }

sql/catalyst/src/main/java/org/apache/spark/sql/sources/v2/FilterV2.java

@@ -0,0 +1,37 @@
+/*
+ * 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.sources.v2;
+
+import org.apache.spark.annotation.Experimental;
+import org.apache.spark.sql.catalog.v2.expressions.NamedReference;
+
+@Experimental
+public abstract class FilterV2 {
+  /**
+   * Returns list of columns that are referenced by this filter.
+   */
+  abstract NamedReference[] references();
+
+  protected NamedReference[] findReferences(Object valve) {
+    if (valve instanceof FilterV2) {
+      return ((FilterV2) valve).references();
+    } else {
+      return new NamedReference[0];
+    }
+  }
+}

sql/catalyst/src/main/java/org/apache/spark/sql/sources/v2/reader/SupportsPushDownFilters.java

@@ -18,7 +18,7 @@
 package org.apache.spark.sql.sources.v2.reader;
 
 import org.apache.spark.annotation.Evolving;
-import org.apache.spark.sql.sources.Filter;
+import org.apache.spark.sql.sources.v2.FilterV2;
 
 /**
  * A mix-in interface for {@link ScanBuilder}. Data sources can implement this interface to
@@ -33,10 +33,10 @@ public interface SupportsPushDownFilters extends ScanBuilder {
    * 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.
    */
-  Filter[] pushFilters(Filter[] filters);
+  FilterV2[] pushFilters(FilterV2[] filters);
 
   /**
-   * Returns the filters that are pushed to the data source via {@link #pushFilters(Filter[])}.
+   * Returns the filters that are pushed to the data source via {@link #pushFilters(FilterV2[])}.
    *
    * There are 3 kinds of filters:
    *  1. pushable filters which don't need to be evaluated again after scanning.
@@ -45,8 +45,8 @@ public interface SupportsPushDownFilters extends ScanBuilder {
    *  3. non-pushable filters.
    * Both case 1 and 2 should be considered as pushed filters and should be returned by this method.
    *
-   * It's possible that there is no filters in the query and {@link #pushFilters(Filter[])}
+   * It's possible that there is no filters in the query and {@link #pushFilters(FilterV2[])}
    * is never called, empty array should be returned for this case.
    */
-  Filter[] pushedFilters();
+  FilterV2[] pushedFilters();
 }

sql/catalyst/src/main/java/org/apache/spark/sql/sources/v2/writer/SupportsOverwrite.java

@@ -17,8 +17,8 @@
 
 package org.apache.spark.sql.sources.v2.writer;
 
-import org.apache.spark.sql.sources.AlwaysTrue$;
-import org.apache.spark.sql.sources.Filter;
+import org.apache.spark.sql.sources.v2.AlwaysTrue$;
+import org.apache.spark.sql.sources.v2.FilterV2;
 
 /**
  * Write builder trait for tables that support overwrite by filter.
@@ -36,10 +36,10 @@ public interface SupportsOverwrite extends WriteBuilder, SupportsTruncate {
    * @param filters filters used to match data to overwrite
    * @return this write builder for method chaining
    */
-  WriteBuilder overwrite(Filter[] filters);
+  WriteBuilder overwrite(FilterV2[] filters);
 
   @Override
   default WriteBuilder truncate() {
-    return overwrite(new Filter[] { AlwaysTrue$.MODULE$ });
+    return overwrite(new FilterV2[] { AlwaysTrue$.MODULE$ });
   }
 }

sql/catalyst/src/main/scala/org/apache/spark/sql/sources/filters.scala

@@ -18,6 +18,9 @@
 package org.apache.spark.sql.sources
 
 import org.apache.spark.annotation.{Evolving, Stable}
+import org.apache.spark.sql.catalog.v2.expressions.FieldReference
+import org.apache.spark.sql.catalog.v2.expressions.NamedReference
+import org.apache.spark.sql.sources.v2.FilterV2
 
 ////////////////////////////////////////////////////////////////////////////////////////////////////
 // This file defines all the filters that we can push down to the data sources.
@@ -40,6 +43,17 @@ abstract class Filter {
     case f: Filter => f.references
     case _ => Array.empty
   }
+
+  private[sql] def toV2: Option[FilterV2] = None
+
+  private[sql] def attToNamedRef(attribute: String): NamedReference = {
+    FieldReference(Seq(attribute))
+  }
+
+  private[sql] def valueConverter(value: Any): Any = value match {
+    case f: Filter => f.toV2
+    case _ => value
+  }
 }
 
 /**
@@ -51,6 +65,8 @@ abstract class Filter {
 @Stable
 case class EqualTo(attribute: String, value: Any) extends Filter {
   override def references: Array[String] = Array(attribute) ++ findReferences(value)
+
+  override private[sql] def toV2 = Some(v2.EqualTo(attToNamedRef(attribute), valueConverter(value)))
 }
 
 /**
@@ -63,6 +79,9 @@ 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 private[sql] def toV2 =
+    Some(v2.EqualNullSafe(attToNamedRef(attribute), valueConverter(value)))
 }
 
 /**
@@ -74,6 +93,9 @@ 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 private[sql] def toV2 =
+    Some(v2.GreaterThan(attToNamedRef(attribute), valueConverter(value)))
 }
 
 /**
@@ -85,6 +107,9 @@ 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 private[sql] def toV2 =
+    Some(v2.GreaterThanOrEqual(attToNamedRef(attribute), valueConverter(value)))
 }
 
 /**
@@ -96,6 +121,9 @@ 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 private[sql] def toV2 =
+    Some(v2.LessThan(attToNamedRef(attribute), valueConverter(value)))
 }
 
 /**
@@ -107,6 +135,9 @@ 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 private[sql] def toV2 =
+    Some(v2.LessThanOrEqual(attToNamedRef(attribute), valueConverter(value)))
 }
 
 /**
@@ -134,6 +165,9 @@ case class In(attribute: String, values: Array[Any]) extends Filter {
   }
 
   override def references: Array[String] = Array(attribute) ++ values.flatMap(findReferences)
+
+  override private[sql] def toV2 =
+    Some(v2.In(attToNamedRef(attribute), values.map(valueConverter)))
 }
 
 /**
@@ -144,6 +178,8 @@ 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 private[sql] def toV2 = Some(v2.IsNull(attToNamedRef(attribute)))
 }
 
 /**
@@ -154,6 +190,8 @@ case class IsNull(attribute: String) extends Filter {
 @Stable
 case class IsNotNull(attribute: String) extends Filter {
   override def references: Array[String] = Array(attribute)
+
+  override private[sql] def toV2 = Some(v2.IsNotNull(attToNamedRef(attribute)))
 }
 
 /**
@@ -164,6 +202,11 @@ 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 private[sql] def toV2 = (left.toV2, right.toV2) match {
+    case (Some(leftV2), Some(rightV2)) => Some(v2.And(leftV2, rightV2))
+    case _ => None
+  }
 }
 
 /**
@@ -174,6 +217,11 @@ 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 private[sql] def toV2 = (left.toV2, right.toV2) match {
+    case (Some(leftV2), Some(rightV2)) => Some(v2.Or(leftV2, rightV2))
+    case _ => None
+  }
 }
 
 /**
@@ -184,6 +232,11 @@ 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 private[sql] def toV2 = child.toV2 match {
+    case Some(childV2) => Some(v2.Not(childV2))
+    case _ => None
+  }
 }
 
 /**
@@ -195,6 +248,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 private[sql] def toV2 = Some(v2.StringStartsWith(attToNamedRef(attribute), value))
 }
 
 /**
@@ -206,6 +261,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 private[sql] def toV2 = Some(v2.StringEndsWith(attToNamedRef(attribute), value))
 }
 
 /**
@@ -217,6 +274,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 private[sql] def toV2 = Some(v2.StringContains(attToNamedRef(attribute), value))
 }
 
 /**
@@ -225,6 +284,8 @@ case class StringContains(attribute: String, value: String) extends Filter {
 @Evolving
 case class AlwaysTrue() extends Filter {
   override def references: Array[String] = Array.empty
+
+  override private[sql] def toV2 = Some(v2.AlwaysTrue)
 }
 
 @Evolving
@@ -237,6 +298,8 @@ object AlwaysTrue extends AlwaysTrue {
 @Evolving
 case class AlwaysFalse() extends Filter {
   override def references: Array[String] = Array.empty
+
+  override private[sql] def toV2 = Some(v2.AlwaysFalse)
 }
 
 @Evolving