From 22cb5db831723d38b58fe3e154e9b8c13bbe6625 Mon Sep 17 00:00:00 2001 From: JingsongLi Date: Wed, 16 Sep 2026 18:12:28 +0800 Subject: [PATCH 1/5] [spark] Support more expressions in V2 filter pushdown --- .../paimon/fileindex/FileIndexReader.java | 5 + .../paimon/globalindex/GlobalIndexReader.java | 6 + .../predicate/DateExtractTransform.java | 145 ++++++++++++++++++ .../apache/paimon/predicate/DayTransform.java | 57 +++++++ .../paimon/predicate/FunctionVisitor.java | 4 + .../paimon/predicate/HourTransform.java | 57 +++++++ .../apache/paimon/predicate/LeafFunction.java | 1 + .../paimon/predicate/LengthTransform.java | 129 ++++++++++++++++ .../org/apache/paimon/predicate/Like.java | 2 +- .../paimon/predicate/MinuteTransform.java | 57 +++++++ .../paimon/predicate/MonthTransform.java | 57 +++++++ .../org/apache/paimon/predicate/NotLike.java | 65 ++++++++ .../OnlyPartitionKeyEqualVisitor.java | 5 + .../paimon/predicate/PredicateBuilder.java | 8 + .../paimon/predicate/SecondTransform.java | 57 +++++++ .../apache/paimon/predicate/Transform.java | 7 + .../paimon/predicate/YearTransform.java | 57 +++++++ .../predicate/DateExtractTransformTest.java | 133 ++++++++++++++++ .../paimon/predicate/LengthTransformTest.java | 82 ++++++++++ .../predicate/PredicateJsonSerdeTest.java | 13 +- .../paimon/predicate/PredicateTest.java | 28 ++++ .../predicate/TransformJsonSerdeTest.java | 40 ++++- .../filter/OrcPredicateFunctionVisitor.java | 7 + .../orc/filter/OrcFilterConverterTest.java | 15 ++ .../format/parquet/ParquetFiltersTest.java | 15 ++ .../paimon/spark/SparkV2FilterConverter.scala | 9 ++ .../spark/util/SparkExpressionConverter.scala | 38 ++++- .../sql/SparkV2FilterConverterTestBase.scala | 102 +++++++++++- 28 files changed, 1190 insertions(+), 11 deletions(-) create mode 100644 paimon-common/src/main/java/org/apache/paimon/predicate/DateExtractTransform.java create mode 100644 paimon-common/src/main/java/org/apache/paimon/predicate/DayTransform.java create mode 100644 paimon-common/src/main/java/org/apache/paimon/predicate/HourTransform.java create mode 100644 paimon-common/src/main/java/org/apache/paimon/predicate/LengthTransform.java create mode 100644 paimon-common/src/main/java/org/apache/paimon/predicate/MinuteTransform.java create mode 100644 paimon-common/src/main/java/org/apache/paimon/predicate/MonthTransform.java create mode 100644 paimon-common/src/main/java/org/apache/paimon/predicate/NotLike.java create mode 100644 paimon-common/src/main/java/org/apache/paimon/predicate/SecondTransform.java create mode 100644 paimon-common/src/main/java/org/apache/paimon/predicate/YearTransform.java create mode 100644 paimon-common/src/test/java/org/apache/paimon/predicate/DateExtractTransformTest.java create mode 100644 paimon-common/src/test/java/org/apache/paimon/predicate/LengthTransformTest.java 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 b67dcaefd4d0..de9d13a3ed2c 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 @@ -83,6 +83,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; 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 f1098cf68e41..2e874173e492 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 @@ -59,6 +59,12 @@ default CompletableFuture> visitArrayContainsAll( return CompletableFuture.completedFuture(Optional.empty()); } + @Override + default CompletableFuture> visitNotLike( + FieldRef fieldRef, Object literal) { + return CompletableFuture.completedFuture(Optional.empty()); + } + @Override default CompletableFuture> visitBetween( FieldRef fieldRef, Object from, Object to) { 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 000000000000..70dfa9e0409d --- /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}. + * + *

