You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tw...@apache.org on 2019/12/19 16:44:09 UTC
[flink] branch master updated: [FLINK-15149][table-common] Merge
InputTypeStrategy and InputTypeValidator
This is an automated email from the ASF dual-hosted git repository.
twalthr pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new e72a5a7 [FLINK-15149][table-common] Merge InputTypeStrategy and InputTypeValidator
e72a5a7 is described below
commit e72a5a7ea41b307a0b6393d1988a5ebf90720e23
Author: Timo Walther <tw...@apache.org>
AuthorDate: Mon Dec 9 13:11:11 2019 +0100
[FLINK-15149][table-common] Merge InputTypeStrategy and InputTypeValidator
This merges the concepts of InputTypeStrategy for inferring input data types
and enriching data types with conversion classes and InputTypeValidator for
validating arguments. It enables implicit casting and simplifies the overall
design.
This closes #10546.
---
.../resolver/rules/ResolveCallByArgumentsRule.java | 49 ++-
.../apache/flink/table/annotation/InputGroup.java | 6 +-
.../table/functions/BuiltInFunctionDefinition.java | 22 +-
.../types/extraction/utils/DataTypeTemplate.java | 16 +-
...ypeValidator.java => ArgumentTypeStrategy.java} | 22 +-
.../flink/table/types/inference/CallContext.java | 17 +
.../table/types/inference/InputTypeStrategies.java | 161 ++++++-
.../table/types/inference/InputTypeStrategy.java | 46 +-
.../table/types/inference/InputTypeValidator.java | 60 ---
.../table/types/inference/InputTypeValidators.java | 164 --------
.../table/types/inference/MutableCallContext.java | 50 ---
.../flink/table/types/inference/Signature.java | 2 +-
.../flink/table/types/inference/TypeInference.java | 90 ++--
.../table/types/inference/TypeInferenceUtil.java | 169 ++++----
.../table/types/inference/TypeStrategies.java | 10 +-
.../strategies/AndArgumentTypeStrategy.java | 111 +++++
.../AnyArgumentTypeStrategy.java} | 38 +-
.../strategies/BridgingInputTypeStrategy.java | 135 ------
.../ExplicitArgumentTypeStrategy.java} | 39 +-
.../strategies/ExplicitInputTypeStrategy.java | 67 ---
.../LiteralArgumentTypeStrategy.java} | 47 +--
...gTypeStrategy.java => MappingTypeStrategy.java} | 47 ++-
.../inference/strategies/NopInputTypeStrategy.java | 45 --
.../strategies/OrArgumentTypeStrategy.java | 122 ++++++
.../OrInputTypeStrategy.java} | 63 ++-
...rategy.java => OutputArgumentTypeStrategy.java} | 37 +-
.../SequenceInputTypeStrategy.java} | 60 +--
.../VaryingSequenceInputTypeStrategy.java | 150 +++++++
.../WildcardInputTypeStrategy.java} | 29 +-
.../types/inference/utils/AdaptedCallContext.java | 38 +-
.../types/inference/utils/UnknownCallContext.java | 15 +
.../validators/AndTypeArgumentValidator.java | 80 ----
.../validators/OrTypeArgumentValidator.java | 86 ----
.../validators/VaryingSequenceTypeValidator.java | 134 ------
.../types/inference/InputTypeStrategiesTest.java | 461 +++++++++++++++++----
.../types/inference/InputTypeValidatorsTest.java | 348 ----------------
.../table/types/inference/TypeStrategiesTest.java | 38 +-
.../types/inference/utils/CallContextMock.java | 19 +-
38 files changed, 1445 insertions(+), 1648 deletions(-)
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/resolver/rules/ResolveCallByArgumentsRule.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/resolver/rules/ResolveCallByArgumentsRule.java
index a31efa4..bc238f5 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/resolver/rules/ResolveCallByArgumentsRule.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/resolver/rules/ResolveCallByArgumentsRule.java
@@ -23,6 +23,8 @@ import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeutils.CompositeType;
import org.apache.flink.table.api.TableException;
import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.catalog.DataTypeLookup;
+import org.apache.flink.table.catalog.UnresolvedIdentifier;
import org.apache.flink.table.delegation.PlannerTypeInferenceUtil;
import org.apache.flink.table.expressions.CallExpression;
import org.apache.flink.table.expressions.Expression;
@@ -184,7 +186,12 @@ final class ResolveCallByArgumentsRule implements ResolverRule {
final Result inferenceResult = TypeInferenceUtil.runTypeInference(
inference,
- new TableApiCallContext(name, unresolvedCall.getFunctionDefinition(), resolvedArgs), surroundingInfo);
+ new TableApiCallContext(
+ new UnsupportedDataTypeLookup(),
+ name,
+ unresolvedCall.getFunctionDefinition(),
+ resolvedArgs),
+ surroundingInfo);
final List<ResolvedExpression> adaptedArguments = adaptArguments(inferenceResult, resolvedArgs);
@@ -268,8 +275,28 @@ final class ResolveCallByArgumentsRule implements ResolverRule {
// --------------------------------------------------------------------------------------------
+ private static class UnsupportedDataTypeLookup implements DataTypeLookup {
+
+ @Override
+ public Optional<DataType> lookupDataType(String name) {
+ throw new TableException("Data type lookup is not supported yet.");
+ }
+
+ @Override
+ public Optional<DataType> lookupDataType(UnresolvedIdentifier identifier) {
+ throw new TableException("Data type lookup is not supported yet.");
+ }
+
+ @Override
+ public DataType resolveRawDataType(Class<?> clazz) {
+ throw new TableException("Data type lookup is not supported yet.");
+ }
+ }
+
private static class TableApiCallContext implements CallContext {
+ private final DataTypeLookup lookup;
+
private final String name;
private final FunctionDefinition definition;
@@ -277,19 +304,19 @@ final class ResolveCallByArgumentsRule implements ResolverRule {
private final List<ResolvedExpression> resolvedArgs;
public TableApiCallContext(
+ DataTypeLookup lookup,
String name,
FunctionDefinition definition,
List<ResolvedExpression> resolvedArgs) {
+ this.lookup = lookup;
this.name = name;
this.definition = definition;
this.resolvedArgs = resolvedArgs;
}
@Override
- public List<DataType> getArgumentDataTypes() {
- return resolvedArgs.stream()
- .map(ResolvedExpression::getOutputDataType)
- .collect(Collectors.toList());
+ public DataTypeLookup getDataTypeLookup() {
+ return lookup;
}
@Override
@@ -321,6 +348,18 @@ final class ResolveCallByArgumentsRule implements ResolverRule {
return name;
}
+ @Override
+ public List<DataType> getArgumentDataTypes() {
+ return resolvedArgs.stream()
+ .map(ResolvedExpression::getOutputDataType)
+ .collect(Collectors.toList());
+ }
+
+ @Override
+ public Optional<DataType> getOutputDataType() {
+ return Optional.empty();
+ }
+
private ResolvedExpression getArgument(int pos) {
if (pos >= resolvedArgs.size()) {
throw new IndexOutOfBoundsException(
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/annotation/InputGroup.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/annotation/InputGroup.java
index 1a9034a..4103579 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/annotation/InputGroup.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/annotation/InputGroup.java
@@ -19,14 +19,14 @@
package org.apache.flink.table.annotation;
import org.apache.flink.annotation.PublicEvolving;
-import org.apache.flink.table.types.inference.InputTypeValidators;
+import org.apache.flink.table.types.inference.InputTypeStrategies;
import org.apache.flink.table.types.logical.LogicalTypeFamily;
/**
* A list of commonly used pre-defined groups of similar types for accepting more than just one data
* type as an input argument in {@link DataTypeHint}s.
*
- * <p>This list exposes a combination of {@link LogicalTypeFamily} and {@link InputTypeValidators}
+ * <p>This list exposes a combination of {@link LogicalTypeFamily} and {@link InputTypeStrategies}
* via annotations for convenient inline usage.
*/
@PublicEvolving
@@ -38,7 +38,7 @@ public enum InputGroup {
UNKNOWN,
/**
- * Enables input wildcards. Any data type can be passed. The behavior is equal to {@link InputTypeValidators#ANY}.
+ * Enables input wildcards. Any data type can be passed. The behavior is equal to {@link InputTypeStrategies#ANY}.
*
* <p>Note: The class of the annotated element must be {@link Object} as this is the super class
* of all possibly passed data types.
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinition.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinition.java
index bfa3fb0..8fdb280 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinition.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinition.java
@@ -20,7 +20,7 @@ package org.apache.flink.table.functions;
import org.apache.flink.annotation.Internal;
import org.apache.flink.table.types.DataType;
-import org.apache.flink.table.types.inference.InputTypeValidator;
+import org.apache.flink.table.types.inference.InputTypeStrategy;
import org.apache.flink.table.types.inference.TypeInference;
import org.apache.flink.table.types.inference.TypeStrategy;
import org.apache.flink.util.Preconditions;
@@ -103,28 +103,28 @@ public final class BuiltInFunctionDefinition implements FunctionDefinition {
return this;
}
- public Builder inputTypeValidator(InputTypeValidator inputTypeValidator) {
- this.typeInferenceBuilder.inputTypeValidator(inputTypeValidator);
+ public Builder namedArguments(List<String> argumentNames) {
+ this.typeInferenceBuilder.namedArguments(argumentNames);
return this;
}
- public Builder accumulatorTypeStrategy(TypeStrategy accumulatorTypeStrategy) {
- this.typeInferenceBuilder.accumulatorTypeStrategy(accumulatorTypeStrategy);
+ public Builder typedArguments(List<DataType> argumentTypes) {
+ this.typeInferenceBuilder.typedArguments(argumentTypes);
return this;
}
- public Builder outputTypeStrategy(TypeStrategy outputTypeStrategy) {
- this.typeInferenceBuilder.outputTypeStrategy(outputTypeStrategy);
+ public Builder inputTypeStrategy(InputTypeStrategy inputTypeStrategy) {
+ this.typeInferenceBuilder.inputTypeStrategy(inputTypeStrategy);
return this;
}
- public Builder namedArguments(List<String> argumentNames) {
- this.typeInferenceBuilder.namedArguments(argumentNames);
+ public Builder accumulatorTypeStrategy(TypeStrategy accumulatorTypeStrategy) {
+ this.typeInferenceBuilder.accumulatorTypeStrategy(accumulatorTypeStrategy);
return this;
}
- public Builder typedArguments(List<DataType> argumentTypes) {
- this.typeInferenceBuilder.typedArguments(argumentTypes);
+ public Builder outputTypeStrategy(TypeStrategy outputTypeStrategy) {
+ this.typeInferenceBuilder.outputTypeStrategy(outputTypeStrategy);
return this;
}
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/extraction/utils/DataTypeTemplate.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/extraction/utils/DataTypeTemplate.java
index 5a73a24..7b31001 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/extraction/utils/DataTypeTemplate.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/extraction/utils/DataTypeTemplate.java
@@ -27,9 +27,9 @@ import org.apache.flink.table.annotation.InputGroup;
import org.apache.flink.table.catalog.DataTypeLookup;
import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.extraction.DataTypeExtractor;
-import org.apache.flink.table.types.inference.ArgumentTypeValidator;
-import org.apache.flink.table.types.inference.InputTypeValidator;
-import org.apache.flink.table.types.inference.InputTypeValidators;
+import org.apache.flink.table.types.inference.ArgumentTypeStrategy;
+import org.apache.flink.table.types.inference.InputTypeStrategies;
+import org.apache.flink.table.types.inference.InputTypeStrategy;
import org.apache.flink.table.types.inference.TypeStrategies;
import org.apache.flink.table.types.inference.TypeStrategy;
@@ -44,7 +44,7 @@ import static org.apache.flink.table.types.extraction.utils.ExtractionUtils.extr
/**
* Internal representation of a {@link DataTypeHint} and template for creating a single {@link DataType}
- * or a {@link InputTypeValidator} for groups of {@link DataType}s.
+ * or a {@link InputTypeStrategy} for groups of {@link DataType}s.
*
* <p>All parameters of a template are optional. An empty annotation results in a template where all
* members are {@code null}.
@@ -212,17 +212,17 @@ public final class DataTypeTemplate {
}
/**
- * Converts this template into an {@link ArgumentTypeValidator}.
+ * Converts this template into an {@link ArgumentTypeStrategy}.
*/
- public ArgumentTypeValidator toArgumentTypeValidator() {
+ public ArgumentTypeStrategy toArgumentTypeStrategy() {
// data type
if (hasDataTypeDefinition()) {
- return InputTypeValidators.explicit(dataType);
+ return InputTypeStrategies.explicit(dataType);
}
// input group
else if (hasInputGroupDefinition()) {
if (inputGroup == InputGroup.ANY) {
- return InputTypeValidators.ANY;
+ return InputTypeStrategies.ANY;
}
}
throw ExtractionUtils.extractionError(
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/ArgumentTypeValidator.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/ArgumentTypeStrategy.java
similarity index 65%
rename from flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/ArgumentTypeValidator.java
rename to flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/ArgumentTypeStrategy.java
index dfc3614..af2fbfc 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/ArgumentTypeValidator.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/ArgumentTypeStrategy.java
@@ -21,24 +21,32 @@ package org.apache.flink.table.types.inference;
import org.apache.flink.annotation.PublicEvolving;
import org.apache.flink.table.api.ValidationException;
import org.apache.flink.table.functions.FunctionDefinition;
+import org.apache.flink.table.types.DataType;
+
+import java.util.Optional;
/**
- * Validator that checks a single input argument type of a function call.
+ * Strategy for inferring and validating a single input argument type of a function call.
*/
@PublicEvolving
-public interface ArgumentTypeValidator {
+public interface ArgumentTypeStrategy {
/**
- * Main logic for validating a single input type. Returns {@code true} if the argument is valid for the
- * given call, {@code false} otherwise.
+ * Main logic for inferring and validating an argument. Returns the data type that is valid for
+ * the given call. If the returned type differs from {@link CallContext#getArgumentDataTypes()} at
+ * {@code argumentPos}, a casting operation can be inserted. An empty result means that the given
+ * input type could not be inferred.
*
* @param callContext provides details about the function call
* @param argumentPos argument index in the {@link CallContext}
* @param throwOnFailure whether this function is allowed to throw an {@link ValidationException}
- * with a meaningful exception in case the validation is not successful or
- * if this function should simply return {@code false}.
+ * with a meaningful exception in case the inference is not successful or
+ * if this function should simply return an empty result.
+ * @return three-state result for either "true, same data type as argument", "true, but argument
+ * must be casted to returned data type", or "false, no inferred data type could be found"
+ * @see CallContext#newValidationError(String, Object...)
*/
- boolean validateArgument(CallContext callContext, int argumentPos, boolean throwOnFailure);
+ Optional<DataType> inferArgumentType(CallContext callContext, int argumentPos, boolean throwOnFailure);
/**
* Returns a summary of the function's expected argument at {@code argumentPos}.
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/CallContext.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/CallContext.java
index 780a12c..f43b29c 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/CallContext.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/CallContext.java
@@ -19,7 +19,9 @@
package org.apache.flink.table.types.inference;
import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.catalog.DataTypeLookup;
import org.apache.flink.table.functions.FunctionDefinition;
import org.apache.flink.table.types.DataType;
@@ -33,6 +35,11 @@ import java.util.Optional;
public interface CallContext {
/**
+ * Enables to lookup types in a catalog and resolve RAW types.
+ */
+ DataTypeLookup getDataTypeLookup();
+
+ /**
* Returns the function definition that defines the function currently being called.
*/
FunctionDefinition getFunctionDefinition();
@@ -72,6 +79,16 @@ public interface CallContext {
List<DataType> getArgumentDataTypes();
/**
+ * Returns the inferred output data type of the function call.
+ *
+ * <p>It does this by inferring the input argument data type using
+ * {@link ArgumentTypeStrategy#inferArgumentType(CallContext, int, boolean)} of a wrapping call (if
+ * available) where this function call is an argument. For example, {@code takes_string(this_function(NULL))}
+ * would lead to a {@link DataTypes#STRING()} because the wrapping call expects a string argument.
+ */
+ Optional<DataType> getOutputDataType();
+
+ /**
* Creates a validation error for exiting the type inference process with a meaningful exception.
*/
default ValidationException newValidationError(String message, Object... args) {
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/InputTypeStrategies.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/InputTypeStrategies.java
index bd45c36..3ca09fa 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/InputTypeStrategies.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/InputTypeStrategies.java
@@ -20,49 +20,170 @@ package org.apache.flink.table.types.inference;
import org.apache.flink.annotation.Internal;
import org.apache.flink.table.types.DataType;
-import org.apache.flink.table.types.inference.strategies.BridgingInputTypeStrategy;
-import org.apache.flink.table.types.inference.strategies.BridgingInputTypeStrategy.BridgingSignature;
-import org.apache.flink.table.types.inference.strategies.ExplicitInputTypeStrategy;
-import org.apache.flink.table.types.inference.strategies.NopInputTypeStrategy;
-import org.apache.flink.table.types.inference.strategies.OutputTypeInputTypeStrategy;
-import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.inference.strategies.AndArgumentTypeStrategy;
+import org.apache.flink.table.types.inference.strategies.AnyArgumentTypeStrategy;
+import org.apache.flink.table.types.inference.strategies.ExplicitArgumentTypeStrategy;
+import org.apache.flink.table.types.inference.strategies.LiteralArgumentTypeStrategy;
+import org.apache.flink.table.types.inference.strategies.OrArgumentTypeStrategy;
+import org.apache.flink.table.types.inference.strategies.OrInputTypeStrategy;
+import org.apache.flink.table.types.inference.strategies.OutputArgumentTypeStrategy;
+import org.apache.flink.table.types.inference.strategies.SequenceInputTypeStrategy;
+import org.apache.flink.table.types.inference.strategies.VaryingSequenceInputTypeStrategy;
+import org.apache.flink.table.types.inference.strategies.WildcardInputTypeStrategy;
import java.util.Arrays;
import java.util.List;
+import java.util.stream.Collectors;
/**
- * Strategies for inferring missing or incomplete input argument data types.
+ * Strategies for inferring and validating input arguments in a function call.
*
* @see InputTypeStrategy
+ * @see ArgumentTypeStrategy
*/
@Internal
public final class InputTypeStrategies {
+ // --------------------------------------------------------------------------------------------
+ // Input type strategies
+ // --------------------------------------------------------------------------------------------
+
+ /**
+ * Strategy that does not perform any modification or validation of the input.
+ */
+ public static final WildcardInputTypeStrategy WILDCARD = new WildcardInputTypeStrategy();
+
+ /**
+ * Strategy for a function signature like {@code f(STRING, NUMERIC)} using a sequence of
+ * {@link ArgumentTypeStrategy}s.
+ */
+ public static InputTypeStrategy sequence(ArgumentTypeStrategy... strategies) {
+ return new SequenceInputTypeStrategy(Arrays.asList(strategies), null);
+ }
+
+ /**
+ * Strategy for a named function signature like {@code f(s STRING, n NUMERIC)} using a sequence
+ * of {@link ArgumentTypeStrategy}s.
+ */
+ public static InputTypeStrategy sequence(String[] argumentNames, ArgumentTypeStrategy[] strategies) {
+ return new SequenceInputTypeStrategy(Arrays.asList(strategies), Arrays.asList(argumentNames));
+ }
+
+ /**
+ * Strategy for a varying function signature like {@code f(INT, STRING, NUMERIC...)} using a
+ * sequence of {@link ArgumentTypeStrategy}s. The first n - 1 arguments must be constant. The
+ * n-th argument can occur 0, 1, or more times.
+ */
+ public static InputTypeStrategy varyingSequence(ArgumentTypeStrategy... strategies) {
+ return new VaryingSequenceInputTypeStrategy(Arrays.asList(strategies), null);
+ }
+
/**
- * Input strategy that does nothing.
+ * Strategy for a varying named function signature like {@code f(i INT, str STRING, num NUMERIC...)}
+ * using a sequence of {@link ArgumentTypeStrategy}s. The first n - 1 arguments must be constant. The
+ * n-th argument can occur 0, 1, or more times.
*/
- public static final InputTypeStrategy NOP = new NopInputTypeStrategy();
+ public static InputTypeStrategy varyingSequence(String[] argumentNames, ArgumentTypeStrategy[] strategies) {
+ return new VaryingSequenceInputTypeStrategy(Arrays.asList(strategies), Arrays.asList(argumentNames));
+ }
+
+ /**
+ * Strategy for a function signature of explicitly defined types like {@code f(STRING, INT)}.
+ * Implicit casts will be inserted if possible.
+ *
+ * <p>This is equivalent to using {@link #sequence(ArgumentTypeStrategy...)} and
+ * {@link #explicit(DataType)}.
+ */
+ public static InputTypeStrategy explicitSequence(DataType... expectedDataTypes) {
+ final List<ArgumentTypeStrategy> strategies = Arrays.stream(expectedDataTypes)
+ .map(InputTypeStrategies::explicit)
+ .collect(Collectors.toList());
+ return new SequenceInputTypeStrategy(strategies, null);
+ }
/**
- * Input type strategy that supplies the function's output {@link DataType} for each unknown
- * argument if available.
+ * Strategy for a named function signature of explicitly defined types like {@code f(s STRING, i INT)}.
+ * Implicit casts will be inserted if possible.
+ *
+ * <p>This is equivalent to using {@link #sequence(String[], ArgumentTypeStrategy[])} and
+ * {@link #explicit(DataType)}.
*/
- public static final OutputTypeInputTypeStrategy OUTPUT_TYPE = new OutputTypeInputTypeStrategy();
+ public static InputTypeStrategy explicitSequence(String[] argumentNames, DataType[] expectedDataTypes) {
+ final List<ArgumentTypeStrategy> strategies = Arrays.stream(expectedDataTypes)
+ .map(InputTypeStrategies::explicit)
+ .collect(Collectors.toList());
+ return new SequenceInputTypeStrategy(strategies, Arrays.asList(argumentNames));
+ }
+
+ /**
+ * Strategy for a disjunction of multiple {@link InputTypeStrategy}s into one like
+ * {@code f(NUMERIC) || f(STRING)}.
+ */
+ public static InputTypeStrategy or(InputTypeStrategy... strategies) {
+ return new OrInputTypeStrategy(Arrays.asList(strategies));
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Argument type strategies
+ // --------------------------------------------------------------------------------------------
+
+ /**
+ * Strategy for inferring an unknown argument type from the function's output {@link DataType}
+ * if available.
+ */
+ public static final OutputArgumentTypeStrategy OUTPUT_IF_NULL = new OutputArgumentTypeStrategy();
+
+ /**
+ * Strategy for an argument that can be of any type.
+ */
+ public static final AnyArgumentTypeStrategy ANY = new AnyArgumentTypeStrategy();
+
+ /**
+ * Strategy that checks if an argument is a literal.
+ */
+ public static final LiteralArgumentTypeStrategy LITERAL = new LiteralArgumentTypeStrategy(false);
+
+ /**
+ * Strategy that checks if an argument is a literal or NULL.
+ */
+ public static final LiteralArgumentTypeStrategy LITERAL_OR_NULL = new LiteralArgumentTypeStrategy(true);
+
+ /**
+ * Strategy for an argument that corresponds to an explicitly defined type casting.
+ * Implicit casts will be inserted if possible.
+ */
+ public static ExplicitArgumentTypeStrategy explicit(DataType expectedDataType) {
+ return new ExplicitArgumentTypeStrategy(expectedDataType);
+ }
/**
- * Input type strategy that supplies a fixed {@link DataType} for each argument.
+ * Strategy for a conjunction of multiple {@link ArgumentTypeStrategy}s into one like
+ * {@code f(NUMERIC && LITERAL)}.
+ *
+ * <p>Some {@link ArgumentTypeStrategy}s cannot contribute an inferred type that is different from
+ * the input type (e.g. {@link #LITERAL}). Therefore, the order {@code f(X && Y)} or {@code f(Y && X)}
+ * matters as it defines the precedence in case the result must be casted to a more specific type.
+ *
+ * <p>This strategy aims to infer the first more specific, casted type or (if this is not possible)
+ * a type that has been inferred from all {@link ArgumentTypeStrategy}s.
*/
- public static ExplicitInputTypeStrategy explicit(DataType... dataTypes) {
- return new ExplicitInputTypeStrategy(Arrays.asList(dataTypes));
+ public static AndArgumentTypeStrategy and(ArgumentTypeStrategy... strategies) {
+ return new AndArgumentTypeStrategy(Arrays.asList(strategies));
}
/**
- * Special input type strategy for enriching data types with an expected conversion class. This
- * is in particular useful when a data type has been created out of a {@link LogicalType} but
- * runtime hints are still missing.
+ * Strategy for a disjunction of multiple {@link ArgumentTypeStrategy}s into one like
+ * {@code f(NUMERIC || STRING)}.
+ *
+ * <p>Some {@link ArgumentTypeStrategy}s cannot contribute an inferred type that is different from
+ * the input type (e.g. {@link #LITERAL}). Therefore, the order {@code f(X || Y)} or {@code f(Y || X)}
+ * matters as it defines the precedence in case the result must be casted to a more specific type.
+ *
+ * <p>This strategy aims to infer a type that is equal to the input type (to prevent unnecessary casting)
+ * or (if this is not possible) the first more specific, casted type.
*/
- public static BridgingInputTypeStrategy bridging(List<BridgingSignature> bridgingSignatures) {
- return new BridgingInputTypeStrategy(bridgingSignatures);
+ public static OrArgumentTypeStrategy or(ArgumentTypeStrategy... strategies) {
+ return new OrArgumentTypeStrategy(Arrays.asList(strategies));
}
// --------------------------------------------------------------------------------------------
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/InputTypeStrategy.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/InputTypeStrategy.java
index dd93afd..545addd 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/InputTypeStrategy.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/InputTypeStrategy.java
@@ -19,21 +19,25 @@
package org.apache.flink.table.types.inference;
import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.functions.FunctionDefinition;
import org.apache.flink.table.types.DataType;
+import java.util.List;
+import java.util.Optional;
+
/**
- * Strategy for inferring missing or incomplete input argument data types in a function call.
+ * Strategy for inferring and validating input arguments in a function call.
+ *
+ * <p>This interface has three responsibilities during the {@link TypeInference} process:
*
- * <p>This interface has two responsibilities:
+ * <p>It can help in resolving the type of untyped {@code NULL} literals.
*
- * <p>In the {@link TypeInference} process, it is called before the validation of input arguments and
- * can help in resolving the type of untyped {@code NULL} literals.
+ * <p>It validates the types of the input arguments.
*
- * <p>During the planning process, it can help in resolving the actual {@link DataType} including the
+ * <p>During the planning process, it can help in resolving the complete {@link DataType}, i.e., the
* conversion class that a function implementation expects from the runtime. This requires that a
- * strategy can also be called on already validated arguments without affecting the logical type. This
- * is different from Calcite where unknown types are resolved first and might be overwritten by more
- * concrete types if available.
+ * strategy can also be called on already validated arguments without affecting the logical type.
*
* <p>Note: Implementations should implement {@link Object#hashCode()} and {@link Object#equals(Object)}.
*
@@ -43,7 +47,29 @@ import org.apache.flink.table.types.DataType;
public interface InputTypeStrategy {
/**
- * Infers the argument types of a function call.
+ * Initial input validation based on the number of arguments.
+ */
+ ArgumentCount getArgumentCount();
+
+ /**
+ * Main logic for inferring and validating the input arguments. Returns a list of argument data types that
+ * are valid for the given call. If the returned types differ from {@link CallContext#getArgumentDataTypes()},
+ * a casting operation can be inserted. An empty result means that the given input is invalid.
+ *
+ * @param callContext provides details about the function call
+ * @param throwOnFailure whether this function is allowed to throw an {@link ValidationException}
+ * with a meaningful exception in case the inference is not successful or
+ * if this function should simply return an empty result.
+ * @return three-state result for either "true, same data types as arguments", "true, but arguments
+ * must be casted to returned data types", or "false, no inferred data types could be found"
+ * @see CallContext#newValidationError(String, Object...)
+ */
+ Optional<List<DataType>> inferInputTypes(CallContext callContext, boolean throwOnFailure);
+
+ /**
+ * Returns a summary of the function's expected signatures.
+ *
+ * @param definition the function definition that defines the function currently being called.
*/
- void inferInputTypes(MutableCallContext callContext);
+ List<Signature> getExpectedSignatures(FunctionDefinition definition);
}
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/InputTypeValidator.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/InputTypeValidator.java
deleted file mode 100644
index 52687fe..0000000
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/InputTypeValidator.java
+++ /dev/null
@@ -1,60 +0,0 @@
-/*
- * 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;
-
-import org.apache.flink.annotation.PublicEvolving;
-import org.apache.flink.table.api.ValidationException;
-import org.apache.flink.table.functions.FunctionDefinition;
-
-import java.util.List;
-
-/**
- * Validator for checking the input data types of a function call.
- *
- * <p>Note: Implementations should implement {@link Object#hashCode()} and {@link Object#equals(Object)}.
- *
- * @see InputTypeValidators
- */
-@PublicEvolving
-public interface InputTypeValidator {
-
- /**
- * Initial input validation based on the number of arguments.
- */
- ArgumentCount getArgumentCount();
-
- /**
- * Main logic for validating the input. Returns {@code true} if the arguments are valid for the
- * given call, {@code false} otherwise.
- *
- * @param callContext provides details about the function call
- * @param throwOnFailure whether this function is allowed to throw an {@link ValidationException}
- * with a meaningful exception in case the validation is not successful or
- * if this function should simply return {@code false}.
- * @see CallContext#newValidationError(String, Object...)
- */
- boolean validate(CallContext callContext, boolean throwOnFailure);
-
- /**
- * Returns a summary of the function's expected signatures.
- *
- * @param definition the function definition that defines the function currently being called.
- */
- List<Signature> getExpectedSignatures(FunctionDefinition definition);
-}
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/InputTypeValidators.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/InputTypeValidators.java
deleted file mode 100644
index 0790147..0000000
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/InputTypeValidators.java
+++ /dev/null
@@ -1,164 +0,0 @@
-/*
- * 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;
-
-import org.apache.flink.annotation.Internal;
-import org.apache.flink.table.types.DataType;
-import org.apache.flink.table.types.inference.validators.AndTypeArgumentValidator;
-import org.apache.flink.table.types.inference.validators.AnyTypeValidator;
-import org.apache.flink.table.types.inference.validators.ExplicitTypeValidator;
-import org.apache.flink.table.types.inference.validators.LiteralTypeValidator;
-import org.apache.flink.table.types.inference.validators.OrTypeArgumentValidator;
-import org.apache.flink.table.types.inference.validators.OrTypeInputValidator;
-import org.apache.flink.table.types.inference.validators.PassingTypeValidator;
-import org.apache.flink.table.types.inference.validators.SequenceInputValidator;
-import org.apache.flink.table.types.inference.validators.VaryingSequenceTypeValidator;
-import org.apache.flink.table.types.logical.LogicalType;
-
-import java.util.Arrays;
-import java.util.List;
-import java.util.stream.Collectors;
-
-/**
- * Validators for checking the input data types of a function call.
- *
- * @see InputTypeValidator
- * @see ArgumentTypeValidator
- */
-@Internal
-public final class InputTypeValidators {
-
- /**
- * Validator that does not perform any validation and always passes.
- */
- public static final PassingTypeValidator PASSING = new PassingTypeValidator();
-
- /**
- * Validator that checks for a single argument that can be of any type.
- */
- public static final AnyTypeValidator ANY = new AnyTypeValidator();
-
- /**
- * Validator that checks if a single argument is a literal.
- */
- public static final LiteralTypeValidator LITERAL = new LiteralTypeValidator(false);
-
- /**
- * Validator that checks if a single argument is a literal or null.
- */
- public static final LiteralTypeValidator LITERAL_OR_NULL = new LiteralTypeValidator(true);
-
- /**
- * Validator that checks if a single argument corresponds to an explicitly defined logical type.
- *
- * <p>Note: The validation happens on {@link LogicalType} level only.
- */
- public static ArgumentTypeValidator explicit(DataType expectedDataType) {
- return new ExplicitTypeValidator(expectedDataType.getLogicalType());
- }
-
- /**
- * Conjunction of multiple {@link ArgumentTypeValidator}s into one like {@code f(NUMERIC && LITERAL)}.
- */
- public static ArgumentTypeValidator and(ArgumentTypeValidator... validators) {
- return new AndTypeArgumentValidator(Arrays.asList(validators));
- }
-
- /**
- * Conjunction of multiple {@link ArgumentTypeValidator}s into one like {@code f(NUMERIC || STRING)}.
- */
- public static ArgumentTypeValidator or(ArgumentTypeValidator... validators) {
- return new OrTypeArgumentValidator(Arrays.asList(validators));
- }
-
- /**
- * Conjunction of multiple {@link InputTypeValidator}s into one like {@code f(NUMERIC) || f(STRING)}.
- */
- public static InputTypeValidator or(InputTypeValidator... validators) {
- return new OrTypeInputValidator(Arrays.asList(validators));
- }
-
- /**
- * Validator that checks if each operand corresponds to an explicitly defined logical type
- * like {@code f(STRING, INT)}.
- *
- * <p>Note: The validation happens on {@link LogicalType} level only.
- */
- public static InputTypeValidator explicitSequence(DataType... expectedDataTypes) {
- final List<ArgumentTypeValidator> validators = Arrays.stream(expectedDataTypes)
- .map(InputTypeValidators::explicit)
- .collect(Collectors.toList());
- return new SequenceInputValidator(validators, null);
- }
-
- /**
- * Validator that checks if each named operand corresponds to an explicitly defined logical type
- * like {@code f(s STRING, i INT)}.
- *
- * <p>Note: The validation happens on {@link LogicalType} level only.
- */
- public static InputTypeValidator explicitSequence(String[] argumentNames, DataType[] expectedDataTypes) {
- final List<ArgumentTypeValidator> validators = Arrays.stream(expectedDataTypes)
- .map(InputTypeValidators::explicit)
- .collect(Collectors.toList());
- return new SequenceInputValidator(validators, Arrays.asList(argumentNames));
- }
-
- /**
- * A sequence of {@link ArgumentTypeValidator}s for validating an entire function signature
- * like {@code f(STRING, NUMERIC)}.
- */
- public static InputTypeValidator sequence(ArgumentTypeValidator... validators) {
- return new SequenceInputValidator(Arrays.asList(validators), null);
- }
-
- /**
- * A sequence of {@link ArgumentTypeValidator}s for validating an entire named function signature
- * like {@code f(s STRING, n NUMERIC)}.
- */
- public static InputTypeValidator sequence(String[] argumentNames, ArgumentTypeValidator[] validators) {
- return new SequenceInputValidator(Arrays.asList(validators), Arrays.asList(argumentNames));
- }
-
- /**
- * A varying sequence of {@link ArgumentTypeValidator}s for validating an entire function signature
- * like {@code f(INT, STRING, NUMERIC...)}. The first n - 1 arguments must be constant and are validated
- * according to {@link #sequence(ArgumentTypeValidator[])}. The n-th argument can occur 0, 1, or
- * more times.
- */
- public static InputTypeValidator varyingSequence(ArgumentTypeValidator... validators) {
- return new VaryingSequenceTypeValidator(Arrays.asList(validators), null);
- }
-
- /**
- * A varying sequence of {@link ArgumentTypeValidator}s for validating an entire function signature
- * like {@code f(i INT, str STRING, n NUMERIC...)}. The first n - 1 arguments must be constant and are validated
- * according to {@link #sequence(String[], ArgumentTypeValidator[])}. The n-th argument can occur 0, 1,
- * or more times.
- */
- public static InputTypeValidator varyingSequence(String[] argumentNames, ArgumentTypeValidator[] validators) {
- return new VaryingSequenceTypeValidator(Arrays.asList(validators), Arrays.asList(argumentNames));
- }
-
- // --------------------------------------------------------------------------------------------
-
- private InputTypeValidators() {
- // no instantiation
- }
-}
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/MutableCallContext.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/MutableCallContext.java
deleted file mode 100644
index d07697a..0000000
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/MutableCallContext.java
+++ /dev/null
@@ -1,50 +0,0 @@
-/*
- * 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;
-
-import org.apache.flink.annotation.PublicEvolving;
-import org.apache.flink.table.api.DataTypes;
-import org.apache.flink.table.types.DataType;
-
-import java.util.Optional;
-
-/**
- * Provides details about a function call during {@link TypeInference} and allows to mutate argument
- * data types for {@link InputTypeStrategy}.
- *
- * <p>Note: In particular, this method allows to enrich {@link DataTypes#NULL()} to a meaningful data
- * type or modify a {@link DataType}'s conversion class.
- */
-@PublicEvolving
-public interface MutableCallContext extends CallContext {
-
- /**
- * Mutates the data type of an argument at the given position.
- */
- void mutateArgumentDataType(int pos, DataType newDataType);
-
- /**
- * Returns the inferred output data type of the function call.
- *
- * <p>It does this by inferring the input argument data type of a wrapping call (if available)
- * where this function call is an argument. For example, {@code takes_string(this_function(NULL))}
- * would lead to a {@link DataTypes#STRING()} because the wrapping call expects a string argument.
- */
- Optional<DataType> getOutputDataType();
-}
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/Signature.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/Signature.java
index cf53f74..4cacd53 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/Signature.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/Signature.java
@@ -32,7 +32,7 @@ import java.util.Optional;
* Describes the signature of a function. It is meant for representing information for debugging
* purposes.
*
- * <p>A signature is returned from {@link InputTypeValidator#getExpectedSignatures(FunctionDefinition)}.
+ * <p>A signature is returned from {@link InputTypeStrategy#getExpectedSignatures(FunctionDefinition)}.
*/
@PublicEvolving
public final class Signature {
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/TypeInference.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/TypeInference.java
index 220121f..1581a84 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/TypeInference.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/TypeInference.java
@@ -30,7 +30,7 @@ import java.util.Optional;
/**
* Provides logic for the type inference of function calls. It includes:
* <ul>
- * <li>explicit input specification for (possibly named) arguments</li>
+ * <li>explicit input specification for (possibly named and/or typed) arguments</li>
* <li>inference of missing or incomplete input types</li>
* <li>validation of input types</li>
* <li>inference of an intermediate accumulation type</li>
@@ -42,37 +42,33 @@ import java.util.Optional;
@PublicEvolving
public final class TypeInference {
- private final @Nullable List<String> argumentNames;
+ private final @Nullable List<String> namedArguments;
- private final @Nullable List<DataType> argumentTypes;
+ private final @Nullable List<DataType> typedArguments;
- private final @Nullable InputTypeStrategy inputTypeStrategy;
-
- private final InputTypeValidator inputTypeValidator;
+ private final InputTypeStrategy inputTypeStrategy;
private final @Nullable TypeStrategy accumulatorTypeStrategy;
private final TypeStrategy outputTypeStrategy;
private TypeInference(
- @Nullable List<String> argumentNames,
- @Nullable List<DataType> argumentTypes,
- @Nullable InputTypeStrategy inputTypeStrategy,
- InputTypeValidator inputTypeValidator,
+ @Nullable List<String> namedArguments,
+ @Nullable List<DataType> typedArguments,
+ InputTypeStrategy inputTypeStrategy,
@Nullable TypeStrategy accumulatorTypeStrategy,
TypeStrategy outputTypeStrategy) {
- this.argumentNames = argumentNames;
- this.argumentTypes = argumentTypes;
+ this.namedArguments = namedArguments;
+ this.typedArguments = typedArguments;
this.inputTypeStrategy = inputTypeStrategy;
- this.inputTypeValidator = inputTypeValidator;
this.accumulatorTypeStrategy = accumulatorTypeStrategy;
this.outputTypeStrategy = outputTypeStrategy;
- if (argumentNames != null && argumentTypes != null && argumentNames.size() != argumentTypes.size()) {
+ if (namedArguments != null && typedArguments != null && namedArguments.size() != typedArguments.size()) {
throw new IllegalArgumentException(
String.format(
- "Mismatch between argument types %d and argument names %d.",
- argumentNames.size(),
- argumentTypes.size()));
+ "Mismatch between typed arguments %d and named argument %d.",
+ namedArguments.size(),
+ typedArguments.size()));
}
}
@@ -83,20 +79,16 @@ public final class TypeInference {
return new TypeInference.Builder();
}
- public Optional<List<String>> getArgumentNames() {
- return Optional.ofNullable(argumentNames);
- }
-
- public Optional<List<DataType>> getArgumentTypes() {
- return Optional.ofNullable(argumentTypes);
+ public Optional<List<String>> getNamedArguments() {
+ return Optional.ofNullable(namedArguments);
}
- public Optional<InputTypeStrategy> getInputTypeStrategy() {
- return Optional.ofNullable(inputTypeStrategy);
+ public Optional<List<DataType>> getTypedArguments() {
+ return Optional.ofNullable(typedArguments);
}
- public InputTypeValidator getInputTypeValidator() {
- return inputTypeValidator;
+ public InputTypeStrategy getInputTypeStrategy() {
+ return inputTypeStrategy;
}
public Optional<TypeStrategy> getAccumulatorTypeStrategy() {
@@ -114,13 +106,11 @@ public final class TypeInference {
*/
public static class Builder {
- private @Nullable List<String> argumentNames;
+ private @Nullable List<String> namedArguments;
- private @Nullable List<DataType> argumentTypes;
+ private @Nullable List<DataType> typedArguments;
- private @Nullable InputTypeStrategy inputTypeStrategy;
-
- private InputTypeValidator inputTypeValidator = InputTypeValidators.PASSING;
+ private InputTypeStrategy inputTypeStrategy = InputTypeStrategies.WILDCARD;
private @Nullable TypeStrategy accumulatorTypeStrategy;
@@ -131,30 +121,36 @@ public final class TypeInference {
}
/**
- * Sets the list of argument names for specifying static input explicitly.
+ * Sets the list of argument names for specifying a fixed, not overloaded, not vararg input
+ * signature explicitly.
*
* <p>This information is useful for SQL's concept of named arguments using the assignment
- * operator (e.g. {@code FUNC(max => 42)}).
+ * operator (e.g. {@code FUNC(max => 42)}). The names are used for reordering the call's
+ * arguments to the formal argument order of the function.
*/
public Builder namedArguments(List<String> argumentNames) {
- this.argumentNames =
+ this.namedArguments =
Preconditions.checkNotNull(argumentNames, "List of argument names must not be null.");
return this;
}
/**
- * Sets the list of argument types for specifying static input explicitly.
+ * Sets the list of argument types for specifying a fixed, not overloaded, not vararg input
+ * signature explicitly.
*
- * <p>This information is useful for implicit and safe casting.
+ * <p>This information is useful for optional arguments with default value. In particular, the
+ * number of arguments that need to be filled with a default value and their types is important.
*/
public Builder typedArguments(List<DataType> argumentTypes) {
- this.argumentTypes =
+ this.typedArguments =
Preconditions.checkNotNull(argumentTypes, "List of argument types must not be null.");
return this;
}
/**
- * Sets the strategy for inferring missing or incomplete input argument data types.
+ * Sets the strategy for inferring and validating input arguments in a function call.
+ *
+ * <p>A {@link InputTypeStrategies#WILDCARD} strategy function is assumed by default.
*/
public Builder inputTypeStrategy(InputTypeStrategy inputTypeStrategy) {
this.inputTypeStrategy =
@@ -163,17 +159,6 @@ public final class TypeInference {
}
/**
- * Sets the validator for checking the input data types of a function call.
- *
- * <p>A always passing function is assumed by default (see {@link InputTypeValidators#PASSING}).
- */
- public Builder inputTypeValidator(InputTypeValidator inputTypeValidator) {
- this.inputTypeValidator =
- Preconditions.checkNotNull(inputTypeValidator, "Input type validator must not be null.");
- return this;
- }
-
- /**
* Sets the strategy for inferring the intermediate accumulator data type of a function call.
*/
public Builder accumulatorTypeStrategy(TypeStrategy accumulatorTypeStrategy) {
@@ -194,11 +179,8 @@ public final class TypeInference {
}
public TypeInference build() {
- return new TypeInference(
- argumentNames,
- argumentTypes,
+ return new TypeInference(namedArguments, typedArguments,
inputTypeStrategy,
- inputTypeValidator,
accumulatorTypeStrategy,
Preconditions.checkNotNull(outputTypeStrategy, "Output type strategy must not be null."));
}
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/TypeInferenceUtil.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/TypeInferenceUtil.java
index ffdd99b..783eb79 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/TypeInferenceUtil.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/TypeInferenceUtil.java
@@ -21,20 +21,22 @@ package org.apache.flink.table.types.inference;
import org.apache.flink.annotation.Internal;
import org.apache.flink.table.api.TableException;
import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.catalog.DataTypeLookup;
import org.apache.flink.table.functions.FunctionDefinition;
import org.apache.flink.table.functions.FunctionKind;
import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.inference.utils.AdaptedCallContext;
import org.apache.flink.table.types.inference.utils.UnknownCallContext;
import org.apache.flink.table.types.logical.LogicalTypeRoot;
-import org.apache.flink.table.types.logical.utils.LogicalTypeCasts;
import javax.annotation.Nullable;
import java.util.List;
import java.util.Optional;
import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import static org.apache.flink.table.types.logical.utils.LogicalTypeCasts.supportsImplicitCast;
import static org.apache.flink.table.types.logical.utils.LogicalTypeChecks.hasRoot;
/**
@@ -45,9 +47,9 @@ import static org.apache.flink.table.types.logical.utils.LogicalTypeChecks.hasRo
*
* <p><ul>
* <li>1. Validate number of arguments.
- * <li>2. (*) Apply assignment operators on the call expression by permuting operands and adding default
- * expressions. These are preparations for {@link CallContext}.
- * <li>3. For unknown (NULL) operands: Access the outer wrapping call and try to get its operand
+ * <li>2. (*) Apply assignment operators on the call by permuting operands and adding default values. These
+ * are preparations for {@link CallContext}.
+ * <li>3. For resolving unknown (NULL) operands: Access the outer wrapping call and try to get its operand
* type for the return type of the actual call. E.g. for {@code takes_string(this_function(NULL))}
* infer operands from {@code takes_string(NULL)} and use the inferred string type as the return
* type of {@code this_function(NULL)}.
@@ -55,16 +57,34 @@ import static org.apache.flink.table.types.logical.utils.LogicalTypeChecks.hasRo
* <li>5. (*) Check the usage of DEFAULT operands are correct using validator.isOptional().
* <li>6. Perform input type validation.
* <li>7. (Optional) Infer accumulator type.
- * <li>8. (*) Check for an implementation evaluation method matching the operands. The matching happens
- * class-based. Thus, for example, eval(Object) is valid for (INT). Or eval(Object...) is valid
- * for (INT, STRING). We rely on the conversion classes specified by DataType.
- * <li>9. Infer return type.
+ * <li>8. Infer return type.
+ * <li>9. (*) In the planner: Call the strategies again at any point in time to enrich a DataType that
+ * has been created from a logical type with a conversion class.
+ * <li>10. (*) In the planner: Check for an implementation evaluation method matching the operands. The
+ * matching happens class-based. Thus, for example, eval(Object) is valid for (INT). Or eval(Object...)
+ * is valid for (INT, STRING). We rely on the conversion classes specified by DataType.
* </ul>
*/
@Internal
public final class TypeInferenceUtil {
/**
+ * Generates a signature of the given {@link FunctionDefinition}.
+ */
+ public static String generateSignature(
+ String name,
+ FunctionDefinition definition,
+ TypeInference typeInference) {
+ if (typeInference.getNamedArguments().isPresent() || typeInference.getTypedArguments().isPresent()) {
+ return formatNamedOrTypedArguments(name, typeInference);
+ }
+ return typeInference.getInputTypeStrategy().getExpectedSignatures(definition)
+ .stream()
+ .map(s -> formatSignature(name, s))
+ .collect(Collectors.joining("\n"));
+ }
+
+ /**
* Runs the type inference process.
*
* @param typeInference type inference of the current call
@@ -98,9 +118,9 @@ public final class TypeInferenceUtil {
/**
* Information what the outer world (i.e. an outer wrapping call) expects from the current
- * function call. This can be helpful for {@link InputTypeStrategy}.
+ * function call. This can be helpful for an {@link InputTypeStrategy}.
*
- * @see MutableCallContext#getOutputDataType()
+ * @see CallContext#getOutputDataType()
*/
public static final class SurroundingInfo {
@@ -127,13 +147,9 @@ public final class TypeInferenceUtil {
this.innerCallPosition = innerCallPosition;
}
- private Optional<DataType> inferOutputType() {
- // no strategy for inference
- if (!typeInference.getInputTypeStrategy().isPresent()) {
- return Optional.empty();
- }
+ private Optional<DataType> inferOutputType(DataTypeLookup lookup) {
final boolean isValidCount = validateArgumentCount(
- typeInference.getInputTypeValidator().getArgumentCount(),
+ typeInference.getInputTypeStrategy().getArgumentCount(),
argumentCount,
false);
if (!isValidCount) {
@@ -141,11 +157,11 @@ public final class TypeInferenceUtil {
}
// for "takes_string(this_function(NULL))" simulate "takes_string(NULL)"
// for retrieving the output type of "this_function(NULL)"
- final CallContext callContext = new UnknownCallContext(name, functionDefinition, argumentCount);
+ final CallContext callContext = new UnknownCallContext(lookup, name, functionDefinition, argumentCount);
final AdaptedCallContext adaptedContext = adaptArguments(typeInference, callContext, null);
- final InputTypeStrategy inputTypeStrategy = typeInference.getInputTypeStrategy().get();
- inputTypeStrategy.inferInputTypes(adaptedContext);
- return Optional.of(adaptedContext.getArgumentDataTypes().get(innerCallPosition));
+ return typeInference.getInputTypeStrategy()
+ .inferInputTypes(adaptedContext, false)
+ .map(dataTypes -> dataTypes.get(innerCallPosition));
}
}
@@ -194,26 +210,26 @@ public final class TypeInferenceUtil {
@Nullable SurroundingInfo surroundingInfo) {
try {
validateArgumentCount(
- typeInference.getInputTypeValidator().getArgumentCount(),
+ typeInference.getInputTypeStrategy().getArgumentCount(),
callContext.getArgumentDataTypes().size(),
true);
} catch (ValidationException e) {
- throw getInvalidInputException(typeInference.getInputTypeValidator(), callContext, e);
+ throw getInvalidInputException(
+ typeInference,
+ callContext,
+ e);
}
- final AdaptedCallContext adaptedCallContext = adaptArguments(
- typeInference,
- callContext,
- surroundingInfo);
-
+ final AdaptedCallContext adaptedCallContext;
try {
- validateInputTypes(
- typeInference.getInputTypeValidator(),
- adaptedCallContext);
+ adaptedCallContext = adaptArguments(
+ typeInference,
+ callContext,
+ surroundingInfo);
} catch (ValidationException e) {
throw getInvalidInputException(
- typeInference.getInputTypeValidator(),
- adaptedCallContext,
+ typeInference,
+ callContext,
e);
}
@@ -224,22 +240,37 @@ public final class TypeInferenceUtil {
}
private static ValidationException getInvalidInputException(
- InputTypeValidator validator,
+ TypeInference typeInference,
CallContext callContext,
ValidationException cause) {
-
- final String expectedSignatures = validator.getExpectedSignatures(callContext.getFunctionDefinition())
- .stream()
- .map(s -> formatSignature(callContext.getName(), s))
- .collect(Collectors.joining("\n"));
return new ValidationException(
String.format(
"Invalid input arguments. Expected signatures are:\n%s",
- expectedSignatures
+ generateSignature(
+ callContext.getName(),
+ callContext.getFunctionDefinition(),
+ typeInference)
),
cause);
}
+ private static String formatNamedOrTypedArguments(String name, TypeInference typeInference) {
+ final Optional<List<String>> optionalNames = typeInference.getNamedArguments();
+ final Optional<List<DataType>> optionalDataTypes = typeInference.getTypedArguments();
+ final int count = Math.max(
+ optionalNames.map(List::size).orElse(0),
+ optionalDataTypes.map(List::size).orElse(0));
+ final String arguments = IntStream.range(0, count)
+ .mapToObj(pos -> {
+ final StringBuilder builder = new StringBuilder();
+ optionalNames.ifPresent(names -> builder.append(names.get(pos)).append(" => "));
+ optionalDataTypes.ifPresent(dataTypes -> builder.append(dataTypes.get(pos).toString()));
+ return builder.toString();
+ })
+ .collect(Collectors.joining(", "));
+ return String.format("%s(%s)", name, arguments);
+ }
+
private static String formatSignature(String name, Signature s) {
final String arguments = s.getArguments()
.stream()
@@ -293,16 +324,6 @@ public final class TypeInferenceUtil {
return true;
}
- private static void validateInputTypes(InputTypeValidator inputTypeValidator, CallContext callContext) {
- // check for unknown types first
- if (callContext.getArgumentDataTypes().stream().anyMatch(TypeInferenceUtil::isUnknown)) {
- throw new ValidationException("Invalid use of untyped NULL in arguments.");
- }
- if (!inputTypeValidator.validate(callContext, true)) {
- throw new ValidationException("Invalid input arguments.");
- }
- }
-
/**
* Adapts the call's argument if necessary.
*
@@ -313,23 +334,30 @@ public final class TypeInferenceUtil {
TypeInference typeInference,
CallContext callContext,
@Nullable SurroundingInfo surroundingInfo) {
+ final List<DataType> actualTypes = callContext.getArgumentDataTypes();
- final List<DataType> expectedOrActualTypes = typeInference.getArgumentTypes()
- .orElse(callContext.getArgumentDataTypes());
+ typeInference.getTypedArguments()
+ .ifPresent((dataTypes) -> {
+ if (actualTypes.size() != dataTypes.size()) {
+ throw new ValidationException(
+ String.format(
+ "Invalid number of arguments. %d arguments expected after argument expansion but %d passed.",
+ dataTypes.size(),
+ actualTypes.size()));
+ }
+ });
final AdaptedCallContext adaptedCallContext = inferInputTypes(
typeInference,
- expectedOrActualTypes,
callContext,
surroundingInfo);
- final List<DataType> actualTypes = callContext.getArgumentDataTypes();
+ // final check if the call is valid after casting
final List<DataType> expectedTypes = adaptedCallContext.getArgumentDataTypes();
for (int pos = 0; pos < actualTypes.size(); pos++) {
final DataType expectedType = expectedTypes.get(pos);
final DataType actualType = actualTypes.get(pos);
-
- if (!actualType.equals(expectedType) && !canCast(actualType, expectedType)) {
+ if (!supportsImplicitCast(actualType.getLogicalType(), expectedType.getLogicalType())) {
throw new ValidationException(
String.format(
"Invalid argument type at position %d. Data type %s expected but %s passed.",
@@ -338,38 +366,39 @@ public final class TypeInferenceUtil {
actualType));
}
}
+
return adaptedCallContext;
}
private static AdaptedCallContext inferInputTypes(
TypeInference typeInference,
- List<DataType> expectedTypes,
CallContext callContext,
@Nullable SurroundingInfo surroundingInfo) {
// use information of surrounding call to determine output type of this call
final DataType outputType;
if (surroundingInfo != null) {
- outputType = surroundingInfo.inferOutputType().orElse(null);
+ outputType = surroundingInfo.inferOutputType(callContext.getDataTypeLookup()).orElse(null);
} else {
outputType = null;
}
- final AdaptedCallContext adaptedCallContext = new AdaptedCallContext(
- callContext,
- expectedTypes,
- outputType);
+ final AdaptedCallContext adaptedCallContext = new AdaptedCallContext(callContext, outputType);
- // further adapt the arguments by calling an input strategy
- typeInference.getInputTypeStrategy()
- .ifPresent(s -> s.inferInputTypes(adaptedCallContext));
+ // typed arguments have highest priority
+ typeInference.getTypedArguments().ifPresent(adaptedCallContext::setExpectedArguments);
- return adaptedCallContext;
- }
+ final List<DataType> inferredDataTypes = typeInference.getInputTypeStrategy()
+ .inferInputTypes(adaptedCallContext, true)
+ .orElseThrow(() -> new ValidationException("Invalid input arguments."));
- private static boolean canCast(DataType sourceDataType, DataType targetDataType) {
- return LogicalTypeCasts.supportsImplicitCast(
- sourceDataType.getLogicalType(),
- targetDataType.getLogicalType());
+ // input must not contain unknown types at this point
+ if (inferredDataTypes.stream().anyMatch(TypeInferenceUtil::isUnknown)) {
+ throw new ValidationException("Invalid use of untyped NULL in arguments.");
+ }
+
+ adaptedCallContext.setExpectedArguments(inferredDataTypes);
+
+ return adaptedCallContext;
}
private static Result inferTypes(
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/TypeStrategies.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/TypeStrategies.java
index b14d8e2..184b33d 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/TypeStrategies.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/TypeStrategies.java
@@ -21,7 +21,7 @@ package org.apache.flink.table.types.inference;
import org.apache.flink.annotation.Internal;
import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.inference.strategies.ExplicitTypeStrategy;
-import org.apache.flink.table.types.inference.strategies.MatchingTypeStrategy;
+import org.apache.flink.table.types.inference.strategies.MappingTypeStrategy;
import org.apache.flink.table.types.inference.strategies.MissingTypeStrategy;
import java.util.Map;
@@ -47,11 +47,11 @@ public final class TypeStrategies {
}
/**
- * Type strategy that maps an {@link InputTypeValidator} to a {@link TypeStrategy} if the validator
- * matches.
+ * Type strategy that maps an {@link InputTypeStrategy} to a {@link TypeStrategy} if the input strategy
+ * infers identical types.
*/
- public static TypeStrategy matching(Map<InputTypeValidator, TypeStrategy> matchers) {
- return new MatchingTypeStrategy(matchers);
+ public static TypeStrategy mapping(Map<InputTypeStrategy, TypeStrategy> mappings) {
+ return new MappingTypeStrategy(mappings);
}
// --------------------------------------------------------------------------------------------
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/AndArgumentTypeStrategy.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/AndArgumentTypeStrategy.java
new file mode 100644
index 0000000..f2d89cc
--- /dev/null
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/AndArgumentTypeStrategy.java
@@ -0,0 +1,111 @@
+/*
+ * 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.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.InputTypeStrategies;
+import org.apache.flink.table.types.inference.Signature;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.util.Preconditions;
+
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.stream.Collectors;
+
+/**
+ * Strategy for inferring and validating an argument using a conjunction of multiple {@link ArgumentTypeStrategy}s
+ * into one like {@code f(NUMERIC && LITERAL)}
+ *
+ * <p>Some {@link ArgumentTypeStrategy}s cannot contribute an inferred type that is different from
+ * the input type (e.g. {@link InputTypeStrategies#LITERAL}). Therefore, the order {@code f(X && Y)} or
+ * {@code f(Y && X)} matters as it defines the precedence in case the result must be casted to a more
+ * specific type.
+ *
+ * <p>The strategy aims to infer the first more specific, casted type or (if this is not possible)
+ * a type that has been inferred from all {@link ArgumentTypeStrategy}s.
+ */
+@Internal
+public final class AndArgumentTypeStrategy implements ArgumentTypeStrategy {
+
+ private final List<? extends ArgumentTypeStrategy> argumentStrategies;
+
+ public AndArgumentTypeStrategy(List<? extends ArgumentTypeStrategy> argumentStrategies) {
+ Preconditions.checkArgument(argumentStrategies.size() > 0);
+ this.argumentStrategies = argumentStrategies;
+ }
+
+ @Override
+ public Optional<DataType> inferArgumentType(CallContext callContext, int argumentPos, boolean throwOnFailure) {
+ final DataType actualDataType = callContext.getArgumentDataTypes().get(argumentPos);
+ final LogicalType actualType = actualDataType.getLogicalType();
+
+ Optional<DataType> closestDataType = Optional.empty();
+ for (ArgumentTypeStrategy strategy : argumentStrategies) {
+ final Optional<DataType> inferredDataType = strategy.inferArgumentType(
+ callContext,
+ argumentPos,
+ throwOnFailure);
+ // argument type does not match at all
+ if (!inferredDataType.isPresent()) {
+ return Optional.empty();
+ }
+ final LogicalType inferredType = inferredDataType.get().getLogicalType();
+ // a more specific, casted argument type is available
+ if (!actualType.equals(inferredType) && !closestDataType.isPresent()) {
+ closestDataType = inferredDataType;
+ }
+ }
+
+ if (closestDataType.isPresent()) {
+ return closestDataType;
+ }
+
+ return Optional.of(actualDataType);
+ }
+
+ @Override
+ public Signature.Argument getExpectedArgument(FunctionDefinition functionDefinition, int argumentPos) {
+ final String argument = argumentStrategies.stream()
+ .map(v -> v.getExpectedArgument(functionDefinition, argumentPos).getType())
+ .collect(Collectors.joining(" & ", "[", "]"));
+ return Signature.Argument.of(argument);
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ AndArgumentTypeStrategy that = (AndArgumentTypeStrategy) o;
+ return Objects.equals(argumentStrategies, that.argumentStrategies);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(argumentStrategies);
+ }
+}
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/validators/AnyTypeValidator.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/AnyArgumentTypeStrategy.java
similarity index 53%
rename from flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/validators/AnyTypeValidator.java
rename to flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/AnyArgumentTypeStrategy.java
index 4bb816e..d912539 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/validators/AnyTypeValidator.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/AnyArgumentTypeStrategy.java
@@ -15,29 +15,26 @@
* limitations under the License.
*/
-package org.apache.flink.table.types.inference.validators;
+package org.apache.flink.table.types.inference.strategies;
import org.apache.flink.annotation.Internal;
import org.apache.flink.table.functions.FunctionDefinition;
-import org.apache.flink.table.types.inference.ArgumentCount;
-import org.apache.flink.table.types.inference.ArgumentTypeValidator;
+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.ConstantArgumentCount;
-import org.apache.flink.table.types.inference.InputTypeValidator;
import org.apache.flink.table.types.inference.Signature;
-import java.util.Collections;
-import java.util.List;
+import java.util.Optional;
/**
- * Validator that checks for a single argument that can be of any type.
+ * Strategy for an argument that can be of any type.
*/
@Internal
-public final class AnyTypeValidator implements ArgumentTypeValidator, InputTypeValidator {
+public final class AnyArgumentTypeStrategy implements ArgumentTypeStrategy {
@Override
- public boolean validateArgument(CallContext callContext, int argumentPos, boolean throwOnFailure) {
- return true;
+ public Optional<DataType> inferArgumentType(CallContext callContext, int argumentPos, boolean throwOnFailure) {
+ return Optional.of(callContext.getArgumentDataTypes().get(argumentPos));
}
@Override
@@ -46,27 +43,12 @@ public final class AnyTypeValidator implements ArgumentTypeValidator, InputTypeV
}
@Override
- public ArgumentCount getArgumentCount() {
- return ConstantArgumentCount.of(1);
- }
-
- @Override
- public boolean validate(CallContext callContext, boolean throwOnFailure) {
- return true;
- }
-
- @Override
- public List<Signature> getExpectedSignatures(FunctionDefinition definition) {
- return Collections.singletonList(Signature.of(getExpectedArgument(definition, 0)));
- }
-
- @Override
public boolean equals(Object o) {
- return this == o || o instanceof AnyTypeValidator;
+ return this == o || o instanceof AnyArgumentTypeStrategy;
}
@Override
public int hashCode() {
- return AnyTypeValidator.class.hashCode();
+ return AnyArgumentTypeStrategy.class.hashCode();
}
}
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/BridgingInputTypeStrategy.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/BridgingInputTypeStrategy.java
deleted file mode 100644
index 4def14d..0000000
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/BridgingInputTypeStrategy.java
+++ /dev/null
@@ -1,135 +0,0 @@
-/*
- * 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.types.DataType;
-import org.apache.flink.table.types.inference.InputTypeStrategy;
-import org.apache.flink.table.types.inference.MutableCallContext;
-import org.apache.flink.table.types.logical.LogicalType;
-
-import javax.annotation.Nullable;
-
-import java.util.List;
-import java.util.Objects;
-
-/**
- * Input type strategy for enriching data types with an expected conversion class. This is in particular
- * useful when a data type has been created out of a {@link LogicalType} but runtime hints are still missing.
- */
-@Internal
-public final class BridgingInputTypeStrategy implements InputTypeStrategy {
-
- private final List<BridgingSignature> bridgingSignatures;
-
- public BridgingInputTypeStrategy(List<BridgingSignature> bridgingSignatures) {
- this.bridgingSignatures = bridgingSignatures;
- }
-
- @Override
- public void inferInputTypes(MutableCallContext callContext) {
- final List<DataType> actualDataTypes = callContext.getArgumentDataTypes();
- for (BridgingSignature bridgingSignature : bridgingSignatures) {
- if (bridgingSignature.matches(actualDataTypes)) {
- bridgingSignature.enrich(actualDataTypes.size(), callContext);
- return; // there should be only one matching signature
- }
- }
- }
-
- @Override
- public boolean equals(Object o) {
- if (this == o) {
- return true;
- }
- if (o == null || getClass() != o.getClass()) {
- return false;
- }
- BridgingInputTypeStrategy that = (BridgingInputTypeStrategy) o;
- return bridgingSignatures.equals(that.bridgingSignatures);
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(bridgingSignatures);
- }
-
- /**
- * Helper class that represents a signature of input arguments.
- *
- * <p>Note: Array elements can be null for skipping the enrichment of certain arguments. If the
- * signature has varargs, the last data type represents the varying argument type.
- */
- public static final class BridgingSignature {
-
- private final DataType[] expectedDataTypes;
-
- private final boolean isVarying;
-
- public BridgingSignature(DataType[] expectedDataTypes, boolean isVarying) {
- this.expectedDataTypes = expectedDataTypes;
- this.isVarying = isVarying;
- }
-
- public boolean matches(List<DataType> actualDataTypes) {
- if (!isValidCount(actualDataTypes.size())) {
- return false;
- }
- for (int i = 0; i < actualDataTypes.size(); i++) {
- final LogicalType actualType = actualDataTypes.get(i).getLogicalType();
- final DataType expectedType = getExpectedDataType(i);
- if (expectedType != null && !actualType.equals(expectedType.getLogicalType())) {
- return false;
- }
- }
- return true;
- }
-
- public void enrich(int argumentCount, MutableCallContext mutableCallContext) {
- for (int i = 0; i < argumentCount; i++) {
- mutableCallContext.mutateArgumentDataType(i, getExpectedDataType(i));
- }
- }
-
- private boolean isValidCount(int actualCount) {
- final int minCount;
- if (isVarying) {
- minCount = expectedDataTypes.length - 1;
- } else {
- minCount = expectedDataTypes.length;
- }
- final int maxCount;
- if (isVarying) {
- maxCount = Integer.MAX_VALUE;
- } else {
- maxCount = expectedDataTypes.length;
- }
- return actualCount >= minCount && actualCount <= maxCount;
- }
-
- private @Nullable DataType getExpectedDataType(int pos) {
- if (pos < expectedDataTypes.length) {
- return expectedDataTypes[pos];
- } else if (isVarying) {
- return expectedDataTypes[expectedDataTypes.length - 1];
- }
- throw new IllegalStateException("Argument count should have been validated before.");
- }
- }
-}
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/validators/ExplicitTypeValidator.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/ExplicitArgumentTypeStrategy.java
similarity index 61%
rename from flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/validators/ExplicitTypeValidator.java
rename to flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/ExplicitArgumentTypeStrategy.java
index 2005846..770558e 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/validators/ExplicitTypeValidator.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/ExplicitArgumentTypeStrategy.java
@@ -16,11 +16,12 @@
* limitations under the License.
*/
-package org.apache.flink.table.types.inference.validators;
+package org.apache.flink.table.types.inference.strategies;
import org.apache.flink.annotation.Internal;
import org.apache.flink.table.functions.FunctionDefinition;
-import org.apache.flink.table.types.inference.ArgumentTypeValidator;
+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.Argument;
import org.apache.flink.table.types.logical.LogicalType;
@@ -28,27 +29,31 @@ import org.apache.flink.table.types.logical.utils.LogicalTypeCasts;
import org.apache.flink.util.Preconditions;
import java.util.Objects;
+import java.util.Optional;
/**
- * Validator that checks if each operand corresponds to an explicitly defined logical type.
+ * Strategy for an argument that corresponds to an explicitly defined type. Implicit casts will be
+ * inserted if possible.
*/
@Internal
-public final class ExplicitTypeValidator implements ArgumentTypeValidator {
+public final class ExplicitArgumentTypeStrategy implements ArgumentTypeStrategy {
- private final LogicalType expectedType;
+ private final DataType expectedDataType;
- public ExplicitTypeValidator(LogicalType expectedType) {
- this.expectedType = Preconditions.checkNotNull(expectedType);
+ public ExplicitArgumentTypeStrategy(DataType expectedDataType) {
+ this.expectedDataType = Preconditions.checkNotNull(expectedDataType);
}
@Override
- public boolean validateArgument(CallContext callContext, int argumentPos, boolean throwOnFailure) {
+ public Optional<DataType> inferArgumentType(CallContext callContext, int argumentPos, boolean throwOnFailure) {
+ final LogicalType expectedType = expectedDataType.getLogicalType();
final LogicalType actualType = callContext.getArgumentDataTypes().get(argumentPos).getLogicalType();
- // quick path
+ // if logical types match, we return the expected data type
+ // for ensuring the expected conversion class
if (expectedType.equals(actualType)) {
- return true;
+ return Optional.of(expectedDataType);
}
- // lenient check
+ // type coercion
if (!LogicalTypeCasts.supportsImplicitCast(actualType, expectedType)) {
if (throwOnFailure) {
throw callContext.newValidationError(
@@ -56,14 +61,14 @@ public final class ExplicitTypeValidator implements ArgumentTypeValidator {
expectedType,
actualType);
}
- return false;
+ return Optional.empty();
}
- return true;
+ return Optional.of(expectedDataType);
}
@Override
public Argument getExpectedArgument(FunctionDefinition functionDefinition, int argumentPos) {
- return Argument.of(expectedType.asSummaryString());
+ return Argument.of(expectedDataType.toString());
}
@Override
@@ -74,12 +79,12 @@ public final class ExplicitTypeValidator implements ArgumentTypeValidator {
if (o == null || getClass() != o.getClass()) {
return false;
}
- ExplicitTypeValidator that = (ExplicitTypeValidator) o;
- return Objects.equals(expectedType, that.expectedType);
+ ExplicitArgumentTypeStrategy that = (ExplicitArgumentTypeStrategy) o;
+ return Objects.equals(expectedDataType, that.expectedDataType);
}
@Override
public int hashCode() {
- return Objects.hash(expectedType);
+ return Objects.hash(expectedDataType);
}
}
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/ExplicitInputTypeStrategy.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/ExplicitInputTypeStrategy.java
deleted file mode 100644
index cf4c354..0000000
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/ExplicitInputTypeStrategy.java
+++ /dev/null
@@ -1,67 +0,0 @@
-/*
- * 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.types.DataType;
-import org.apache.flink.table.types.inference.InputTypeStrategy;
-import org.apache.flink.table.types.inference.MutableCallContext;
-
-import java.util.List;
-import java.util.Objects;
-
-/**
- * Input type strategy that supplies a fixed {@link DataType} for each argument.
- */
-@Internal
-public final class ExplicitInputTypeStrategy implements InputTypeStrategy {
-
- private final List<DataType> dataTypes;
-
- public ExplicitInputTypeStrategy(List<DataType> dataTypes) {
- this.dataTypes = dataTypes;
- }
-
- @Override
- public void inferInputTypes(MutableCallContext callContext) {
- if (callContext.getArgumentDataTypes().size() != dataTypes.size()) {
- return;
- }
- for (int i = 0; i < dataTypes.size(); i++) {
- callContext.mutateArgumentDataType(i, dataTypes.get(i));
- }
- }
-
- @Override
- public boolean equals(Object o) {
- if (this == o) {
- return true;
- }
- if (o == null || getClass() != o.getClass()) {
- return false;
- }
- ExplicitInputTypeStrategy that = (ExplicitInputTypeStrategy) o;
- return dataTypes.equals(that.dataTypes);
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(dataTypes);
- }
-}
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/validators/LiteralTypeValidator.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/LiteralArgumentTypeStrategy.java
similarity index 59%
rename from flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/validators/LiteralTypeValidator.java
rename to flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/LiteralArgumentTypeStrategy.java
index 1da7c18..607d962 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/validators/LiteralTypeValidator.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/LiteralArgumentTypeStrategy.java
@@ -16,68 +16,53 @@
* limitations under the License.
*/
-package org.apache.flink.table.types.inference.validators;
+package org.apache.flink.table.types.inference.strategies;
import org.apache.flink.annotation.Internal;
import org.apache.flink.table.functions.FunctionDefinition;
-import org.apache.flink.table.types.inference.ArgumentCount;
-import org.apache.flink.table.types.inference.ArgumentTypeValidator;
+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.ConstantArgumentCount;
-import org.apache.flink.table.types.inference.InputTypeValidator;
import org.apache.flink.table.types.inference.Signature;
-import java.util.Collections;
-import java.util.List;
import java.util.Objects;
+import java.util.Optional;
/**
- * Validator that checks if a single argument is a literal.
+ * Strategy that checks if an argument is a literal.
*/
@Internal
-public final class LiteralTypeValidator implements ArgumentTypeValidator, InputTypeValidator {
+public final class LiteralArgumentTypeStrategy implements ArgumentTypeStrategy {
private final boolean allowNull;
- public LiteralTypeValidator(boolean allowNull) {
+ public LiteralArgumentTypeStrategy(boolean allowNull) {
this.allowNull = allowNull;
}
@Override
- public boolean validateArgument(CallContext callContext, int argumentPos, boolean throwOnFailure) {
+ public Optional<DataType> inferArgumentType(CallContext callContext, int argumentPos, boolean throwOnFailure) {
if (!callContext.isArgumentLiteral(argumentPos)) {
if (throwOnFailure) {
throw callContext.newValidationError("Literal expected.");
}
- return false;
+ return Optional.empty();
}
if (callContext.isArgumentNull(argumentPos) && !allowNull) {
if (throwOnFailure) {
throw callContext.newValidationError("Literal must not be NULL.");
}
- return false;
+ return Optional.empty();
}
- return true;
+ return Optional.of(callContext.getArgumentDataTypes().get(argumentPos));
}
@Override
public Signature.Argument getExpectedArgument(FunctionDefinition functionDefinition, int argumentPos) {
- return Signature.Argument.of("<LITERAL>");
- }
-
- @Override
- public ArgumentCount getArgumentCount() {
- return ConstantArgumentCount.of(1);
- }
-
- @Override
- public boolean validate(CallContext callContext, boolean throwOnFailure) {
- return validateArgument(callContext, 0, throwOnFailure);
- }
-
- @Override
- public List<Signature> getExpectedSignatures(FunctionDefinition definition) {
- return Collections.singletonList(Signature.of(getExpectedArgument(definition, 0)));
+ if (allowNull) {
+ return Signature.Argument.of("<LITERAL>");
+ }
+ return Signature.Argument.of("<LITERAL NOT NULL>");
}
@Override
@@ -88,7 +73,7 @@ public final class LiteralTypeValidator implements ArgumentTypeValidator, InputT
if (o == null || getClass() != o.getClass()) {
return false;
}
- LiteralTypeValidator that = (LiteralTypeValidator) o;
+ LiteralArgumentTypeStrategy that = (LiteralArgumentTypeStrategy) o;
return allowNull == that.allowNull;
}
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/MatchingTypeStrategy.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/MappingTypeStrategy.java
similarity index 50%
rename from flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/MatchingTypeStrategy.java
rename to flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/MappingTypeStrategy.java
index 1129307..b4047b9 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/MatchingTypeStrategy.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/MappingTypeStrategy.java
@@ -21,33 +21,50 @@ package org.apache.flink.table.types.inference.strategies;
import org.apache.flink.annotation.Internal;
import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.inference.CallContext;
-import org.apache.flink.table.types.inference.InputTypeValidator;
+import org.apache.flink.table.types.inference.InputTypeStrategy;
import org.apache.flink.table.types.inference.TypeStrategy;
+import org.apache.flink.table.types.logical.LogicalType;
+import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
+import java.util.stream.Collectors;
/**
- * Type strategy that maps an {@link InputTypeValidator} to a {@link TypeStrategy} if the validator
- * matches.
+ * Type strategy that maps an {@link InputTypeStrategy} to a {@link TypeStrategy} if the input strategy
+ * infers identical types.
*/
@Internal
-public final class MatchingTypeStrategy implements TypeStrategy {
+public final class MappingTypeStrategy implements TypeStrategy {
- private final Map<InputTypeValidator, TypeStrategy> matchers;
+ private final Map<InputTypeStrategy, TypeStrategy> mappings;
- public MatchingTypeStrategy(Map<InputTypeValidator, TypeStrategy> matchers) {
- this.matchers = matchers;
+ public MappingTypeStrategy(Map<InputTypeStrategy, TypeStrategy> mappings) {
+ this.mappings = mappings;
}
@Override
public Optional<DataType> inferType(CallContext callContext) {
- for (Map.Entry<InputTypeValidator, TypeStrategy> matcher : matchers.entrySet()) {
- final InputTypeValidator validator = matcher.getKey();
- final TypeStrategy strategy = matcher.getValue();
- if (validator.getArgumentCount().isValidCount(callContext.getArgumentDataTypes().size()) &&
- validator.validate(callContext, false)) {
+ final List<LogicalType> actualTypes = callContext.getArgumentDataTypes().stream()
+ .map(DataType::getLogicalType)
+ .collect(Collectors.toList());
+ for (Map.Entry<InputTypeStrategy, TypeStrategy> mapping : mappings.entrySet()) {
+ final InputTypeStrategy inputStrategy = mapping.getKey();
+ final TypeStrategy strategy = mapping.getValue();
+ if (!inputStrategy.getArgumentCount().isValidCount(actualTypes.size())) {
+ continue;
+ }
+ final Optional<List<DataType>> inferredDataTypes = inputStrategy.inferInputTypes(
+ callContext,
+ false);
+ if (!inferredDataTypes.isPresent()) {
+ continue;
+ }
+ final List<LogicalType> inferredTypes = inferredDataTypes.get().stream()
+ .map(DataType::getLogicalType)
+ .collect(Collectors.toList());
+ if (actualTypes.equals(inferredTypes)) {
return strategy.inferType(callContext);
}
}
@@ -62,12 +79,12 @@ public final class MatchingTypeStrategy implements TypeStrategy {
if (o == null || getClass() != o.getClass()) {
return false;
}
- MatchingTypeStrategy that = (MatchingTypeStrategy) o;
- return matchers.equals(that.matchers);
+ MappingTypeStrategy that = (MappingTypeStrategy) o;
+ return mappings.equals(that.mappings);
}
@Override
public int hashCode() {
- return Objects.hash(matchers);
+ return Objects.hash(mappings);
}
}
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/NopInputTypeStrategy.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/NopInputTypeStrategy.java
deleted file mode 100644
index 8c33c39..0000000
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/NopInputTypeStrategy.java
+++ /dev/null
@@ -1,45 +0,0 @@
-/*
- * 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.types.inference.InputTypeStrategy;
-import org.apache.flink.table.types.inference.MutableCallContext;
-
-/**
- * Input strategy that does nothing.
- */
-@Internal
-public final class NopInputTypeStrategy implements InputTypeStrategy {
-
- @Override
- public void inferInputTypes(MutableCallContext callContext) {
- // nothing to do
- }
-
- @Override
- public boolean equals(Object o) {
- return this == o || o instanceof NopInputTypeStrategy;
- }
-
- @Override
- public int hashCode() {
- return NopInputTypeStrategy.class.hashCode();
- }
-}
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/OrArgumentTypeStrategy.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/OrArgumentTypeStrategy.java
new file mode 100644
index 0000000..67a7738
--- /dev/null
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/OrArgumentTypeStrategy.java
@@ -0,0 +1,122 @@
+/*
+ * 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.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.InputTypeStrategies;
+import org.apache.flink.table.types.inference.Signature;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.util.Preconditions;
+
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.stream.Collectors;
+
+/**
+ * Strategy for inferring and validating an argument using a disjunction of multiple {@link ArgumentTypeStrategy}s
+ * into one like {@code f(NUMERIC || STRING)}.
+ *
+ * <p>Some {@link ArgumentTypeStrategy}s cannot contribute an inferred type that is different from
+ * the input type (e.g. {@link InputTypeStrategies#LITERAL}). Therefore, the order {@code f(X || Y)} or
+ * {@code f(Y || X)} matters as it defines the precedence in case the result must be casted to a more
+ * specific type.
+ *
+ * <p>This strategy aims to infer a type that is equal to the input type (to prevent unnecessary casting)
+ * or (if this is not possible) the first more specific, casted type.
+ */
+@Internal
+public final class OrArgumentTypeStrategy implements ArgumentTypeStrategy {
+
+ private final List<? extends ArgumentTypeStrategy> argumentStrategies;
+
+ public OrArgumentTypeStrategy(List<? extends ArgumentTypeStrategy> argumentStrategies) {
+ Preconditions.checkArgument(argumentStrategies.size() > 0);
+ this.argumentStrategies = argumentStrategies;
+ }
+
+ @Override
+ public Optional<DataType> inferArgumentType(CallContext callContext, int argumentPos, boolean throwOnFailure) {
+ final LogicalType actualType = callContext.getArgumentDataTypes().get(argumentPos).getLogicalType();
+
+ Optional<DataType> closestDataType = Optional.empty();
+ for (ArgumentTypeStrategy strategy : argumentStrategies) {
+ final Optional<DataType> inferredDataType = strategy.inferArgumentType(
+ callContext,
+ argumentPos,
+ false);
+ // argument type does not match at all
+ if (!inferredDataType.isPresent()) {
+ continue;
+ }
+ final LogicalType inferredType = inferredDataType.get().getLogicalType();
+ // argument type matches exactly
+ // we prefer a strategy that does not require an implicit cast
+ if (actualType.equals(inferredType)) {
+ return inferredDataType;
+ }
+ // argument type requires a more specific, casted type
+ else if (!closestDataType.isPresent()) {
+ closestDataType = inferredDataType;
+ }
+ }
+
+ if (closestDataType.isPresent()) {
+ return closestDataType;
+ }
+
+ // generate a helpful exception if possible
+ if (throwOnFailure) {
+ for (ArgumentTypeStrategy strategy : argumentStrategies) {
+ strategy.inferArgumentType(callContext, argumentPos, true);
+ }
+ }
+
+ return Optional.empty();
+ }
+
+ @Override
+ public Signature.Argument getExpectedArgument(FunctionDefinition functionDefinition, int argumentPos) {
+ final String argument = argumentStrategies.stream()
+ .map(v -> v.getExpectedArgument(functionDefinition, argumentPos).getType())
+ .collect(Collectors.joining(" | ", "[", "]"));
+ return Signature.Argument.of(argument);
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ OrArgumentTypeStrategy that = (OrArgumentTypeStrategy) o;
+ return Objects.equals(argumentStrategies, that.argumentStrategies);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(argumentStrategies);
+ }
+}
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/validators/OrTypeInputValidator.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/OrInputTypeStrategy.java
similarity index 67%
rename from flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/validators/OrTypeInputValidator.java
rename to flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/OrInputTypeStrategy.java
index c1cbcf6..5c861bb 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/validators/OrTypeInputValidator.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/OrInputTypeStrategy.java
@@ -16,15 +16,17 @@
* limitations under the License.
*/
-package org.apache.flink.table.types.inference.validators;
+package org.apache.flink.table.types.inference.strategies;
import org.apache.flink.annotation.Internal;
import org.apache.flink.table.functions.FunctionDefinition;
+import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.inference.ArgumentCount;
import org.apache.flink.table.types.inference.CallContext;
import org.apache.flink.table.types.inference.ConstantArgumentCount;
-import org.apache.flink.table.types.inference.InputTypeValidator;
+import org.apache.flink.table.types.inference.InputTypeStrategy;
import org.apache.flink.table.types.inference.Signature;
+import org.apache.flink.table.types.logical.LogicalType;
import org.apache.flink.util.Preconditions;
import javax.annotation.Nullable;
@@ -36,28 +38,28 @@ import java.util.Optional;
import java.util.stream.Collectors;
/**
- * Validator that checks for a disjunction of multiple {@link InputTypeValidator}s into one like
- * {@code f(NUMERIC) || f(STRING)}.
+ * Strategy for inferring and validating the input using a disjunction of multiple {@link InputTypeStrategy}s
+ * into one like {@code f(NUMERIC) || f(STRING)}.
*/
@Internal
-public final class OrTypeInputValidator implements InputTypeValidator {
+public final class OrInputTypeStrategy implements InputTypeStrategy {
- private final List<? extends InputTypeValidator> validators;
+ private final List<? extends InputTypeStrategy> inputStrategies;
- public OrTypeInputValidator(List<? extends InputTypeValidator> validators) {
- Preconditions.checkArgument(validators.size() > 0);
- this.validators = validators;
+ public OrInputTypeStrategy(List<? extends InputTypeStrategy> inputStrategies) {
+ Preconditions.checkArgument(inputStrategies.size() > 0);
+ this.inputStrategies = inputStrategies;
}
@Override
public ArgumentCount getArgumentCount() {
final List<ArgumentCount> counts = new AbstractList<ArgumentCount>() {
public ArgumentCount get(int index) {
- return validators.get(index).getArgumentCount();
+ return inputStrategies.get(index).getArgumentCount();
}
public int size() {
- return validators.size();
+ return inputStrategies.size();
}
};
final Integer min = commonMin(counts);
@@ -102,18 +104,39 @@ public final class OrTypeInputValidator implements InputTypeValidator {
}
@Override
- public boolean validate(CallContext callContext, boolean throwOnFailure) {
- for (final InputTypeValidator validator : validators) {
- if (validator.validate(callContext, false)) {
- return true;
+ public Optional<List<DataType>> inferInputTypes(CallContext callContext, boolean throwOnFailure) {
+ final List<LogicalType> actualTypes = callContext.getArgumentDataTypes().stream()
+ .map(DataType::getLogicalType)
+ .collect(Collectors.toList());
+
+ Optional<List<DataType>> closestDataTypes = Optional.empty();
+ for (InputTypeStrategy inputStrategy : inputStrategies) {
+ final Optional<List<DataType>> inferredDataTypes = inputStrategy.inferInputTypes(
+ callContext,
+ false);
+ // types do not match at all
+ if (!inferredDataTypes.isPresent()) {
+ continue;
+ }
+ final List<LogicalType> inferredTypes = inferredDataTypes.get().stream()
+ .map(DataType::getLogicalType)
+ .collect(Collectors.toList());
+ // types match exactly
+ if (actualTypes.equals(inferredTypes)) {
+ return inferredDataTypes;
+ }
+ // type matches with some casting
+ else if (!closestDataTypes.isPresent()) {
+ closestDataTypes = inferredDataTypes;
}
}
- return false;
+
+ return closestDataTypes;
}
@Override
public List<Signature> getExpectedSignatures(FunctionDefinition definition) {
- return validators.stream()
+ return inputStrategies.stream()
.flatMap(v -> v.getExpectedSignatures(definition).stream())
.collect(Collectors.toList());
}
@@ -126,13 +149,13 @@ public final class OrTypeInputValidator implements InputTypeValidator {
if (o == null || getClass() != o.getClass()) {
return false;
}
- OrTypeInputValidator that = (OrTypeInputValidator) o;
- return Objects.equals(validators, that.validators);
+ OrInputTypeStrategy that = (OrInputTypeStrategy) o;
+ return Objects.equals(inputStrategies, that.inputStrategies);
}
@Override
public int hashCode() {
- return Objects.hash(validators);
+ return Objects.hash(inputStrategies);
}
// --------------------------------------------------------------------------------------------
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/OutputTypeInputTypeStrategy.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/OutputArgumentTypeStrategy.java
similarity index 52%
rename from flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/OutputTypeInputTypeStrategy.java
rename to flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/OutputArgumentTypeStrategy.java
index fa154a0..21ab05c 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/OutputTypeInputTypeStrategy.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/OutputArgumentTypeStrategy.java
@@ -18,41 +18,44 @@
package org.apache.flink.table.types.inference.strategies;
import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.functions.FunctionDefinition;
import org.apache.flink.table.types.DataType;
-import org.apache.flink.table.types.inference.InputTypeStrategy;
-import org.apache.flink.table.types.inference.MutableCallContext;
+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.LogicalTypeRoot;
-import java.util.List;
+import java.util.Optional;
import static org.apache.flink.table.types.logical.utils.LogicalTypeChecks.hasRoot;
/**
- * Input type strategy that supplies the function's output {@link DataType} for each unknown
- * argument if available.
+ * Strategy for inferring an unknown argument type from the function's output {@link DataType} if available.
*/
@Internal
-public final class OutputTypeInputTypeStrategy implements InputTypeStrategy {
+public final class OutputArgumentTypeStrategy implements ArgumentTypeStrategy {
@Override
- public void inferInputTypes(MutableCallContext callContext) {
- callContext.getOutputDataType().ifPresent(t -> {
- final List<DataType> dataTypes = callContext.getArgumentDataTypes();
- for (int i = 0; i < dataTypes.size(); i++) {
- if (hasRoot(dataTypes.get(i).getLogicalType(), LogicalTypeRoot.NULL)) {
- callContext.mutateArgumentDataType(i, t);
- }
- }
- });
+ public Optional<DataType> inferArgumentType(CallContext callContext, int argumentPos, boolean throwOnFailure) {
+ final DataType actualDataType = callContext.getArgumentDataTypes().get(argumentPos);
+ if (hasRoot(actualDataType.getLogicalType(), LogicalTypeRoot.NULL)) {
+ return callContext.getOutputDataType();
+ }
+ return Optional.of(actualDataType);
+ }
+
+ @Override
+ public Signature.Argument getExpectedArgument(FunctionDefinition functionDefinition, int argumentPos) {
+ return Signature.Argument.of("<OUTPUT>");
}
@Override
public boolean equals(Object o) {
- return this == o || o instanceof OutputTypeInputTypeStrategy;
+ return this == o || o instanceof OutputArgumentTypeStrategy;
}
@Override
public int hashCode() {
- return OutputTypeInputTypeStrategy.class.hashCode();
+ return OutputArgumentTypeStrategy.class.hashCode();
}
}
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/validators/SequenceInputValidator.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/SequenceInputTypeStrategy.java
similarity index 55%
rename from flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/validators/SequenceInputValidator.java
rename to flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/SequenceInputTypeStrategy.java
index db1486f..d5e7c90 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/validators/SequenceInputValidator.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/SequenceInputTypeStrategy.java
@@ -16,16 +16,16 @@
* limitations under the License.
*/
-package org.apache.flink.table.types.inference.validators;
+package org.apache.flink.table.types.inference.strategies;
import org.apache.flink.annotation.Internal;
import org.apache.flink.table.functions.FunctionDefinition;
import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.inference.ArgumentCount;
-import org.apache.flink.table.types.inference.ArgumentTypeValidator;
+import org.apache.flink.table.types.inference.ArgumentTypeStrategy;
import org.apache.flink.table.types.inference.CallContext;
import org.apache.flink.table.types.inference.ConstantArgumentCount;
-import org.apache.flink.table.types.inference.InputTypeValidator;
+import org.apache.flink.table.types.inference.InputTypeStrategy;
import org.apache.flink.table.types.inference.Signature;
import org.apache.flink.util.Preconditions;
@@ -35,57 +35,63 @@ import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Objects;
+import java.util.Optional;
/**
- * Validator that checks for a sequence of {@link ArgumentTypeValidator}s for validating an entire
- * function signature like {@code f(STRING, NUMERIC)} or {@code f(s STRING, n NUMERIC)}.
+ * Strategy for inferring and validating a function signature like {@code f(STRING, NUMERIC)}
+ * or {@code f(s STRING, n NUMERIC)} using a sequence of {@link ArgumentTypeStrategy}s.
*/
@Internal
-public final class SequenceInputValidator implements InputTypeValidator {
+public final class SequenceInputTypeStrategy implements InputTypeStrategy {
- private final List<? extends ArgumentTypeValidator> validators;
+ private final List<? extends ArgumentTypeStrategy> argumentStrategies;
private final @Nullable List<String> argumentNames;
- public SequenceInputValidator(
- List<? extends ArgumentTypeValidator> validators,
+ public SequenceInputTypeStrategy(
+ List<? extends ArgumentTypeStrategy> argumentStrategies,
@Nullable List<String> argumentNames) {
- Preconditions.checkArgument(validators.size() > 0);
- Preconditions.checkArgument(argumentNames == null || argumentNames.size() == validators.size());
- this.validators = validators;
+ Preconditions.checkArgument(argumentNames == null || argumentNames.size() == argumentStrategies.size());
+ this.argumentStrategies = argumentStrategies;
this.argumentNames = argumentNames;
}
@Override
public ArgumentCount getArgumentCount() {
- return ConstantArgumentCount.of(validators.size());
+ return ConstantArgumentCount.of(argumentStrategies.size());
}
@Override
- public boolean validate(CallContext callContext, boolean throwOnFailure) {
+ public Optional<List<DataType>> inferInputTypes(CallContext callContext, boolean throwOnFailure) {
final List<DataType> dataTypes = callContext.getArgumentDataTypes();
- if (dataTypes.size() != validators.size()) {
- return false;
+ if (dataTypes.size() != argumentStrategies.size()) {
+ return Optional.empty();
}
- for (int i = 0; i < validators.size(); i++) {
- final ArgumentTypeValidator validator = validators.get(i);
- if (!validator.validateArgument(callContext, i, throwOnFailure)) {
- return false;
+ final List<DataType> inferredDataTypes = new ArrayList<>(dataTypes.size());
+ for (int i = 0; i < argumentStrategies.size(); i++) {
+ final ArgumentTypeStrategy argumentTypeStrategy = argumentStrategies.get(i);
+ final Optional<DataType> inferredDataType = argumentTypeStrategy.inferArgumentType(
+ callContext,
+ i,
+ throwOnFailure);
+ if (!inferredDataType.isPresent()) {
+ return Optional.empty();
}
+ inferredDataTypes.add(inferredDataType.get());
}
- return true;
+ return Optional.of(inferredDataTypes);
}
@Override
public List<Signature> getExpectedSignatures(FunctionDefinition definition) {
final List<Signature.Argument> arguments = new ArrayList<>();
- for (int i = 0; i < validators.size(); i++) {
+ for (int i = 0; i < argumentStrategies.size(); i++) {
if (argumentNames == null) {
- arguments.add(validators.get(i).getExpectedArgument(definition, i));
+ arguments.add(argumentStrategies.get(i).getExpectedArgument(definition, i));
} else {
arguments.add(Signature.Argument.of(
argumentNames.get(i),
- validators.get(i).getExpectedArgument(definition, i).getType()));
+ argumentStrategies.get(i).getExpectedArgument(definition, i).getType()));
}
}
@@ -100,13 +106,13 @@ public final class SequenceInputValidator implements InputTypeValidator {
if (o == null || getClass() != o.getClass()) {
return false;
}
- SequenceInputValidator that = (SequenceInputValidator) o;
- return Objects.equals(validators, that.validators) &&
+ SequenceInputTypeStrategy that = (SequenceInputTypeStrategy) o;
+ return Objects.equals(argumentStrategies, that.argumentStrategies) &&
Objects.equals(argumentNames, that.argumentNames);
}
@Override
public int hashCode() {
- return Objects.hash(validators, argumentNames);
+ return Objects.hash(argumentStrategies, argumentNames);
}
}
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/VaryingSequenceInputTypeStrategy.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/VaryingSequenceInputTypeStrategy.java
new file mode 100644
index 0000000..70f85f5
--- /dev/null
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/VaryingSequenceInputTypeStrategy.java
@@ -0,0 +1,150 @@
+/*
+ * 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.functions.FunctionDefinition;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.inference.ArgumentCount;
+import org.apache.flink.table.types.inference.ArgumentTypeStrategy;
+import org.apache.flink.table.types.inference.CallContext;
+import org.apache.flink.table.types.inference.ConstantArgumentCount;
+import org.apache.flink.table.types.inference.InputTypeStrategy;
+import org.apache.flink.table.types.inference.Signature;
+import org.apache.flink.util.Preconditions;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+
+/**
+ * Strategy for inferring and validating a varying function signature like {@code f(INT, STRING, NUMERIC...)}
+ * or {@code f(i INT, str STRING, num NUMERIC...)} using a sequence of {@link ArgumentTypeStrategy}s. The
+ * first n - 1 arguments must be constant. The n-th argument can occur 0, 1, or more times.
+ */
+@Internal
+public final class VaryingSequenceInputTypeStrategy implements InputTypeStrategy {
+
+ private final int constantArgumentCount;
+
+ private final List<ArgumentTypeStrategy> constantArgumentStrategies;
+
+ private final ArgumentTypeStrategy varyingArgumentStrategy;
+
+ private final @Nullable List<String> argumentNames;
+
+ public VaryingSequenceInputTypeStrategy(
+ List<ArgumentTypeStrategy> argumentStrategies,
+ @Nullable List<String> argumentNames) {
+ Preconditions.checkArgument(argumentStrategies.size() > 0);
+ Preconditions.checkArgument(argumentNames == null || argumentNames.size() == argumentStrategies.size());
+ constantArgumentCount = argumentStrategies.size() - 1;
+ constantArgumentStrategies = argumentStrategies.subList(0, constantArgumentCount);
+ varyingArgumentStrategy = argumentStrategies.get(constantArgumentCount);
+ this.argumentNames = argumentNames;
+ }
+
+ @Override
+ public ArgumentCount getArgumentCount() {
+ return ConstantArgumentCount.from(constantArgumentCount);
+ }
+
+ @Override
+ public Optional<List<DataType>> inferInputTypes(CallContext callContext, boolean throwOnFailure) {
+ final List<DataType> dataTypes = callContext.getArgumentDataTypes();
+ if (dataTypes.size() < constantArgumentCount) {
+ return Optional.empty();
+ }
+ final List<DataType> inferredDataTypes = new ArrayList<>(dataTypes.size());
+ for (int i = 0; i < callContext.getArgumentDataTypes().size(); i++) {
+ final ArgumentTypeStrategy argumentTypeStrategy;
+ if (i < constantArgumentCount) {
+ argumentTypeStrategy = constantArgumentStrategies.get(i);
+ } else {
+ argumentTypeStrategy = varyingArgumentStrategy;
+ }
+ final Optional<DataType> inferredDataType = argumentTypeStrategy.inferArgumentType(
+ callContext,
+ i,
+ throwOnFailure);
+ if (!inferredDataType.isPresent()) {
+ return Optional.empty();
+ }
+ inferredDataTypes.add(inferredDataType.get());
+ }
+ return Optional.of(inferredDataTypes);
+ }
+
+ @Override
+ public List<Signature> getExpectedSignatures(FunctionDefinition definition) {
+ final Signature.Argument varyingArgument = varyingArgumentStrategy.getExpectedArgument(
+ definition,
+ constantArgumentCount);
+ final Signature.Argument newArg;
+ final String type = varyingArgument.getType();
+ if (argumentNames == null) {
+ newArg = Signature.Argument.of(type + "...");
+ } else {
+ newArg = Signature.Argument.of(argumentNames.get(constantArgumentCount), type + "...");
+ }
+
+ final List<Signature.Argument> arguments = new ArrayList<>();
+ for (int i = 0; i < constantArgumentCount; i++) {
+ if (argumentNames == null) {
+ arguments.add(constantArgumentStrategies.get(i).getExpectedArgument(definition, i));
+ } else {
+ arguments.add(Signature.Argument.of(
+ argumentNames.get(i),
+ constantArgumentStrategies.get(i).getExpectedArgument(definition, i).getType()));
+ }
+ }
+
+ arguments.add(newArg);
+
+ return Collections.singletonList(Signature.of(arguments));
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ VaryingSequenceInputTypeStrategy that = (VaryingSequenceInputTypeStrategy) o;
+ return constantArgumentCount == that.constantArgumentCount &&
+ Objects.equals(constantArgumentStrategies, that.constantArgumentStrategies) &&
+ Objects.equals(varyingArgumentStrategy, that.varyingArgumentStrategy) &&
+ Objects.equals(argumentNames, that.argumentNames);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(
+ constantArgumentCount,
+ constantArgumentStrategies,
+ varyingArgumentStrategy,
+ argumentNames);
+ }
+}
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/validators/PassingTypeValidator.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/WildcardInputTypeStrategy.java
similarity index 68%
rename from flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/validators/PassingTypeValidator.java
rename to flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/WildcardInputTypeStrategy.java
index ecc272d..7a45939 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/validators/PassingTypeValidator.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/WildcardInputTypeStrategy.java
@@ -16,26 +16,27 @@
* limitations under the License.
*/
-package org.apache.flink.table.types.inference.validators;
+package org.apache.flink.table.types.inference.strategies;
import org.apache.flink.annotation.Internal;
import org.apache.flink.table.functions.FunctionDefinition;
+import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.inference.ArgumentCount;
-import org.apache.flink.table.types.inference.ArgumentTypeValidator;
import org.apache.flink.table.types.inference.CallContext;
import org.apache.flink.table.types.inference.ConstantArgumentCount;
-import org.apache.flink.table.types.inference.InputTypeValidator;
+import org.apache.flink.table.types.inference.InputTypeStrategy;
import org.apache.flink.table.types.inference.Signature;
import org.apache.flink.table.types.inference.Signature.Argument;
import java.util.Collections;
import java.util.List;
+import java.util.Optional;
/**
- * Validator that does not perform any validation and always passes.
+ * Strategy that does not perform any modification or validation of the input.
*/
@Internal
-public final class PassingTypeValidator implements InputTypeValidator, ArgumentTypeValidator {
+public final class WildcardInputTypeStrategy implements InputTypeStrategy {
private static final ArgumentCount PASSING_ARGUMENT_COUNT = ConstantArgumentCount.any();
@@ -45,8 +46,8 @@ public final class PassingTypeValidator implements InputTypeValidator, ArgumentT
}
@Override
- public boolean validate(CallContext callContext, boolean throwOnFailure) {
- return true;
+ public Optional<List<DataType>> inferInputTypes(CallContext callContext, boolean throwOnFailure) {
+ return Optional.of(callContext.getArgumentDataTypes());
}
@Override
@@ -56,21 +57,11 @@ public final class PassingTypeValidator implements InputTypeValidator, ArgumentT
@Override
public boolean equals(Object o) {
- return this == o || o instanceof PassingTypeValidator;
+ return this == o || o instanceof WildcardInputTypeStrategy;
}
@Override
public int hashCode() {
- return PassingTypeValidator.class.hashCode();
- }
-
- @Override
- public boolean validateArgument(CallContext callContext, int argumentPos, boolean throwOnFailure) {
- return true;
- }
-
- @Override
- public Argument getExpectedArgument(FunctionDefinition functionDefinition, int argumentPos) {
- return Argument.of("*");
+ return WildcardInputTypeStrategy.class.hashCode();
}
}
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/utils/AdaptedCallContext.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/utils/AdaptedCallContext.java
index 5240a64..8f7e1a3 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/utils/AdaptedCallContext.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/utils/AdaptedCallContext.java
@@ -19,14 +19,14 @@
package org.apache.flink.table.types.inference.utils;
import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.catalog.DataTypeLookup;
import org.apache.flink.table.functions.FunctionDefinition;
import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.inference.CallContext;
-import org.apache.flink.table.types.inference.MutableCallContext;
+import org.apache.flink.util.Preconditions;
import javax.annotation.Nullable;
-import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
@@ -37,26 +37,28 @@ import java.util.Optional;
* literal before is not a literal anymore in this call context.
*/
@Internal
-public final class AdaptedCallContext implements MutableCallContext {
+public final class AdaptedCallContext implements CallContext {
private final CallContext originalContext;
- private final List<DataType> expectedArguments;
-
private final @Nullable DataType outputDataType;
- public AdaptedCallContext(
- CallContext originalContext,
- List<DataType> castedArguments,
- @Nullable DataType outputDataType) {
+ private List<DataType> expectedArguments;
+
+ public AdaptedCallContext(CallContext originalContext, @Nullable DataType outputDataType) {
this.originalContext = originalContext;
- this.expectedArguments = new ArrayList<>(castedArguments);
+ this.expectedArguments = originalContext.getArgumentDataTypes();
this.outputDataType = outputDataType;
}
+ public void setExpectedArguments(List<DataType> expectedArguments) {
+ Preconditions.checkArgument(this.expectedArguments.size() == expectedArguments.size());
+ this.expectedArguments = expectedArguments;
+ }
+
@Override
- public List<DataType> getArgumentDataTypes() {
- return expectedArguments;
+ public DataTypeLookup getDataTypeLookup() {
+ return originalContext.getDataTypeLookup();
}
@Override
@@ -91,17 +93,17 @@ public final class AdaptedCallContext implements MutableCallContext {
return originalContext.getName();
}
- private boolean isCasted(int pos) {
- return !originalContext.getArgumentDataTypes().get(pos).equals(expectedArguments.get(pos));
- }
-
@Override
- public void mutateArgumentDataType(int pos, DataType newDataType) {
- expectedArguments.set(pos, newDataType);
+ public List<DataType> getArgumentDataTypes() {
+ return expectedArguments;
}
@Override
public Optional<DataType> getOutputDataType() {
return Optional.ofNullable(outputDataType);
}
+
+ private boolean isCasted(int pos) {
+ return !originalContext.getArgumentDataTypes().get(pos).equals(expectedArguments.get(pos));
+ }
}
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/utils/UnknownCallContext.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/utils/UnknownCallContext.java
index f4ba594..503c732 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/utils/UnknownCallContext.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/utils/UnknownCallContext.java
@@ -20,6 +20,7 @@ package org.apache.flink.table.types.inference.utils;
import org.apache.flink.annotation.Internal;
import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.catalog.DataTypeLookup;
import org.apache.flink.table.functions.FunctionDefinition;
import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.inference.CallContext;
@@ -36,6 +37,8 @@ public final class UnknownCallContext implements CallContext {
private static final DataType NULL = DataTypes.NULL();
+ private final DataTypeLookup lookup;
+
private final String name;
private final FunctionDefinition functionDefinition;
@@ -43,9 +46,11 @@ public final class UnknownCallContext implements CallContext {
private final List<DataType> argumentDataTypes;
public UnknownCallContext(
+ DataTypeLookup lookup,
String name,
FunctionDefinition functionDefinition,
int argumentCount) {
+ this.lookup = lookup;
this.name = name;
this.functionDefinition = functionDefinition;
this.argumentDataTypes = new AbstractList<DataType>() {
@@ -62,6 +67,11 @@ public final class UnknownCallContext implements CallContext {
}
@Override
+ public DataTypeLookup getDataTypeLookup() {
+ return lookup;
+ }
+
+ @Override
public FunctionDefinition getFunctionDefinition() {
return functionDefinition;
}
@@ -90,4 +100,9 @@ public final class UnknownCallContext implements CallContext {
public List<DataType> getArgumentDataTypes() {
return argumentDataTypes;
}
+
+ @Override
+ public Optional<DataType> getOutputDataType() {
+ return Optional.empty();
+ }
}
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/validators/AndTypeArgumentValidator.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/validators/AndTypeArgumentValidator.java
deleted file mode 100644
index 43b9335..0000000
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/validators/AndTypeArgumentValidator.java
+++ /dev/null
@@ -1,80 +0,0 @@
-/*
- * 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.validators;
-
-import org.apache.flink.annotation.Internal;
-import org.apache.flink.table.functions.FunctionDefinition;
-import org.apache.flink.table.types.inference.ArgumentTypeValidator;
-import org.apache.flink.table.types.inference.CallContext;
-import org.apache.flink.table.types.inference.Signature;
-import org.apache.flink.util.Preconditions;
-
-import java.util.List;
-import java.util.Objects;
-import java.util.stream.Collectors;
-
-/**
- * Validator that checks for a conjunction of multiple {@link ArgumentTypeValidator}s into one like
- * {@code f(NUMERIC && LITERAL)}.
- */
-@Internal
-public final class AndTypeArgumentValidator implements ArgumentTypeValidator {
-
- private final List<? extends ArgumentTypeValidator> validators;
-
- public AndTypeArgumentValidator(List<? extends ArgumentTypeValidator> validators) {
- Preconditions.checkArgument(validators.size() > 0);
- this.validators = validators;
- }
-
- @Override
- public boolean validateArgument(CallContext callContext, int argumentPos, boolean throwOnFailure) {
- for (ArgumentTypeValidator validator : validators) {
- if (!validator.validateArgument(callContext, argumentPos, throwOnFailure)) {
- return false;
- }
- }
- return true;
- }
-
- @Override
- public Signature.Argument getExpectedArgument(FunctionDefinition functionDefinition, int argumentPos) {
- final String argument = validators.stream()
- .map(v -> v.getExpectedArgument(functionDefinition, argumentPos).getType())
- .collect(Collectors.joining(" & ", "[", "]"));
- return Signature.Argument.of(argument);
- }
-
- @Override
- public boolean equals(Object o) {
- if (this == o) {
- return true;
- }
- if (o == null || getClass() != o.getClass()) {
- return false;
- }
- AndTypeArgumentValidator that = (AndTypeArgumentValidator) o;
- return Objects.equals(validators, that.validators);
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(validators);
- }
-}
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/validators/OrTypeArgumentValidator.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/validators/OrTypeArgumentValidator.java
deleted file mode 100644
index eb91b8d..0000000
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/validators/OrTypeArgumentValidator.java
+++ /dev/null
@@ -1,86 +0,0 @@
-/*
- * 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.validators;
-
-import org.apache.flink.annotation.Internal;
-import org.apache.flink.table.functions.FunctionDefinition;
-import org.apache.flink.table.types.inference.ArgumentTypeValidator;
-import org.apache.flink.table.types.inference.CallContext;
-import org.apache.flink.table.types.inference.Signature;
-import org.apache.flink.util.Preconditions;
-
-import java.util.List;
-import java.util.Objects;
-import java.util.stream.Collectors;
-
-/**
- * Validator that checks for a disjunction of multiple {@link ArgumentTypeValidator}s into one like
- * {@code f(NUMERIC || STRING)}.
- */
-@Internal
-public final class OrTypeArgumentValidator implements ArgumentTypeValidator {
-
- private final List<? extends ArgumentTypeValidator> validators;
-
- public OrTypeArgumentValidator(List<? extends ArgumentTypeValidator> validators) {
- Preconditions.checkArgument(validators.size() > 0);
- this.validators = validators;
- }
-
- @Override
- public boolean validateArgument(CallContext callContext, int argumentPos, boolean throwOnFailure) {
- for (ArgumentTypeValidator validator : validators) {
- if (validator.validateArgument(callContext, argumentPos, false)) {
- return true;
- }
- }
- // generate a helpful exception if possible
- if (throwOnFailure) {
- for (ArgumentTypeValidator validator : validators) {
- validator.validateArgument(callContext, argumentPos, true);
- }
- }
- return false;
- }
-
- @Override
- public Signature.Argument getExpectedArgument(FunctionDefinition functionDefinition, int argumentPos) {
- final String argument = validators.stream()
- .map(v -> v.getExpectedArgument(functionDefinition, argumentPos).getType())
- .collect(Collectors.joining(" | ", "[", "]"));
- return Signature.Argument.of(argument);
- }
-
- @Override
- public boolean equals(Object o) {
- if (this == o) {
- return true;
- }
- if (o == null || getClass() != o.getClass()) {
- return false;
- }
- OrTypeArgumentValidator that = (OrTypeArgumentValidator) o;
- return Objects.equals(validators, that.validators);
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(validators);
- }
-}
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/validators/VaryingSequenceTypeValidator.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/validators/VaryingSequenceTypeValidator.java
deleted file mode 100644
index 41b2812..0000000
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/validators/VaryingSequenceTypeValidator.java
+++ /dev/null
@@ -1,134 +0,0 @@
-/*
- * 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.validators;
-
-import org.apache.flink.annotation.Internal;
-import org.apache.flink.table.functions.FunctionDefinition;
-import org.apache.flink.table.types.inference.ArgumentCount;
-import org.apache.flink.table.types.inference.ArgumentTypeValidator;
-import org.apache.flink.table.types.inference.CallContext;
-import org.apache.flink.table.types.inference.ConstantArgumentCount;
-import org.apache.flink.table.types.inference.InputTypeValidator;
-import org.apache.flink.table.types.inference.Signature;
-import org.apache.flink.util.Preconditions;
-
-import javax.annotation.Nullable;
-
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.List;
-import java.util.Objects;
-
-/**
- * A varying sequence of {@link ArgumentTypeValidator}s for validating an entire function signature
- * like {@code f(INT, STRING, NUMERIC...)}. The first n - 1 arguments must be constant. The n-th
- * argument can occur 0, 1, or more times.
- */
-@Internal
-public final class VaryingSequenceTypeValidator implements InputTypeValidator {
-
- private final int constantArgumentCount;
-
- private final List<ArgumentTypeValidator> constantValidators;
-
- private final ArgumentTypeValidator varyingValidator;
-
- private final @Nullable List<String> argumentNames;
-
- public VaryingSequenceTypeValidator(
- List<ArgumentTypeValidator> validators,
- @Nullable List<String> argumentNames) {
- Preconditions.checkArgument(validators.size() > 0);
- Preconditions.checkArgument(argumentNames == null || argumentNames.size() == validators.size());
- constantArgumentCount = validators.size() - 1;
- constantValidators = validators.subList(0, constantArgumentCount);
- varyingValidator = validators.get(constantArgumentCount);
- this.argumentNames = argumentNames;
- }
-
- @Override
- public ArgumentCount getArgumentCount() {
- return ConstantArgumentCount.from(constantArgumentCount); // at least the constant part
- }
-
- @Override
- public boolean validate(CallContext callContext, boolean throwOnFailure) {
- for (int i = 0; i < callContext.getArgumentDataTypes().size(); i++) {
- if (i < constantArgumentCount) {
- if (!constantValidators.get(i).validateArgument(callContext, i, throwOnFailure)) {
- return false;
- }
- } else {
- if (!varyingValidator.validateArgument(callContext, i, throwOnFailure)) {
- return false;
- }
- }
- }
- return true;
- }
-
- @Override
- public List<Signature> getExpectedSignatures(FunctionDefinition definition) {
- final Signature.Argument varyingArgument = varyingValidator.getExpectedArgument(
- definition,
- constantValidators.size());
- final Signature.Argument newArg;
- final String type = varyingArgument.getType();
- if (argumentNames == null) {
- newArg = Signature.Argument.of(type + "...");
- } else {
- newArg = Signature.Argument.of(argumentNames.get(argumentNames.size() - 1), type + "...");
- }
-
- final List<Signature.Argument> arguments = new ArrayList<>();
- for (int i = 0; i < constantValidators.size(); i++) {
- if (argumentNames == null) {
- arguments.add(constantValidators.get(i).getExpectedArgument(definition, i));
- } else {
- arguments.add(Signature.Argument.of(
- argumentNames.get(i),
- constantValidators.get(i).getExpectedArgument(definition, i).getType()));
- }
- }
-
- arguments.add(newArg);
-
- return Collections.singletonList(Signature.of(arguments));
- }
-
- @Override
- public boolean equals(Object o) {
- if (this == o) {
- return true;
- }
- if (o == null || getClass() != o.getClass()) {
- return false;
- }
- VaryingSequenceTypeValidator that = (VaryingSequenceTypeValidator) o;
- return constantArgumentCount == that.constantArgumentCount &&
- Objects.equals(constantValidators, that.constantValidators) &&
- Objects.equals(varyingValidator, that.varyingValidator) &&
- Objects.equals(argumentNames, that.argumentNames);
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(constantArgumentCount, constantValidators, varyingValidator, argumentNames);
- }
-}
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 42c8f4e..bacbc42 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
@@ -1,12 +1,13 @@
/*
- * 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
+ * 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
+ * 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,
@@ -21,11 +22,10 @@ import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.ValidationException;
import org.apache.flink.table.functions.FunctionKind;
import org.apache.flink.table.types.DataType;
-import org.apache.flink.table.types.inference.strategies.BridgingInputTypeStrategy.BridgingSignature;
import org.apache.flink.table.types.inference.utils.CallContextMock;
+import org.apache.flink.table.types.inference.utils.DataTypeLookupMock;
import org.apache.flink.table.types.inference.utils.FunctionDefinitionMock;
-import org.junit.Assert;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.ExpectedException;
@@ -36,11 +36,28 @@ import org.junit.runners.Parameterized.Parameters;
import javax.annotation.Nullable;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
-
+import java.util.Optional;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import static java.util.Arrays.asList;
+import static org.apache.flink.table.types.inference.InputTypeStrategies.ANY;
+import static org.apache.flink.table.types.inference.InputTypeStrategies.LITERAL;
+import static org.apache.flink.table.types.inference.InputTypeStrategies.LITERAL_OR_NULL;
+import static org.apache.flink.table.types.inference.InputTypeStrategies.OUTPUT_IF_NULL;
+import static org.apache.flink.table.types.inference.InputTypeStrategies.WILDCARD;
+import static org.apache.flink.table.types.inference.InputTypeStrategies.and;
+import static org.apache.flink.table.types.inference.InputTypeStrategies.explicit;
+import static org.apache.flink.table.types.inference.InputTypeStrategies.explicitSequence;
+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.util.CoreMatchers.containsCause;
import static org.hamcrest.CoreMatchers.equalTo;
+import static org.junit.Assert.assertThat;
/**
* Tests for built-in {@link InputTypeStrategies}.
@@ -50,53 +67,319 @@ public class InputTypeStrategiesTest {
@Parameters
public static List<TestSpec> testData() {
- return Arrays.asList(
- // no inference
+ return asList(
+ // wildcard with 2 arguments
+ TestSpec
+ .forStrategy(WILDCARD)
+ .calledWithArgumentTypes(DataTypes.INT(), DataTypes.INT())
+ .expectSignature("f(*)")
+ .expectArgumentTypes(DataTypes.INT(), DataTypes.INT()),
+
+ // wildcard with 0 arguments
+ TestSpec
+ .forStrategy(WILDCARD)
+ .calledWithArgumentTypes()
+ .expectSignature("f(*)")
+ .expectArgumentTypes(),
+
+ // explicit sequence
+ TestSpec
+ .forStrategy(explicitSequence(DataTypes.INT().bridgedTo(int.class), DataTypes.BOOLEAN()))
+ .calledWithArgumentTypes(DataTypes.INT(), DataTypes.BOOLEAN())
+ .expectSignature("f(INT, BOOLEAN)")
+ .expectArgumentTypes(DataTypes.INT().bridgedTo(int.class), DataTypes.BOOLEAN()),
+
+ // invalid named sequence
+ TestSpec
+ .forStrategy(
+ explicitSequence(
+ new String[]{"i", "s"},
+ new DataType[]{DataTypes.INT(), DataTypes.STRING()}))
+ .calledWithArgumentTypes(DataTypes.INT())
+ .expectErrorMessage("Invalid input arguments. Expected signatures are:\nf(i INT, s STRING)"),
+
+ // incompatible nullability
+ TestSpec
+ .forStrategy(explicitSequence(DataTypes.BIGINT().notNull()))
+ .calledWithArgumentTypes(DataTypes.BIGINT())
+ .expectErrorMessage("Unsupported argument type. Expected type 'BIGINT NOT NULL' but actual type was 'BIGINT'."),
+
+ // implicit cast
+ TestSpec
+ .forStrategy(explicitSequence(DataTypes.BIGINT()))
+ .calledWithArgumentTypes(DataTypes.INT())
+ .expectArgumentTypes(DataTypes.BIGINT()),
+
+ // incompatible types
TestSpec
- .forInputStrategy(InputTypeStrategies.NOP)
+ .forStrategy(explicitSequence(DataTypes.BIGINT()))
+ .calledWithArgumentTypes(DataTypes.STRING())
+ .expectErrorMessage("Unsupported argument type. Expected type 'BIGINT' but actual type was 'STRING'."),
+
+ // incompatible number of arguments
+ TestSpec
+ .forStrategy(explicitSequence(DataTypes.BIGINT(), DataTypes.BIGINT()))
+ .calledWithArgumentTypes(DataTypes.BIGINT())
+ .expectErrorMessage("Invalid number of arguments. At least 2 arguments expected but 1 passed."),
+
+ // any type
+ TestSpec
+ .forStrategy(sequence(ANY))
+ .calledWithArgumentTypes(DataTypes.BIGINT())
+ .expectSignature("f(<ANY>)")
+ .expectArgumentTypes(DataTypes.BIGINT()),
+
+ // incompatible number of arguments
+ TestSpec
+ .forStrategy(sequence(ANY))
+ .calledWithArgumentTypes(DataTypes.BIGINT(), DataTypes.BIGINT())
+ .expectErrorMessage("Invalid number of arguments. At most 1 arguments expected but 2 passed."),
+
+ // OR with bridging class
+ TestSpec
+ .forStrategy(
+ or(
+ explicitSequence(DataTypes.STRING()),
+ explicitSequence(DataTypes.INT().bridgedTo(int.class)),
+ explicitSequence(DataTypes.BOOLEAN())))
+ .calledWithArgumentTypes(DataTypes.INT())
+ .calledWithArgumentTypes(DataTypes.TINYINT())
+ .expectSignature("f(STRING)\nf(INT)\nf(BOOLEAN)")
+ .expectArgumentTypes(DataTypes.INT().bridgedTo(int.class)),
+
+ // OR with implicit casting
+ TestSpec
+ .forStrategy(
+ or(
+ explicitSequence(DataTypes.TINYINT()),
+ explicitSequence(DataTypes.INT()),
+ explicitSequence(DataTypes.BIGINT())))
+ .calledWithArgumentTypes(DataTypes.SMALLINT())
+ .expectArgumentTypes(DataTypes.INT()),
+
+ // invalid type in OR
+ TestSpec
+ .forStrategy(or(explicitSequence(DataTypes.INT()), explicitSequence(DataTypes.STRING())))
+ .calledWithArgumentTypes(DataTypes.BOOLEAN())
+ .expectErrorMessage("Invalid input arguments. Expected signatures are:\nf(INT)\nf(STRING)"),
+
+ // invalid typed sequence
+ TestSpec
+ .forStrategy(explicitSequence(DataTypes.INT(), DataTypes.BOOLEAN()))
+ .calledWithArgumentTypes(DataTypes.BOOLEAN(), DataTypes.INT())
+ .expectErrorMessage("Invalid input arguments. Expected signatures are:\nf(INT, BOOLEAN)"),
+
+ // sequence with wildcard
+ TestSpec
+ .forStrategy(sequence(ANY, explicit(DataTypes.INT())))
+ .calledWithArgumentTypes(DataTypes.BOOLEAN(), DataTypes.INT())
+ .calledWithArgumentTypes(DataTypes.BOOLEAN(), DataTypes.TINYINT())
+ .expectArgumentTypes(DataTypes.BOOLEAN(), DataTypes.INT()),
+
+ // invalid named sequence
+ TestSpec
+ .forStrategy(sequence(
+ new String[]{"any", "int"},
+ new ArgumentTypeStrategy[]{ANY, explicit(DataTypes.INT())}))
+ .calledWithArgumentTypes(DataTypes.STRING(), DataTypes.BOOLEAN())
+ .expectErrorMessage("Invalid input arguments. Expected signatures are:\nf(any <ANY>, int INT)"),
+
+ // sequence with OR and implicit casting
+ TestSpec
+ .forStrategy(
+ sequence(
+ explicit(DataTypes.INT()),
+ or(explicit(DataTypes.BOOLEAN()), explicit(DataTypes.INT()))))
+ .expectSignature("f(INT, [BOOLEAN | INT])")
+ .calledWithArgumentTypes(DataTypes.INT(), DataTypes.INT())
+ .calledWithArgumentTypes(DataTypes.TINYINT(), DataTypes.TINYINT())
+ .expectArgumentTypes(DataTypes.INT(), DataTypes.INT()),
+
+ // sequence with OR
+ TestSpec
+ .forStrategy(
+ sequence(
+ explicit(DataTypes.INT()),
+ or(explicit(DataTypes.BOOLEAN()), explicit(DataTypes.STRING()))))
+ .calledWithArgumentTypes(DataTypes.INT(), DataTypes.BIGINT())
+ .expectErrorMessage("Invalid input arguments. Expected signatures are:\nf(INT, [BOOLEAN | STRING])"),
+
+ // sequence with literal
+ TestSpec
+ .forStrategy(sequence(LITERAL))
+ .calledWithLiteralAt(0)
+ .calledWithArgumentTypes(DataTypes.INT())
+ .expectArgumentTypes(DataTypes.INT()),
+
+ // sequence with literal
+ TestSpec
+ .forStrategy(sequence(and(LITERAL, explicit(DataTypes.STRING())), explicit(DataTypes.INT())))
+ .calledWithLiteralAt(0)
+ .calledWithArgumentTypes(DataTypes.STRING(), DataTypes.INT())
+ .expectSignature("f([<LITERAL NOT NULL> & STRING], INT)")
+ .expectArgumentTypes(DataTypes.STRING(), DataTypes.INT()),
+
+ // sequence with missing literal
+ TestSpec
+ .forStrategy(sequence(and(explicit(DataTypes.STRING()), LITERAL_OR_NULL), explicit(DataTypes.INT())))
+ .calledWithArgumentTypes(DataTypes.STRING(), DataTypes.INT())
+ .expectErrorMessage("Invalid input arguments. Expected signatures are:\nf([STRING & <LITERAL>], INT)"),
+
+ // vararg sequence
+ TestSpec
+ .forStrategy(
+ varyingSequence(
+ new String[]{"i", "s", "var"},
+ new ArgumentTypeStrategy[]{
+ explicit(DataTypes.INT()),
+ explicit(DataTypes.STRING()),
+ explicit(DataTypes.BOOLEAN())}))
+ .calledWithArgumentTypes(
+ DataTypes.INT(),
+ DataTypes.STRING(),
+ DataTypes.BOOLEAN(),
+ DataTypes.BOOLEAN(),
+ DataTypes.BOOLEAN())
+ .expectArgumentTypes(
+ DataTypes.INT(),
+ DataTypes.STRING(),
+ DataTypes.BOOLEAN(),
+ DataTypes.BOOLEAN(),
+ DataTypes.BOOLEAN()),
+
+ // vararg sequence with conversion class
+ TestSpec
+ .forStrategy(
+ varyingSequence(
+ new String[]{"var"},
+ new ArgumentTypeStrategy[]{explicit(DataTypes.BOOLEAN().bridgedTo(boolean.class))}))
+ .calledWithArgumentTypes(
+ DataTypes.BOOLEAN(),
+ DataTypes.BOOLEAN(),
+ DataTypes.BOOLEAN())
+ .expectSignature("f(var BOOLEAN...)")
+ .expectArgumentTypes(
+ DataTypes.BOOLEAN().bridgedTo(boolean.class),
+ DataTypes.BOOLEAN().bridgedTo(boolean.class),
+ DataTypes.BOOLEAN().bridgedTo(boolean.class)),
+
+ // vararg sequence
+ TestSpec
+ .forStrategy(
+ varyingSequence(
+ new String[]{"i", "s", "var"},
+ new ArgumentTypeStrategy[]{
+ explicit(DataTypes.INT()),
+ explicit(DataTypes.STRING()),
+ explicit(DataTypes.BOOLEAN())}))
.calledWithArgumentTypes(DataTypes.INT(), DataTypes.STRING())
.expectArgumentTypes(DataTypes.INT(), DataTypes.STRING()),
+ // invalid vararg type
+ TestSpec
+ .forStrategy(
+ varyingSequence(
+ new String[]{"i", "s", "var"},
+ new ArgumentTypeStrategy[]{
+ explicit(DataTypes.INT()),
+ explicit(DataTypes.STRING()),
+ explicit(DataTypes.BOOLEAN())}))
+ .calledWithArgumentTypes(DataTypes.INT(), DataTypes.STRING(), DataTypes.STRING())
+ .expectErrorMessage("Invalid input arguments. Expected signatures are:\nf(i INT, s STRING, var BOOLEAN...)"),
+
+ // invalid non-vararg type
+ TestSpec
+ .forStrategy(
+ varyingSequence(
+ new String[]{"i", "s", "var"},
+ new ArgumentTypeStrategy[]{
+ explicit(DataTypes.INT()),
+ explicit(DataTypes.STRING()),
+ explicit(DataTypes.BOOLEAN())}))
+ .calledWithArgumentTypes(DataTypes.INT(), DataTypes.INT(), DataTypes.BOOLEAN())
+ .expectErrorMessage("Unsupported argument type. Expected type 'STRING' but actual type was 'INT'."),
+
+ // OR in vararg type
+ TestSpec
+ .forStrategy(
+ varyingSequence(
+ new String[]{"i", "s", "var"},
+ new ArgumentTypeStrategy[]{
+ explicit(DataTypes.INT()),
+ explicit(DataTypes.STRING()),
+ or(explicit(DataTypes.BOOLEAN()), explicit(DataTypes.INT()))}))
+ .calledWithArgumentTypes(DataTypes.INT(), DataTypes.STRING(), DataTypes.INT(), DataTypes.BOOLEAN())
+ .expectArgumentTypes(DataTypes.INT(), DataTypes.STRING(), DataTypes.INT(), DataTypes.BOOLEAN()),
+
+ // invalid OR in vararg type
+ TestSpec
+ .forStrategy(
+ varyingSequence(
+ new String[]{"i", "s", "var"},
+ new ArgumentTypeStrategy[]{
+ explicit(DataTypes.INT()),
+ explicit(DataTypes.STRING()),
+ or(explicit(DataTypes.BOOLEAN()), explicit(DataTypes.INT()))}))
+ .calledWithArgumentTypes(DataTypes.INT(), DataTypes.STRING(), DataTypes.STRING(), DataTypes.STRING())
+ .expectErrorMessage("Invalid input arguments. Expected signatures are:\nf(i INT, s STRING, var [BOOLEAN | INT]...)"),
+
// incomplete inference
TestSpec
- .forInputStrategy(InputTypeStrategies.NOP)
+ .forStrategy(WILDCARD)
.calledWithArgumentTypes(DataTypes.NULL(), DataTypes.STRING(), DataTypes.NULL())
.expectErrorMessage("Invalid use of untyped NULL in arguments."),
// typed arguments help inferring a type
TestSpec
- .forInputStrategy(InputTypeStrategies.NOP)
- .typedArguments(DataTypes.INT(), DataTypes.STRING(), DataTypes.BOOLEAN())
+ .forStrategy(WILDCARD)
+ .typedArguments(DataTypes.INT().bridgedTo(int.class), DataTypes.STRING(), DataTypes.BOOLEAN())
.calledWithArgumentTypes(DataTypes.NULL(), DataTypes.STRING(), DataTypes.NULL())
- .expectArgumentTypes(DataTypes.INT(), DataTypes.STRING(), DataTypes.BOOLEAN()),
+ .expectArgumentTypes(DataTypes.INT().bridgedTo(int.class), DataTypes.STRING(), DataTypes.BOOLEAN()),
// surrounding function helps inferring a type
TestSpec
- .forInputStrategy(InputTypeStrategies.OUTPUT_TYPE)
- .surroundingStrategy(InputTypeStrategies.explicit(DataTypes.BOOLEAN()))
+ .forStrategy(sequence(OUTPUT_IF_NULL, OUTPUT_IF_NULL, OUTPUT_IF_NULL))
+ .surroundingStrategy(explicitSequence(DataTypes.BOOLEAN()))
.calledWithArgumentTypes(DataTypes.NULL(), DataTypes.STRING(), DataTypes.NULL())
+ .expectSignature("f(<OUTPUT>, <OUTPUT>, <OUTPUT>)")
.expectArgumentTypes(DataTypes.BOOLEAN(), DataTypes.STRING(), DataTypes.BOOLEAN()),
- // enrich data type with conversion class of INT
+ // surrounding function helps inferring a type
TestSpec
- .forInputStrategy(createBridgingInputTypeStrategy())
- .calledWithArgumentTypes(DataTypes.STRING(), DataTypes.INT())
- .expectArgumentTypes(DataTypes.STRING(), DataTypes.INT().bridgedTo(int.class)),
+ .forStrategy(sequence(or(OUTPUT_IF_NULL, explicit(DataTypes.INT()))))
+ .surroundingStrategy(explicitSequence(DataTypes.BOOLEAN()))
+ .calledWithArgumentTypes(DataTypes.NULL())
+ .expectSignature("f([<OUTPUT> | INT])")
+ .expectArgumentTypes(DataTypes.BOOLEAN()),
- // enrich data type with conversion class of varying INT
+ // surrounding function does not help inferring a type
TestSpec
- .forInputStrategy(createBridgingInputTypeStrategy())
- .calledWithArgumentTypes(DataTypes.INT(), DataTypes.INT(), DataTypes.INT())
- .expectArgumentTypes(
- DataTypes.INT().bridgedTo(int.class),
- DataTypes.INT().bridgedTo(int.class),
- DataTypes.INT().bridgedTo(int.class)),
+ .forStrategy(sequence(or(OUTPUT_IF_NULL, explicit(DataTypes.INT()))))
+ .calledWithArgumentTypes(DataTypes.NULL())
+ .expectSignature("f([<OUTPUT> | INT])")
+ .expectArgumentTypes(DataTypes.INT()),
- // check function without arguments
+ // typed arguments only with casting
TestSpec
- .forInputStrategy(createBridgingInputTypeStrategy())
- .calledWithArgumentTypes()
- .expectArgumentTypes()
+ .forStrategy(WILDCARD)
+ .typedArguments(DataTypes.INT(), DataTypes.STRING())
+ .calledWithArgumentTypes(DataTypes.TINYINT(), DataTypes.STRING())
+ .expectSignature("f(INT, STRING)")
+ .expectArgumentTypes(DataTypes.INT(), DataTypes.STRING()),
+
+ // invalid typed arguments
+ TestSpec
+ .forStrategy(WILDCARD)
+ .typedArguments(DataTypes.INT(), DataTypes.STRING())
+ .calledWithArgumentTypes(DataTypes.STRING(), DataTypes.STRING())
+ .expectErrorMessage("Invalid argument type at position 0. Data type INT expected but STRING passed."),
+
+ // named arguments
+ TestSpec
+ .forStrategy(WILDCARD)
+ .namedArguments("i", "s")
+ .typedArguments(DataTypes.INT(), DataTypes.STRING())
+ .expectSignature("f(i => INT, s => STRING)")
);
}
@@ -107,35 +390,49 @@ public class InputTypeStrategiesTest {
public ExpectedException thrown = ExpectedException.none();
@Test
- public void testInputTypeStrategy() {
+ public void testStrategy() {
+ if (testSpec.expectedSignature != null) {
+ assertThat(
+ generateSignature(),
+ equalTo(testSpec.expectedSignature));
+ }
if (testSpec.expectedErrorMessage != null) {
thrown.expect(ValidationException.class);
thrown.expectCause(containsCause(new ValidationException(testSpec.expectedErrorMessage)));
}
- TypeInferenceUtil.Result result = runTypeInference();
- if (testSpec.expectedArgumentTypes != null) {
- Assert.assertThat(result.getExpectedArgumentTypes(), equalTo(testSpec.expectedArgumentTypes));
+ for (List<DataType> actualArgumentTypes : testSpec.actualArgumentTypes) {
+ TypeInferenceUtil.Result result = runTypeInference(actualArgumentTypes);
+ if (testSpec.expectedArgumentTypes != null) {
+ assertThat(result.getExpectedArgumentTypes(), equalTo(testSpec.expectedArgumentTypes));
+ }
}
}
// --------------------------------------------------------------------------------------------
- private TypeInferenceUtil.Result runTypeInference() {
+ private String generateSignature() {
final FunctionDefinitionMock functionDefinitionMock = new FunctionDefinitionMock();
functionDefinitionMock.functionKind = FunctionKind.SCALAR;
+ return TypeInferenceUtil.generateSignature("f", functionDefinitionMock, createTypeInference());
+ }
+
+ private TypeInferenceUtil.Result runTypeInference(List<DataType> actualArgumentTypes) {
+ final FunctionDefinitionMock functionDefinitionMock = new FunctionDefinitionMock();
+ functionDefinitionMock.functionKind = FunctionKind.SCALAR;
+
+ final DataTypeLookupMock dataTypeLookupMock = new DataTypeLookupMock();
+
final CallContextMock callContextMock = new CallContextMock();
callContextMock.functionDefinition = functionDefinitionMock;
- callContextMock.argumentDataTypes = testSpec.actualArgumentTypes;
+ callContextMock.argumentDataTypes = actualArgumentTypes;
+ callContextMock.argumentLiterals = IntStream.range(0, actualArgumentTypes.size())
+ .mapToObj(i -> testSpec.literalPos != null && i == testSpec.literalPos)
+ .collect(Collectors.toList());
+ callContextMock.argumentNulls = IntStream.range(0, actualArgumentTypes.size())
+ .mapToObj(i -> false)
+ .collect(Collectors.toList());
callContextMock.name = "f";
-
- final TypeInference.Builder builder = TypeInference.newBuilder()
- .inputTypeStrategy(testSpec.strategy)
- .inputTypeValidator(InputTypeValidators.PASSING)
- .outputTypeStrategy(TypeStrategies.explicit(DataTypes.BOOLEAN()));
-
- if (testSpec.typedArguments != null) {
- builder.typedArguments(testSpec.typedArguments);
- }
+ callContextMock.outputDataType = Optional.empty();
final TypeInferenceUtil.SurroundingInfo surroundingInfo;
if (testSpec.surroundingStrategy != null) {
@@ -152,7 +449,24 @@ public class InputTypeStrategiesTest {
} else {
surroundingInfo = null;
}
- return TypeInferenceUtil.runTypeInference(builder.build(), callContextMock, surroundingInfo);
+ return TypeInferenceUtil.runTypeInference(
+ createTypeInference(),
+ callContextMock,
+ surroundingInfo);
+ }
+
+ private TypeInference createTypeInference() {
+ final TypeInference.Builder builder = TypeInference.newBuilder()
+ .inputTypeStrategy(testSpec.strategy)
+ .outputTypeStrategy(TypeStrategies.explicit(DataTypes.BOOLEAN()));
+
+ if (testSpec.namedArguments != null) {
+ builder.namedArguments(testSpec.namedArguments);
+ }
+ if (testSpec.typedArguments != null) {
+ builder.typedArguments(testSpec.typedArguments);
+ }
+ return builder.build();
}
// --------------------------------------------------------------------------------------------
@@ -161,12 +475,17 @@ public class InputTypeStrategiesTest {
private final InputTypeStrategy strategy;
- // types explicitly expected by the type inference
+ private @Nullable List<String> namedArguments;
+
private @Nullable List<DataType> typedArguments;
+ private List<List<DataType>> actualArgumentTypes = new ArrayList<>();
+
+ private @Nullable Integer literalPos;
+
private @Nullable InputTypeStrategy surroundingStrategy;
- private @Nullable List<DataType> actualArgumentTypes;
+ private @Nullable String expectedSignature;
private @Nullable List<DataType> expectedArgumentTypes;
@@ -176,10 +495,15 @@ public class InputTypeStrategiesTest {
this.strategy = strategy;
}
- static TestSpec forInputStrategy(InputTypeStrategy strategy) {
+ static TestSpec forStrategy(InputTypeStrategy strategy) {
return new TestSpec(strategy);
}
+ TestSpec namedArguments(String... names) {
+ this.namedArguments = Arrays.asList(names);
+ return this;
+ }
+
TestSpec typedArguments(DataType... dataTypes) {
this.typedArguments = Arrays.asList(dataTypes);
return this;
@@ -191,7 +515,17 @@ public class InputTypeStrategiesTest {
}
TestSpec calledWithArgumentTypes(DataType... dataTypes) {
- this.actualArgumentTypes = Arrays.asList(dataTypes);
+ this.actualArgumentTypes.add(Arrays.asList(dataTypes));
+ return this;
+ }
+
+ TestSpec calledWithLiteralAt(int pos) {
+ this.literalPos = pos;
+ return this;
+ }
+
+ TestSpec expectSignature(String signature) {
+ this.expectedSignature = signature;
return this;
}
@@ -205,23 +539,4 @@ public class InputTypeStrategiesTest {
return this;
}
}
-
- private static InputTypeStrategy createBridgingInputTypeStrategy() {
- return InputTypeStrategies.bridging(
- Arrays.asList(
- new BridgingSignature(
- new DataType[]{DataTypes.STRING(), DataTypes.INT().bridgedTo(int.class)}, false
- ),
- new BridgingSignature(
- new DataType[]{DataTypes.TIMESTAMP(3).bridgedTo(java.sql.Timestamp.class), DataTypes.STRING()}, false
- ),
- new BridgingSignature(
- new DataType[]{}, false
- ),
- new BridgingSignature(
- new DataType[]{DataTypes.INT().bridgedTo(int.class)}, true
- )
- )
- );
- }
}
diff --git a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/InputTypeValidatorsTest.java b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/InputTypeValidatorsTest.java
deleted file mode 100644
index 0ac4de1..0000000
--- a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/InputTypeValidatorsTest.java
+++ /dev/null
@@ -1,348 +0,0 @@
-/*
- * 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;
-
-import org.apache.flink.table.api.DataTypes;
-import org.apache.flink.table.api.ValidationException;
-import org.apache.flink.table.functions.FunctionKind;
-import org.apache.flink.table.types.DataType;
-import org.apache.flink.table.types.inference.utils.CallContextMock;
-import org.apache.flink.table.types.inference.utils.FunctionDefinitionMock;
-
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.rules.ExpectedException;
-import org.junit.runner.RunWith;
-import org.junit.runners.Parameterized;
-import org.junit.runners.Parameterized.Parameter;
-import org.junit.runners.Parameterized.Parameters;
-
-import javax.annotation.Nullable;
-
-import java.util.List;
-import java.util.stream.Collectors;
-import java.util.stream.IntStream;
-
-import static java.util.Arrays.asList;
-import static org.apache.flink.table.types.inference.InputTypeValidators.ANY;
-import static org.apache.flink.table.types.inference.InputTypeValidators.LITERAL;
-import static org.apache.flink.table.types.inference.InputTypeValidators.LITERAL_OR_NULL;
-import static org.apache.flink.table.types.inference.InputTypeValidators.PASSING;
-import static org.apache.flink.table.types.inference.InputTypeValidators.and;
-import static org.apache.flink.table.types.inference.InputTypeValidators.explicit;
-import static org.apache.flink.table.types.inference.InputTypeValidators.explicitSequence;
-import static org.apache.flink.table.types.inference.InputTypeValidators.or;
-import static org.apache.flink.table.types.inference.InputTypeValidators.sequence;
-import static org.apache.flink.table.types.inference.InputTypeValidators.varyingSequence;
-import static org.apache.flink.util.CoreMatchers.containsCause;
-
-/**
- * Tests for built-in {@link InputTypeValidators}.
- */
-@RunWith(Parameterized.class)
-public class InputTypeValidatorsTest {
-
- @Parameters
- public static List<TestSpec> testData() {
- return asList(
- // 2 arguments
- TestSpec
- .forValidator(PASSING)
- .inputTypes(DataTypes.INT(), DataTypes.INT())
- .expectSuccess(),
-
- // 0 arguments
- TestSpec
- .forValidator(PASSING)
- .inputTypes()
- .expectSuccess(),
-
- // full equivalence
- TestSpec
- .forValidator(explicitSequence(DataTypes.INT()))
- .inputTypes(DataTypes.INT())
- .expectSuccess(),
-
- // invalid named sequence
- TestSpec
- .forValidator(
- explicitSequence(
- new String[]{"i", "s"},
- new DataType[]{DataTypes.INT(), DataTypes.STRING()}))
- .inputTypes(DataTypes.INT())
- .expectErrorMessage("Invalid input arguments. Expected signatures are:\nf(i INT, s STRING)"),
-
- // incompatible nullability
- TestSpec
- .forValidator(explicitSequence(DataTypes.BIGINT().notNull()))
- .inputTypes(DataTypes.BIGINT())
- .expectErrorMessage("Unsupported argument type. Expected type 'BIGINT NOT NULL' but actual type was 'BIGINT'."),
-
- // implicit cast
- TestSpec
- .forValidator(explicitSequence(DataTypes.BIGINT()))
- .inputTypes(DataTypes.INT())
- .expectSuccess(),
-
- // incompatible types
- TestSpec
- .forValidator(explicitSequence(DataTypes.BIGINT()))
- .inputTypes(DataTypes.STRING())
- .expectErrorMessage("Unsupported argument type. Expected type 'BIGINT' but actual type was 'STRING'."),
-
- // incompatible number of arguments
- TestSpec
- .forValidator(explicitSequence(DataTypes.BIGINT(), DataTypes.BIGINT()))
- .inputTypes(DataTypes.BIGINT())
- .expectErrorMessage("Invalid number of arguments. At least 2 arguments expected but 1 passed."),
-
- // any type
- TestSpec
- .forValidator(ANY)
- .inputTypes(DataTypes.BIGINT())
- .expectSuccess(),
-
- // incompatible number of arguments
- TestSpec
- .forValidator(ANY)
- .inputTypes(DataTypes.BIGINT(), DataTypes.BIGINT())
- .expectErrorMessage("Invalid number of arguments. At most 1 arguments expected but 2 passed."),
-
- // left of OR
- TestSpec
- .forValidator(or(explicitSequence(DataTypes.INT()), explicitSequence(DataTypes.BOOLEAN())))
- .inputTypes(DataTypes.INT())
- .expectSuccess(),
-
- // right of OR
- TestSpec
- .forValidator(or(explicitSequence(DataTypes.INT()), explicitSequence(DataTypes.BOOLEAN())))
- .inputTypes(DataTypes.BOOLEAN())
- .expectSuccess(),
-
- // invalid type in OR
- TestSpec
- .forValidator(or(explicitSequence(DataTypes.INT()), explicitSequence(DataTypes.STRING())))
- .inputTypes(DataTypes.BOOLEAN())
- .expectErrorMessage("Invalid input arguments. Expected signatures are:\nf(INT)\nf(STRING)"),
-
- // explicit sequence
- TestSpec
- .forValidator(sequence(explicit(DataTypes.INT()), explicit(DataTypes.BOOLEAN())))
- .inputTypes(DataTypes.INT(), DataTypes.BOOLEAN())
- .expectSuccess(),
-
- // invalid sequence
- TestSpec
- .forValidator(sequence(explicit(DataTypes.INT()), explicit(DataTypes.BOOLEAN())))
- .inputTypes(DataTypes.BOOLEAN(), DataTypes.INT())
- .expectErrorMessage("Invalid input arguments. Expected signatures are:\nf(INT, BOOLEAN)"),
-
- // correct sequence with wildcard
- TestSpec
- .forValidator(sequence(ANY, explicit(DataTypes.INT())))
- .inputTypes(DataTypes.BOOLEAN(), DataTypes.INT())
- .expectSuccess(),
-
- // invalid named sequence
- TestSpec
- .forValidator(sequence(
- new String[]{"any", "int"},
- new ArgumentTypeValidator[]{ANY, explicit(DataTypes.INT())}))
- .inputTypes(DataTypes.STRING(), DataTypes.BOOLEAN())
- .expectErrorMessage("Invalid input arguments. Expected signatures are:\nf(any <ANY>, int INT)"),
-
- // sequence with OR
- TestSpec
- .forValidator(sequence(explicit(DataTypes.INT()), or(explicit(DataTypes.BOOLEAN()), explicit(DataTypes.STRING()))))
- .inputTypes(DataTypes.INT(), DataTypes.BIGINT())
- .expectErrorMessage("Invalid input arguments. Expected signatures are:\nf(INT, [BOOLEAN | STRING])"),
-
- // literal
- TestSpec
- .forValidator(LITERAL)
- .inputTypes(DataTypes.INT())
- .literalAt(0)
- .expectSuccess(),
-
- // sequence with literal
- TestSpec
- .forValidator(sequence(and(LITERAL, explicit(DataTypes.STRING())), explicit(DataTypes.INT())))
- .inputTypes(DataTypes.STRING(), DataTypes.INT())
- .literalAt(0)
- .expectSuccess(),
-
- // sequence with missing literal
- TestSpec
- .forValidator(sequence(and(LITERAL_OR_NULL, explicit(DataTypes.STRING())), explicit(DataTypes.INT())))
- .inputTypes(DataTypes.STRING(), DataTypes.INT())
- .expectErrorMessage("Invalid input arguments. Expected signatures are:\nf([<LITERAL> & STRING], INT)"),
-
- // vararg sequence
- TestSpec
- .forValidator(
- varyingSequence(
- new String[]{"i", "s", "var"},
- new ArgumentTypeValidator[]{
- explicit(DataTypes.INT()),
- explicit(DataTypes.STRING()),
- explicit(DataTypes.BOOLEAN())}))
- .inputTypes(DataTypes.INT(), DataTypes.STRING(), DataTypes.BOOLEAN(), DataTypes.BOOLEAN(), DataTypes.BOOLEAN())
- .expectSuccess(),
-
- // vararg sequence
- TestSpec
- .forValidator(
- varyingSequence(
- new String[]{"i", "s", "var"},
- new ArgumentTypeValidator[]{
- explicit(DataTypes.INT()),
- explicit(DataTypes.STRING()),
- explicit(DataTypes.BOOLEAN())}))
- .inputTypes(DataTypes.INT(), DataTypes.STRING())
- .expectSuccess(),
-
- // invalid vararg type
- TestSpec
- .forValidator(
- varyingSequence(
- new String[]{"i", "s", "var"},
- new ArgumentTypeValidator[]{
- explicit(DataTypes.INT()),
- explicit(DataTypes.STRING()),
- explicit(DataTypes.BOOLEAN())}))
- .inputTypes(DataTypes.INT(), DataTypes.STRING(), DataTypes.STRING())
- .expectErrorMessage("Invalid input arguments. Expected signatures are:\nf(i INT, s STRING, var BOOLEAN...)"),
-
- // invalid non-vararg type
- TestSpec
- .forValidator(
- varyingSequence(
- new String[]{"i", "s", "var"},
- new ArgumentTypeValidator[]{
- explicit(DataTypes.INT()),
- explicit(DataTypes.STRING()),
- explicit(DataTypes.BOOLEAN())}))
- .inputTypes(DataTypes.INT(), DataTypes.INT(), DataTypes.BOOLEAN())
- .expectErrorMessage("Unsupported argument type. Expected type 'STRING' but actual type was 'INT'."),
-
- // OR in vararg type
- TestSpec
- .forValidator(
- varyingSequence(
- new String[]{"i", "s", "var"},
- new ArgumentTypeValidator[]{
- explicit(DataTypes.INT()),
- explicit(DataTypes.STRING()),
- or(explicit(DataTypes.BOOLEAN()), explicit(DataTypes.INT()))}))
- .inputTypes(DataTypes.INT(), DataTypes.STRING(), DataTypes.INT(), DataTypes.BOOLEAN())
- .expectSuccess(),
-
- // invalid OR in vararg type
- TestSpec
- .forValidator(
- varyingSequence(
- new String[]{"i", "s", "var"},
- new ArgumentTypeValidator[]{
- explicit(DataTypes.INT()),
- explicit(DataTypes.STRING()),
- or(explicit(DataTypes.BOOLEAN()), explicit(DataTypes.INT()))}))
- .inputTypes(DataTypes.INT(), DataTypes.STRING(), DataTypes.STRING(), DataTypes.STRING())
- .expectErrorMessage("Invalid input arguments. Expected signatures are:\nf(i INT, s STRING, var [BOOLEAN | INT]...)")
- );
- }
-
- @Parameter
- public TestSpec testSpec;
-
- @Rule
- public ExpectedException thrown = ExpectedException.none();
-
- @Test
- public void testValidator() {
- if (testSpec.expectedErrorMessage != null) {
- thrown.expect(ValidationException.class);
- thrown.expectCause(containsCause(new ValidationException(testSpec.expectedErrorMessage)));
- }
- runTypeInference();
- }
-
- // --------------------------------------------------------------------------------------------
-
- private void runTypeInference() {
- final FunctionDefinitionMock functionDefinitionMock = new FunctionDefinitionMock();
- functionDefinitionMock.functionKind = FunctionKind.SCALAR;
- final CallContextMock callContextMock = new CallContextMock();
- callContextMock.functionDefinition = functionDefinitionMock;
- callContextMock.argumentDataTypes = testSpec.inputTypes;
- callContextMock.argumentLiterals = IntStream.range(0, testSpec.inputTypes.size())
- .mapToObj(i -> testSpec.literalPos != null && i == testSpec.literalPos)
- .collect(Collectors.toList());
- callContextMock.argumentNulls = IntStream.range(0, testSpec.inputTypes.size())
- .mapToObj(i -> false)
- .collect(Collectors.toList());
- callContextMock.name = "f";
-
- final TypeInference typeInference = TypeInference.newBuilder()
- .inputTypeValidator(testSpec.validator)
- .outputTypeStrategy(TypeStrategies.explicit(DataTypes.BOOLEAN()))
- .build();
- TypeInferenceUtil.runTypeInference(typeInference, callContextMock, null);
- }
-
- // --------------------------------------------------------------------------------------------
-
- private static class TestSpec {
-
- private final InputTypeValidator validator;
-
- private List<DataType> inputTypes;
-
- private @Nullable Integer literalPos;
-
- private @Nullable String expectedErrorMessage;
-
- private TestSpec(InputTypeValidator validator) {
- this.validator = validator;
- }
-
- static TestSpec forValidator(InputTypeValidator validator) {
- return new TestSpec(validator);
- }
-
- TestSpec inputTypes(DataType... dataTypes) {
- this.inputTypes = asList(dataTypes);
- return this;
- }
-
- TestSpec literalAt(int pos) {
- this.literalPos = pos;
- return this;
- }
-
- TestSpec expectSuccess() {
- return this;
- }
-
- TestSpec expectErrorMessage(String expectedErrorMessage) {
- this.expectedErrorMessage = expectedErrorMessage;
- return this;
- }
- }
-}
diff --git a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/TypeStrategiesTest.java b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/TypeStrategiesTest.java
index 64e7a11..07fdeff 100644
--- a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/TypeStrategiesTest.java
+++ b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/TypeStrategiesTest.java
@@ -40,6 +40,7 @@ import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Optional;
import static org.apache.flink.table.types.inference.TypeStrategies.MISSING;
import static org.apache.flink.table.types.inference.TypeStrategies.explicit;
@@ -69,19 +70,19 @@ public class TypeStrategiesTest {
// (INT, BOOLEAN) -> STRING
TestSpec
- .forStrategy(createMatchingTypeStrategy())
+ .forStrategy(createMappingTypeStrategy())
.inputTypes(DataTypes.INT(), DataTypes.BOOLEAN())
.expectDataType(DataTypes.STRING()),
// (INT, STRING) -> BOOLEAN
TestSpec
- .forStrategy(createMatchingTypeStrategy())
+ .forStrategy(createMappingTypeStrategy())
.inputTypes(DataTypes.INT(), DataTypes.STRING())
- .expectDataType(DataTypes.BOOLEAN()),
+ .expectDataType(DataTypes.BOOLEAN().bridgedTo(boolean.class)),
- // invalid matching strategy
+ // invalid mapping strategy
TestSpec
- .forStrategy(createMatchingTypeStrategy())
+ .forStrategy(createMappingTypeStrategy())
.inputTypes(DataTypes.INT(), DataTypes.INT())
.expectErrorMessage("Could not infer an output type for the given arguments."),
@@ -120,9 +121,10 @@ public class TypeStrategiesTest {
callContextMock.functionDefinition = functionDefinitionMock;
callContextMock.argumentDataTypes = testSpec.inputTypes;
callContextMock.name = "f";
+ callContextMock.outputDataType = Optional.empty();
final TypeInference typeInference = TypeInference.newBuilder()
- .inputTypeValidator(InputTypeValidators.PASSING)
+ .inputTypeStrategy(InputTypeStrategies.WILDCARD)
.outputTypeStrategy(testSpec.strategy)
.build();
return TypeInferenceUtil.runTypeInference(typeInference, callContextMock, null);
@@ -164,18 +166,18 @@ public class TypeStrategiesTest {
}
}
- private static TypeStrategy createMatchingTypeStrategy() {
- final Map<InputTypeValidator, TypeStrategy> matchers = new HashMap<>();
- matchers.put(
- InputTypeValidators.sequence(
- InputTypeValidators.explicit(DataTypes.INT()),
- InputTypeValidators.explicit(DataTypes.STRING())),
- TypeStrategies.explicit(DataTypes.BOOLEAN()));
- matchers.put(
- InputTypeValidators.sequence(
- InputTypeValidators.explicit(DataTypes.INT()),
- InputTypeValidators.explicit(DataTypes.BOOLEAN())),
+ private static TypeStrategy createMappingTypeStrategy() {
+ final Map<InputTypeStrategy, TypeStrategy> mappings = new HashMap<>();
+ mappings.put(
+ InputTypeStrategies.sequence(
+ InputTypeStrategies.explicit(DataTypes.INT()),
+ InputTypeStrategies.explicit(DataTypes.STRING())),
+ TypeStrategies.explicit(DataTypes.BOOLEAN().bridgedTo(boolean.class)));
+ mappings.put(
+ InputTypeStrategies.sequence(
+ InputTypeStrategies.explicit(DataTypes.INT()),
+ InputTypeStrategies.explicit(DataTypes.BOOLEAN())),
TypeStrategies.explicit(DataTypes.STRING()));
- return TypeStrategies.matching(matchers);
+ return TypeStrategies.mapping(mappings);
}
}
diff --git a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/utils/CallContextMock.java b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/utils/CallContextMock.java
index bea7cf8..0fb5635 100644
--- a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/utils/CallContextMock.java
+++ b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/utils/CallContextMock.java
@@ -18,6 +18,7 @@
package org.apache.flink.table.types.inference.utils;
+import org.apache.flink.table.catalog.DataTypeLookup;
import org.apache.flink.table.functions.FunctionDefinition;
import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.inference.CallContext;
@@ -30,6 +31,8 @@ import java.util.Optional;
*/
public class CallContextMock implements CallContext {
+ public DataTypeLookup lookup;
+
public List<DataType> argumentDataTypes;
public FunctionDefinition functionDefinition;
@@ -42,9 +45,11 @@ public class CallContextMock implements CallContext {
public String name;
+ public Optional<DataType> outputDataType;
+
@Override
- public List<DataType> getArgumentDataTypes() {
- return argumentDataTypes;
+ public DataTypeLookup getDataTypeLookup() {
+ return lookup;
}
@Override
@@ -72,4 +77,14 @@ public class CallContextMock implements CallContext {
public String getName() {
return name;
}
+
+ @Override
+ public List<DataType> getArgumentDataTypes() {
+ return argumentDataTypes;
+ }
+
+ @Override
+ public Optional<DataType> getOutputDataType() {
+ return outputDataType;
+ }
}