This is an automated email from the ASF dual-hosted git repository. XiaoHongbo-Hope pushed a commit to branch release-2.1 in repository https://gitbox.apache.org/repos/asf/paimon.git
commit a5fb087bdeb500e68513e5a79f2a7fd0614fb2f4 Author: Jingsong Lee <[email protected]> AuthorDate: Thu Sep 17 11:23:09 2026 +0800 [spark][core] Support more scalar transforms in V2 filter pushdown (#9905) --- .../paimon/predicate/BitLengthTransform.java | 104 +++++++++++++ .../apache/paimon/predicate/DateAddTransform.java | 105 +++++++++++++ .../apache/paimon/predicate/DateDiffTransform.java | 105 +++++++++++++ .../paimon/predicate/DateExtractTransform.java | 4 +- .../paimon/predicate/DateTruncTransform.java | 135 ++++++++++++++++ .../paimon/predicate/DayOfWeekTransform.java | 58 +++++++ .../paimon/predicate/DayOfYearTransform.java | 58 +++++++ .../paimon/predicate/IsoDayOfWeekTransform.java | 58 +++++++ .../apache/paimon/predicate/OverlayTransform.java | 124 +++++++++++++++ .../org/apache/paimon/predicate/PadTransform.java | 145 +++++++++++++++++ .../apache/paimon/predicate/QuarterTransform.java | 57 +++++++ .../org/apache/paimon/predicate/Transform.java | 14 ++ .../paimon/predicate/TransformInputUtils.java | 77 ++++++++++ .../paimon/predicate/TranslateTransform.java | 95 ++++++++++++ .../org/apache/paimon/predicate/WeekTransform.java | 58 +++++++ .../apache/paimon/predicate/WeekdayTransform.java | 57 +++++++ .../paimon/predicate/YearOfWeekTransform.java | 59 +++++++ .../paimon/predicate/DateExtractTransformTest.java | 17 ++ .../paimon/predicate/ScalarTransformTest.java | 171 +++++++++++++++++++++ .../paimon/predicate/TransformJsonSerdeTest.java | 78 ++++++++++ .../spark/util/SparkExpressionConverter.scala | 78 ++++++++++ .../spark/sql/SparkV2FilterConverterTestBase.scala | 133 +++++++++++++++- 22 files changed, 1788 insertions(+), 2 deletions(-) diff --git a/paimon-common/src/main/java/org/apache/paimon/predicate/BitLengthTransform.java b/paimon-common/src/main/java/org/apache/paimon/predicate/BitLengthTransform.java new file mode 100644 index 0000000000..08baec3242 --- /dev/null +++ b/paimon-common/src/main/java/org/apache/paimon/predicate/BitLengthTransform.java @@ -0,0 +1,104 @@ +/* + * 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.utils.Preconditions.checkArgument; + +/** Transform that returns the UTF-8 bit length of a string. */ +public class BitLengthTransform implements Transform { + + private static final long serialVersionUID = 1L; + + public static final String NAME = "BIT_LENGTH"; + + private final List<Object> inputs; + + @JsonCreator + public BitLengthTransform( + @JsonProperty(StringTransform.FIELD_INPUTS) + @JsonDeserialize(contentUsing = StringTransform.InputDeserializer.class) + List<Object> inputs) { + checkArgument(inputs.size() == 1, "BIT_LENGTH requires exactly one input"); + TransformInputUtils.checkString(inputs.get(0), "BIT_LENGTH input must be a string"); + this.inputs = inputs; + } + + @Override + public String name() { + return NAME; + } + + @Override + @JsonIgnore + public List<Object> inputs() { + return inputs; + } + + @JsonGetter(StringTransform.FIELD_INPUTS) + public List<Object> inputsForJson() { + return StringTransform.inputsForJson(inputs); + } + + @Override + public DataType outputType() { + return DataTypes.INT(); + } + + @Override + public Object transform(InternalRow row) { + BinaryString value = TransformInputUtils.string(inputs.get(0), row); + return value == null ? null : value.getSizeInBytes() * 8; + } + + @Override + public Transform copyWithNewInputs(List<Object> inputs) { + return new BitLengthTransform(inputs); + } + + @Override + public boolean equals(Object o) { + return o != null + && getClass() == o.getClass() + && Objects.equals(inputs, ((BitLengthTransform) o).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/DateAddTransform.java b/paimon-common/src/main/java/org/apache/paimon/predicate/DateAddTransform.java new file mode 100644 index 0000000000..c663e04a6d --- /dev/null +++ b/paimon-common/src/main/java/org/apache/paimon/predicate/DateAddTransform.java @@ -0,0 +1,105 @@ +/* + * 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.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.utils.Preconditions.checkArgument; + +/** SQL {@code DATE_ADD(date, days)} transform. */ +public class DateAddTransform implements Transform { + + private static final long serialVersionUID = 1L; + + public static final String NAME = "DATE_ADD"; + + private final List<Object> inputs; + + @JsonCreator + public DateAddTransform( + @JsonProperty(StringTransform.FIELD_INPUTS) + @JsonDeserialize(contentUsing = SubstringTransform.InputDeserializer.class) + List<Object> inputs) { + checkArgument(inputs.size() == 2, "DATE_ADD requires exactly two inputs"); + TransformInputUtils.checkDate(inputs.get(0), "DATE_ADD first input must be a date"); + TransformInputUtils.checkInteger(inputs.get(1), "DATE_ADD second input must be an integer"); + this.inputs = inputs; + } + + @Override + public String name() { + return NAME; + } + + @Override + @JsonIgnore + public List<Object> inputs() { + return inputs; + } + + @JsonGetter(StringTransform.FIELD_INPUTS) + public List<Object> inputsForJson() { + return StringTransform.inputsForJson(inputs); + } + + @Override + public DataType outputType() { + return DataTypes.DATE(); + } + + @Override + public Object transform(InternalRow row) { + Integer date = TransformInputUtils.integer(inputs.get(0), row); + Integer days = TransformInputUtils.integer(inputs.get(1), row); + return date == null || days == null ? null : date + days; + } + + @Override + public Transform copyWithNewInputs(List<Object> inputs) { + return new DateAddTransform(inputs); + } + + @Override + public boolean equals(Object o) { + return o != null + && getClass() == o.getClass() + && Objects.equals(inputs, ((DateAddTransform) o).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/DateDiffTransform.java b/paimon-common/src/main/java/org/apache/paimon/predicate/DateDiffTransform.java new file mode 100644 index 0000000000..268822fe1e --- /dev/null +++ b/paimon-common/src/main/java/org/apache/paimon/predicate/DateDiffTransform.java @@ -0,0 +1,105 @@ +/* + * 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.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.utils.Preconditions.checkArgument; + +/** SQL {@code DATE_DIFF(end_date, start_date)} transform. */ +public class DateDiffTransform implements Transform { + + private static final long serialVersionUID = 1L; + + public static final String NAME = "DATE_DIFF"; + + private final List<Object> inputs; + + @JsonCreator + public DateDiffTransform( + @JsonProperty(StringTransform.FIELD_INPUTS) + @JsonDeserialize(contentUsing = SubstringTransform.InputDeserializer.class) + List<Object> inputs) { + checkArgument(inputs.size() == 2, "DATE_DIFF requires exactly two inputs"); + TransformInputUtils.checkDate(inputs.get(0), "DATE_DIFF first input must be a date"); + TransformInputUtils.checkDate(inputs.get(1), "DATE_DIFF second input must be a date"); + this.inputs = inputs; + } + + @Override + public String name() { + return NAME; + } + + @Override + @JsonIgnore + public List<Object> inputs() { + return inputs; + } + + @JsonGetter(StringTransform.FIELD_INPUTS) + public List<Object> inputsForJson() { + return StringTransform.inputsForJson(inputs); + } + + @Override + public DataType outputType() { + return DataTypes.INT(); + } + + @Override + public Object transform(InternalRow row) { + Integer end = TransformInputUtils.integer(inputs.get(0), row); + Integer start = TransformInputUtils.integer(inputs.get(1), row); + return end == null || start == null ? null : end - start; + } + + @Override + public Transform copyWithNewInputs(List<Object> inputs) { + return new DateDiffTransform(inputs); + } + + @Override + public boolean equals(Object o) { + return o != null + && getClass() == o.getClass() + && Objects.equals(inputs, ((DateDiffTransform) o).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/DateExtractTransform.java b/paimon-common/src/main/java/org/apache/paimon/predicate/DateExtractTransform.java index 70dfa9e040..5da9fddcdd 100644 --- a/paimon-common/src/main/java/org/apache/paimon/predicate/DateExtractTransform.java +++ b/paimon-common/src/main/java/org/apache/paimon/predicate/DateExtractTransform.java @@ -41,7 +41,9 @@ 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}. + * HourTransform}, {@link MinuteTransform}, {@link SecondTransform}, {@link QuarterTransform}, + * {@link IsoDayOfWeekTransform}, {@link DayOfWeekTransform}, {@link WeekdayTransform}, {@link + * DayOfYearTransform}, {@link WeekTransform} and {@link YearOfWeekTransform}. * * <p>{@code TIMESTAMP WITH LOCAL TIME ZONE} is deliberately not supported: extracting a calendar * field from it depends on a session time zone, which the reader does not know, so such a predicate diff --git a/paimon-common/src/main/java/org/apache/paimon/predicate/DateTruncTransform.java b/paimon-common/src/main/java/org/apache/paimon/predicate/DateTruncTransform.java new file mode 100644 index 0000000000..cb0125c5c9 --- /dev/null +++ b/paimon-common/src/main/java/org/apache/paimon/predicate/DateTruncTransform.java @@ -0,0 +1,135 @@ +/* + * 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.time.DayOfWeek; +import java.time.LocalDate; +import java.time.temporal.TemporalAdjusters; +import java.util.List; +import java.util.Locale; +import java.util.Objects; + +import static org.apache.paimon.utils.Preconditions.checkArgument; + +/** SQL {@code TRUNC(date, format)} transform. */ +public class DateTruncTransform implements Transform { + + private static final long serialVersionUID = 1L; + + public static final String NAME = "DATE_TRUNC"; + + private final List<Object> inputs; + + @JsonCreator + public DateTruncTransform( + @JsonProperty(StringTransform.FIELD_INPUTS) + @JsonDeserialize(contentUsing = SubstringTransform.InputDeserializer.class) + List<Object> inputs) { + checkArgument(inputs.size() == 2, "DATE_TRUNC requires exactly two inputs"); + TransformInputUtils.checkDate(inputs.get(0), "DATE_TRUNC first input must be a date"); + TransformInputUtils.checkString(inputs.get(1), "DATE_TRUNC format must be a string"); + this.inputs = inputs; + } + + @Override + public String name() { + return NAME; + } + + @Override + @JsonIgnore + public List<Object> inputs() { + return inputs; + } + + @JsonGetter(StringTransform.FIELD_INPUTS) + public List<Object> inputsForJson() { + return StringTransform.inputsForJson(inputs); + } + + @Override + public DataType outputType() { + return DataTypes.DATE(); + } + + @Override + public Object transform(InternalRow row) { + Integer epochDay = TransformInputUtils.integer(inputs.get(0), row); + BinaryString format = TransformInputUtils.string(inputs.get(1), row); + if (epochDay == null || format == null) { + return null; + } + + LocalDate date = LocalDate.ofEpochDay(epochDay); + switch (format.toString().toUpperCase(Locale.ROOT)) { + case "WEEK": + date = date.with(TemporalAdjusters.previousOrSame(DayOfWeek.MONDAY)); + break; + case "MON": + case "MONTH": + case "MM": + date = date.withDayOfMonth(1); + break; + case "QUARTER": + date = date.withMonth(((date.getMonthValue() - 1) / 3) * 3 + 1).withDayOfMonth(1); + break; + case "YEAR": + case "YYYY": + case "YY": + date = date.withDayOfYear(1); + break; + default: + return null; + } + return Math.toIntExact(date.toEpochDay()); + } + + @Override + public Transform copyWithNewInputs(List<Object> inputs) { + return new DateTruncTransform(inputs); + } + + @Override + public boolean equals(Object o) { + return o != null + && getClass() == o.getClass() + && Objects.equals(inputs, ((DateTruncTransform) o).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/DayOfWeekTransform.java b/paimon-common/src/main/java/org/apache/paimon/predicate/DayOfWeekTransform.java new file mode 100644 index 0000000000..41d4f2032c --- /dev/null +++ b/paimon-common/src/main/java/org/apache/paimon/predicate/DayOfWeekTransform.java @@ -0,0 +1,58 @@ +/* + * 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 week, Sunday = 1 through Saturday = 7. */ +public class DayOfWeekTransform extends DateExtractTransform { + + private static final long serialVersionUID = 1L; + + public static final String NAME = "DAY_OF_WEEK"; + + @JsonCreator + public DayOfWeekTransform( + @JsonProperty(DateExtractTransform.FIELD_FIELD_REF) FieldRef fieldRef) { + super(fieldRef); + } + + public static Optional<Transform> tryCreate(FieldRef fieldRef) { + return DateExtractTransform.tryCreate(fieldRef, DayOfWeekTransform::new); + } + + @Override + public String name() { + return NAME; + } + + @Override + protected Integer extract(LocalDateTime dateTime) { + return dateTime.getDayOfWeek().getValue() % 7 + 1; + } + + @Override + protected Transform copy(FieldRef fieldRef) { + return new DayOfWeekTransform(fieldRef); + } +} diff --git a/paimon-common/src/main/java/org/apache/paimon/predicate/DayOfYearTransform.java b/paimon-common/src/main/java/org/apache/paimon/predicate/DayOfYearTransform.java new file mode 100644 index 0000000000..846915ada5 --- /dev/null +++ b/paimon-common/src/main/java/org/apache/paimon/predicate/DayOfYearTransform.java @@ -0,0 +1,58 @@ +/* + * 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 year, like SQL {@code EXTRACT(DAY_OF_YEAR FROM d)}. */ +public class DayOfYearTransform extends DateExtractTransform { + + private static final long serialVersionUID = 1L; + + public static final String NAME = "DAY_OF_YEAR"; + + @JsonCreator + public DayOfYearTransform( + @JsonProperty(DateExtractTransform.FIELD_FIELD_REF) FieldRef fieldRef) { + super(fieldRef); + } + + public static Optional<Transform> tryCreate(FieldRef fieldRef) { + return DateExtractTransform.tryCreate(fieldRef, DayOfYearTransform::new); + } + + @Override + public String name() { + return NAME; + } + + @Override + protected Integer extract(LocalDateTime dateTime) { + return dateTime.getDayOfYear(); + } + + @Override + protected Transform copy(FieldRef fieldRef) { + return new DayOfYearTransform(fieldRef); + } +} diff --git a/paimon-common/src/main/java/org/apache/paimon/predicate/IsoDayOfWeekTransform.java b/paimon-common/src/main/java/org/apache/paimon/predicate/IsoDayOfWeekTransform.java new file mode 100644 index 0000000000..afb53f242c --- /dev/null +++ b/paimon-common/src/main/java/org/apache/paimon/predicate/IsoDayOfWeekTransform.java @@ -0,0 +1,58 @@ +/* + * 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 ISO day of week, Monday = 1 through Sunday = 7. */ +public class IsoDayOfWeekTransform extends DateExtractTransform { + + private static final long serialVersionUID = 1L; + + public static final String NAME = "ISO_DAY_OF_WEEK"; + + @JsonCreator + public IsoDayOfWeekTransform( + @JsonProperty(DateExtractTransform.FIELD_FIELD_REF) FieldRef fieldRef) { + super(fieldRef); + } + + public static Optional<Transform> tryCreate(FieldRef fieldRef) { + return DateExtractTransform.tryCreate(fieldRef, IsoDayOfWeekTransform::new); + } + + @Override + public String name() { + return NAME; + } + + @Override + protected Integer extract(LocalDateTime dateTime) { + return dateTime.getDayOfWeek().getValue(); + } + + @Override + protected Transform copy(FieldRef fieldRef) { + return new IsoDayOfWeekTransform(fieldRef); + } +} diff --git a/paimon-common/src/main/java/org/apache/paimon/predicate/OverlayTransform.java b/paimon-common/src/main/java/org/apache/paimon/predicate/OverlayTransform.java new file mode 100644 index 0000000000..1122f1da20 --- /dev/null +++ b/paimon-common/src/main/java/org/apache/paimon/predicate/OverlayTransform.java @@ -0,0 +1,124 @@ +/* + * 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.utils.Preconditions.checkArgument; + +/** SQL {@code OVERLAY(input PLACING replacement FROM position [FOR length])} transform. */ +public class OverlayTransform implements Transform { + + private static final long serialVersionUID = 1L; + + public static final String NAME = "OVERLAY"; + + private final List<Object> inputs; + + @JsonCreator + public OverlayTransform( + @JsonProperty(StringTransform.FIELD_INPUTS) + @JsonDeserialize(contentUsing = SubstringTransform.InputDeserializer.class) + List<Object> inputs) { + checkArgument( + inputs.size() == 3 || inputs.size() == 4, "OVERLAY requires three or four inputs"); + TransformInputUtils.checkString(inputs.get(0), "OVERLAY input must be a string"); + TransformInputUtils.checkString(inputs.get(1), "OVERLAY replacement must be a string"); + TransformInputUtils.checkInteger(inputs.get(2), "OVERLAY position must be an integer"); + if (inputs.size() == 4) { + TransformInputUtils.checkInteger(inputs.get(3), "OVERLAY length must be an integer"); + } + this.inputs = inputs; + } + + @Override + public String name() { + return NAME; + } + + @Override + @JsonIgnore + public List<Object> inputs() { + return inputs; + } + + @JsonGetter(StringTransform.FIELD_INPUTS) + public List<Object> inputsForJson() { + return StringTransform.inputsForJson(inputs); + } + + @Override + public DataType outputType() { + return DataTypes.STRING(); + } + + @Override + public Object transform(InternalRow row) { + BinaryString source = TransformInputUtils.string(inputs.get(0), row); + BinaryString replacement = TransformInputUtils.string(inputs.get(1), row); + Integer position = TransformInputUtils.integer(inputs.get(2), row); + Integer length = + inputs.size() == 4 ? TransformInputUtils.integer(inputs.get(3), row) : null; + if (source == null || replacement == null || position == null) { + return null; + } + if (inputs.size() == 4 && length == null) { + return null; + } + int replacedLength = length == null || length < 0 ? replacement.numChars() : length; + return BinaryString.concat( + source.substringSQL(1, position - 1), + replacement, + source.substringSQL(position + replacedLength, Integer.MAX_VALUE)); + } + + @Override + public Transform copyWithNewInputs(List<Object> inputs) { + return new OverlayTransform(inputs); + } + + @Override + public boolean equals(Object o) { + return o != null + && getClass() == o.getClass() + && Objects.equals(inputs, ((OverlayTransform) o).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/PadTransform.java b/paimon-common/src/main/java/org/apache/paimon/predicate/PadTransform.java new file mode 100644 index 0000000000..6b7478c4a3 --- /dev/null +++ b/paimon-common/src/main/java/org/apache/paimon/predicate/PadTransform.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.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.utils.Preconditions.checkArgument; +import static org.apache.paimon.utils.Preconditions.checkNotNull; + +/** SQL {@code LPAD} and {@code RPAD} transform. */ +public class PadTransform implements Transform { + + private static final long serialVersionUID = 1L; + + public static final String NAME = "PAD"; + public static final String FIELD_DIRECTION = "direction"; + + private final List<Object> inputs; + private final Direction direction; + + @JsonCreator + public PadTransform( + @JsonProperty(StringTransform.FIELD_INPUTS) + @JsonDeserialize(contentUsing = SubstringTransform.InputDeserializer.class) + List<Object> inputs, + @JsonProperty(FIELD_DIRECTION) Direction direction) { + checkArgument(inputs.size() == 3, "PAD requires exactly three inputs"); + TransformInputUtils.checkString(inputs.get(0), "PAD input must be a string"); + TransformInputUtils.checkInteger(inputs.get(1), "PAD length must be an integer"); + TransformInputUtils.checkString(inputs.get(2), "PAD string must be a string"); + this.inputs = inputs; + this.direction = checkNotNull(direction, "direction must not be null"); + } + + @Override + public String name() { + return NAME; + } + + @Override + @JsonIgnore + public List<Object> inputs() { + return inputs; + } + + @JsonGetter(StringTransform.FIELD_INPUTS) + public List<Object> inputsForJson() { + return StringTransform.inputsForJson(inputs); + } + + @JsonGetter(FIELD_DIRECTION) + public Direction direction() { + return direction; + } + + @Override + public DataType outputType() { + return DataTypes.STRING(); + } + + @Override + public Object transform(InternalRow row) { + BinaryString source = TransformInputUtils.string(inputs.get(0), row); + Integer length = TransformInputUtils.integer(inputs.get(1), row); + BinaryString pad = TransformInputUtils.string(inputs.get(2), row); + if (source == null || length == null || pad == null) { + return null; + } + + int needed = length - source.numChars(); + if (needed <= 0 || pad.getSizeInBytes() == 0) { + return source.substring(0, length); + } + + int padChars = pad.numChars(); + StringBuilder padding = new StringBuilder(); + for (int i = 0; i < needed / padChars; i++) { + padding.append(pad); + } + padding.append(pad.substring(0, needed % padChars)); + BinaryString padValue = BinaryString.fromString(padding.toString()); + return direction == Direction.LEFT + ? BinaryString.concat(padValue, source) + : BinaryString.concat(source, padValue); + } + + @Override + public Transform copyWithNewInputs(List<Object> inputs) { + return new PadTransform(inputs, direction); + } + + @Override + public boolean equals(Object o) { + if (o == null || getClass() != o.getClass()) { + return false; + } + PadTransform that = (PadTransform) o; + return Objects.equals(inputs, that.inputs) && direction == that.direction; + } + + @Override + public int hashCode() { + return Objects.hash(inputs, direction); + } + + @Override + public String toString() { + return (direction == Direction.LEFT ? "LPAD" : "RPAD") + + StringTransform.formatCall("", inputs); + } + + /** Pad side. */ + public enum Direction { + LEFT, + RIGHT + } +} diff --git a/paimon-common/src/main/java/org/apache/paimon/predicate/QuarterTransform.java b/paimon-common/src/main/java/org/apache/paimon/predicate/QuarterTransform.java new file mode 100644 index 0000000000..199f552cde --- /dev/null +++ b/paimon-common/src/main/java/org/apache/paimon/predicate/QuarterTransform.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 quarter of year, like SQL {@code EXTRACT(QUARTER FROM d)}. */ +public class QuarterTransform extends DateExtractTransform { + + private static final long serialVersionUID = 1L; + + public static final String NAME = "QUARTER"; + + @JsonCreator + public QuarterTransform(@JsonProperty(DateExtractTransform.FIELD_FIELD_REF) FieldRef fieldRef) { + super(fieldRef); + } + + public static Optional<Transform> tryCreate(FieldRef fieldRef) { + return DateExtractTransform.tryCreate(fieldRef, QuarterTransform::new); + } + + @Override + public String name() { + return NAME; + } + + @Override + protected Integer extract(LocalDateTime dateTime) { + return (dateTime.getMonthValue() - 1) / 3 + 1; + } + + @Override + protected Transform copy(FieldRef fieldRef) { + return new QuarterTransform(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 dffa541397..f0aefe1b2d 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 @@ -43,11 +43,25 @@ import java.util.List; @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 = QuarterTransform.class, name = QuarterTransform.NAME), + @JsonSubTypes.Type(value = IsoDayOfWeekTransform.class, name = IsoDayOfWeekTransform.NAME), + @JsonSubTypes.Type(value = DayOfWeekTransform.class, name = DayOfWeekTransform.NAME), + @JsonSubTypes.Type(value = WeekdayTransform.class, name = WeekdayTransform.NAME), + @JsonSubTypes.Type(value = DayOfYearTransform.class, name = DayOfYearTransform.NAME), + @JsonSubTypes.Type(value = WeekTransform.class, name = WeekTransform.NAME), + @JsonSubTypes.Type(value = YearOfWeekTransform.class, name = YearOfWeekTransform.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 = BitLengthTransform.class, name = BitLengthTransform.NAME), + @JsonSubTypes.Type(value = TranslateTransform.class, name = TranslateTransform.NAME), + @JsonSubTypes.Type(value = OverlayTransform.class, name = OverlayTransform.NAME), + @JsonSubTypes.Type(value = PadTransform.class, name = PadTransform.NAME), + @JsonSubTypes.Type(value = DateAddTransform.class, name = DateAddTransform.NAME), + @JsonSubTypes.Type(value = DateDiffTransform.class, name = DateDiffTransform.NAME), + @JsonSubTypes.Type(value = DateTruncTransform.class, name = DateTruncTransform.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/TransformInputUtils.java b/paimon-common/src/main/java/org/apache/paimon/predicate/TransformInputUtils.java new file mode 100644 index 0000000000..11673a7c77 --- /dev/null +++ b/paimon-common/src/main/java/org/apache/paimon/predicate/TransformInputUtils.java @@ -0,0 +1,77 @@ +/* + * 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 static org.apache.paimon.types.DataTypeFamily.CHARACTER_STRING; +import static org.apache.paimon.types.DataTypeFamily.INTEGER_NUMERIC; +import static org.apache.paimon.utils.InternalRowUtils.get; +import static org.apache.paimon.utils.Preconditions.checkArgument; + +/** Utilities shared by transforms with mixed field-reference and literal inputs. */ +final class TransformInputUtils { + + private TransformInputUtils() {} + + static Object value(Object input, InternalRow row) { + if (input instanceof FieldRef) { + FieldRef ref = (FieldRef) input; + return get(row, ref.index(), ref.type()); + } + return input; + } + + static BinaryString string(Object input, InternalRow row) { + return (BinaryString) value(input, row); + } + + static Integer integer(Object input, InternalRow row) { + Object value = value(input, row); + return value == null ? null : Math.toIntExact(((Number) value).longValue()); + } + + static void checkString(Object input, String message) { + if (input instanceof FieldRef) { + checkArgument(((FieldRef) input).type().is(CHARACTER_STRING), message); + } else { + checkArgument(input == null || input instanceof BinaryString, message); + } + } + + static void checkInteger(Object input, String message) { + if (input instanceof FieldRef) { + checkArgument(((FieldRef) input).type().is(INTEGER_NUMERIC), message); + } else { + checkArgument(input == null || input instanceof Number, message); + } + } + + static void checkDate(Object input, String message) { + if (input instanceof FieldRef) { + checkArgument( + ((FieldRef) input).type().getTypeRoot() + == org.apache.paimon.types.DataTypeRoot.DATE, + message); + } else { + checkArgument(input == null || input instanceof Integer, message); + } + } +} diff --git a/paimon-common/src/main/java/org/apache/paimon/predicate/TranslateTransform.java b/paimon-common/src/main/java/org/apache/paimon/predicate/TranslateTransform.java new file mode 100644 index 0000000000..909f4f7134 --- /dev/null +++ b/paimon-common/src/main/java/org/apache/paimon/predicate/TranslateTransform.java @@ -0,0 +1,95 @@ +/* + * 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.shade.jackson2.com.fasterxml.jackson.annotation.JsonCreator; +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.HashMap; +import java.util.List; +import java.util.Map; + +import static org.apache.paimon.utils.Preconditions.checkArgument; + +/** SQL {@code TRANSLATE(input, from, to)} transform. */ +public class TranslateTransform extends StringTransform { + + private static final long serialVersionUID = 1L; + + public static final String NAME = "TRANSLATE"; + + @JsonCreator + public TranslateTransform( + @JsonProperty(StringTransform.FIELD_INPUTS) + @JsonDeserialize(contentUsing = StringTransform.InputDeserializer.class) + List<Object> inputs) { + super(inputs); + checkArgument(inputs.size() == 3, "TRANSLATE requires exactly three inputs"); + } + + @Override + public String name() { + return NAME; + } + + @Override + protected BinaryString transform(List<BinaryString> inputs) { + BinaryString source = inputs.get(0); + BinaryString matching = inputs.get(1); + BinaryString replacement = inputs.get(2); + if (source == null || matching == null || replacement == null) { + return null; + } + + int[] matches = matching.toString().codePoints().toArray(); + int[] replacements = replacement.toString().codePoints().toArray(); + Map<Integer, Integer> dictionary = new HashMap<>(); + for (int i = 0; i < matches.length; i++) { + // Keep the first mapping for duplicate characters, including a first mapping to + // deletion. Map.putIfAbsent cannot express that because it treats a null value as + // absent. + if (!dictionary.containsKey(matches[i])) { + dictionary.put( + matches[i], + i < replacements.length && replacements[i] != 0 ? replacements[i] : null); + } + } + + StringBuilder result = new StringBuilder(); + source.toString() + .codePoints() + .forEach( + codePoint -> { + if (!dictionary.containsKey(codePoint)) { + result.appendCodePoint(codePoint); + } else if (dictionary.get(codePoint) != null) { + result.appendCodePoint(dictionary.get(codePoint)); + } + }); + return BinaryString.fromString(result.toString()); + } + + @Override + public Transform copyWithNewInputs(List<Object> inputs) { + return new TranslateTransform(inputs); + } +} diff --git a/paimon-common/src/main/java/org/apache/paimon/predicate/WeekTransform.java b/paimon-common/src/main/java/org/apache/paimon/predicate/WeekTransform.java new file mode 100644 index 0000000000..3f370ebdd7 --- /dev/null +++ b/paimon-common/src/main/java/org/apache/paimon/predicate/WeekTransform.java @@ -0,0 +1,58 @@ +/* + * 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.time.temporal.IsoFields; +import java.util.Optional; + +/** Extracts the ISO week of the week-based year. */ +public class WeekTransform extends DateExtractTransform { + + private static final long serialVersionUID = 1L; + + public static final String NAME = "WEEK"; + + @JsonCreator + public WeekTransform(@JsonProperty(DateExtractTransform.FIELD_FIELD_REF) FieldRef fieldRef) { + super(fieldRef); + } + + public static Optional<Transform> tryCreate(FieldRef fieldRef) { + return DateExtractTransform.tryCreate(fieldRef, WeekTransform::new); + } + + @Override + public String name() { + return NAME; + } + + @Override + protected Integer extract(LocalDateTime dateTime) { + return dateTime.get(IsoFields.WEEK_OF_WEEK_BASED_YEAR); + } + + @Override + protected Transform copy(FieldRef fieldRef) { + return new WeekTransform(fieldRef); + } +} diff --git a/paimon-common/src/main/java/org/apache/paimon/predicate/WeekdayTransform.java b/paimon-common/src/main/java/org/apache/paimon/predicate/WeekdayTransform.java new file mode 100644 index 0000000000..5e961887cd --- /dev/null +++ b/paimon-common/src/main/java/org/apache/paimon/predicate/WeekdayTransform.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 weekday, Monday = 0 through Sunday = 6. */ +public class WeekdayTransform extends DateExtractTransform { + + private static final long serialVersionUID = 1L; + + public static final String NAME = "WEEKDAY"; + + @JsonCreator + public WeekdayTransform(@JsonProperty(DateExtractTransform.FIELD_FIELD_REF) FieldRef fieldRef) { + super(fieldRef); + } + + public static Optional<Transform> tryCreate(FieldRef fieldRef) { + return DateExtractTransform.tryCreate(fieldRef, WeekdayTransform::new); + } + + @Override + public String name() { + return NAME; + } + + @Override + protected Integer extract(LocalDateTime dateTime) { + return dateTime.getDayOfWeek().getValue() - 1; + } + + @Override + protected Transform copy(FieldRef fieldRef) { + return new WeekdayTransform(fieldRef); + } +} diff --git a/paimon-common/src/main/java/org/apache/paimon/predicate/YearOfWeekTransform.java b/paimon-common/src/main/java/org/apache/paimon/predicate/YearOfWeekTransform.java new file mode 100644 index 0000000000..06b947bd5c --- /dev/null +++ b/paimon-common/src/main/java/org/apache/paimon/predicate/YearOfWeekTransform.java @@ -0,0 +1,59 @@ +/* + * 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.time.temporal.IsoFields; +import java.util.Optional; + +/** Extracts the ISO week-based year. */ +public class YearOfWeekTransform extends DateExtractTransform { + + private static final long serialVersionUID = 1L; + + public static final String NAME = "YEAR_OF_WEEK"; + + @JsonCreator + public YearOfWeekTransform( + @JsonProperty(DateExtractTransform.FIELD_FIELD_REF) FieldRef fieldRef) { + super(fieldRef); + } + + public static Optional<Transform> tryCreate(FieldRef fieldRef) { + return DateExtractTransform.tryCreate(fieldRef, YearOfWeekTransform::new); + } + + @Override + public String name() { + return NAME; + } + + @Override + protected Integer extract(LocalDateTime dateTime) { + return dateTime.get(IsoFields.WEEK_BASED_YEAR); + } + + @Override + protected Transform copy(FieldRef fieldRef) { + return new YearOfWeekTransform(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 index 0a472c78e4..09baa8ad29 100644 --- a/paimon-common/src/test/java/org/apache/paimon/predicate/DateExtractTransformTest.java +++ b/paimon-common/src/test/java/org/apache/paimon/predicate/DateExtractTransformTest.java @@ -52,6 +52,23 @@ class DateExtractTransformTest { 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); + assertThat(extract(DataTypes.DATE(), QuarterTransform::new, value)).isEqualTo(3); + assertThat(extract(DataTypes.DATE(), IsoDayOfWeekTransform::new, value)).isEqualTo(6); + assertThat(extract(DataTypes.DATE(), DayOfWeekTransform::new, value)).isEqualTo(7); + assertThat(extract(DataTypes.DATE(), WeekdayTransform::new, value)).isEqualTo(5); + assertThat(extract(DataTypes.DATE(), DayOfYearTransform::new, value)).isEqualTo(196); + assertThat(extract(DataTypes.DATE(), WeekTransform::new, value)).isEqualTo(28); + assertThat(extract(DataTypes.DATE(), YearOfWeekTransform::new, value)).isEqualTo(2023); + } + + @Test + public void testIsoWeekCrossesCalendarYear() { + int value = epochDay(2024, 12, 30); + assertThat(extract(DataTypes.DATE(), WeekTransform::new, value)).isEqualTo(1); + assertThat(extract(DataTypes.DATE(), YearOfWeekTransform::new, value)).isEqualTo(2025); + assertThat(extract(DataTypes.DATE(), IsoDayOfWeekTransform::new, value)).isEqualTo(1); + assertThat(extract(DataTypes.DATE(), DayOfWeekTransform::new, value)).isEqualTo(2); + assertThat(extract(DataTypes.DATE(), WeekdayTransform::new, value)).isEqualTo(0); } @Test diff --git a/paimon-common/src/test/java/org/apache/paimon/predicate/ScalarTransformTest.java b/paimon-common/src/test/java/org/apache/paimon/predicate/ScalarTransformTest.java new file mode 100644 index 0000000000..918a4798f1 --- /dev/null +++ b/paimon-common/src/test/java/org/apache/paimon/predicate/ScalarTransformTest.java @@ -0,0 +1,171 @@ +/* + * 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.time.LocalDate; +import java.util.Arrays; +import java.util.Collections; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Behavioral tests for scalar transforms. */ +class ScalarTransformTest { + + @Test + void testBitLengthUsesUtf8BytesAndPropagatesNull() { + BitLengthTransform transform = + new BitLengthTransform( + Collections.singletonList(new FieldRef(0, "s", DataTypes.STRING()))); + + assertThat(transform.transform(GenericRow.of(string("é😀")))).isEqualTo(48); + assertThat(transform.transform(GenericRow.of((Object) null))).isNull(); + } + + @Test + void testTranslateCodePointAndDuplicateSemantics() { + assertThat( + new TranslateTransform( + Arrays.asList(string("A😀B"), string("😀"), string("界"))) + .transform(GenericRow.of())) + .isEqualTo(string("A界B")); + + // The first duplicate mapping wins even when it maps to deletion. + assertThat( + new TranslateTransform( + Arrays.asList(string("aba"), string("aa"), string(""))) + .transform(GenericRow.of())) + .isEqualTo(string("b")); + + // A NUL replacement code point acts as the deletion sentinel. + assertThat( + new TranslateTransform( + Arrays.asList(string("aba"), string("a"), string("\u0000"))) + .transform(GenericRow.of())) + .isEqualTo(string("b")); + + assertThat( + new TranslateTransform(Arrays.asList(null, string("a"), string("b"))) + .transform(GenericRow.of())) + .isNull(); + } + + @Test + void testOverlayDefaultExplicitAndUnicodeSemantics() { + assertThat( + new OverlayTransform(Arrays.asList(string("Hello SQL"), string("_"), 6)) + .transform(GenericRow.of())) + .isEqualTo(string("Hello_SQL")); + assertThat( + new OverlayTransform( + Arrays.asList(string("Hello SQL"), string("ANSI "), 7, 0)) + .transform(GenericRow.of())) + .isEqualTo(string("Hello ANSI SQL")); + assertThat( + new OverlayTransform(Arrays.asList(string("a😀c"), string("界"), 2, 1)) + .transform(GenericRow.of())) + .isEqualTo(string("a界c")); + assertThat( + new OverlayTransform(Arrays.asList(string("abc"), null, 1, 1)) + .transform(GenericRow.of())) + .isNull(); + } + + @Test + void testPadForPaddingTruncationAndEmptyPad() { + assertThat(pad(PadTransform.Direction.LEFT, "hi", 5, "??")).isEqualTo(string("???hi")); + assertThat(pad(PadTransform.Direction.RIGHT, "hi", 5, "??")).isEqualTo(string("hi???")); + assertThat(pad(PadTransform.Direction.LEFT, "a😀c", 2, "x")).isEqualTo(string("a😀")); + assertThat(pad(PadTransform.Direction.RIGHT, "hi", 5, "")).isEqualTo(string("hi")); + assertThat(pad(PadTransform.Direction.LEFT, "hi", -1, "?")).isEqualTo(string("")); + } + + @Test + void testDateAddAndDiffWithLiteralAndFieldInputs() { + FieldRef date = new FieldRef(0, "date", DataTypes.DATE()); + FieldRef days = new FieldRef(1, "days", DataTypes.SMALLINT()); + int january15 = epochDay("2025-01-15"); + + DateAddTransform add = new DateAddTransform(Arrays.asList(date, days)); + assertThat(add.transform(GenericRow.of(january15, (short) 2))) + .isEqualTo(epochDay("2025-01-17")); + assertThat(add.transform(GenericRow.of(january15, null))).isNull(); + + DateDiffTransform diff = new DateDiffTransform(Arrays.asList(date, epochDay("2025-01-01"))); + assertThat(diff.transform(GenericRow.of(january15))).isEqualTo(14); + assertThat(diff.transform(GenericRow.of((Object) null))).isNull(); + } + + @Test + void testDateTruncSupportsEveryDateLevel() { + FieldRef date = new FieldRef(0, "date", DataTypes.DATE()); + GenericRow row = GenericRow.of(epochDay("2025-05-18")); + + assertThat(trunc(date, "week").transform(row)).isEqualTo(epochDay("2025-05-12")); + assertThat(trunc(date, "mon").transform(row)).isEqualTo(epochDay("2025-05-01")); + assertThat(trunc(date, "quarter").transform(row)).isEqualTo(epochDay("2025-04-01")); + assertThat(trunc(date, "yyyy").transform(row)).isEqualTo(epochDay("2025-01-01")); + assertThat(trunc(date, "day").transform(row)).isNull(); + assertThat( + new DateTruncTransform(Arrays.asList(date, null)) + .transform(GenericRow.of(epochDay("2025-05-18")))) + .isNull(); + } + + @Test + void testInputTypesAreValidated() { + assertThatThrownBy( + () -> + new DateAddTransform( + Arrays.asList(new FieldRef(0, "s", DataTypes.STRING()), 1))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("first input must be a date"); + assertThatThrownBy( + () -> + new PadTransform( + Arrays.asList(string("x"), string("1"), string("_")), + PadTransform.Direction.LEFT)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("length must be an integer"); + } + + private static Object pad( + PadTransform.Direction direction, String value, int length, String padding) { + return new PadTransform(Arrays.asList(string(value), length, string(padding)), direction) + .transform(GenericRow.of()); + } + + private static DateTruncTransform trunc(FieldRef date, String format) { + return new DateTruncTransform(Arrays.asList(date, string(format))); + } + + private static BinaryString string(String value) { + return BinaryString.fromString(value); + } + + private static int epochDay(String value) { + return Math.toIntExact(LocalDate.parse(value).toEpochDay()); + } +} 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 f2a711e01e..47fac07736 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 @@ -131,6 +131,31 @@ class TransformJsonSerdeTest { new MinuteTransform(new FieldRef(1, "t1", DataTypes.TIMESTAMP(3)))) .expectJson( "{\"name\":\"MINUTE\",\"fieldRef\":{\"index\":1,\"name\":\"t1\",\"type\":\"TIMESTAMP(3)\"}}"), + TestSpec.forTransform(new QuarterTransform(new FieldRef(0, "d0", DataTypes.DATE()))) + .expectJson( + "{\"name\":\"QUARTER\",\"fieldRef\":{\"index\":0,\"name\":\"d0\",\"type\":\"DATE\"}}"), + TestSpec.forTransform( + new IsoDayOfWeekTransform(new FieldRef(0, "d0", DataTypes.DATE()))) + .expectJson( + "{\"name\":\"ISO_DAY_OF_WEEK\",\"fieldRef\":{\"index\":0,\"name\":\"d0\",\"type\":\"DATE\"}}"), + TestSpec.forTransform( + new DayOfWeekTransform(new FieldRef(0, "d0", DataTypes.DATE()))) + .expectJson( + "{\"name\":\"DAY_OF_WEEK\",\"fieldRef\":{\"index\":0,\"name\":\"d0\",\"type\":\"DATE\"}}"), + TestSpec.forTransform(new WeekdayTransform(new FieldRef(0, "d0", DataTypes.DATE()))) + .expectJson( + "{\"name\":\"WEEKDAY\",\"fieldRef\":{\"index\":0,\"name\":\"d0\",\"type\":\"DATE\"}}"), + TestSpec.forTransform( + new DayOfYearTransform(new FieldRef(0, "d0", DataTypes.DATE()))) + .expectJson( + "{\"name\":\"DAY_OF_YEAR\",\"fieldRef\":{\"index\":0,\"name\":\"d0\",\"type\":\"DATE\"}}"), + TestSpec.forTransform(new WeekTransform(new FieldRef(0, "d0", DataTypes.DATE()))) + .expectJson( + "{\"name\":\"WEEK\",\"fieldRef\":{\"index\":0,\"name\":\"d0\",\"type\":\"DATE\"}}"), + TestSpec.forTransform( + new YearOfWeekTransform(new FieldRef(0, "d0", DataTypes.DATE()))) + .expectJson( + "{\"name\":\"YEAR_OF_WEEK\",\"fieldRef\":{\"index\":0,\"name\":\"d0\",\"type\":\"DATE\"}}"), TestSpec.forTransform( new SubstringTransform( Arrays.asList( @@ -195,6 +220,59 @@ class TransformJsonSerdeTest { TestSpec.forTransform(new LengthTransform(Collections.singletonList(null))) .expectJson("{\"name\":\"LENGTH\",\"inputs\":[null]}"), + // Additional scalar transforms + TestSpec.forTransform( + new BitLengthTransform( + Collections.singletonList( + new FieldRef(1, "f1", DataTypes.STRING())))) + .expectJson( + "{\"name\":\"BIT_LENGTH\",\"inputs\":[{\"index\":1,\"name\":\"f1\",\"type\":\"STRING\"}]}"), + TestSpec.forTransform( + new TranslateTransform( + Arrays.asList( + new FieldRef(1, "f1", DataTypes.STRING()), + BinaryString.fromString("ab"), + BinaryString.fromString("xy")))) + .expectJson( + "{\"name\":\"TRANSLATE\",\"inputs\":[{\"index\":1,\"name\":\"f1\",\"type\":\"STRING\"},\"ab\",\"xy\"]}"), + TestSpec.forTransform( + new OverlayTransform( + Arrays.asList( + new FieldRef(1, "f1", DataTypes.STRING()), + BinaryString.fromString("x"), + 2, + 1))) + .expectJson( + "{\"name\":\"OVERLAY\",\"inputs\":[{\"index\":1,\"name\":\"f1\",\"type\":\"STRING\"},\"x\",2,1]}"), + TestSpec.forTransform( + new PadTransform( + Arrays.asList( + new FieldRef(1, "f1", DataTypes.STRING()), + 5, + BinaryString.fromString("_")), + PadTransform.Direction.LEFT)) + .expectJson( + "{\"name\":\"PAD\",\"inputs\":[{\"index\":1,\"name\":\"f1\",\"type\":\"STRING\"},5,\"_\"],\"direction\":\"LEFT\"}"), + TestSpec.forTransform( + new DateAddTransform( + Arrays.asList(new FieldRef(0, "d0", DataTypes.DATE()), 1))) + .expectJson( + "{\"name\":\"DATE_ADD\",\"inputs\":[{\"index\":0,\"name\":\"d0\",\"type\":\"DATE\"},1]}"), + TestSpec.forTransform( + new DateDiffTransform( + Arrays.asList( + new FieldRef(0, "d0", DataTypes.DATE()), + new FieldRef(1, "d1", DataTypes.DATE())))) + .expectJson( + "{\"name\":\"DATE_DIFF\",\"inputs\":[{\"index\":0,\"name\":\"d0\",\"type\":\"DATE\"},{\"index\":1,\"name\":\"d1\",\"type\":\"DATE\"}]}"), + TestSpec.forTransform( + new DateTruncTransform( + Arrays.asList( + new FieldRef(0, "d0", DataTypes.DATE()), + BinaryString.fromString("MONTH")))) + .expectJson( + "{\"name\":\"DATE_TRUNC\",\"inputs\":[{\"index\":0,\"name\":\"d0\",\"type\":\"DATE\"},\"MONTH\"]}"), + // error message testing TestSpec.forJson("{\"name\":\"invalid\"}") .expectErrorMessage("Could not resolve type id 'invalid'"), 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 b90f4902c4..0e54b10075 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 @@ -41,9 +41,22 @@ object SparkExpressionConverter { private val LOWER = "LOWER" private val SUBSTRING = "SUBSTRING" private val CHAR_LENGTH = "CHAR_LENGTH" + private val BIT_LENGTH = "BIT_LENGTH" + private val TRANSLATE = "TRANSLATE" + private val OVERLAY = "OVERLAY" + private val LPAD = "LPAD" + private val RPAD = "RPAD" private val TRIM = "TRIM" private val LTRIM = "LTRIM" private val RTRIM = "RTRIM" + private val DATE_ADD = "DATE_ADD" + private val DATE_DIFF = "DATE_DIFF" + private val TRUNC = "TRUNC" + + // Spark encodes dayofweek and weekday as arithmetic over an ISO EXTRACT(DAY_OF_WEEK). + private val ADD = "+" + private val SUBTRACT = "-" + private val REMAINDER = "%" // Supported fields of the EXTRACT expression private val EXTRACT_YEAR = "YEAR" @@ -52,6 +65,11 @@ object SparkExpressionConverter { private val EXTRACT_HOUR = "HOUR" private val EXTRACT_MINUTE = "MINUTE" private val EXTRACT_SECOND = "SECOND" + private val EXTRACT_QUARTER = "QUARTER" + private val EXTRACT_DAY_OF_WEEK = "DAY_OF_WEEK" + private val EXTRACT_DAY_OF_YEAR = "DAY_OF_YEAR" + private val EXTRACT_WEEK = "WEEK" + private val EXTRACT_YEAR_OF_WEEK = "YEAR_OF_WEEK" /** Convert Spark [[Expression]] to Paimon [[Transform]], return None if not supported. */ def toPaimonTransform(exp: Expression, rowType: RowType): Option[Transform] = { @@ -69,6 +87,48 @@ object SparkExpressionConverter { } } + def literalEquals(exp: Expression, value: Int): Boolean = exp match { + case l: Literal[_] => + l.value() match { + case i: Int => i == value + case _ => false + } + case _ => false + } + + def dayOfWeekField(exp: Expression): Option[FieldRef] = { + if (org.apache.spark.SPARK_VERSION < "3.4") { + None + } else { + exp match { + case extract: Extract if extract.field() == EXTRACT_DAY_OF_WEEK => + extract.source() match { + case n: NamedReference => Some(toPaimonFieldRef(n, rowType)) + case _ => None + } + case _ => None + } + } + } + + def sparkDayOfWeek(children: Seq[Expression]): Option[Transform] = children match { + case Seq(remainder: GeneralScalarExpression, one) + if literalEquals(one, 1) && + remainder.name() == REMAINDER => + remainder.children().toSeq match { + case Seq(extract, seven) if literalEquals(seven, 7) => + dayOfWeekField(extract).flatMap(DayOfWeekTransform.tryCreate) + case _ => None + } + case _ => None + } + + def sparkWeekday(children: Seq[Expression]): Option[Transform] = children match { + case Seq(extract, one) if literalEquals(one, 1) => + dayOfWeekField(extract).flatMap(WeekdayTransform.tryCreate) + case _ => None + } + exp match { case n: NamedReference => Some(new FieldTransform(toPaimonFieldRef(n, rowType))) case s: GeneralScalarExpression => @@ -78,6 +138,14 @@ object SparkExpressionConverter { 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 BIT_LENGTH => convertChildren(s.children()).map(i => new BitLengthTransform(i)) + case TRANSLATE => convertChildren(s.children()).map(i => new TranslateTransform(i)) + case OVERLAY => convertChildren(s.children()).map(i => new OverlayTransform(i)) + case LPAD => + convertChildren(s.children()).map(i => new PadTransform(i, PadTransform.Direction.LEFT)) + case RPAD => + convertChildren(s.children()).map( + i => new PadTransform(i, PadTransform.Direction.RIGHT)) case TRIM => convertChildren(s.children()).map(i => new TrimTransform(i, TrimTransform.Flag.BOTH)) case LTRIM => @@ -85,6 +153,11 @@ object SparkExpressionConverter { case RTRIM => convertChildren(s.children()).map( i => new TrimTransform(i, TrimTransform.Flag.TRAILING)) + case DATE_ADD => convertChildren(s.children()).map(i => new DateAddTransform(i)) + case DATE_DIFF => convertChildren(s.children()).map(i => new DateDiffTransform(i)) + case TRUNC => convertChildren(s.children()).map(i => new DateTruncTransform(i)) + case ADD => sparkDayOfWeek(s.children()) + case SUBTRACT => sparkWeekday(s.children()) case _ => None } case c: Cast => @@ -119,6 +192,11 @@ object SparkExpressionConverter { case EXTRACT_HOUR => HourTransform.tryCreate(fieldRef) case EXTRACT_MINUTE => MinuteTransform.tryCreate(fieldRef) case EXTRACT_SECOND => SecondTransform.tryCreate(fieldRef) + case EXTRACT_QUARTER => QuarterTransform.tryCreate(fieldRef) + case EXTRACT_DAY_OF_WEEK => IsoDayOfWeekTransform.tryCreate(fieldRef) + case EXTRACT_DAY_OF_YEAR => DayOfYearTransform.tryCreate(fieldRef) + case EXTRACT_WEEK => WeekTransform.tryCreate(fieldRef) + case EXTRACT_YEAR_OF_WEEK => YearOfWeekTransform.tryCreate(fieldRef) 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 7a87bd0666..6321a9d5d7 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.{DayTransform, FieldRef, HourTransform, LengthTransform, MinuteTransform, MonthTransform, PredicateBuilder, SecondTransform, YearTransform} +import org.apache.paimon.predicate.{BitLengthTransform, DateAddTransform, DateDiffTransform, DateTruncTransform, DayOfWeekTransform, DayOfYearTransform, DayTransform, FieldRef, HourTransform, LengthTransform, MinuteTransform, MonthTransform, OverlayTransform, PadTransform, PredicateBuilder, QuarterTransform, SecondTransform, Transform, TranslateTransform, WeekdayTransform, WeekTransform, YearOfWeekTransform, YearTransform} import org.apache.paimon.spark.{PaimonSparkTestBase, SparkV2FilterConverter} import org.apache.paimon.spark.util.shim.TypeUtils.treatPaimonTimestampTypeAsSparkTimestampType import org.apache.paimon.table.source.DataSplit @@ -431,6 +431,76 @@ abstract class SparkV2FilterConverterTestBase extends PaimonSparkTestBase { } } + test("V2Filter: remaining Spark string scalar expressions") { + if (gteqSpark3_4) { + val stringFieldRef = new FieldRef(0, "string_col", rowType.getTypeAt(0)) + val specs: Seq[(String, Transform, Object, Seq[Row])] = Seq( + ( + "bit_length(string_col) = 16", + new BitLengthTransform(List[Object](stringFieldRef).asJava), + Int.box(16), + Seq(Row("hi"))), + ( + "translate(string_col, 'hel', 'xyz') = 'xyzzo'", + new TranslateTransform( + List[Object]( + stringFieldRef, + BinaryString.fromString("hel"), + BinaryString.fromString("xyz")).asJava), + BinaryString.fromString("xyzzo"), + Seq(Row("hello"))), + ( + "overlay(string_col, 'XX', 2, 2) = 'hXXlo'", + new OverlayTransform( + List[Object]( + stringFieldRef, + BinaryString.fromString("XX"), + Int.box(2), + Int.box(2)).asJava), + BinaryString.fromString("hXXlo"), + Seq(Row("hello"))) + ) + specs.foreach { + case (filter, transform, expectedValue, expectedRows) => + assert( + converter.convert(v2Filter(filter)).contains(builder.equal(transform, expectedValue)), + filter) + checkAnswer( + sql(s"SELECT string_col FROM test_tbl WHERE $filter ORDER BY string_col"), + expectedRows) + } + } + } + + test("V2Filter: Spark 4.1 pad expressions") { + if (gteqSpark4_1) { + val stringFieldRef = new FieldRef(0, "string_col", rowType.getTypeAt(0)) + Seq( + ( + "lpad(string_col, 7, '_') = '__hello'", + PadTransform.Direction.LEFT, + BinaryString.fromString("__hello"), + Seq(Row("hello"))), + ( + "rpad(string_col, 7, '_') = 'hello__'", + PadTransform.Direction.RIGHT, + BinaryString.fromString("hello__"), + Seq(Row("hello"))) + ).foreach { + case (filter, direction, expectedValue, expectedRows) => + val transform = new PadTransform( + List[Object](stringFieldRef, Int.box(7), BinaryString.fromString("_")).asJava, + direction) + assert( + converter.convert(v2Filter(filter)).contains(builder.equal(transform, expectedValue)), + filter) + checkAnswer( + sql(s"SELECT string_col FROM test_tbl WHERE $filter ORDER BY string_col"), + expectedRows) + } + } + } + test("V2Filter: Year, Month and Day") { if (gteqSpark3_4) { val dateFieldRef = new FieldRef(9, "date_col", rowType.getTypeAt(9)) @@ -508,6 +578,67 @@ abstract class SparkV2FilterConverterTestBase extends PaimonSparkTestBase { } } + test("V2Filter: remaining date fields") { + if (gteqSpark3_4) { + val dateFieldRef = new FieldRef(9, "date_col", rowType.getTypeAt(9)) + val specs: Seq[(String, Transform, Object, Int)] = Seq( + ("quarter(date_col) = 1", new QuarterTransform(dateFieldRef), Int.box(1), 4), + ("dayofweek(date_col) = 4", new DayOfWeekTransform(dateFieldRef), Int.box(4), 1), + ("weekday(date_col) = 2", new WeekdayTransform(dateFieldRef), Int.box(2), 1), + ("dayofyear(date_col) = 15", new DayOfYearTransform(dateFieldRef), Int.box(15), 1), + ("weekofyear(date_col) = 3", new WeekTransform(dateFieldRef), Int.box(3), 4), + ( + "extract(YEAROFWEEK FROM date_col) = 2025", + new YearOfWeekTransform(dateFieldRef), + Int.box(2025), + 4) + ) + specs.foreach { + case (filter, transform, expectedValue, expectedCount) => + assert( + converter.convert(v2Filter(filter)).contains(builder.equal(transform, expectedValue)), + filter) + assert(sql(s"SELECT date_col FROM test_tbl WHERE $filter").count() == expectedCount) + } + } + } + + test("V2Filter: date scalar expressions") { + if (gteqSpark3_4) { + val dateFieldRef = new FieldRef(9, "date_col", rowType.getTypeAt(9)) + val specs: Seq[(String, Transform, Object, String)] = Seq( + ( + "date_add(date_col, 1) = date('2025-01-16')", + new DateAddTransform(List[Object](dateFieldRef, Int.box(1)).asJava), + Int.box(LocalDate.parse("2025-01-16").toEpochDay.toInt), + "SELECT date('2025-01-15')"), + ( + "datediff(date_col, date('2025-01-01')) = 14", + new DateDiffTransform( + List[Object]( + dateFieldRef, + Int.box(LocalDate.parse("2025-01-01").toEpochDay.toInt)).asJava), + Int.box(14), + "SELECT date('2025-01-15')"), + ( + "trunc(date_col, 'MONTH') = date('2025-01-01')", + new DateTruncTransform( + List[Object](dateFieldRef, BinaryString.fromString("MONTH")).asJava), + Int.box(LocalDate.parse("2025-01-01").toEpochDay.toInt), + "SELECT date_col FROM test_tbl ORDER BY date_col") + ) + specs.foreach { + case (filter, transform, expectedValue, expectedSql) => + assert( + converter.convert(v2Filter(filter)).contains(builder.equal(transform, expectedValue)), + filter) + checkAnswer( + sql(s"SELECT date_col FROM test_tbl WHERE $filter ORDER BY date_col"), + sql(expectedSql)) + } + } + } + test("V2Filter: StartWith") { val filter = "string_col LIKE 'h%'" val actual = converter.convert(v2Filter(filter)).get