{@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 tryCreate( + FieldRef fieldRef, Function 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 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 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 000000000000..9b7e60aa84a3 --- /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 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/FunctionVisitor.java b/paimon-common/src/main/java/org/apache/paimon/predicate/FunctionVisitor.java index 6ffa97cee284..ca0423ad675d 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 @@ -80,6 +80,10 @@ default T visitArrayContainsAll(FieldRef fieldRef, List 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); 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 000000000000..86edcd344b72 --- /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 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 77326abd32f3..20cf2064fddc 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 @@ -67,6 +67,7 @@ private static Map createRegistry() { 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 000000000000..b4fc6e057220 --- /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}. + * + *

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 inputs; + + @JsonCreator + public LengthTransform( + @JsonProperty(StringTransform.FIELD_INPUTS) + @JsonDeserialize(contentUsing = StringTransform.InputDeserializer.class) + List 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 inputs() { + return inputs; + } + + @JsonGetter(StringTransform.FIELD_INPUTS) + public List 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 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 29281fe3b488..8737fb8afe4b 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 boolean test( @Override public Optional 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 000000000000..431231abb90a --- /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 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 000000000000..cd4a37458f5e --- /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 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/NotLike.java b/paimon-common/src/main/java/org/apache/paimon/predicate/NotLike.java new file mode 100644 index 000000000000..c9c6cf376d00 --- /dev/null +++ b/paimon-common/src/main/java/org/apache/paimon/predicate/NotLike.java @@ -0,0 +1,65 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.paimon.predicate; + +import org.apache.paimon.types.DataType; + +import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonCreator; + +import java.util.List; +import java.util.Optional; + +/** A {@link LeafBinaryFunction} to evaluate {@code field not like pattern}. */ +public class NotLike extends LeafBinaryFunction { + + private static final long serialVersionUID = 1L; + + public static final String NAME = "NOT_LIKE"; + + public static final NotLike INSTANCE = new NotLike(); + + @JsonCreator + private NotLike() {} + + @Override + public boolean test(DataType type, Object field, Object patternLiteral) { + return !Like.INSTANCE.test(type, field, patternLiteral); + } + + @Override + public boolean test( + DataType type, long rowCount, Object min, Object max, Long nullCount, Object literal) { + return true; + } + + @Override + public Optional negate() { + return Optional.of(Like.INSTANCE); + } + + @Override + public T visit(FunctionVisitor visitor, FieldRef fieldRef, List literals) { + return visitor.visitNotLike(fieldRef, literals.get(0)); + } + + @Override + public String toJson() { + return NAME; + } +} 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 fb6e6e1c85ff..4d46606ff803 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 @@ -87,6 +87,11 @@ public Boolean visitLike(FieldRef fieldRef, Object literal) { 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 04e813cdcff9..f576e48fc102 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 Predicate like(Transform transform, Object patternLiteral) { 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 000000000000..ca676dd43a33 --- /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 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/Transform.java b/paimon-common/src/main/java/org/apache/paimon/predicate/Transform.java index ad01afcfb7be..dffa541397c6 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 @@ @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 000000000000..aa9460e09e78 --- /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 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 000000000000..0a472c78e474 --- /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 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 000000000000..9abf4cc3eda0 --- /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 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 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 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 4a62f0008e76..964324517064 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 @@ private static Stream testData() { .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( 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 9585bb7199de..1b048f437e18 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 @@ -622,6 +622,34 @@ public void testLikeSingleCharacterWildcardMatchesLineTerminators() { .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 3e6b97a2a228..f2a711e01e15 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 @@ private static Stream testData() { 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 @@ private static Stream testData() { .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 @@ private static Stream testData() { .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 f72478c8a109..243e05259faf 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 @@ -104,6 +104,13 @@ public Optional visitLike(FieldRef fieldRef, Object litera return Optional.empty(); } + @Override + public Optional 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 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 40ceaaa0b3d8..03b3e88727a6 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 void testIsNaN() { .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 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 fd677c3849cd..b1119bc345e9 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 @@ 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 41e5c47c4a21..6f5d215130d9 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 00898c98ca53..49dce0bcb50a 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.{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 => @@ -77,12 +87,26 @@ object SparkExpressionConverter { i => new TrimTransform(i, TrimTransform.Flag.TRAILING)) case _ => None } - case c: Cast => - c.expression() match { - case n: NamedReference => - CastTransform.tryCreate( - toPaimonFieldRef(n, rowType), - SparkTypeUtils.toPaimonType(c.dataType())) + // 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) + 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/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 a865bce9ec34..53291e4e1bb4 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 @@ -380,6 +380,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 From 6d14883023e642a9d3fd24eff42d6909bcd3bb26 Mon Sep 17 00:00:00 2001 From: JingsongLi Date: Wed, 16 Sep 2026 18:34:50 +0800 Subject: [PATCH 2/5] [spark] Push down NOT LIKE predicates --- .../paimon/fileindex/FileIndexReader.java | 15 ++++ .../paimon/globalindex/GlobalIndexReader.java | 18 +++++ .../org/apache/paimon/predicate/Contains.java | 2 +- .../org/apache/paimon/predicate/EndsWith.java | 2 +- .../paimon/predicate/FunctionVisitor.java | 12 ++++ .../apache/paimon/predicate/LeafFunction.java | 3 + .../apache/paimon/predicate/NotContains.java | 70 +++++++++++++++++++ .../apache/paimon/predicate/NotEndsWith.java | 70 +++++++++++++++++++ .../paimon/predicate/NotStartsWith.java | 70 +++++++++++++++++++ .../OnlyPartitionKeyEqualVisitor.java | 15 ++++ .../apache/paimon/predicate/StartsWith.java | 2 +- .../predicate/PredicateJsonSerdeTest.java | 18 +++++ .../paimon/predicate/PredicateTest.java | 32 +++++++++ .../filter/OrcPredicateFunctionVisitor.java | 15 ++++ .../sql/SparkV2FilterConverterTestBase.scala | 23 ++++++ 15 files changed, 364 insertions(+), 3 deletions(-) create mode 100644 paimon-common/src/main/java/org/apache/paimon/predicate/NotContains.java create mode 100644 paimon-common/src/main/java/org/apache/paimon/predicate/NotEndsWith.java create mode 100644 paimon-common/src/main/java/org/apache/paimon/predicate/NotStartsWith.java 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 de9d13a3ed2c..5dba86fbbb99 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 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; 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 2e874173e492..c6901c176b89 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 @@ default CompletableFuture> visitIsNaN(FieldRef field return CompletableFuture.completedFuture(Optional.empty()); } + @Override + default CompletableFuture> visitNotStartsWith( + FieldRef fieldRef, Object literal) { + return CompletableFuture.completedFuture(Optional.empty()); + } + + @Override + default CompletableFuture> visitNotEndsWith( + FieldRef fieldRef, Object literal) { + return CompletableFuture.completedFuture(Optional.empty()); + } + + @Override + default CompletableFuture> visitNotContains( + FieldRef fieldRef, Object literal) { + return CompletableFuture.completedFuture(Optional.empty()); + } + @Override default CompletableFuture> visitArrayContains( FieldRef fieldRef, Object literal) { 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 aa4c0d4c2189..9cf65896a5e5 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 boolean test( @Override public Optional negate() { - return Optional.empty(); + return Optional.of(NotContains.INSTANCE); } @Override 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 2757ade2d526..676e1ffbbb5f 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 boolean test( @Override public Optional 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 ca0423ad675d..a8b0440bbfb2 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 @@ 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(); } 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 20cf2064fddc..5c88bb220828 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,8 +61,11 @@ private static Map createRegistry() { 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); diff --git a/paimon-common/src/main/java/org/apache/paimon/predicate/NotContains.java b/paimon-common/src/main/java/org/apache/paimon/predicate/NotContains.java new file mode 100644 index 000000000000..add19d04e1a8 --- /dev/null +++ b/paimon-common/src/main/java/org/apache/paimon/predicate/NotContains.java @@ -0,0 +1,70 @@ +/* + * 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.types.DataType; + +import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonCreator; + +import java.util.List; +import java.util.Optional; + +/** A {@link LeafBinaryFunction} to evaluate {@code NOT CONTAINS(field, literal)}. */ +public class NotContains extends LeafBinaryFunction { + + private static final long serialVersionUID = 1L; + + public static final String NAME = "NOT_CONTAINS"; + + public static final NotContains INSTANCE = new NotContains(); + + @JsonCreator + private NotContains() {} + + @Override + public boolean test(DataType type, Object field, Object patternLiteral) { + return !Contains.INSTANCE.test(type, field, patternLiteral); + } + + @Override + public boolean test( + DataType type, + long rowCount, + Object min, + Object max, + Long nullCount, + Object patternLiteral) { + return true; + } + + @Override + public Optional negate() { + return Optional.of(Contains.INSTANCE); + } + + @Override + public T visit(FunctionVisitor visitor, FieldRef fieldRef, List literals) { + return visitor.visitNotContains(fieldRef, literals.get(0)); + } + + @Override + public String toJson() { + return NAME; + } +} diff --git a/paimon-common/src/main/java/org/apache/paimon/predicate/NotEndsWith.java b/paimon-common/src/main/java/org/apache/paimon/predicate/NotEndsWith.java new file mode 100644 index 000000000000..98e37959e828 --- /dev/null +++ b/paimon-common/src/main/java/org/apache/paimon/predicate/NotEndsWith.java @@ -0,0 +1,70 @@ +/* + * 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.types.DataType; + +import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonCreator; + +import java.util.List; +import java.util.Optional; + +/** A {@link LeafBinaryFunction} to evaluate {@code NOT ENDS_WITH(field, literal)}. */ +public class NotEndsWith extends LeafBinaryFunction { + + private static final long serialVersionUID = 1L; + + public static final String NAME = "NOT_ENDS_WITH"; + + public static final NotEndsWith INSTANCE = new NotEndsWith(); + + @JsonCreator + private NotEndsWith() {} + + @Override + public boolean test(DataType type, Object field, Object patternLiteral) { + return !EndsWith.INSTANCE.test(type, field, patternLiteral); + } + + @Override + public boolean test( + DataType type, + long rowCount, + Object min, + Object max, + Long nullCount, + Object patternLiteral) { + return true; + } + + @Override + public Optional negate() { + return Optional.of(EndsWith.INSTANCE); + } + + @Override + public T visit(FunctionVisitor visitor, FieldRef fieldRef, List literals) { + return visitor.visitNotEndsWith(fieldRef, literals.get(0)); + } + + @Override + public String toJson() { + return NAME; + } +} diff --git a/paimon-common/src/main/java/org/apache/paimon/predicate/NotStartsWith.java b/paimon-common/src/main/java/org/apache/paimon/predicate/NotStartsWith.java new file mode 100644 index 000000000000..6817f5a3ec79 --- /dev/null +++ b/paimon-common/src/main/java/org/apache/paimon/predicate/NotStartsWith.java @@ -0,0 +1,70 @@ +/* + * 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.types.DataType; + +import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonCreator; + +import java.util.List; +import java.util.Optional; + +/** A {@link LeafBinaryFunction} to evaluate {@code NOT STARTS_WITH(field, literal)}. */ +public class NotStartsWith extends LeafBinaryFunction { + + private static final long serialVersionUID = 1L; + + public static final String NAME = "NOT_STARTS_WITH"; + + public static final NotStartsWith INSTANCE = new NotStartsWith(); + + @JsonCreator + private NotStartsWith() {} + + @Override + public boolean test(DataType type, Object field, Object patternLiteral) { + return !StartsWith.INSTANCE.test(type, field, patternLiteral); + } + + @Override + public boolean test( + DataType type, + long rowCount, + Object min, + Object max, + Long nullCount, + Object patternLiteral) { + return true; + } + + @Override + public Optional negate() { + return Optional.of(StartsWith.INSTANCE); + } + + @Override + public T visit(FunctionVisitor visitor, FieldRef fieldRef, List literals) { + return visitor.visitNotStartsWith(fieldRef, literals.get(0)); + } + + @Override + public String toJson() { + return NAME; + } +} 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 4d46606ff803..57bd3bd5055d 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 Boolean visitStartsWith(FieldRef fieldRef, Object literal) { 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; 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 26b9689006ca..ba855e85d7c6 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 boolean test( @Override public Optional negate() { - return Optional.empty(); + return Optional.of(NotStartsWith.INSTANCE); } @Override 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 964324517064..a0ff9592e216 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 @@ -189,6 +189,24 @@ private static Stream testData() { .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 1b048f437e18..f87ba4095ab3 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 void testEndsWith() { assertThat(predicate2.test(10, min, max, new GenericArray(nullCount))).isEqualTo(true); } + @Test + public void testNegatedStringPredicates() { + PredicateBuilder builder = new PredicateBuilder(RowType.of(new VarCharType())); + List 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())); 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 243e05259faf..3ee59e2c50de 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 Optional visitStartsWith(FieldRef fieldRef, Object return Optional.empty(); } + @Override + public Optional visitNotStartsWith(FieldRef fieldRef, Object literal) { + return Optional.empty(); + } + @Override public Optional visitEndsWith(FieldRef fieldRef, Object literal) { return Optional.empty(); } + @Override + public Optional visitNotEndsWith(FieldRef fieldRef, Object literal) { + return Optional.empty(); + } + @Override public Optional visitContains(FieldRef fieldRef, Object literal) { return Optional.empty(); } + @Override + public Optional visitNotContains(FieldRef fieldRef, Object literal) { + return Optional.empty(); + } + @Override public Optional visitArrayContains(FieldRef fieldRef, Object literal) { return Optional.empty(); 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 53291e4e1bb4..5055ec6b6d2e 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 @@ -512,6 +512,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() From 345260b08a4f270fcbad805ebb469e2f897ca67f Mon Sep 17 00:00:00 2001 From: JingsongLi Date: Wed, 16 Sep 2026 19:39:22 +0800 Subject: [PATCH 3/5] [spark] Avoid timestamp extract pushdown with legacy mapping --- .../spark/util/SparkExpressionConverter.scala | 25 +++++++++++------ .../spark/SparkFilterConverterTest.java | 17 +++++++---- .../sql/SparkV2FilterConverterTestBase.scala | 28 +++++++++++++++++++ 3 files changed, 56 insertions(+), 14 deletions(-) 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 49dce0bcb50a..639fb7e0d90e 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 @@ -96,14 +96,23 @@ object SparkExpressionConverter { extract.source() match { case n: NamedReference => val fieldRef = toPaimonFieldRef(n, rowType) - 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 + 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 } 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 8b5457c9dff6..2e5cec362e9f 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 void testEqualToNaN() { @Test public void testIgnoreFailure() { List 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 5055ec6b6d2e..7a87bd066649 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 @@ -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 From cac3962ec111d7ec1baca50ff5682fe1dbd700a9 Mon Sep 17 00:00:00 2001 From: JingsongLi Date: Wed, 16 Sep 2026 21:17:04 +0800 Subject: [PATCH 4/5] [spark] Fix CAST pushdown expectation --- .../org/apache/paimon/spark/sql/PaimonPushDownTestBase.scala | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PaimonPushDownTestBase.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PaimonPushDownTestBase.scala index 38ec748709bc..e4ca39ca18d9 100644 --- a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PaimonPushDownTestBase.scala +++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PaimonPushDownTestBase.scala @@ -298,7 +298,7 @@ abstract class PaimonPushDownTestBase extends PaimonSparkTestBase with AdaptiveS } } - test(s"Paimon push down: apply CAST") { + test(s"Paimon push down: do not apply CAST") { if (gteqSpark3_4) { withSparkSQLConf("spark.sql.ansi.enabled" -> "true") { withTable("t") { @@ -314,7 +314,7 @@ abstract class PaimonPushDownTestBase extends PaimonSparkTestBase with AdaptiveS |""".stripMargin) val q = "SELECT * FROM t WHERE dt = 1" - assert(!checkFilterExists(q)) + assert(checkFilterExists(q)) checkAnswer(sql(q), Seq(Row(1, 100, "1"))) } } From c89f119cb979bb030552578294b30c5c70739fb8 Mon Sep 17 00:00:00 2001 From: JingsongLi Date: Wed, 16 Sep 2026 21:52:00 +0800 Subject: [PATCH 5/5] [spark] Restore CAST filter pushdown --- .../paimon/spark/util/SparkExpressionConverter.scala | 10 +++++++++- .../paimon/spark/sql/PaimonPushDownTestBase.scala | 4 ++-- 2 files changed, 11 insertions(+), 3 deletions(-) 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 639fb7e0d90e..b90f4902c461 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.{Expression, Extract, 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._ @@ -87,6 +87,14 @@ object SparkExpressionConverter { i => new TrimTransform(i, TrimTransform.Flag.TRAILING)) case _ => None } + case c: Cast => + c.expression() match { + case n: NamedReference => + CastTransform.tryCreate( + toPaimonFieldRef(n, rowType), + 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. diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PaimonPushDownTestBase.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PaimonPushDownTestBase.scala index e4ca39ca18d9..38ec748709bc 100644 --- a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PaimonPushDownTestBase.scala +++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PaimonPushDownTestBase.scala @@ -298,7 +298,7 @@ abstract class PaimonPushDownTestBase extends PaimonSparkTestBase with AdaptiveS } } - test(s"Paimon push down: do not apply CAST") { + test(s"Paimon push down: apply CAST") { if (gteqSpark3_4) { withSparkSQLConf("spark.sql.ansi.enabled" -> "true") { withTable("t") { @@ -314,7 +314,7 @@ abstract class PaimonPushDownTestBase extends PaimonSparkTestBase with AdaptiveS |""".stripMargin) val q = "SELECT * FROM t WHERE dt = 1" - assert(checkFilterExists(q)) + assert(!checkFilterExists(q)) checkAnswer(sql(q), Seq(Row(1, 100, "1"))) } }