[spark][core] Support more expressions in V2 filter pushdown

paimon-common/src/main/java/org/apache/paimon/fileindex/FileIndexReader.java

@@ -53,16 +53,31 @@ public FileIndexResult visitStartsWith(FieldRef fieldRef, Object literal) {
         return REMAIN;
     }
 
+    @Override
+    public FileIndexResult visitNotStartsWith(FieldRef fieldRef, Object literal) {
+        return REMAIN;
+    }
+
     @Override
     public FileIndexResult visitEndsWith(FieldRef fieldRef, Object literal) {
         return REMAIN;
     }
 
+    @Override
+    public FileIndexResult visitNotEndsWith(FieldRef fieldRef, Object literal) {
+        return REMAIN;
+    }
+
     @Override
     public FileIndexResult visitContains(FieldRef fieldRef, Object literal) {
         return REMAIN;
     }
 
+    @Override
+    public FileIndexResult visitNotContains(FieldRef fieldRef, Object literal) {
+        return REMAIN;
+    }
+
     @Override
     public FileIndexResult visitArrayContains(FieldRef fieldRef, Object literal) {
         return REMAIN;
@@ -83,6 +98,11 @@ public FileIndexResult visitLike(FieldRef fieldRef, Object literal) {
         return REMAIN;
     }
 
+    @Override
+    public FileIndexResult visitNotLike(FieldRef fieldRef, Object literal) {
+        return REMAIN;
+    }
+
     @Override
     public FileIndexResult visitLessThan(FieldRef fieldRef, Object literal) {
         return REMAIN;

paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexReader.java

@@ -41,6 +41,24 @@ default CompletableFuture<Optional<GlobalIndexResult>> visitIsNaN(FieldRef field
         return CompletableFuture.completedFuture(Optional.empty());
     }
 
+    @Override
+    default CompletableFuture<Optional<GlobalIndexResult>> visitNotStartsWith(
+            FieldRef fieldRef, Object literal) {
+        return CompletableFuture.completedFuture(Optional.empty());
+    }
+
+    @Override
+    default CompletableFuture<Optional<GlobalIndexResult>> visitNotEndsWith(
+            FieldRef fieldRef, Object literal) {
+        return CompletableFuture.completedFuture(Optional.empty());
+    }
+
+    @Override
+    default CompletableFuture<Optional<GlobalIndexResult>> visitNotContains(
+            FieldRef fieldRef, Object literal) {
+        return CompletableFuture.completedFuture(Optional.empty());
+    }
+
     @Override
     default CompletableFuture<Optional<GlobalIndexResult>> visitArrayContains(
             FieldRef fieldRef, Object literal) {
@@ -59,6 +77,12 @@ default CompletableFuture<Optional<GlobalIndexResult>> visitArrayContainsAll(
         return CompletableFuture.completedFuture(Optional.empty());
     }
 
+    @Override
+    default CompletableFuture<Optional<GlobalIndexResult>> visitNotLike(
+            FieldRef fieldRef, Object literal) {
+        return CompletableFuture.completedFuture(Optional.empty());
+    }
+
     @Override
     default CompletableFuture<Optional<GlobalIndexResult>> visitBetween(
             FieldRef fieldRef, Object from, Object to) {

paimon-common/src/main/java/org/apache/paimon/predicate/Contains.java

@@ -55,7 +55,7 @@ public boolean test(
 
     @Override
     public Optional<LeafFunction> negate() {
-        return Optional.empty();
+        return Optional.of(NotContains.INSTANCE);
     }
 
     @Override

paimon-common/src/main/java/org/apache/paimon/predicate/DateExtractTransform.java

@@ -0,0 +1,145 @@
+/*
+ * 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.paimon.predicate;
+
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.data.Timestamp;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DataTypes;
+
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonGetter;
+
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.function.Function;
+
+import static org.apache.paimon.utils.InternalRowUtils.get;
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+import static org.apache.paimon.utils.Preconditions.checkNotNull;
+
+/**
+ * Base {@link Transform} that extracts a calendar field from a {@code DATE} or {@code TIMESTAMP}
+ * field, like SQL {@code EXTRACT(YEAR FROM d)} and the {@code year}, {@code month}, ... functions.
+ * See the subclasses {@link YearTransform}, {@link MonthTransform}, {@link DayTransform}, {@link
+ * HourTransform}, {@link MinuteTransform} and {@link SecondTransform}.
+ *
+ * <p>{@code TIMESTAMP WITH LOCAL TIME ZONE} is deliberately not supported: extracting a calendar
+ * field from it depends on a session time zone, which the reader does not know, so such a predicate
+ * must stay in the query engine.
+ */
+public abstract class DateExtractTransform implements Transform {
+
+    private static final long serialVersionUID = 1L;
+
+    public static final String FIELD_FIELD_REF = "fieldRef";
+
+    private final FieldRef fieldRef;
+
+    protected DateExtractTransform(FieldRef fieldRef) {
+        this.fieldRef = checkNotNull(fieldRef, "fieldRef must not be null");
+        checkArgument(
+                supported(fieldRef.type()),
+                "%s requires a DATE or TIMESTAMP field, found %s",
+                name(),
+                fieldRef.type());
+    }
+
+    /** Creates a transform if {@code fieldRef} is a DATE or TIMESTAMP field. */
+    protected static Optional<Transform> tryCreate(
+            FieldRef fieldRef, Function<FieldRef, Transform> factory) {
+        if (fieldRef == null || !supported(fieldRef.type())) {
+            return Optional.empty();
+        }
+        return Optional.of(factory.apply(fieldRef));
+    }
+
+    private static boolean supported(DataType type) {
+        switch (type.getTypeRoot()) {
+            case DATE:
+            case TIMESTAMP_WITHOUT_TIME_ZONE:
+                return true;
+            default:
+                return false;
+        }
+    }
+
+    @JsonGetter(FIELD_FIELD_REF)
+    public FieldRef fieldRef() {
+        return fieldRef;
+    }
+
+    @Override
+    public List<Object> inputs() {
+        return Collections.singletonList(fieldRef);
+    }
+
+    @Override
+    public DataType outputType() {
+        return DataTypes.INT();
+    }
+
+    @Override
+    public Object transform(InternalRow row) {
+        Object value = get(row, fieldRef.index(), fieldRef.type());
+        if (value == null) {
+            return null;
+        }
+        LocalDateTime dateTime;
+        if (value instanceof Timestamp) {
+            dateTime = ((Timestamp) value).toLocalDateTime();
+        } else {
+            dateTime = LocalDate.ofEpochDay((Integer) value).atStartOfDay();
+        }
+        return extract(dateTime);
+    }
+
+    /** Extracts the calendar field this transform stands for. */
+    protected abstract Integer extract(LocalDateTime dateTime);
+
+    @Override
+    public Transform copyWithNewInputs(List<Object> inputs) {
+        checkArgument(inputs.size() == 1, "%s requires exactly one input", name());
+        return copy((FieldRef) inputs.get(0));
+    }
+
+    /** Rebuilds this transform over a new field reference. */
+    protected abstract Transform copy(FieldRef fieldRef);
+
+    @Override
+    public boolean equals(Object o) {
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        return Objects.equals(fieldRef, ((DateExtractTransform) o).fieldRef);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hashCode(fieldRef);
+    }
+
+    @Override
+    public String toString() {
+        return name() + "(" + fieldRef + ")";
+    }
+}

paimon-common/src/main/java/org/apache/paimon/predicate/DayTransform.java

@@ -0,0 +1,57 @@
+/*
+ * 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.paimon.predicate;
+
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+
+import java.time.LocalDateTime;
+import java.util.Optional;
+
+/** Extracts the day of month, like SQL {@code EXTRACT(DAY FROM d)}. */
+public class DayTransform extends DateExtractTransform {
+
+    private static final long serialVersionUID = 1L;
+
+    public static final String NAME = "DAY";
+
+    @JsonCreator
+    public DayTransform(@JsonProperty(DateExtractTransform.FIELD_FIELD_REF) FieldRef fieldRef) {
+        super(fieldRef);
+    }
+
+    public static Optional<Transform> tryCreate(FieldRef fieldRef) {
+        return DateExtractTransform.tryCreate(fieldRef, DayTransform::new);
+    }
+
+    @Override
+    public String name() {
+        return NAME;
+    }
+
+    @Override
+    protected Integer extract(LocalDateTime dateTime) {
+        return dateTime.getDayOfMonth();
+    }
+
+    @Override
+    protected Transform copy(FieldRef fieldRef) {
+        return new DayTransform(fieldRef);
+    }
+}

paimon-common/src/main/java/org/apache/paimon/predicate/EndsWith.java

@@ -55,7 +55,7 @@ public boolean test(
 
     @Override
     public Optional<LeafFunction> negate() {
-        return Optional.empty();
+        return Optional.of(NotEndsWith.INSTANCE);
     }
 
     @Override

paimon-common/src/main/java/org/apache/paimon/predicate/FunctionVisitor.java

@@ -62,10 +62,22 @@ default T visitIsNaN(FieldRef fieldRef) {
 
     T visitStartsWith(FieldRef fieldRef, Object literal);
 
+    default T visitNotStartsWith(FieldRef fieldRef, Object literal) {
+        throw new UnsupportedOperationException();
+    }
+
     T visitEndsWith(FieldRef fieldRef, Object literal);
 
+    default T visitNotEndsWith(FieldRef fieldRef, Object literal) {
+        throw new UnsupportedOperationException();
+    }
+
     T visitContains(FieldRef fieldRef, Object literal);
 
+    default T visitNotContains(FieldRef fieldRef, Object literal) {
+        throw new UnsupportedOperationException();
+    }
+
     default T visitArrayContains(FieldRef fieldRef, Object literal) {
         throw new UnsupportedOperationException();
     }
@@ -80,6 +92,10 @@ default T visitArrayContainsAll(FieldRef fieldRef, List<Object> literals) {
 
     T visitLike(FieldRef fieldRef, Object literal);
 
+    default T visitNotLike(FieldRef fieldRef, Object literal) {
+        throw new UnsupportedOperationException();
+    }
+
     T visitLessThan(FieldRef fieldRef, Object literal);
 
     T visitGreaterOrEqual(FieldRef fieldRef, Object literal);

paimon-common/src/main/java/org/apache/paimon/predicate/HourTransform.java

@@ -0,0 +1,57 @@
+/*
+ * 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.paimon.predicate;
+
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+
+import java.time.LocalDateTime;
+import java.util.Optional;
+
+/** Extracts the hour of day, like SQL {@code EXTRACT(HOUR FROM t)}. */
+public class HourTransform extends DateExtractTransform {
+
+    private static final long serialVersionUID = 1L;
+
+    public static final String NAME = "HOUR";
+
+    @JsonCreator
+    public HourTransform(@JsonProperty(DateExtractTransform.FIELD_FIELD_REF) FieldRef fieldRef) {
+        super(fieldRef);
+    }
+
+    public static Optional<Transform> tryCreate(FieldRef fieldRef) {
+        return DateExtractTransform.tryCreate(fieldRef, HourTransform::new);
+    }
+
+    @Override
+    public String name() {
+        return NAME;
+    }
+
+    @Override
+    protected Integer extract(LocalDateTime dateTime) {
+        return dateTime.getHour();
+    }
+
+    @Override
+    protected Transform copy(FieldRef fieldRef) {
+        return new HourTransform(fieldRef);
+    }
+}