[SPARK-36556][SQL] Add DSV2 filters

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

@@ -0,0 +1,50 @@
+/*
+ * 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 always evaluates to {@code false}.
+ *
+ * @since 3.3.0
+ */
+@Evolving
+public final class AlwaysFalse extends Filter {
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) return true;
+    if (o == null || getClass() != o.getClass()) return false;
+    return true;
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash();
+  }
+
+  @Override
+  public String toString() { return "FALSE"; }
+
+  @Override
+  public NamedReference[] references() { return EMPTY_REFERENCE; }
+}

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

@@ -0,0 +1,50 @@
+/*
+ * 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 always evaluates to {@code true}.
+ *
+ * @since 3.3.0
+ */
+@Evolving
+public final class AlwaysTrue extends Filter {
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) return true;
+    if (o == null || getClass() != o.getClass()) return false;
+    return true;
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash();
+  }
+
+  @Override
+  public String toString() { return "TRUE"; }
+
+  @Override
+  public NamedReference[] references() { return EMPTY_REFERENCE; }
+}

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

@@ -0,0 +1,39 @@
+/*
+ * 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;
+
+/**
+ * A filter 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 And(Filter left, Filter right) {
+    super(left, right);
+  }
+
+  @Override
+  public String toString() {
+    return String.format("(%s) AND (%s)", left.describe(), right.describe());
+  }
+}

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

@@ -0,0 +1,60 @@
+/*
+ * 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
+public 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 }; }
+}

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

@@ -0,0 +1,65 @@
+/*
+ * 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
+public 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;
+  }
+}

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

@@ -0,0 +1,40 @@
+/*
+ * 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(); }
+}

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

@@ -0,0 +1,39 @@
+/*
+ * 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(); }
+}

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

@@ -0,0 +1,41 @@
+/*
+ * 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.Expression;
+import org.apache.spark.sql.connector.expressions.NamedReference;
+
+/**
+ * Filter base class
+ *
+ * @since 3.3.0
+ */
+@Evolving
+public abstract class Filter implements Expression {
+
+  protected static final NamedReference[] EMPTY_REFERENCE = new NamedReference[0];
+
+  /**
+   * Returns list of columns that are referenced by this filter.
+   */
+  public abstract NamedReference[] references();
+
+  @Override
+  public String describe() { return this.toString(); }
+}