You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by li...@apache.org on 2024/09/11 08:44:40 UTC
(flink) 01/04: [FLINK-36123][table-common] Add ArgumentTypeStrategy for percentage (array) value
This is an automated email from the ASF dual-hosted git repository.
lincoln pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 6404214eeac97510556e4aaf880e36481936ee21
Author: dylanhz <dy...@163.com>
AuthorDate: Wed Aug 28 11:39:33 2024 +0800
[FLINK-36123][table-common] Add ArgumentTypeStrategy for percentage (array) value
---
.../strategies/PercentageArgumentTypeStrategy.java | 80 ++++++++++++++++
.../PercentageArrayArgumentTypeStrategy.java | 103 +++++++++++++++++++++
.../strategies/SpecificInputTypeStrategies.java | 13 +++
.../types/inference/InputTypeStrategiesTest.java | 102 +++++++++++++++++++-
4 files changed, 297 insertions(+), 1 deletion(-)
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/PercentageArgumentTypeStrategy.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/PercentageArgumentTypeStrategy.java
new file mode 100644
index 00000000000..b04d4ef4fa5
--- /dev/null
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/PercentageArgumentTypeStrategy.java
@@ -0,0 +1,80 @@
+/*
+ * 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.flink.table.types.inference.strategies;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.functions.FunctionDefinition;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.inference.ArgumentTypeStrategy;
+import org.apache.flink.table.types.inference.CallContext;
+import org.apache.flink.table.types.inference.Signature;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.LogicalTypeFamily;
+
+import java.util.Optional;
+
+/** An {@link ArgumentTypeStrategy} that expects a percentage value between [0.0, 1.0]. */
+@Internal
+public final class PercentageArgumentTypeStrategy implements ArgumentTypeStrategy {
+
+ private final boolean expectedNullability;
+
+ public PercentageArgumentTypeStrategy(boolean nullable) {
+ this.expectedNullability = nullable;
+ }
+
+ @Override
+ public Optional<DataType> inferArgumentType(
+ CallContext callContext, int argumentPos, boolean throwOnFailure) {
+ final LogicalType actualType =
+ callContext.getArgumentDataTypes().get(argumentPos).getLogicalType();
+
+ if (!actualType.is(LogicalTypeFamily.NUMERIC)) {
+ return callContext.fail(throwOnFailure, "Percentage must be of NUMERIC type.");
+ }
+ if (!expectedNullability && actualType.isNullable()) {
+ return callContext.fail(throwOnFailure, "Percentage must be of NOT NULL type.");
+ }
+
+ if (callContext.isArgumentLiteral(argumentPos)) {
+ Optional<Number> literalVal = callContext.getArgumentValue(argumentPos, Number.class);
+
+ Double val = null;
+ if (literalVal.isPresent()) {
+ val = literalVal.get().doubleValue();
+ }
+
+ if (val == null || val < 0.0 || val > 1.0) {
+ return callContext.fail(
+ throwOnFailure,
+ "Percentage must be between [0.0, 1.0], but was '%s'.",
+ val);
+ }
+ }
+
+ return Optional.of(expectedNullability ? DataTypes.DOUBLE() : DataTypes.DOUBLE().notNull());
+ }
+
+ @Override
+ public Signature.Argument getExpectedArgument(
+ FunctionDefinition functionDefinition, int argumentPos) {
+ return Signature.Argument.of(expectedNullability ? "<NUMERIC>" : "<NUMERIC NOT NULL>");
+ }
+}
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/PercentageArrayArgumentTypeStrategy.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/PercentageArrayArgumentTypeStrategy.java
new file mode 100644
index 00000000000..1d5560bfcc0
--- /dev/null
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/PercentageArrayArgumentTypeStrategy.java
@@ -0,0 +1,103 @@
+/*
+ * 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.flink.table.types.inference.strategies;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.functions.FunctionDefinition;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.inference.ArgumentTypeStrategy;
+import org.apache.flink.table.types.inference.CallContext;
+import org.apache.flink.table.types.inference.Signature;
+import org.apache.flink.table.types.logical.ArrayType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.LogicalTypeFamily;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
+
+import java.util.Optional;
+
+/**
+ * An {@link ArgumentTypeStrategy} that expects an array of percentages with each element between
+ * [0.0, 1.0].
+ */
+@Internal
+public final class PercentageArrayArgumentTypeStrategy implements ArgumentTypeStrategy {
+
+ private final boolean expectedNullability;
+
+ public PercentageArrayArgumentTypeStrategy(boolean nullable) {
+ this.expectedNullability = nullable;
+ }
+
+ @Override
+ public Optional<DataType> inferArgumentType(
+ CallContext callContext, int argumentPos, boolean throwOnFailure) {
+ final LogicalType actualType =
+ callContext.getArgumentDataTypes().get(argumentPos).getLogicalType();
+
+ if (!actualType.is(LogicalTypeRoot.ARRAY)) {
+ return callContext.fail(throwOnFailure, "Percentage must be an array.");
+ }
+ if (!expectedNullability && actualType.isNullable()) {
+ return callContext.fail(throwOnFailure, "Percentage must be a non-null array.");
+ }
+
+ LogicalType elementType = ((ArrayType) actualType).getElementType();
+ if (!elementType.is(LogicalTypeFamily.NUMERIC)) {
+ return callContext.fail(
+ throwOnFailure, "Value in the percentage array must be of NUMERIC type.");
+ }
+ if (!expectedNullability && elementType.isNullable()) {
+ return callContext.fail(
+ throwOnFailure, "Value in the percentage array must be of NOT NULL type.");
+ }
+
+ if (callContext.isArgumentLiteral(argumentPos)) {
+ Optional<Number[]> literalVal =
+ callContext.getArgumentValue(argumentPos, Number[].class);
+
+ if (!literalVal.isPresent()) {
+ return callContext.fail(
+ throwOnFailure,
+ "Percentage must be an array of NUMERIC values between [0.0, 1.0].");
+ }
+
+ for (Number value : literalVal.get()) {
+ if (value.doubleValue() < 0.0 || value.doubleValue() > 1.0) {
+ return callContext.fail(
+ throwOnFailure,
+ "Value in the percentage array must be between [0.0, 1.0], but was '%s'.",
+ value.doubleValue());
+ }
+ }
+ }
+
+ return Optional.of(
+ expectedNullability
+ ? DataTypes.ARRAY(DataTypes.DOUBLE())
+ : DataTypes.ARRAY(DataTypes.DOUBLE().notNull()).notNull());
+ }
+
+ @Override
+ public Signature.Argument getExpectedArgument(
+ FunctionDefinition functionDefinition, int argumentPos) {
+ return Signature.Argument.of(
+ expectedNullability ? "ARRAY<NUMERIC>" : "ARRAY<NUMERIC NOT NULL> NOT NULL");
+ }
+}
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/SpecificInputTypeStrategies.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/SpecificInputTypeStrategies.java
index 26ee877d1f4..1c20f91c6ee 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/SpecificInputTypeStrategies.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/SpecificInputTypeStrategies.java
@@ -125,6 +125,19 @@ public final class SpecificInputTypeStrategies {
*/
public static final ArgumentTypeStrategy INDEX = new IndexArgumentTypeStrategy();
+ /** An {@link ArgumentTypeStrategy} that expects a percentage value between [0.0, 1.0]. */
+ public static ArgumentTypeStrategy percentage(boolean expectedNullability) {
+ return new PercentageArgumentTypeStrategy(expectedNullability);
+ }
+
+ /**
+ * An {@link ArgumentTypeStrategy} that expects an array of percentages with each element
+ * between [0.0, 1.0].
+ */
+ public static ArgumentTypeStrategy percentageArray(boolean expectedNullability) {
+ return new PercentageArrayArgumentTypeStrategy(expectedNullability);
+ }
+
// --------------------------------------------------------------------------------------------
// Strategies composed of other strategies
// --------------------------------------------------------------------------------------------
diff --git a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/InputTypeStrategiesTest.java b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/InputTypeStrategiesTest.java
index 91844cab069..9a63dbb0cb4 100644
--- a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/InputTypeStrategiesTest.java
+++ b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/InputTypeStrategiesTest.java
@@ -27,6 +27,7 @@ import org.apache.flink.table.types.logical.LogicalTypeRoot;
import org.apache.flink.table.types.logical.TimestampKind;
import org.apache.flink.table.types.utils.TypeConversions;
+import java.math.BigDecimal;
import java.util.stream.Stream;
import static org.apache.flink.table.types.inference.InputTypeStrategies.ANY;
@@ -43,6 +44,8 @@ import static org.apache.flink.table.types.inference.InputTypeStrategies.or;
import static org.apache.flink.table.types.inference.InputTypeStrategies.sequence;
import static org.apache.flink.table.types.inference.InputTypeStrategies.varyingSequence;
import static org.apache.flink.table.types.inference.strategies.SpecificInputTypeStrategies.INDEX;
+import static org.apache.flink.table.types.inference.strategies.SpecificInputTypeStrategies.percentage;
+import static org.apache.flink.table.types.inference.strategies.SpecificInputTypeStrategies.percentageArray;
/** Tests for built-in {@link InputTypeStrategies}. */
class InputTypeStrategiesTest extends InputTypeStrategiesTestBase {
@@ -751,7 +754,104 @@ class InputTypeStrategiesTest extends InputTypeStrategiesTestBase {
"Index must be an integer starting from '0', but was '-1'."),
TestSpec.forStrategy("IndexArgumentTypeStrategy index type", sequence(INDEX))
.calledWithArgumentTypes(DataTypes.DECIMAL(10, 5))
- .expectErrorMessage("Index can only be an INTEGER NUMERIC type."));
+ .expectErrorMessage("Index can only be an INTEGER NUMERIC type."),
+
+ // Percentage ArgumentStrategy
+ TestSpec.forStrategy("normal", sequence(percentage(true)))
+ .calledWithArgumentTypes(DataTypes.DOUBLE())
+ .expectSignature("f(<NUMERIC>)")
+ .expectArgumentTypes(DataTypes.DOUBLE()),
+ TestSpec.forStrategy("implicit cast", sequence(percentage(false)))
+ .calledWithArgumentTypes(DataTypes.DECIMAL(5, 2).notNull())
+ .expectSignature("f(<NUMERIC NOT NULL>)")
+ .expectArgumentTypes(DataTypes.DOUBLE().notNull()),
+ TestSpec.forStrategy("literal", sequence(percentage(true)))
+ .calledWithArgumentTypes(DataTypes.DECIMAL(2, 2))
+ .calledWithLiteralAt(0, BigDecimal.valueOf(45, 2))
+ .expectArgumentTypes(DataTypes.DOUBLE()),
+ TestSpec.forStrategy("literal", sequence(percentage(false)))
+ .calledWithArgumentTypes(DataTypes.INT().notNull())
+ .calledWithLiteralAt(0, 1)
+ .expectArgumentTypes(DataTypes.DOUBLE().notNull()),
+ TestSpec.forStrategy("invalid type", sequence(percentage(true)))
+ .calledWithArgumentTypes(DataTypes.STRING())
+ .expectErrorMessage("Percentage must be of NUMERIC type."),
+ TestSpec.forStrategy("invalid nullability", sequence(percentage(false)))
+ .calledWithArgumentTypes(DataTypes.DOUBLE())
+ .expectErrorMessage("Percentage must be of NOT NULL type."),
+ TestSpec.forStrategy("invalid literal value", sequence(percentage(false)))
+ .calledWithArgumentTypes(DataTypes.DECIMAL(2, 1).notNull())
+ .calledWithLiteralAt(0, BigDecimal.valueOf(20, 1))
+ .expectErrorMessage(
+ "Percentage must be between [0.0, 1.0], but was '2.0'."),
+ TestSpec.forStrategy("invalid literal value", sequence(percentage(false)))
+ .calledWithArgumentTypes(DataTypes.DECIMAL(2, 1).notNull())
+ .calledWithLiteralAt(0, BigDecimal.valueOf(-5, 1))
+ .expectErrorMessage(
+ "Percentage must be between [0.0, 1.0], but was '-0.5'."),
+
+ // Percentage Array ArgumentStrategy
+ TestSpec.forStrategy("normal", sequence(percentageArray(true)))
+ .calledWithArgumentTypes(DataTypes.ARRAY(DataTypes.DOUBLE()))
+ .expectSignature("f(ARRAY<NUMERIC>)")
+ .expectArgumentTypes(DataTypes.ARRAY(DataTypes.DOUBLE())),
+ TestSpec.forStrategy("implicit cast", sequence(percentageArray(false)))
+ .calledWithArgumentTypes(
+ DataTypes.ARRAY(DataTypes.DECIMAL(5, 2).notNull()).notNull())
+ .expectSignature("f(ARRAY<NUMERIC NOT NULL> NOT NULL)")
+ .expectArgumentTypes(
+ DataTypes.ARRAY(DataTypes.DOUBLE().notNull()).notNull()),
+ TestSpec.forStrategy("literal", sequence(percentageArray(true)))
+ .calledWithArgumentTypes(DataTypes.ARRAY(DataTypes.DOUBLE()))
+ .calledWithLiteralAt(0, new Double[] {0.45, 0.55})
+ .expectArgumentTypes(DataTypes.ARRAY(DataTypes.DOUBLE())),
+ TestSpec.forStrategy("literal", sequence(percentageArray(false)))
+ .calledWithArgumentTypes(
+ DataTypes.ARRAY(DataTypes.DECIMAL(2, 2).notNull()).notNull())
+ .calledWithLiteralAt(
+ 0,
+ new BigDecimal[] {
+ BigDecimal.valueOf(45, 2), BigDecimal.valueOf(55, 2)
+ })
+ .expectArgumentTypes(
+ DataTypes.ARRAY(DataTypes.DOUBLE().notNull()).notNull()),
+ TestSpec.forStrategy("literal", sequence(percentageArray(true)))
+ .calledWithArgumentTypes(DataTypes.ARRAY(DataTypes.INT()))
+ .calledWithLiteralAt(0, new Integer[] {0, 1})
+ .expectArgumentTypes(DataTypes.ARRAY(DataTypes.DOUBLE())),
+ TestSpec.forStrategy("empty literal array", sequence(percentageArray(true)))
+ .calledWithArgumentTypes(DataTypes.ARRAY(DataTypes.DOUBLE()))
+ .calledWithLiteralAt(0, new Double[0])
+ .expectArgumentTypes(DataTypes.ARRAY(DataTypes.DOUBLE())),
+ TestSpec.forStrategy("not array", sequence(percentageArray(true)))
+ .calledWithArgumentTypes(DataTypes.DOUBLE())
+ .expectErrorMessage("Percentage must be an array."),
+ TestSpec.forStrategy("invalid array nullability", sequence(percentageArray(false)))
+ .calledWithArgumentTypes(DataTypes.ARRAY(DataTypes.STRING().notNull()))
+ .expectErrorMessage("Percentage must be a non-null array."),
+ TestSpec.forStrategy("invalid element type", sequence(percentageArray(true)))
+ .calledWithArgumentTypes(DataTypes.ARRAY(DataTypes.STRING()))
+ .expectErrorMessage(
+ "Value in the percentage array must be of NUMERIC type."),
+ TestSpec.forStrategy(
+ "invalid element nullability", sequence(percentageArray(false)))
+ .calledWithArgumentTypes(DataTypes.ARRAY(DataTypes.DOUBLE()).notNull())
+ .expectErrorMessage(
+ "Value in the percentage array must be of NOT NULL type."),
+ TestSpec.forStrategy("invalid literal", sequence(percentageArray(true)))
+ .calledWithArgumentTypes(DataTypes.ARRAY(DataTypes.DOUBLE()))
+ .calledWithLiteralAt(0, new Double[] {0.5, 1.5})
+ .expectErrorMessage(
+ "Value in the percentage array must be between [0.0, 1.0], but was '1.5'."),
+ TestSpec.forStrategy("invalid literal", sequence(percentageArray(true)))
+ .calledWithArgumentTypes(DataTypes.ARRAY(DataTypes.DECIMAL(3, 2)))
+ .calledWithLiteralAt(
+ 0,
+ new BigDecimal[] {
+ BigDecimal.valueOf(-1, 1), BigDecimal.valueOf(5, 1)
+ })
+ .expectErrorMessage(
+ "Value in the percentage array must be between [0.0, 1.0], but was '-0.1'."));
}
private static DataType timeIndicatorType(TimestampKind timestampKind) {