This is an automated email from the ASF dual-hosted git repository. XiaoHongbo-Hope pushed a commit to branch release-2.1 in repository https://gitbox.apache.org/repos/asf/paimon.git
commit ce81e89b9d390c09bc72701abf0d9a08bdcf3b35 Author: Jingsong Lee <[email protected]> AuthorDate: Wed Sep 16 23:15:17 2026 +0800 [spark][core] Support more expressions in V2 filter pushdown (#9891) --- .../apache/paimon/fileindex/FileIndexReader.java | 20 +++ .../paimon/globalindex/GlobalIndexReader.java | 24 ++++ .../java/org/apache/paimon/predicate/Contains.java | 2 +- .../paimon/predicate/DateExtractTransform.java | 145 +++++++++++++++++++ .../org/apache/paimon/predicate/DayTransform.java | 57 ++++++++ .../java/org/apache/paimon/predicate/EndsWith.java | 2 +- .../apache/paimon/predicate/FunctionVisitor.java | 16 +++ .../org/apache/paimon/predicate/HourTransform.java | 57 ++++++++ .../org/apache/paimon/predicate/LeafFunction.java | 4 + .../apache/paimon/predicate/LengthTransform.java | 129 +++++++++++++++++ .../java/org/apache/paimon/predicate/Like.java | 2 +- .../apache/paimon/predicate/MinuteTransform.java | 57 ++++++++ .../apache/paimon/predicate/MonthTransform.java | 57 ++++++++ .../predicate/{Contains.java => NotContains.java} | 20 +-- .../predicate/{Contains.java => NotEndsWith.java} | 20 +-- .../predicate/{Contains.java => NotLike.java} | 27 ++-- .../{Contains.java => NotStartsWith.java} | 20 +-- .../predicate/OnlyPartitionKeyEqualVisitor.java | 20 +++ .../apache/paimon/predicate/PredicateBuilder.java | 8 ++ .../apache/paimon/predicate/SecondTransform.java | 57 ++++++++ .../org/apache/paimon/predicate/StartsWith.java | 2 +- .../org/apache/paimon/predicate/Transform.java | 7 + .../org/apache/paimon/predicate/YearTransform.java | 57 ++++++++ .../paimon/predicate/DateExtractTransformTest.java | 133 ++++++++++++++++++ .../paimon/predicate/LengthTransformTest.java | 82 +++++++++++ .../paimon/predicate/PredicateJsonSerdeTest.java | 31 ++++- .../org/apache/paimon/predicate/PredicateTest.java | 60 ++++++++ .../paimon/predicate/TransformJsonSerdeTest.java | 40 +++++- .../orc/filter/OrcPredicateFunctionVisitor.java | 22 +++ .../format/orc/filter/OrcFilterConverterTest.java | 15 ++ .../paimon/format/parquet/ParquetFiltersTest.java | 15 ++ .../paimon/spark/SparkV2FilterConverter.scala | 9 ++ .../spark/util/SparkExpressionConverter.scala | 43 +++++- .../paimon/spark/SparkFilterConverterTest.java | 17 ++- .../spark/sql/SparkV2FilterConverterTestBase.scala | 153 ++++++++++++++++++++- 35 files changed, 1370 insertions(+), 60 deletions(-) diff --git a/paimon-common/src/main/java/org/apache/paimon/fileindex/FileIndexReader.java b/paimon-common/src/main/java/org/apache/paimon/fileindex/FileIndexReader.java index b67dcaefd4..5dba86fbbb 100644 --- a/paimon-common/src/main/java/org/apache/paimon/fileindex/FileIndexReader.java +++ b/paimon-common/src/main/java/org/apache/paimon/fileindex/FileIndexReader.java @@ -53,16 +53,31 @@ public abstract class FileIndexReader implements FunctionVisitor<FileIndexResult 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 abstract class FileIndexReader implements FunctionVisitor<FileIndexResult return REMAIN; } + @Override + public FileIndexResult visitNotLike(FieldRef fieldRef, Object literal) { + return REMAIN; + } + @Override public FileIndexResult visitLessThan(FieldRef fieldRef, Object literal) { return REMAIN; diff --git a/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexReader.java b/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexReader.java index f1098cf68e..c6901c176b 100644 --- a/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexReader.java +++ b/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexReader.java @@ -41,6 +41,24 @@ public interface GlobalIndexReader 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 @@ public interface GlobalIndexReader 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) { diff --git a/paimon-common/src/main/java/org/apache/paimon/predicate/Contains.java b/paimon-common/src/main/java/org/apache/paimon/predicate/Contains.java index aa4c0d4c21..9cf65896a5 100644 --- a/paimon-common/src/main/java/org/apache/paimon/predicate/Contains.java +++ b/paimon-common/src/main/java/org/apache/paimon/predicate/Contains.java @@ -55,7 +55,7 @@ public class Contains extends LeafBinaryFunction { @Override public Optional<LeafFunction> negate() { - return Optional.empty(); + return Optional.of(NotContains.INSTANCE); } @Override diff --git a/paimon-common/src/main/java/org/apache/paimon/predicate/DateExtractTransform.java b/paimon-common/src/main/java/org/apache/paimon/predicate/DateExtractTransform.java new file mode 100644 index 0000000000..70dfa9e040 --- /dev/null +++ b/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 + ")"; + } +} diff --git a/paimon-common/src/main/java/org/apache/paimon/predicate/DayTransform.java b/paimon-common/src/main/java/org/apache/paimon/predicate/DayTransform.java new file mode 100644 index 0000000000..9b7e60aa84 --- /dev/null +++ b/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); + } +} diff --git a/paimon-common/src/main/java/org/apache/paimon/predicate/EndsWith.java b/paimon-common/src/main/java/org/apache/paimon/predicate/EndsWith.java index 2757ade2d5..676e1ffbbb 100644 --- a/paimon-common/src/main/java/org/apache/paimon/predicate/EndsWith.java +++ b/paimon-common/src/main/java/org/apache/paimon/predicate/EndsWith.java @@ -55,7 +55,7 @@ public class EndsWith extends LeafBinaryFunction { @Override public Optional<LeafFunction> negate() { - return Optional.empty(); + return Optional.of(NotEndsWith.INSTANCE); } @Override diff --git a/paimon-common/src/main/java/org/apache/paimon/predicate/FunctionVisitor.java b/paimon-common/src/main/java/org/apache/paimon/predicate/FunctionVisitor.java index 6ffa97cee2..a8b0440bbf 100644 --- a/paimon-common/src/main/java/org/apache/paimon/predicate/FunctionVisitor.java +++ b/paimon-common/src/main/java/org/apache/paimon/predicate/FunctionVisitor.java @@ -62,10 +62,22 @@ public interface FunctionVisitor<T> extends PredicateVisitor<T> { 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 @@ public interface FunctionVisitor<T> extends PredicateVisitor<T> { 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); diff --git a/paimon-common/src/main/java/org/apache/paimon/predicate/HourTransform.java b/paimon-common/src/main/java/org/apache/paimon/predicate/HourTransform.java new file mode 100644 index 0000000000..86edcd344b --- /dev/null +++ b/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); + } +} diff --git a/paimon-common/src/main/java/org/apache/paimon/predicate/LeafFunction.java b/paimon-common/src/main/java/org/apache/paimon/predicate/LeafFunction.java index 77326abd32..5c88bb2208 100644 --- a/paimon-common/src/main/java/org/apache/paimon/predicate/LeafFunction.java +++ b/paimon-common/src/main/java/org/apache/paimon/predicate/LeafFunction.java @@ -61,12 +61,16 @@ public abstract class LeafFunction implements Serializable { registry.put(IsNull.NAME, IsNull.INSTANCE); registry.put(IsNotNull.NAME, IsNotNull.INSTANCE); registry.put(StartsWith.NAME, StartsWith.INSTANCE); + registry.put(NotStartsWith.NAME, NotStartsWith.INSTANCE); registry.put(EndsWith.NAME, EndsWith.INSTANCE); + registry.put(NotEndsWith.NAME, NotEndsWith.INSTANCE); registry.put(Contains.NAME, Contains.INSTANCE); + registry.put(NotContains.NAME, NotContains.INSTANCE); registry.put(ArrayContains.NAME, ArrayContains.INSTANCE); registry.put(ArraysOverlap.NAME, ArraysOverlap.INSTANCE); registry.put(ArrayContainsAll.NAME, ArrayContainsAll.INSTANCE); registry.put(Like.NAME, Like.INSTANCE); + registry.put(NotLike.NAME, NotLike.INSTANCE); registry.put(In.NAME, In.INSTANCE); registry.put(NotIn.NAME, NotIn.INSTANCE); registry.put(Between.NAME, Between.INSTANCE); diff --git a/paimon-common/src/main/java/org/apache/paimon/predicate/LengthTransform.java b/paimon-common/src/main/java/org/apache/paimon/predicate/LengthTransform.java new file mode 100644 index 0000000000..b4fc6e0572 --- /dev/null +++ b/paimon-common/src/main/java/org/apache/paimon/predicate/LengthTransform.java @@ -0,0 +1,129 @@ +/* + * 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.BinaryString; +import org.apache.paimon.data.InternalRow; +import org.apache.paimon.types.DataType; +import org.apache.paimon.types.DataTypes; + +import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonCreator; +import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonGetter; +import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonIgnore; +import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonProperty; +import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.annotation.JsonDeserialize; + +import java.util.List; +import java.util.Objects; + +import static org.apache.paimon.types.DataTypeFamily.CHARACTER_STRING; +import static org.apache.paimon.utils.Preconditions.checkArgument; + +/** + * Transform that returns the number of characters of a string, like SQL {@code CHAR_LENGTH}. + * + * <p>Unlike {@link StringTransform} its output is an {@code INT}, so it is a standalone {@link + * Transform}. Input and JSON conventions follow {@link StringTransform}. + */ +public class LengthTransform implements Transform { + + private static final long serialVersionUID = 1L; + + public static final String NAME = "LENGTH"; + + private final List<Object> inputs; + + @JsonCreator + public LengthTransform( + @JsonProperty(StringTransform.FIELD_INPUTS) + @JsonDeserialize(contentUsing = StringTransform.InputDeserializer.class) + List<Object> inputs) { + checkArgument(inputs.size() == 1, "LENGTH requires exactly one input"); + Object input = inputs.get(0); + if (input instanceof FieldRef) { + checkArgument( + ((FieldRef) input).type().is(CHARACTER_STRING), + "LENGTH input must be a string field"); + } else { + checkArgument( + input == null || input instanceof BinaryString, + "LENGTH input literal must be a string"); + } + this.inputs = inputs; + } + + @Override + public String name() { + return NAME; + } + + @Override + @JsonIgnore + public List<Object> inputs() { + return inputs; + } + + @JsonGetter(StringTransform.FIELD_INPUTS) + public List<Object> inputsForJson() { + return StringTransform.inputsForJson(inputs); + } + + @Override + public DataType outputType() { + return DataTypes.INT(); + } + + @Override + public Object transform(InternalRow row) { + Object input = inputs.get(0); + BinaryString value; + if (input instanceof FieldRef) { + FieldRef ref = (FieldRef) input; + int i = ref.index(); + value = row.isNullAt(i) ? null : row.getString(i); + } else { + value = (BinaryString) input; + } + return value == null ? null : value.numChars(); + } + + @Override + public Transform copyWithNewInputs(List<Object> inputs) { + return new LengthTransform(inputs); + } + + @Override + public boolean equals(Object o) { + if (o == null || getClass() != o.getClass()) { + return false; + } + LengthTransform that = (LengthTransform) o; + return Objects.equals(inputs, that.inputs); + } + + @Override + public int hashCode() { + return Objects.hashCode(inputs); + } + + @Override + public String toString() { + return StringTransform.formatCall(name(), inputs); + } +} diff --git a/paimon-common/src/main/java/org/apache/paimon/predicate/Like.java b/paimon-common/src/main/java/org/apache/paimon/predicate/Like.java index 29281fe3b4..8737fb8afe 100644 --- a/paimon-common/src/main/java/org/apache/paimon/predicate/Like.java +++ b/paimon-common/src/main/java/org/apache/paimon/predicate/Like.java @@ -148,7 +148,7 @@ public class Like extends LeafBinaryFunction { @Override public Optional<LeafFunction> negate() { - return Optional.empty(); + return Optional.of(NotLike.INSTANCE); } @Override diff --git a/paimon-common/src/main/java/org/apache/paimon/predicate/MinuteTransform.java b/paimon-common/src/main/java/org/apache/paimon/predicate/MinuteTransform.java new file mode 100644 index 0000000000..431231abb9 --- /dev/null +++ b/paimon-common/src/main/java/org/apache/paimon/predicate/MinuteTransform.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 minute of hour, like SQL {@code EXTRACT(MINUTE FROM t)}. */ +public class MinuteTransform extends DateExtractTransform { + + private static final long serialVersionUID = 1L; + + public static final String NAME = "MINUTE"; + + @JsonCreator + public MinuteTransform(@JsonProperty(DateExtractTransform.FIELD_FIELD_REF) FieldRef fieldRef) { + super(fieldRef); + } + + public static Optional<Transform> tryCreate(FieldRef fieldRef) { + return DateExtractTransform.tryCreate(fieldRef, MinuteTransform::new); + } + + @Override + public String name() { + return NAME; + } + + @Override + protected Integer extract(LocalDateTime dateTime) { + return dateTime.getMinute(); + } + + @Override + protected Transform copy(FieldRef fieldRef) { + return new MinuteTransform(fieldRef); + } +} diff --git a/paimon-common/src/main/java/org/apache/paimon/predicate/MonthTransform.java b/paimon-common/src/main/java/org/apache/paimon/predicate/MonthTransform.java new file mode 100644 index 0000000000..cd4a37458f --- /dev/null +++ b/paimon-common/src/main/java/org/apache/paimon/predicate/MonthTransform.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 month of year, like SQL {@code EXTRACT(MONTH FROM d)}. */ +public class MonthTransform extends DateExtractTransform { + + private static final long serialVersionUID = 1L; + + public static final String NAME = "MONTH"; + + @JsonCreator + public MonthTransform(@JsonProperty(DateExtractTransform.FIELD_FIELD_REF) FieldRef fieldRef) { + super(fieldRef); + } + + public static Optional<Transform> tryCreate(FieldRef fieldRef) { + return DateExtractTransform.tryCreate(fieldRef, MonthTransform::new); + } + + @Override + public String name() { + return NAME; + } + + @Override + protected Integer extract(LocalDateTime dateTime) { + return dateTime.getMonthValue(); + } + + @Override + protected Transform copy(FieldRef fieldRef) { + return new MonthTransform(fieldRef); + } +} diff --git a/paimon-common/src/main/java/org/apache/paimon/predicate/Contains.java b/paimon-common/src/main/java/org/apache/paimon/predicate/NotContains.java similarity index 75% copy from paimon-common/src/main/java/org/apache/paimon/predicate/Contains.java copy to paimon-common/src/main/java/org/apache/paimon/predicate/NotContains.java index aa4c0d4c21..add19d04e1 100644 --- a/paimon-common/src/main/java/org/apache/paimon/predicate/Contains.java +++ b/paimon-common/src/main/java/org/apache/paimon/predicate/NotContains.java @@ -18,7 +18,6 @@ package org.apache.paimon.predicate; -import org.apache.paimon.data.BinaryString; import org.apache.paimon.types.DataType; import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonCreator; @@ -26,20 +25,21 @@ import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonCre import java.util.List; import java.util.Optional; -/** A {@link LeafBinaryFunction} to evaluate {@code filter like '%abc%'}. */ -public class Contains extends LeafBinaryFunction { +/** A {@link LeafBinaryFunction} to evaluate {@code NOT CONTAINS(field, literal)}. */ +public class NotContains extends LeafBinaryFunction { - public static final String NAME = "CONTAINS"; + private static final long serialVersionUID = 1L; - public static final Contains INSTANCE = new Contains(); + public static final String NAME = "NOT_CONTAINS"; + + public static final NotContains INSTANCE = new NotContains(); @JsonCreator - private Contains() {} + private NotContains() {} @Override public boolean test(DataType type, Object field, Object patternLiteral) { - BinaryString fieldString = (BinaryString) field; - return fieldString.contains((BinaryString) patternLiteral); + return !Contains.INSTANCE.test(type, field, patternLiteral); } @Override @@ -55,12 +55,12 @@ public class Contains extends LeafBinaryFunction { @Override public Optional<LeafFunction> negate() { - return Optional.empty(); + return Optional.of(Contains.INSTANCE); } @Override public <T> T visit(FunctionVisitor<T> visitor, FieldRef fieldRef, List<Object> literals) { - return visitor.visitContains(fieldRef, literals.get(0)); + return visitor.visitNotContains(fieldRef, literals.get(0)); } @Override diff --git a/paimon-common/src/main/java/org/apache/paimon/predicate/Contains.java b/paimon-common/src/main/java/org/apache/paimon/predicate/NotEndsWith.java similarity index 75% copy from paimon-common/src/main/java/org/apache/paimon/predicate/Contains.java copy to paimon-common/src/main/java/org/apache/paimon/predicate/NotEndsWith.java index aa4c0d4c21..98e37959e8 100644 --- a/paimon-common/src/main/java/org/apache/paimon/predicate/Contains.java +++ b/paimon-common/src/main/java/org/apache/paimon/predicate/NotEndsWith.java @@ -18,7 +18,6 @@ package org.apache.paimon.predicate; -import org.apache.paimon.data.BinaryString; import org.apache.paimon.types.DataType; import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonCreator; @@ -26,20 +25,21 @@ import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonCre import java.util.List; import java.util.Optional; -/** A {@link LeafBinaryFunction} to evaluate {@code filter like '%abc%'}. */ -public class Contains extends LeafBinaryFunction { +/** A {@link LeafBinaryFunction} to evaluate {@code NOT ENDS_WITH(field, literal)}. */ +public class NotEndsWith extends LeafBinaryFunction { - public static final String NAME = "CONTAINS"; + private static final long serialVersionUID = 1L; - public static final Contains INSTANCE = new Contains(); + public static final String NAME = "NOT_ENDS_WITH"; + + public static final NotEndsWith INSTANCE = new NotEndsWith(); @JsonCreator - private Contains() {} + private NotEndsWith() {} @Override public boolean test(DataType type, Object field, Object patternLiteral) { - BinaryString fieldString = (BinaryString) field; - return fieldString.contains((BinaryString) patternLiteral); + return !EndsWith.INSTANCE.test(type, field, patternLiteral); } @Override @@ -55,12 +55,12 @@ public class Contains extends LeafBinaryFunction { @Override public Optional<LeafFunction> negate() { - return Optional.empty(); + return Optional.of(EndsWith.INSTANCE); } @Override public <T> T visit(FunctionVisitor<T> visitor, FieldRef fieldRef, List<Object> literals) { - return visitor.visitContains(fieldRef, literals.get(0)); + return visitor.visitNotEndsWith(fieldRef, literals.get(0)); } @Override diff --git a/paimon-common/src/main/java/org/apache/paimon/predicate/Contains.java b/paimon-common/src/main/java/org/apache/paimon/predicate/NotLike.java similarity index 67% copy from paimon-common/src/main/java/org/apache/paimon/predicate/Contains.java copy to paimon-common/src/main/java/org/apache/paimon/predicate/NotLike.java index aa4c0d4c21..c9c6cf376d 100644 --- a/paimon-common/src/main/java/org/apache/paimon/predicate/Contains.java +++ b/paimon-common/src/main/java/org/apache/paimon/predicate/NotLike.java @@ -18,7 +18,6 @@ package org.apache.paimon.predicate; -import org.apache.paimon.data.BinaryString; import org.apache.paimon.types.DataType; import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonCreator; @@ -26,41 +25,37 @@ import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonCre import java.util.List; import java.util.Optional; -/** A {@link LeafBinaryFunction} to evaluate {@code filter like '%abc%'}. */ -public class Contains extends LeafBinaryFunction { +/** A {@link LeafBinaryFunction} to evaluate {@code field not like pattern}. */ +public class NotLike extends LeafBinaryFunction { - public static final String NAME = "CONTAINS"; + private static final long serialVersionUID = 1L; - public static final Contains INSTANCE = new Contains(); + public static final String NAME = "NOT_LIKE"; + + public static final NotLike INSTANCE = new NotLike(); @JsonCreator - private Contains() {} + private NotLike() {} @Override public boolean test(DataType type, Object field, Object patternLiteral) { - BinaryString fieldString = (BinaryString) field; - return fieldString.contains((BinaryString) patternLiteral); + return !Like.INSTANCE.test(type, field, patternLiteral); } @Override public boolean test( - DataType type, - long rowCount, - Object min, - Object max, - Long nullCount, - Object patternLiteral) { + DataType type, long rowCount, Object min, Object max, Long nullCount, Object literal) { return true; } @Override public Optional<LeafFunction> negate() { - return Optional.empty(); + return Optional.of(Like.INSTANCE); } @Override public <T> T visit(FunctionVisitor<T> visitor, FieldRef fieldRef, List<Object> literals) { - return visitor.visitContains(fieldRef, literals.get(0)); + return visitor.visitNotLike(fieldRef, literals.get(0)); } @Override diff --git a/paimon-common/src/main/java/org/apache/paimon/predicate/Contains.java b/paimon-common/src/main/java/org/apache/paimon/predicate/NotStartsWith.java similarity index 74% copy from paimon-common/src/main/java/org/apache/paimon/predicate/Contains.java copy to paimon-common/src/main/java/org/apache/paimon/predicate/NotStartsWith.java index aa4c0d4c21..6817f5a3ec 100644 --- a/paimon-common/src/main/java/org/apache/paimon/predicate/Contains.java +++ b/paimon-common/src/main/java/org/apache/paimon/predicate/NotStartsWith.java @@ -18,7 +18,6 @@ package org.apache.paimon.predicate; -import org.apache.paimon.data.BinaryString; import org.apache.paimon.types.DataType; import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonCreator; @@ -26,20 +25,21 @@ import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonCre import java.util.List; import java.util.Optional; -/** A {@link LeafBinaryFunction} to evaluate {@code filter like '%abc%'}. */ -public class Contains extends LeafBinaryFunction { +/** A {@link LeafBinaryFunction} to evaluate {@code NOT STARTS_WITH(field, literal)}. */ +public class NotStartsWith extends LeafBinaryFunction { - public static final String NAME = "CONTAINS"; + private static final long serialVersionUID = 1L; - public static final Contains INSTANCE = new Contains(); + public static final String NAME = "NOT_STARTS_WITH"; + + public static final NotStartsWith INSTANCE = new NotStartsWith(); @JsonCreator - private Contains() {} + private NotStartsWith() {} @Override public boolean test(DataType type, Object field, Object patternLiteral) { - BinaryString fieldString = (BinaryString) field; - return fieldString.contains((BinaryString) patternLiteral); + return !StartsWith.INSTANCE.test(type, field, patternLiteral); } @Override @@ -55,12 +55,12 @@ public class Contains extends LeafBinaryFunction { @Override public Optional<LeafFunction> negate() { - return Optional.empty(); + return Optional.of(StartsWith.INSTANCE); } @Override public <T> T visit(FunctionVisitor<T> visitor, FieldRef fieldRef, List<Object> literals) { - return visitor.visitContains(fieldRef, literals.get(0)); + return visitor.visitNotStartsWith(fieldRef, literals.get(0)); } @Override diff --git a/paimon-common/src/main/java/org/apache/paimon/predicate/OnlyPartitionKeyEqualVisitor.java b/paimon-common/src/main/java/org/apache/paimon/predicate/OnlyPartitionKeyEqualVisitor.java index fb6e6e1c85..57bd3bd505 100644 --- a/paimon-common/src/main/java/org/apache/paimon/predicate/OnlyPartitionKeyEqualVisitor.java +++ b/paimon-common/src/main/java/org/apache/paimon/predicate/OnlyPartitionKeyEqualVisitor.java @@ -57,16 +57,31 @@ public class OnlyPartitionKeyEqualVisitor implements FunctionVisitor<Boolean> { return false; } + @Override + public Boolean visitNotStartsWith(FieldRef fieldRef, Object literal) { + return false; + } + @Override public Boolean visitEndsWith(FieldRef fieldRef, Object literal) { return false; } + @Override + public Boolean visitNotEndsWith(FieldRef fieldRef, Object literal) { + return false; + } + @Override public Boolean visitContains(FieldRef fieldRef, Object literal) { return false; } + @Override + public Boolean visitNotContains(FieldRef fieldRef, Object literal) { + return false; + } + @Override public Boolean visitArrayContains(FieldRef fieldRef, Object literal) { return false; @@ -87,6 +102,11 @@ public class OnlyPartitionKeyEqualVisitor implements FunctionVisitor<Boolean> { return false; } + @Override + public Boolean visitNotLike(FieldRef fieldRef, Object literal) { + return false; + } + @Override public Boolean visitLessThan(FieldRef fieldRef, Object literal) { return false; diff --git a/paimon-common/src/main/java/org/apache/paimon/predicate/PredicateBuilder.java b/paimon-common/src/main/java/org/apache/paimon/predicate/PredicateBuilder.java index 04e813cdcf..f576e48fc1 100644 --- a/paimon-common/src/main/java/org/apache/paimon/predicate/PredicateBuilder.java +++ b/paimon-common/src/main/java/org/apache/paimon/predicate/PredicateBuilder.java @@ -217,6 +217,14 @@ public class PredicateBuilder { return leaf(optimized.getKey(), transform, optimized.getValue()); } + public Predicate notLike(int idx, Object patternLiteral) { + return leaf(NotLike.INSTANCE, idx, patternLiteral); + } + + public Predicate notLike(Transform transform, Object patternLiteral) { + return leaf(NotLike.INSTANCE, transform, patternLiteral); + } + private Predicate leaf(LeafFunction function, int idx, Object literal) { DataField field = rowType.getFields().get(idx); return new LeafPredicate(function, field.type(), idx, field.name(), singletonList(literal)); diff --git a/paimon-common/src/main/java/org/apache/paimon/predicate/SecondTransform.java b/paimon-common/src/main/java/org/apache/paimon/predicate/SecondTransform.java new file mode 100644 index 0000000000..ca676dd43a --- /dev/null +++ b/paimon-common/src/main/java/org/apache/paimon/predicate/SecondTransform.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 second of minute, like SQL {@code EXTRACT(SECOND FROM t)}. */ +public class SecondTransform extends DateExtractTransform { + + private static final long serialVersionUID = 1L; + + public static final String NAME = "SECOND"; + + @JsonCreator + public SecondTransform(@JsonProperty(DateExtractTransform.FIELD_FIELD_REF) FieldRef fieldRef) { + super(fieldRef); + } + + public static Optional<Transform> tryCreate(FieldRef fieldRef) { + return DateExtractTransform.tryCreate(fieldRef, SecondTransform::new); + } + + @Override + public String name() { + return NAME; + } + + @Override + protected Integer extract(LocalDateTime dateTime) { + return dateTime.getSecond(); + } + + @Override + protected Transform copy(FieldRef fieldRef) { + return new SecondTransform(fieldRef); + } +} diff --git a/paimon-common/src/main/java/org/apache/paimon/predicate/StartsWith.java b/paimon-common/src/main/java/org/apache/paimon/predicate/StartsWith.java index 26b9689006..ba855e85d7 100644 --- a/paimon-common/src/main/java/org/apache/paimon/predicate/StartsWith.java +++ b/paimon-common/src/main/java/org/apache/paimon/predicate/StartsWith.java @@ -59,7 +59,7 @@ public class StartsWith extends LeafBinaryFunction { @Override public Optional<LeafFunction> negate() { - return Optional.empty(); + return Optional.of(NotStartsWith.INSTANCE); } @Override diff --git a/paimon-common/src/main/java/org/apache/paimon/predicate/Transform.java b/paimon-common/src/main/java/org/apache/paimon/predicate/Transform.java index ad01afcfb7..dffa541397 100644 --- a/paimon-common/src/main/java/org/apache/paimon/predicate/Transform.java +++ b/paimon-common/src/main/java/org/apache/paimon/predicate/Transform.java @@ -37,10 +37,17 @@ import java.util.List; @JsonSubTypes.Type(value = CastTransform.class, name = CastTransform.NAME), @JsonSubTypes.Type(value = ConcatTransform.class, name = ConcatTransform.NAME), @JsonSubTypes.Type(value = ConcatWsTransform.class, name = ConcatWsTransform.NAME), + @JsonSubTypes.Type(value = YearTransform.class, name = YearTransform.NAME), + @JsonSubTypes.Type(value = MonthTransform.class, name = MonthTransform.NAME), + @JsonSubTypes.Type(value = DayTransform.class, name = DayTransform.NAME), + @JsonSubTypes.Type(value = HourTransform.class, name = HourTransform.NAME), + @JsonSubTypes.Type(value = MinuteTransform.class, name = MinuteTransform.NAME), + @JsonSubTypes.Type(value = SecondTransform.class, name = SecondTransform.NAME), @JsonSubTypes.Type(value = UpperTransform.class, name = UpperTransform.NAME), @JsonSubTypes.Type(value = LowerTransform.class, name = LowerTransform.NAME), @JsonSubTypes.Type(value = SubstringTransform.class, name = SubstringTransform.NAME), @JsonSubTypes.Type(value = TrimTransform.class, name = TrimTransform.NAME), + @JsonSubTypes.Type(value = LengthTransform.class, name = LengthTransform.NAME), @JsonSubTypes.Type(value = NullTransform.class, name = NullTransform.NAME) }) public interface Transform extends Serializable { diff --git a/paimon-common/src/main/java/org/apache/paimon/predicate/YearTransform.java b/paimon-common/src/main/java/org/apache/paimon/predicate/YearTransform.java new file mode 100644 index 0000000000..aa9460e09e --- /dev/null +++ b/paimon-common/src/main/java/org/apache/paimon/predicate/YearTransform.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 year, like SQL {@code EXTRACT(YEAR FROM d)}. */ +public class YearTransform extends DateExtractTransform { + + private static final long serialVersionUID = 1L; + + public static final String NAME = "YEAR"; + + @JsonCreator + public YearTransform(@JsonProperty(DateExtractTransform.FIELD_FIELD_REF) FieldRef fieldRef) { + super(fieldRef); + } + + public static Optional<Transform> tryCreate(FieldRef fieldRef) { + return DateExtractTransform.tryCreate(fieldRef, YearTransform::new); + } + + @Override + public String name() { + return NAME; + } + + @Override + protected Integer extract(LocalDateTime dateTime) { + return dateTime.getYear(); + } + + @Override + protected Transform copy(FieldRef fieldRef) { + return new YearTransform(fieldRef); + } +} diff --git a/paimon-common/src/test/java/org/apache/paimon/predicate/DateExtractTransformTest.java b/paimon-common/src/test/java/org/apache/paimon/predicate/DateExtractTransformTest.java new file mode 100644 index 0000000000..0a472c78e4 --- /dev/null +++ b/paimon-common/src/test/java/org/apache/paimon/predicate/DateExtractTransformTest.java @@ -0,0 +1,133 @@ +/* + * 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.GenericRow; +import org.apache.paimon.data.Timestamp; +import org.apache.paimon.types.DataType; +import org.apache.paimon.types.DataTypes; +import org.apache.paimon.types.RowType; + +import org.junit.jupiter.api.Test; + +import java.time.LocalDate; +import java.time.LocalDateTime; +import java.util.Collections; +import java.util.Optional; +import java.util.function.Function; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +class DateExtractTransformTest { + + private static int epochDay(int year, int month, int day) { + return (int) LocalDate.of(year, month, day).toEpochDay(); + } + + private static Object extract( + DataType type, Function<FieldRef, DateExtractTransform> factory, Object value) { + return factory.apply(new FieldRef(0, "f0", type)).transform(GenericRow.of(value)); + } + + @Test + public void testExtractFromDate() { + int value = epochDay(2023, 7, 15); + assertThat(extract(DataTypes.DATE(), YearTransform::new, value)).isEqualTo(2023); + assertThat(extract(DataTypes.DATE(), MonthTransform::new, value)).isEqualTo(7); + assertThat(extract(DataTypes.DATE(), DayTransform::new, value)).isEqualTo(15); + } + + @Test + public void testExtractFromTimestamp() { + Timestamp value = Timestamp.fromLocalDateTime(LocalDateTime.of(2024, 3, 5, 14, 30, 45)); + assertThat(extract(DataTypes.TIMESTAMP(3), YearTransform::new, value)).isEqualTo(2024); + assertThat(extract(DataTypes.TIMESTAMP(3), MonthTransform::new, value)).isEqualTo(3); + assertThat(extract(DataTypes.TIMESTAMP(3), DayTransform::new, value)).isEqualTo(5); + assertThat(extract(DataTypes.TIMESTAMP(3), HourTransform::new, value)).isEqualTo(14); + assertThat(extract(DataTypes.TIMESTAMP(3), MinuteTransform::new, value)).isEqualTo(30); + assertThat(extract(DataTypes.TIMESTAMP(3), SecondTransform::new, value)).isEqualTo(45); + } + + @Test + public void testTimeFieldsOfDateAreZero() { + int value = epochDay(2024, 1, 1); + assertThat(extract(DataTypes.DATE(), HourTransform::new, value)).isEqualTo(0); + assertThat(extract(DataTypes.DATE(), MinuteTransform::new, value)).isEqualTo(0); + assertThat(extract(DataTypes.DATE(), SecondTransform::new, value)).isEqualTo(0); + } + + @Test + public void testNullYieldsNull() { + DateExtractTransform transform = new YearTransform(new FieldRef(0, "d0", DataTypes.DATE())); + assertThat(transform.transform(GenericRow.of((Object) null))).isNull(); + } + + @Test + public void testOutputTypeIsInt() { + DateExtractTransform transform = + new HourTransform(new FieldRef(0, "t0", DataTypes.TIMESTAMP(6))); + assertThat(transform.outputType()).isEqualTo(DataTypes.INT()); + } + + @Test + public void testExtractPredicateFiltersRows() { + PredicateBuilder builder = + new PredicateBuilder(RowType.of(DataTypes.DATE(), DataTypes.TIMESTAMP(3))); + Predicate predicate = + builder.equal(new YearTransform(new FieldRef(0, "d0", DataTypes.DATE())), 2023); + + assertThat(predicate.test(GenericRow.of(epochDay(2023, 7, 15), null))).isTrue(); + assertThat(predicate.test(GenericRow.of(epochDay(2024, 1, 1), null))).isFalse(); + assertThat(predicate.test(GenericRow.of((Object) null, null))).isFalse(); + } + + @Test + public void testCopyWithNewInputsRemapsField() { + DateExtractTransform transform = + new MonthTransform(new FieldRef(0, "d0", DataTypes.DATE())); + DateExtractTransform copied = + (DateExtractTransform) + transform.copyWithNewInputs( + Collections.singletonList(new FieldRef(1, "d1", DataTypes.DATE()))); + assertThat(copied.transform(GenericRow.of((Object) null, epochDay(2024, 5, 20)))) + .isEqualTo(5); + } + + @Test + public void testUnsupportedFieldTypeIsRejected() { + for (DataType type : + new DataType[] {DataTypes.STRING(), DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE(3)}) { + FieldRef fieldRef = new FieldRef(0, "f0", type); + assertThatThrownBy(() -> new YearTransform(fieldRef)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("YEAR requires a DATE or TIMESTAMP field, found " + type); + assertThat(YearTransform.tryCreate(fieldRef)).isEqualTo(Optional.empty()); + } + + FieldRef dateRef = new FieldRef(0, "d0", DataTypes.DATE()); + assertThat(YearTransform.tryCreate(dateRef).isPresent()).isTrue(); + } + + @Test + public void testToString() { + DateExtractTransform transform = new YearTransform(new FieldRef(0, "d0", DataTypes.DATE())); + assertThat(transform.toString()).isEqualTo("YEAR(d0)"); + } +} diff --git a/paimon-common/src/test/java/org/apache/paimon/predicate/LengthTransformTest.java b/paimon-common/src/test/java/org/apache/paimon/predicate/LengthTransformTest.java new file mode 100644 index 0000000000..9abf4cc3ed --- /dev/null +++ b/paimon-common/src/test/java/org/apache/paimon/predicate/LengthTransformTest.java @@ -0,0 +1,82 @@ +/* + * 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.BinaryString; +import org.apache.paimon.data.GenericRow; +import org.apache.paimon.types.DataTypes; + +import org.junit.jupiter.api.Test; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +class LengthTransformTest { + + @Test + public void testLiteralInput() { + List<Object> inputs = new ArrayList<>(); + inputs.add(BinaryString.fromString("hello")); + LengthTransform transform = new LengthTransform(inputs); + assertThat(transform.transform(GenericRow.of())).isEqualTo(5); + assertThat(transform.outputType()).isEqualTo(DataTypes.INT()); + } + + @Test + public void testFieldInput() { + List<Object> inputs = new ArrayList<>(); + inputs.add(new FieldRef(0, "f0", DataTypes.STRING())); + LengthTransform transform = new LengthTransform(inputs); + assertThat(transform.transform(GenericRow.of(BinaryString.fromString("paimon")))) + .isEqualTo(6); + assertThat(transform.transform(GenericRow.of((Object) null))).isNull(); + } + + @Test + public void testNullInputSlot() { + LengthTransform transform = new LengthTransform(Collections.singletonList(null)); + assertThat(transform.transform(GenericRow.of())).isNull(); + } + + @Test + public void testIllegalInputs() { + // wrong arity + List<Object> two = new ArrayList<>(); + two.add(BinaryString.fromString("hello")); + two.add(BinaryString.fromString("hi")); + assertThatThrownBy(() -> new LengthTransform(two)) + .isInstanceOf(IllegalArgumentException.class); + + // non-string field + assertThatThrownBy( + () -> + new LengthTransform( + Collections.singletonList( + new FieldRef(0, "f0", DataTypes.INT())))) + .isInstanceOf(IllegalArgumentException.class); + + // non-string literal + assertThatThrownBy(() -> new LengthTransform(Collections.singletonList((Object) 5))) + .isInstanceOf(IllegalArgumentException.class); + } +} diff --git a/paimon-common/src/test/java/org/apache/paimon/predicate/PredicateJsonSerdeTest.java b/paimon-common/src/test/java/org/apache/paimon/predicate/PredicateJsonSerdeTest.java index 4a62f0008e..a0ff9592e2 100644 --- a/paimon-common/src/test/java/org/apache/paimon/predicate/PredicateJsonSerdeTest.java +++ b/paimon-common/src/test/java/org/apache/paimon/predicate/PredicateJsonSerdeTest.java @@ -158,11 +158,22 @@ class PredicateJsonSerdeTest { .expectJson( "{\"kind\":\"LEAF\",\"transform\":{\"name\":\"CONCAT_WS\",\"inputs\":[\"|\",{\"index\":1,\"name\":\"f1\",\"type\":\"STRING\"},\"X\",null,{\"index\":2,\"name\":\"f2\",\"type\":\"STRING\"}]},\"function\":\"ENDS_WITH\",\"literals\":[\"z\"]}"), - // LeafPredicate - Like (non-negatable) + // LeafPredicate - Like TestSpec.forPredicate(builder.like(2, BinaryString.fromString("%a%b%"))) .expectJson( "{\"kind\":\"LEAF\",\"transform\":{\"name\":\"FIELD_REF\",\"fieldRef\":{\"index\":2,\"name\":\"f2\",\"type\":\"STRING\"}},\"function\":\"LIKE\",\"literals\":[\"%a%b%\"]}"), + // LeafPredicate - NotLike + TestSpec.forPredicate(builder.notLike(2, BinaryString.fromString("%a%b%"))) + .expectJson( + "{\"kind\":\"LEAF\",\"transform\":{\"name\":\"FIELD_REF\",\"fieldRef\":{\"index\":2,\"name\":\"f2\",\"type\":\"STRING\"}},\"function\":\"NOT_LIKE\",\"literals\":[\"%a%b%\"]}"), + + // LeafPredicate - NotLike (negate of Like) + TestSpec.forPredicate( + builder.like(2, BinaryString.fromString("%a%b%")).negate().get()) + .expectJson( + "{\"kind\":\"LEAF\",\"transform\":{\"name\":\"FIELD_REF\",\"fieldRef\":{\"index\":2,\"name\":\"f2\",\"type\":\"STRING\"}},\"function\":\"NOT_LIKE\",\"literals\":[\"%a%b%\"]}"), + // LeafPredicate - StartsWith (field index) TestSpec.forPredicate(builder.startsWith(2, BinaryString.fromString("hello"))) .expectJson( @@ -178,6 +189,24 @@ class PredicateJsonSerdeTest { .expectJson( "{\"kind\":\"LEAF\",\"transform\":{\"name\":\"FIELD_REF\",\"fieldRef\":{\"index\":2,\"name\":\"f2\",\"type\":\"STRING\"}},\"function\":\"CONTAINS\",\"literals\":[\"foo\"]}"), + // LeafPredicate - negated string predicates + TestSpec.forPredicate( + builder.startsWith(2, BinaryString.fromString("hello")) + .negate() + .get()) + .expectJson( + "{\"kind\":\"LEAF\",\"transform\":{\"name\":\"FIELD_REF\",\"fieldRef\":{\"index\":2,\"name\":\"f2\",\"type\":\"STRING\"}},\"function\":\"NOT_STARTS_WITH\",\"literals\":[\"hello\"]}"), + TestSpec.forPredicate( + builder.endsWith(2, BinaryString.fromString("world")) + .negate() + .get()) + .expectJson( + "{\"kind\":\"LEAF\",\"transform\":{\"name\":\"FIELD_REF\",\"fieldRef\":{\"index\":2,\"name\":\"f2\",\"type\":\"STRING\"}},\"function\":\"NOT_ENDS_WITH\",\"literals\":[\"world\"]}"), + TestSpec.forPredicate( + builder.contains(2, BinaryString.fromString("foo")).negate().get()) + .expectJson( + "{\"kind\":\"LEAF\",\"transform\":{\"name\":\"FIELD_REF\",\"fieldRef\":{\"index\":2,\"name\":\"f2\",\"type\":\"STRING\"}},\"function\":\"NOT_CONTAINS\",\"literals\":[\"foo\"]}"), + // LeafPredicate - ArrayContains uses the element type for literal serde TestSpec.forPredicate(builder.arrayContains(4, BinaryString.fromString("vip"))) .expectJson( diff --git a/paimon-common/src/test/java/org/apache/paimon/predicate/PredicateTest.java b/paimon-common/src/test/java/org/apache/paimon/predicate/PredicateTest.java index 9585bb7199..f87ba4095a 100644 --- a/paimon-common/src/test/java/org/apache/paimon/predicate/PredicateTest.java +++ b/paimon-common/src/test/java/org/apache/paimon/predicate/PredicateTest.java @@ -449,6 +449,38 @@ public class PredicateTest { assertThat(predicate2.test(10, min, max, new GenericArray(nullCount))).isEqualTo(true); } + @Test + public void testNegatedStringPredicates() { + PredicateBuilder builder = new PredicateBuilder(RowType.of(new VarCharType())); + List<Predicate> positives = + Arrays.asList( + builder.startsWith(0, fromString("he")), + builder.endsWith(0, fromString("lo")), + builder.contains(0, fromString("ell"))); + + for (Predicate positive : positives) { + Predicate negative = positive.negate().get(); + assertThat(negative.test(GenericRow.of(fromString("hello")))).isFalse(); + assertThat(negative.test(GenericRow.of(fromString("world")))).isTrue(); + assertThat(negative.test(GenericRow.of((Object) null))).isFalse(); + assertThat(negative.negate()).contains(positive); + + // Negative string predicates cannot use min/max statistics, but an all-null column + // still cannot contain a matching row. + assertThat( + test( + negative, + 3, + new SimpleColStats[] { + new SimpleColStats( + fromString("hello"), fromString("world"), 0L) + })) + .isTrue(); + assertThat(test(negative, 1, new SimpleColStats[] {new SimpleColStats(null, null, 1L)})) + .isFalse(); + } + } + @Test public void testLargeIn() { PredicateBuilder builder = new PredicateBuilder(RowType.of(new IntType())); @@ -622,6 +654,34 @@ public class PredicateTest { .isTrue(); } + @Test + public void testNotLike() { + PredicateBuilder builder = new PredicateBuilder(RowType.of(new VarCharType())); + Predicate predicate = builder.notLike(0, fromString("h%")); + + assertThat(predicate.test(GenericRow.of(fromString("hello")))).isEqualTo(false); + assertThat(predicate.test(GenericRow.of(fromString("world")))).isEqualTo(true); + assertThat(predicate.test(GenericRow.of((Object) null))).isEqualTo(false); + + // unknown stats cannot prune + assertThat(test(predicate, 3, new SimpleColStats[] {new SimpleColStats(null, null, 1L)})) + .isEqualTo(true); + assertThat( + test( + predicate, + 3, + new SimpleColStats[] { + new SimpleColStats(fromString("a"), fromString("z"), 0L) + })) + .isEqualTo(true); + + // like and not like negate each other, 'a_c' cannot be optimized to starts/ends/contains + assertThat(builder.like(0, fromString("a_c")).negate().orElse(null)) + .isEqualTo(builder.notLike(0, fromString("a_c"))); + assertThat(builder.notLike(0, fromString("a_c")).negate().orElse(null)) + .isEqualTo(builder.like(0, fromString("a_c"))); + } + private boolean executeLike(String s, String pattern) { ThreadLocalRandom rnd = ThreadLocalRandom.current(); if (rnd.nextBoolean()) { diff --git a/paimon-common/src/test/java/org/apache/paimon/predicate/TransformJsonSerdeTest.java b/paimon-common/src/test/java/org/apache/paimon/predicate/TransformJsonSerdeTest.java index 3e6b97a2a2..f2a711e01e 100644 --- a/paimon-common/src/test/java/org/apache/paimon/predicate/TransformJsonSerdeTest.java +++ b/paimon-common/src/test/java/org/apache/paimon/predicate/TransformJsonSerdeTest.java @@ -122,6 +122,15 @@ class TransformJsonSerdeTest { new FieldRef(2, "f2", DataTypes.STRING())))) .expectJson( "{\"name\":\"CONCAT_WS\",\"inputs\":[\"|\",{\"index\":1,\"name\":\"f1\",\"type\":\"STRING\"},\"X\",null,{\"index\":2,\"name\":\"f2\",\"type\":\"STRING\"}]}"), + + // DateExtractTransform - YEAR on DATE, MINUTE on TIMESTAMP + TestSpec.forTransform(new YearTransform(new FieldRef(0, "d0", DataTypes.DATE()))) + .expectJson( + "{\"name\":\"YEAR\",\"fieldRef\":{\"index\":0,\"name\":\"d0\",\"type\":\"DATE\"}}"), + TestSpec.forTransform( + new MinuteTransform(new FieldRef(1, "t1", DataTypes.TIMESTAMP(3)))) + .expectJson( + "{\"name\":\"MINUTE\",\"fieldRef\":{\"index\":1,\"name\":\"t1\",\"type\":\"TIMESTAMP(3)\"}}"), TestSpec.forTransform( new SubstringTransform( Arrays.asList( @@ -171,6 +180,21 @@ class TransformJsonSerdeTest { .expectJson( "{\"name\":\"TRIM\",\"inputs\":[{\"index\":1,\"name\":\"f1\",\"type\":\"STRING\"},\"x\"],\"trimFlag\":\"TRAILING\"}"), + // LengthTransform + TestSpec.forTransform( + new LengthTransform( + Collections.singletonList( + new FieldRef(1, "f1", DataTypes.STRING())))) + .expectJson( + "{\"name\":\"LENGTH\",\"inputs\":[{\"index\":1,\"name\":\"f1\",\"type\":\"STRING\"}]}"), + TestSpec.forTransform( + new LengthTransform( + Collections.singletonList( + BinaryString.fromString("hello")))) + .expectJson("{\"name\":\"LENGTH\",\"inputs\":[\"hello\"]}"), + TestSpec.forTransform(new LengthTransform(Collections.singletonList(null))) + .expectJson("{\"name\":\"LENGTH\",\"inputs\":[null]}"), + // error message testing TestSpec.forJson("{\"name\":\"invalid\"}") .expectErrorMessage("Could not resolve type id 'invalid'"), @@ -184,7 +208,21 @@ class TransformJsonSerdeTest { .expectErrorMessage("position must be an integer"), TestSpec.forJson("{\"name\":\"SUBSTRING\",\"inputs\":[123,1,1]}") .expectErrorMessage( - "SUBSTRING source must be a string or a field reference")); + "SUBSTRING source must be a string or a field reference"), + TestSpec.forJson( + "{\"name\":\"YEAR\",\"fieldRef\":{\"index\":0,\"name\":\"f0\",\"type\":\"STRING\"}}") + .expectErrorMessage( + "YEAR requires a DATE or TIMESTAMP field, found STRING"), + TestSpec.forJson("{\"name\":\"LENGTH\",\"inputs\":[]}") + .expectErrorMessage("LENGTH requires exactly one input"), + TestSpec.forJson( + "{\"name\":\"LENGTH\",\"inputs\":[{\"index\":0,\"name\":\"f0\",\"type\":\"INT\"}]}") + .expectErrorMessage("LENGTH input must be a string field"), + TestSpec.forJson( + "{\"name\":\"LENGTH\",\"inputs\":[{\"index\":0,\"name\":\"f0\",\"type\":\"INT\"},{\"index\":1,\"name\":\"f1\",\"type\":\"STRING\"}]}") + .expectErrorMessage("LENGTH requires exactly one input"), + TestSpec.forJson("{\"name\":\"LENGTH\",\"inputs\":[5]}") + .expectErrorMessage("Unsupported StringTransform input JSON")); } @ParameterizedTest(name = "{index}: {0}") diff --git a/paimon-format/src/main/java/org/apache/paimon/format/orc/filter/OrcPredicateFunctionVisitor.java b/paimon-format/src/main/java/org/apache/paimon/format/orc/filter/OrcPredicateFunctionVisitor.java index f72478c8a1..3ee59e2c50 100644 --- a/paimon-format/src/main/java/org/apache/paimon/format/orc/filter/OrcPredicateFunctionVisitor.java +++ b/paimon-format/src/main/java/org/apache/paimon/format/orc/filter/OrcPredicateFunctionVisitor.java @@ -72,16 +72,31 @@ public class OrcPredicateFunctionVisitor return Optional.empty(); } + @Override + public Optional<OrcFilters.Predicate> visitNotStartsWith(FieldRef fieldRef, Object literal) { + return Optional.empty(); + } + @Override public Optional<OrcFilters.Predicate> visitEndsWith(FieldRef fieldRef, Object literal) { return Optional.empty(); } + @Override + public Optional<OrcFilters.Predicate> visitNotEndsWith(FieldRef fieldRef, Object literal) { + return Optional.empty(); + } + @Override public Optional<OrcFilters.Predicate> visitContains(FieldRef fieldRef, Object literal) { return Optional.empty(); } + @Override + public Optional<OrcFilters.Predicate> visitNotContains(FieldRef fieldRef, Object literal) { + return Optional.empty(); + } + @Override public Optional<OrcFilters.Predicate> visitArrayContains(FieldRef fieldRef, Object literal) { return Optional.empty(); @@ -104,6 +119,13 @@ public class OrcPredicateFunctionVisitor return Optional.empty(); } + @Override + public Optional<OrcFilters.Predicate> visitNotLike(FieldRef fieldRef, Object literal) { + // ORC SearchArgument has no not-like leaf, so skip push-down and let the engine + // evaluate the filter (consistent with the Parquet path). + return Optional.empty(); + } + @Override public Optional<OrcFilters.Predicate> visitLessThan(FieldRef fieldRef, Object literal) { return convertBinary(fieldRef, literal, OrcFilters.LessThan::new); diff --git a/paimon-format/src/test/java/org/apache/paimon/format/orc/filter/OrcFilterConverterTest.java b/paimon-format/src/test/java/org/apache/paimon/format/orc/filter/OrcFilterConverterTest.java index 40ceaaa0b3..03b3e88727 100644 --- a/paimon-format/src/test/java/org/apache/paimon/format/orc/filter/OrcFilterConverterTest.java +++ b/paimon-format/src/test/java/org/apache/paimon/format/orc/filter/OrcFilterConverterTest.java @@ -18,6 +18,7 @@ package org.apache.paimon.format.orc.filter; +import org.apache.paimon.data.BinaryString; import org.apache.paimon.data.Decimal; import org.apache.paimon.predicate.LeafPredicate; import org.apache.paimon.predicate.Predicate; @@ -253,6 +254,20 @@ public class OrcFilterConverterTest { .isEqualTo(Optional.empty()); } + @Test + public void testNotLike() { + PredicateBuilder builder = + new PredicateBuilder( + new RowType( + Collections.singletonList( + new DataField(0, "stringField", new VarCharType())))); + + assertThat( + builder.notLike(0, BinaryString.fromString("%value%")) + .visit(OrcPredicateFunctionVisitor.VISITOR)) + .isEqualTo(Optional.empty()); + } + private void test(Predicate predicate, OrcFilters.Predicate orcPredicate, boolean canPushDown) { Optional<OrcFilters.Predicate> optionalPredicate = predicate.visit(OrcPredicateFunctionVisitor.VISITOR); diff --git a/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetFiltersTest.java b/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetFiltersTest.java index fd677c3849..b1119bc345 100644 --- a/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetFiltersTest.java +++ b/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetFiltersTest.java @@ -18,6 +18,7 @@ package org.apache.paimon.format.parquet; +import org.apache.paimon.data.BinaryString; import org.apache.paimon.data.Decimal; import org.apache.paimon.data.Timestamp; import org.apache.paimon.predicate.Predicate; @@ -70,6 +71,20 @@ import static org.assertj.core.api.Assertions.assertThat; class ParquetFiltersTest { + @Test + public void testNotLikeIsNotPushedDown() { + RowType rowType = + new RowType( + Collections.singletonList(new DataField(0, "string1", new VarCharType()))); + MessageType schema = ParquetSchemaConverter.convertToParquetMessageType(rowType); + Predicate predicate = + new PredicateBuilder(rowType).notLike(0, BinaryString.fromString("%unsupported%")); + + FilterCompat.Filter filter = + ParquetFilters.convert(PredicateBuilder.splitAnd(predicate), schema, true); + assertThat(filter).isEqualTo(FilterCompat.NOOP); + } + @Test public void testBoolean() { RowType rowType = diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/SparkV2FilterConverter.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/SparkV2FilterConverter.scala index 41e5c47c4a..6f5d215130 100644 --- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/SparkV2FilterConverter.scala +++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/SparkV2FilterConverter.scala @@ -73,6 +73,14 @@ case class SparkV2FilterConverter(rowType: RowType) extends Logging { throw new UnsupportedOperationException(s"Convert $sparkPredicate is unsupported.") } + case NOT_EQUAL => + sparkPredicate match { + case BinaryPredicate(transform, literal) => + builder.notEqual(transform, literal) + case _ => + throw new UnsupportedOperationException(s"Convert $sparkPredicate is unsupported.") + } + case GREATER_THAN => sparkPredicate match { case BinaryPredicate(transform, literal) => @@ -242,6 +250,7 @@ object SparkV2FilterConverter extends Logging { private val EQUAL_TO = "=" private val EQUAL_NULL_SAFE = "<=>" + private val NOT_EQUAL = "<>" private val GREATER_THAN = ">" private val GREATER_THAN_OR_EQUAL = ">=" private val LESS_THAN = "<" diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/util/SparkExpressionConverter.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/util/SparkExpressionConverter.scala index 00898c98ca..b90f4902c4 100644 --- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/util/SparkExpressionConverter.scala +++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/util/SparkExpressionConverter.scala @@ -26,7 +26,7 @@ import org.apache.paimon.types.{DecimalType, RowType} import org.apache.paimon.types.DataTypeRoot._ import org.apache.spark.sql.catalyst.util.{ArrayData, DateTimeUtils} -import org.apache.spark.sql.connector.expressions.{Cast, Expression, GeneralScalarExpression, Literal, NamedReference} +import org.apache.spark.sql.connector.expressions.{Cast, Expression, Extract, GeneralScalarExpression, Literal, NamedReference} import org.apache.spark.sql.types.{ArrayType => SparkArrayType, DataType => SparkDataType} import scala.collection.JavaConverters._ @@ -40,10 +40,19 @@ object SparkExpressionConverter { private val UPPER = "UPPER" private val LOWER = "LOWER" private val SUBSTRING = "SUBSTRING" + private val CHAR_LENGTH = "CHAR_LENGTH" private val TRIM = "TRIM" private val LTRIM = "LTRIM" private val RTRIM = "RTRIM" + // Supported fields of the EXTRACT expression + private val EXTRACT_YEAR = "YEAR" + private val EXTRACT_MONTH = "MONTH" + private val EXTRACT_DAY = "DAY" + private val EXTRACT_HOUR = "HOUR" + private val EXTRACT_MINUTE = "MINUTE" + private val EXTRACT_SECOND = "SECOND" + /** Convert Spark [[Expression]] to Paimon [[Transform]], return None if not supported. */ def toPaimonTransform(exp: Expression, rowType: RowType): Option[Transform] = { @@ -68,6 +77,7 @@ object SparkExpressionConverter { case UPPER => convertChildren(s.children()).map(i => new UpperTransform(i)) case LOWER => convertChildren(s.children()).map(i => new LowerTransform(i)) case SUBSTRING => convertChildren(s.children()).map(i => new SubstringTransform(i)) + case CHAR_LENGTH => convertChildren(s.children()).map(i => new LengthTransform(i)) case TRIM => convertChildren(s.children()).map(i => new TrimTransform(i, TrimTransform.Flag.BOTH)) case LTRIM => @@ -85,6 +95,37 @@ object SparkExpressionConverter { SparkTypeUtils.toPaimonType(c.dataType())) case _ => None } + // The connector `Extract` expression was added in Spark 3.4 and does not exist on + // Spark 3.2/3.3 runtimes, so its type test must stay behind this version gate to avoid + // a NoClassDefFoundError when linking the class there. + case e if org.apache.spark.SPARK_VERSION >= "3.4" => + e match { + case extract: Extract => + extract.source() match { + case n: NamedReference => + val fieldRef = toPaimonFieldRef(n, rowType) + if ( + fieldRef.`type`().getTypeRoot == TIMESTAMP_WITHOUT_TIME_ZONE && + treatPaimonTimestampTypeAsSparkTimestampType() + ) { + // Legacy mapping exposes this Paimon type as Spark TIMESTAMP, whose extract + // semantics depend on the Spark session time zone. + None + } else { + extract.field() match { + case EXTRACT_YEAR => YearTransform.tryCreate(fieldRef) + case EXTRACT_MONTH => MonthTransform.tryCreate(fieldRef) + case EXTRACT_DAY => DayTransform.tryCreate(fieldRef) + case EXTRACT_HOUR => HourTransform.tryCreate(fieldRef) + case EXTRACT_MINUTE => MinuteTransform.tryCreate(fieldRef) + case EXTRACT_SECOND => SecondTransform.tryCreate(fieldRef) + case _ => None + } + } + case _ => None + } + case _ => None + } case _ => None } } diff --git a/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkFilterConverterTest.java b/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkFilterConverterTest.java index 8b5457c9df..2e5cec362e 100644 --- a/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkFilterConverterTest.java +++ b/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkFilterConverterTest.java @@ -266,15 +266,20 @@ public class SparkFilterConverterTest { @Test public void testIgnoreFailure() { List<DataField> dataFields = new ArrayList<>(); - dataFields.add(new DataField(0, "id", new IntType())); + dataFields.add(new DataField(0, "id", new FloatType())); dataFields.add(new DataField(1, "name", new VarCharType(VarCharType.MAX_LENGTH))); RowType rowType = new RowType(dataFields); SparkFilterConverter converter = new SparkFilterConverter(rowType); + PredicateBuilder builder = new PredicateBuilder(rowType); + + Not notStartsWith = Not.apply(StringStartsWith.apply("name", "paimon")); + assertThat(converter.convert(notStartsWith)) + .isEqualTo(builder.startsWith(1, fromString("paimon")).negate().get()); - Not not = Not.apply(StringStartsWith.apply("name", "paimon")); - assertThatThrownBy(() -> converter.convert(not, false)) - .hasMessageContaining("Not(StringStartsWith(name,paimon)) is unsupported."); - assertThat(converter.convert(not, true)).isNull(); - assertThat(converter.convertIgnoreFailure(not)).isNull(); + Not unsupported = Not.apply(EqualTo.apply("id", Float.NaN)); + assertThatThrownBy(() -> converter.convert(unsupported, false)) + .hasMessageContaining("is unsupported."); + assertThat(converter.convert(unsupported, true)).isNull(); + assertThat(converter.convertIgnoreFailure(unsupported)).isNull(); } } diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/SparkV2FilterConverterTestBase.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/SparkV2FilterConverterTestBase.scala index a865bce9ec..7a87bd0666 100644 --- a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/SparkV2FilterConverterTestBase.scala +++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/SparkV2FilterConverterTestBase.scala @@ -19,7 +19,7 @@ package org.apache.paimon.spark.sql import org.apache.paimon.data.{BinaryString, Decimal, Timestamp} -import org.apache.paimon.predicate.PredicateBuilder +import org.apache.paimon.predicate.{DayTransform, FieldRef, HourTransform, LengthTransform, MinuteTransform, MonthTransform, PredicateBuilder, SecondTransform, YearTransform} import org.apache.paimon.spark.{PaimonSparkTestBase, SparkV2FilterConverter} import org.apache.paimon.spark.util.shim.TypeUtils.treatPaimonTimestampTypeAsSparkTimestampType import org.apache.paimon.table.source.DataSplit @@ -90,6 +90,34 @@ abstract class SparkV2FilterConverterTestBase extends PaimonSparkTestBase { lazy val converter: SparkV2FilterConverter = SparkV2FilterConverter(rowType) + test("V2Filter: legacy timestamp mapping does not push down extract") { + if (gteqSpark3_4) { + withTimeZone("UTC") { + withSparkSQLConf( + "spark.paimon.legacy-timestamp-mapping.enabled" -> "true", + "spark.sql.session.timeZone" -> "America/Los_Angeles") { + withTable("legacy_extract") { + sql(""" + |CREATE TABLE legacy_extract (id INT, ts TIMESTAMP) + |USING paimon PARTITIONED BY (ts) + |""".stripMargin) + + val legacyRowType = loadTable("legacy_extract").rowType() + val legacyConverter = SparkV2FilterConverter(legacyRowType) + val filter = "hour(ts) = 1" + val condition = + sql(s"SELECT * FROM legacy_extract WHERE $filter").queryExecution.analyzed + .collectFirst { case f: Filter => f } + .get + .condition + val sparkPredicate = translateFilterV2(condition).get + assert(legacyConverter.convert(sparkPredicate).isEmpty) + } + } + } + } + } + test("V2Filter: all types") { var filter = "string_col = 'hello'" var actual = converter.convert(v2Filter(filter)).get @@ -380,6 +408,106 @@ abstract class SparkV2FilterConverterTestBase extends PaimonSparkTestBase { assert(scanFilesCount(filter) == 2) } + test("V2Filter: NotEqual") { + val filter = "int_col <> 1" + val actual = converter.convert(v2Filter(filter)).get + assert(actual.equals(builder.notEqual(3, 1))) + checkAnswer( + sql(s"SELECT int_col from test_tbl WHERE $filter ORDER BY int_col"), + Seq(Row(2), Row(3))) + assert(scanFilesCount(filter) == 2) + } + + test("V2Filter: CharLength") { + if (gteqSpark3_4) { + val filter = "char_length(string_col) = 2" + val transform = new LengthTransform( + List[Object](new FieldRef(0, "string_col", rowType.getTypeAt(0))).asJava) + val actual = converter.convert(v2Filter(filter)).get + assert(actual.equals(builder.equal(transform, 2))) + checkAnswer(sql(s"SELECT string_col from test_tbl WHERE $filter"), Seq(Row("hi"))) + // CHAR_LENGTH cannot prune files by column stats. + assert(scanFilesCount(filter) == 4) + } + } + + test("V2Filter: Year, Month and Day") { + if (gteqSpark3_4) { + val dateFieldRef = new FieldRef(9, "date_col", rowType.getTypeAt(9)) + + var filter = "year(date_col) = 2025" + var actual = converter.convert(v2Filter(filter)).get + assert(actual.equals(builder.equal(new YearTransform(dateFieldRef), 2025))) + checkAnswer( + sql(s"SELECT date_col from test_tbl WHERE $filter"), + sql("SELECT date_col from test_tbl")) + // Extracted fields cannot prune files by column stats. + assert(scanFilesCount(filter) == 4) + + filter = "month(date_col) = 1" + actual = converter.convert(v2Filter(filter)).get + assert(actual.equals(builder.equal(new MonthTransform(dateFieldRef), 1))) + checkAnswer( + sql(s"SELECT date_col from test_tbl WHERE $filter"), + sql("SELECT date_col from test_tbl")) + assert(scanFilesCount(filter) == 4) + + filter = "day(date_col) = 15" + actual = converter.convert(v2Filter(filter)).get + assert(actual.equals(builder.equal(new DayTransform(dateFieldRef), 15))) + checkAnswer( + sql(s"SELECT date_col from test_tbl WHERE $filter"), + sql("SELECT date('2025-01-15')")) + assert(scanFilesCount(filter) == 4) + } + } + + test("V2Filter: Hour, Minute and Second") { + if (gteqSpark3_4) { + withTable("extract_tbl", "extract_ltz_tbl") { + sql("CREATE TABLE extract_tbl (ts_col TIMESTAMP_NTZ) USING paimon") + sql("INSERT INTO extract_tbl VALUES (timestamp_ntz'2025-01-15 01:02:03')") + sql("INSERT INTO extract_tbl VALUES (timestamp_ntz'2025-01-16 04:05:06')") + + val ntzRowType = loadTable("extract_tbl").rowType() + val ntzBuilder = new PredicateBuilder(ntzRowType) + val ntzConverter = SparkV2FilterConverter(ntzRowType) + val tsFieldRef = new FieldRef(0, "ts_col", ntzRowType.getTypeAt(0)) + + var filter = "hour(ts_col) = 1" + var actual = ntzConverter.convert(v2Filter(filter, "extract_tbl")).get + assert(actual.equals(ntzBuilder.equal(new HourTransform(tsFieldRef), 1))) + checkAnswer( + sql(s"SELECT ts_col from extract_tbl WHERE $filter"), + sql("SELECT timestamp_ntz'2025-01-15 01:02:03'")) + assert(scanFilesCount(filter, "extract_tbl") == 2) + + filter = "minute(ts_col) = 2" + actual = ntzConverter.convert(v2Filter(filter, "extract_tbl")).get + assert(actual.equals(ntzBuilder.equal(new MinuteTransform(tsFieldRef), 2))) + checkAnswer( + sql(s"SELECT ts_col from extract_tbl WHERE $filter"), + sql("SELECT timestamp_ntz'2025-01-15 01:02:03'")) + assert(scanFilesCount(filter, "extract_tbl") == 2) + + filter = "second(ts_col) = 3" + actual = ntzConverter.convert(v2Filter(filter, "extract_tbl")).get + assert(actual.equals(ntzBuilder.equal(new SecondTransform(tsFieldRef), 3))) + checkAnswer( + sql(s"SELECT ts_col from extract_tbl WHERE $filter"), + sql("SELECT timestamp_ntz'2025-01-15 01:02:03'")) + assert(scanFilesCount(filter, "extract_tbl") == 2) + + // Spark TIMESTAMP maps to a Paimon local-time-zone timestamp by default, which the + // date/time extract transforms do not support, so the conversion degrades. + sql("CREATE TABLE extract_ltz_tbl (ts_col TIMESTAMP) USING paimon") + val ltzRowType = loadTable("extract_ltz_tbl").rowType() + val ltzConverter = SparkV2FilterConverter(ltzRowType) + assert(ltzConverter.convert(v2Filter("hour(ts_col) = 1", "extract_ltz_tbl")).isEmpty) + } + } + } + test("V2Filter: StartWith") { val filter = "string_col LIKE 'h%'" val actual = converter.convert(v2Filter(filter)).get @@ -412,6 +540,29 @@ abstract class SparkV2FilterConverterTestBase extends PaimonSparkTestBase { assert(scanFilesCount(filter) == 4) } + test("V2Filter: Not string predicates") { + Seq( + ( + "string_col NOT LIKE 'h%'", + builder.startsWith(0, BinaryString.fromString("h")).negate().get(), + Seq(Row("paimon"), Row("world"))), + ( + "string_col NOT LIKE '%d'", + builder.endsWith(0, BinaryString.fromString("d")).negate().get(), + Seq(Row("hello"), Row("hi"), Row("paimon"))), + ( + "string_col NOT LIKE '%orl%'", + builder.contains(0, BinaryString.fromString("orl")).negate().get(), + Seq(Row("hello"), Row("hi"), Row("paimon"))) + ).foreach { + case (filter, expectedPredicate, expectedRows) => + assert(converter.convert(v2Filter(filter)).contains(expectedPredicate)) + checkAnswer( + sql(s"SELECT string_col from test_tbl WHERE $filter ORDER BY string_col"), + expectedRows) + } + } + private def paimonAlwaysTrue: org.apache.paimon.predicate.Predicate = PredicateBuilder.alwaysTrue()
