You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2020/06/03 13:41:26 UTC
[GitHub] [flink] twalthr commented on a change in pull request #12410: [FLINK-13782][table-api] Implement type strategies for IF ELSE expression
twalthr commented on a change in pull request #12410:
URL: https://github.com/apache/flink/pull/12410#discussion_r434557424
##########
File path: flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/SameArgumentsInputTypeStrategy.java
##########
@@ -23,37 +23,47 @@
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.InputTypeStrategy;
import org.apache.flink.table.types.inference.Signature;
+import org.apache.flink.table.types.logical.LegacyTypeInformationType;
import org.apache.flink.table.types.logical.LogicalType;
import org.apache.flink.table.types.logical.utils.LogicalTypeGeneralization;
import org.apache.flink.table.types.utils.TypeConversions;
+import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Optional;
import java.util.stream.Collectors;
+import java.util.stream.IntStream;
/**
- * {@link InputTypeStrategy} specific for {@link org.apache.flink.table.functions.BuiltInFunctionDefinitions#ARRAY}.
- *
- * <p>It expects at least one argument. All the arguments must have a common super type.
+ * An {@link InputTypeStrategy} that expects that all arguments have a common type.
*/
@Internal
-public class ArrayInputTypeStrategy implements InputTypeStrategy {
+public class SameArgumentsInputTypeStrategy implements InputTypeStrategy {
Review comment:
nit: call it `CommonInputTypeStrategy`?
##########
File path: flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/SameArgumentsInputTypeStrategy.java
##########
@@ -63,13 +73,40 @@ public ArgumentCount getArgumentCount() {
.collect(Collectors.toList())
);
+ if (!commonType.isPresent()) {
+ throw callContext.newValidationError("Could not find a common type for arguments: %s", argumentDataTypes);
+ }
+
return commonType.map(type -> Collections.nCopies(
argumentDataTypes.size(),
TypeConversions.fromLogicalToDataType(type)));
}
@Override
public List<Signature> getExpectedSignatures(FunctionDefinition definition) {
- return Collections.singletonList(Signature.of(Signature.Argument.of("*")));
+ Optional<Integer> minCount = argumentCount.getMinCount();
+ Optional<Integer> maxCount = argumentCount.getMaxCount();
+
+ int numberOfMandatoryArguments = 0;
Review comment:
nit: I'm a fan of immutable variables, can't we just do:
```
argumentCount.getMinCount().orElse(0)
```
##########
File path: flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/SubSequenceInputTypeStrategy.java
##########
@@ -0,0 +1,260 @@
+/*
+ * 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.annotation.PublicEvolving;
+import org.apache.flink.table.catalog.DataTypeFactory;
+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 javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+/**
+ * An {@link InputTypeStrategy} that lets you apply other strategies for subsequences of
+ * the actual arguments.
+ *
+ * <p>The {@link SequenceInputTypeStrategy} should be preferred in most of the cases. Use this strategy
+ * only if you need to apply a common logic to a subsequence of the arguments.
+ */
+@Internal
+public final class SubSequenceInputTypeStrategy implements InputTypeStrategy {
+
+ private final List<ArgumentsSplit> argumentsSplits;
+ private final ArgumentCount argumentCount;
+
+ private SubSequenceInputTypeStrategy(List<ArgumentsSplit> argumentsSplits, ArgumentCount argumentCount) {
+ this.argumentsSplits = argumentsSplits;
+ this.argumentCount = argumentCount;
+ }
+
+ @Override
+ public ArgumentCount getArgumentCount() {
+ return argumentCount;
+ }
+
+ @Override
+ public Optional<List<DataType>> inferInputTypes(
+ CallContext callContext,
+ boolean throwOnFailure) {
+
+ List<DataType> result = new ArrayList<>();
+
+ for (ArgumentsSplit argumentsSplit : argumentsSplits) {
+ Optional<List<DataType>> splitDataTypes = argumentsSplit.inputTypeStrategy
+ .inferInputTypes(new SplitCallContext(argumentsSplit, callContext), throwOnFailure);
+
+ if (splitDataTypes.isPresent()) {
+ result.addAll(splitDataTypes.get());
+ } else {
+ if (throwOnFailure) {
+ throw callContext.newValidationError(
+ "Could not infer arguments in range: [%d, %d].",
+ argumentsSplit.startIndex,
+ argumentsSplit.endIndex);
+ }
+ return Optional.empty();
+ }
+ }
+
+ return Optional.of(result);
+ }
+
+ @Override
+ public List<Signature> getExpectedSignatures(FunctionDefinition definition) {
+ return argumentsSplits.stream()
+ .map(computeSubSignatures(definition))
+ .reduce(this::crossJoin)
+ .get()
+ .stream().map(Signature::of)
+ .collect(Collectors.toList());
+ }
+
+ private Function<ArgumentsSplit, List<List<Signature.Argument>>> computeSubSignatures(
+ FunctionDefinition definition) {
+ return argumentsSplit -> argumentsSplit.inputTypeStrategy.getExpectedSignatures(definition)
+ .stream()
+ .map(Signature::getArguments)
+ .collect(Collectors.toList());
+ }
+
+ private List<List<Signature.Argument>> crossJoin(
+ List<List<Signature.Argument>> signatureStarts,
+ List<List<Signature.Argument>> signatureEnds) {
+ return signatureStarts.stream()
+ .flatMap(x ->
+ signatureEnds.stream().map(y -> {
+ ArrayList<Signature.Argument> joined = new ArrayList<>(x);
+ joined.addAll(y);
+ return joined;
+ }))
+ .collect(Collectors.toList());
+ }
+
+ private static final class ArgumentsSplit {
+ private final int startIndex;
+ private final @Nullable Integer endIndex;
+
+ private final InputTypeStrategy inputTypeStrategy;
+
+ public ArgumentsSplit(
+ int startIndex,
+ Integer endIndex,
+ InputTypeStrategy inputTypeStrategy) {
+ this.startIndex = startIndex;
+ this.endIndex = endIndex;
+ this.inputTypeStrategy = inputTypeStrategy;
+ }
+ }
+
+ private static final class SplitCallContext implements CallContext {
+
+ private final ArgumentsSplit split;
+ private final CallContext originalCallContext;
+
+ private SplitCallContext(
+ ArgumentsSplit split,
+ CallContext originalCallContext) {
+ this.split = split;
+ this.originalCallContext = originalCallContext;
+ }
+
+ @Override
+ public DataTypeFactory getDataTypeFactory() {
+ return originalCallContext.getDataTypeFactory();
+ }
+
+ @Override
+ public FunctionDefinition getFunctionDefinition() {
+ return originalCallContext.getFunctionDefinition();
+ }
+
+ @Override
+ public boolean isArgumentLiteral(int pos) {
+ return originalCallContext.isArgumentLiteral(pos + split.startIndex);
+ }
+
+ @Override
+ public boolean isArgumentNull(int pos) {
+ return originalCallContext.isArgumentLiteral(pos + split.startIndex);
+ }
+
+ @Override
+ public <T> Optional<T> getArgumentValue(int pos, Class<T> clazz) {
+ return originalCallContext.getArgumentValue(pos + split.startIndex, clazz);
+ }
+
+ @Override
+ public String getName() {
+ return originalCallContext.getName();
+ }
+
+ @Override
+ public List<DataType> getArgumentDataTypes() {
+ List<DataType> originalArgumentDataTypes = originalCallContext.getArgumentDataTypes();
+ return originalArgumentDataTypes.subList(
+ split.startIndex,
+ split.endIndex != null ? split.endIndex : originalArgumentDataTypes.size());
+ }
+
+ @Override
+ public Optional<DataType> getOutputDataType() {
+ return originalCallContext.getOutputDataType();
+ }
+ }
+
+ /**
+ * A Builder for {@link SubSequenceInputTypeStrategy}.
+ */
+ @PublicEvolving
+ public static class SubSequenceStrategyBuilder {
+ private final List<ArgumentsSplit> argumentsSplits = new ArrayList<>();
+ private int currentPos = 0;
+
+ /**
+ * Defines that we expect a single argument at the next position.
+ */
+ public SubSequenceStrategyBuilder argument(ArgumentTypeStrategy argumentTypeStrategy) {
+ SequenceInputTypeStrategy singleArgumentStrategy = new SequenceInputTypeStrategy(
+ Collections.singletonList(argumentTypeStrategy), null);
+ argumentsSplits.add(new ArgumentsSplit(currentPos, currentPos + 1, singleArgumentStrategy));
+ currentPos += 1;
+ return this;
+ }
+
+ /**
+ * Defines a common {@link InputTypeStrategy} for the next arguments. Given
+ * input strategy must expect a constant number of arguments. That means that both
+ * the minimum and maximum number of arguments must be defined and equal to each other.
+ *
+ * <p>If you need a varying logic use {@link #finishWithvarying(InputTypeStrategy)}.
+ */
+ public SubSequenceStrategyBuilder subSequence(InputTypeStrategy inputTypeStrategy) {
+ Optional<Integer> maxCount = inputTypeStrategy.getArgumentCount().getMaxCount();
Review comment:
add a precondition to check for `ConstantArgumentCount`
##########
File path: flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/SubSequenceInputTypeStrategyTest.java
##########
@@ -0,0 +1,105 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.types.inference;
+
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.types.inference.strategies.SubSequenceInputTypeStrategy;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
+
+import org.junit.runners.Parameterized;
+
+import java.util.List;
+
+import static java.util.Arrays.asList;
+import static org.apache.flink.table.types.inference.InputTypeStrategies.commonType;
+import static org.apache.flink.table.types.inference.InputTypeStrategies.logical;
+import static org.apache.flink.table.types.inference.InputTypeStrategies.varyingSequence;
+
+/**
+ * Tests for {@link SubSequenceInputTypeStrategy}
+ */
+public class SubSequenceInputTypeStrategyTest extends InputTypeStrategiesTestBase {
+ @Parameterized.Parameters(name = "{index}: {0}")
+ public static List<TestSpec> testData() {
+ return asList(
+ TestSpec
+ .forStrategy(
+ "A strategy used for IF ELSE with valid arguments",
+ InputTypeStrategies.startSequences()
+ .argument(logical(LogicalTypeRoot.BOOLEAN))
+ .subSequence(commonType(2))
+ .finish()
+ )
+ .calledWithArgumentTypes(
+ DataTypes.BOOLEAN(),
+ DataTypes.SMALLINT(),
+ DataTypes.DECIMAL(10, 2)
+ )
+ .expectSignature(
+ "f(<BOOLEAN>, <COMMON>, <COMMON>)")
+ .expectArgumentTypes(
+ DataTypes.BOOLEAN(),
+ DataTypes.DECIMAL(10, 2),
+ DataTypes.DECIMAL(10, 2)
+ ),
+
+ TestSpec
+ .forStrategy(
+ "Strategy fails if any of the nested strategies fail",
+ InputTypeStrategies.startSequences()
+ .argument(logical(LogicalTypeRoot.BOOLEAN))
+ .subSequence(commonType(2))
+ .finish()
+ )
+ .calledWithArgumentTypes(
+ DataTypes.BOOLEAN(),
+ DataTypes.VARCHAR(3),
+ DataTypes.DECIMAL(10, 2)
+ )
+ .expectErrorMessage("Could not find a common type for arguments: [VARCHAR(3), DECIMAL(10, 2)]"),
+
+ TestSpec
+ .forStrategy(
+ "Strategy with a varying argument",
+ InputTypeStrategies.startSequences()
+ .argument(logical(LogicalTypeRoot.BOOLEAN))
+ .subSequence(commonType(2))
Review comment:
can we have one more sophisticated test where another subsequence and another argument follows here?
##########
File path: flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/SubSequenceInputTypeStrategy.java
##########
@@ -0,0 +1,260 @@
+/*
+ * 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.annotation.PublicEvolving;
+import org.apache.flink.table.catalog.DataTypeFactory;
+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 javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+/**
+ * An {@link InputTypeStrategy} that lets you apply other strategies for subsequences of
+ * the actual arguments.
+ *
+ * <p>The {@link SequenceInputTypeStrategy} should be preferred in most of the cases. Use this strategy
+ * only if you need to apply a common logic to a subsequence of the arguments.
+ */
+@Internal
+public final class SubSequenceInputTypeStrategy implements InputTypeStrategy {
+
+ private final List<ArgumentsSplit> argumentsSplits;
+ private final ArgumentCount argumentCount;
+
+ private SubSequenceInputTypeStrategy(List<ArgumentsSplit> argumentsSplits, ArgumentCount argumentCount) {
Review comment:
use `ConstantArgumentCount`?
##########
File path: flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/SubSequenceInputTypeStrategy.java
##########
@@ -0,0 +1,260 @@
+/*
+ * 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.annotation.PublicEvolving;
+import org.apache.flink.table.catalog.DataTypeFactory;
+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 javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+/**
+ * An {@link InputTypeStrategy} that lets you apply other strategies for subsequences of
+ * the actual arguments.
+ *
+ * <p>The {@link SequenceInputTypeStrategy} should be preferred in most of the cases. Use this strategy
+ * only if you need to apply a common logic to a subsequence of the arguments.
+ */
+@Internal
+public final class SubSequenceInputTypeStrategy implements InputTypeStrategy {
+
+ private final List<ArgumentsSplit> argumentsSplits;
+ private final ArgumentCount argumentCount;
+
+ private SubSequenceInputTypeStrategy(List<ArgumentsSplit> argumentsSplits, ArgumentCount argumentCount) {
+ this.argumentsSplits = argumentsSplits;
+ this.argumentCount = argumentCount;
+ }
+
+ @Override
+ public ArgumentCount getArgumentCount() {
+ return argumentCount;
+ }
+
+ @Override
+ public Optional<List<DataType>> inferInputTypes(
+ CallContext callContext,
+ boolean throwOnFailure) {
+
+ List<DataType> result = new ArrayList<>();
+
+ for (ArgumentsSplit argumentsSplit : argumentsSplits) {
+ Optional<List<DataType>> splitDataTypes = argumentsSplit.inputTypeStrategy
+ .inferInputTypes(new SplitCallContext(argumentsSplit, callContext), throwOnFailure);
+
+ if (splitDataTypes.isPresent()) {
+ result.addAll(splitDataTypes.get());
+ } else {
+ if (throwOnFailure) {
+ throw callContext.newValidationError(
+ "Could not infer arguments in range: [%d, %d].",
+ argumentsSplit.startIndex,
+ argumentsSplit.endIndex);
+ }
+ return Optional.empty();
+ }
+ }
+
+ return Optional.of(result);
+ }
+
+ @Override
+ public List<Signature> getExpectedSignatures(FunctionDefinition definition) {
+ return argumentsSplits.stream()
+ .map(computeSubSignatures(definition))
+ .reduce(this::crossJoin)
+ .get()
+ .stream().map(Signature::of)
+ .collect(Collectors.toList());
+ }
+
+ private Function<ArgumentsSplit, List<List<Signature.Argument>>> computeSubSignatures(
+ FunctionDefinition definition) {
+ return argumentsSplit -> argumentsSplit.inputTypeStrategy.getExpectedSignatures(definition)
+ .stream()
+ .map(Signature::getArguments)
+ .collect(Collectors.toList());
+ }
+
+ private List<List<Signature.Argument>> crossJoin(
+ List<List<Signature.Argument>> signatureStarts,
+ List<List<Signature.Argument>> signatureEnds) {
+ return signatureStarts.stream()
+ .flatMap(x ->
+ signatureEnds.stream().map(y -> {
+ ArrayList<Signature.Argument> joined = new ArrayList<>(x);
+ joined.addAll(y);
+ return joined;
+ }))
+ .collect(Collectors.toList());
+ }
+
+ private static final class ArgumentsSplit {
+ private final int startIndex;
+ private final @Nullable Integer endIndex;
+
+ private final InputTypeStrategy inputTypeStrategy;
+
+ public ArgumentsSplit(
+ int startIndex,
+ Integer endIndex,
+ InputTypeStrategy inputTypeStrategy) {
+ this.startIndex = startIndex;
+ this.endIndex = endIndex;
+ this.inputTypeStrategy = inputTypeStrategy;
+ }
+ }
+
+ private static final class SplitCallContext implements CallContext {
+
+ private final ArgumentsSplit split;
+ private final CallContext originalCallContext;
+
+ private SplitCallContext(
+ ArgumentsSplit split,
+ CallContext originalCallContext) {
+ this.split = split;
+ this.originalCallContext = originalCallContext;
+ }
+
+ @Override
+ public DataTypeFactory getDataTypeFactory() {
+ return originalCallContext.getDataTypeFactory();
+ }
+
+ @Override
+ public FunctionDefinition getFunctionDefinition() {
+ return originalCallContext.getFunctionDefinition();
+ }
+
+ @Override
+ public boolean isArgumentLiteral(int pos) {
+ return originalCallContext.isArgumentLiteral(pos + split.startIndex);
+ }
+
+ @Override
+ public boolean isArgumentNull(int pos) {
+ return originalCallContext.isArgumentLiteral(pos + split.startIndex);
+ }
+
+ @Override
+ public <T> Optional<T> getArgumentValue(int pos, Class<T> clazz) {
+ return originalCallContext.getArgumentValue(pos + split.startIndex, clazz);
+ }
+
+ @Override
+ public String getName() {
+ return originalCallContext.getName();
+ }
+
+ @Override
+ public List<DataType> getArgumentDataTypes() {
+ List<DataType> originalArgumentDataTypes = originalCallContext.getArgumentDataTypes();
+ return originalArgumentDataTypes.subList(
+ split.startIndex,
+ split.endIndex != null ? split.endIndex : originalArgumentDataTypes.size());
+ }
+
+ @Override
+ public Optional<DataType> getOutputDataType() {
+ return originalCallContext.getOutputDataType();
+ }
+ }
+
+ /**
+ * A Builder for {@link SubSequenceInputTypeStrategy}.
+ */
+ @PublicEvolving
+ public static class SubSequenceStrategyBuilder {
+ private final List<ArgumentsSplit> argumentsSplits = new ArrayList<>();
+ private int currentPos = 0;
+
+ /**
+ * Defines that we expect a single argument at the next position.
+ */
+ public SubSequenceStrategyBuilder argument(ArgumentTypeStrategy argumentTypeStrategy) {
+ SequenceInputTypeStrategy singleArgumentStrategy = new SequenceInputTypeStrategy(
+ Collections.singletonList(argumentTypeStrategy), null);
+ argumentsSplits.add(new ArgumentsSplit(currentPos, currentPos + 1, singleArgumentStrategy));
+ currentPos += 1;
+ return this;
+ }
+
+ /**
+ * Defines a common {@link InputTypeStrategy} for the next arguments. Given
+ * input strategy must expect a constant number of arguments. That means that both
+ * the minimum and maximum number of arguments must be defined and equal to each other.
+ *
+ * <p>If you need a varying logic use {@link #finishWithvarying(InputTypeStrategy)}.
+ */
+ public SubSequenceStrategyBuilder subSequence(InputTypeStrategy inputTypeStrategy) {
+ Optional<Integer> maxCount = inputTypeStrategy.getArgumentCount().getMaxCount();
+ Optional<Integer> minCount = inputTypeStrategy.getArgumentCount().getMinCount();
+ if (!maxCount.isPresent() || !minCount.isPresent() || !maxCount.get().equals(minCount.get())) {
+ throw new IllegalArgumentException("Both the minimum and maximum number of expected arguments must" +
+ " be defined and equal to each other.");
+ }
+ argumentsSplits.add(new ArgumentsSplit(currentPos, currentPos + maxCount.get(), inputTypeStrategy));
+ currentPos += maxCount.get();
+ return this;
+ }
+
+ /**
+ * Defines a common {@link InputTypeStrategy} for the next arguments. Given
+ * input strategy must expect a varying number of arguments. That means that the
+ * maximum number of arguments must not be defined.
+ */
+ public InputTypeStrategy finishWithvarying(InputTypeStrategy inputTypeStrategy) {
Review comment:
`finishWithVarying `
##########
File path: flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/SameArgumentsInputTypeStrategy.java
##########
@@ -23,37 +23,47 @@
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.InputTypeStrategy;
import org.apache.flink.table.types.inference.Signature;
+import org.apache.flink.table.types.logical.LegacyTypeInformationType;
import org.apache.flink.table.types.logical.LogicalType;
import org.apache.flink.table.types.logical.utils.LogicalTypeGeneralization;
import org.apache.flink.table.types.utils.TypeConversions;
+import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Optional;
import java.util.stream.Collectors;
+import java.util.stream.IntStream;
/**
- * {@link InputTypeStrategy} specific for {@link org.apache.flink.table.functions.BuiltInFunctionDefinitions#ARRAY}.
- *
- * <p>It expects at least one argument. All the arguments must have a common super type.
+ * An {@link InputTypeStrategy} that expects that all arguments have a common type.
*/
@Internal
-public class ArrayInputTypeStrategy implements InputTypeStrategy {
+public class SameArgumentsInputTypeStrategy implements InputTypeStrategy {
+ private static final Signature.Argument COMMON_ARGUMENT = Signature.Argument.of("<COMMON>");
+ private final ArgumentCount argumentCount;
+
+ public SameArgumentsInputTypeStrategy(ArgumentCount argumentCount) {
+ this.argumentCount = argumentCount;
+ }
+
@Override
public ArgumentCount getArgumentCount() {
- return ConstantArgumentCount.from(1);
+ return argumentCount;
}
@Override
public Optional<List<DataType>> inferInputTypes(
- CallContext callContext,
- boolean throwOnFailure) {
+ CallContext callContext,
+ boolean throwOnFailure) {
List<DataType> argumentDataTypes = callContext.getArgumentDataTypes();
- if (argumentDataTypes.size() == 0) {
- return Optional.empty();
+
+ if (callContext.getArgumentDataTypes()
Review comment:
use the variable above
##########
File path: flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/SameArgumentsInputTypeStrategy.java
##########
@@ -23,37 +23,47 @@
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.InputTypeStrategy;
import org.apache.flink.table.types.inference.Signature;
+import org.apache.flink.table.types.logical.LegacyTypeInformationType;
import org.apache.flink.table.types.logical.LogicalType;
import org.apache.flink.table.types.logical.utils.LogicalTypeGeneralization;
import org.apache.flink.table.types.utils.TypeConversions;
+import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Optional;
import java.util.stream.Collectors;
+import java.util.stream.IntStream;
/**
- * {@link InputTypeStrategy} specific for {@link org.apache.flink.table.functions.BuiltInFunctionDefinitions#ARRAY}.
- *
- * <p>It expects at least one argument. All the arguments must have a common super type.
+ * An {@link InputTypeStrategy} that expects that all arguments have a common type.
*/
@Internal
-public class ArrayInputTypeStrategy implements InputTypeStrategy {
+public class SameArgumentsInputTypeStrategy implements InputTypeStrategy {
+ private static final Signature.Argument COMMON_ARGUMENT = Signature.Argument.of("<COMMON>");
+ private final ArgumentCount argumentCount;
+
+ public SameArgumentsInputTypeStrategy(ArgumentCount argumentCount) {
Review comment:
Use `ConstantArgumentCount`?
##########
File path: flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/InputTypeStrategies.java
##########
@@ -61,6 +62,17 @@
*/
public static final WildcardInputTypeStrategy WILDCARD = new WildcardInputTypeStrategy();
+ /**
+ * An strategy that lets you apply other strategies for subsequences of
+ * the actual arguments.
+ *
+ * <p>The {@link SequenceInputTypeStrategy} should be preferred in most of the cases. Use this strategy
Review comment:
nit: link to the static method instead
##########
File path: flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/SubSequenceInputTypeStrategy.java
##########
@@ -0,0 +1,260 @@
+/*
+ * 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.annotation.PublicEvolving;
+import org.apache.flink.table.catalog.DataTypeFactory;
+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 javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+/**
+ * An {@link InputTypeStrategy} that lets you apply other strategies for subsequences of
+ * the actual arguments.
+ *
+ * <p>The {@link SequenceInputTypeStrategy} should be preferred in most of the cases. Use this strategy
+ * only if you need to apply a common logic to a subsequence of the arguments.
+ */
+@Internal
+public final class SubSequenceInputTypeStrategy implements InputTypeStrategy {
+
+ private final List<ArgumentsSplit> argumentsSplits;
+ private final ArgumentCount argumentCount;
+
+ private SubSequenceInputTypeStrategy(List<ArgumentsSplit> argumentsSplits, ArgumentCount argumentCount) {
+ this.argumentsSplits = argumentsSplits;
+ this.argumentCount = argumentCount;
+ }
+
+ @Override
+ public ArgumentCount getArgumentCount() {
+ return argumentCount;
+ }
+
+ @Override
+ public Optional<List<DataType>> inferInputTypes(
+ CallContext callContext,
+ boolean throwOnFailure) {
+
+ List<DataType> result = new ArrayList<>();
+
+ for (ArgumentsSplit argumentsSplit : argumentsSplits) {
+ Optional<List<DataType>> splitDataTypes = argumentsSplit.inputTypeStrategy
+ .inferInputTypes(new SplitCallContext(argumentsSplit, callContext), throwOnFailure);
+
+ if (splitDataTypes.isPresent()) {
+ result.addAll(splitDataTypes.get());
+ } else {
+ if (throwOnFailure) {
+ throw callContext.newValidationError(
+ "Could not infer arguments in range: [%d, %d].",
+ argumentsSplit.startIndex,
+ argumentsSplit.endIndex);
+ }
+ return Optional.empty();
+ }
+ }
+
+ return Optional.of(result);
+ }
+
+ @Override
+ public List<Signature> getExpectedSignatures(FunctionDefinition definition) {
+ return argumentsSplits.stream()
+ .map(computeSubSignatures(definition))
+ .reduce(this::crossJoin)
+ .get()
+ .stream().map(Signature::of)
+ .collect(Collectors.toList());
+ }
+
+ private Function<ArgumentsSplit, List<List<Signature.Argument>>> computeSubSignatures(
+ FunctionDefinition definition) {
+ return argumentsSplit -> argumentsSplit.inputTypeStrategy.getExpectedSignatures(definition)
+ .stream()
+ .map(Signature::getArguments)
+ .collect(Collectors.toList());
+ }
+
+ private List<List<Signature.Argument>> crossJoin(
+ List<List<Signature.Argument>> signatureStarts,
+ List<List<Signature.Argument>> signatureEnds) {
+ return signatureStarts.stream()
+ .flatMap(x ->
+ signatureEnds.stream().map(y -> {
+ ArrayList<Signature.Argument> joined = new ArrayList<>(x);
+ joined.addAll(y);
+ return joined;
+ }))
+ .collect(Collectors.toList());
+ }
+
+ private static final class ArgumentsSplit {
+ private final int startIndex;
+ private final @Nullable Integer endIndex;
+
+ private final InputTypeStrategy inputTypeStrategy;
+
+ public ArgumentsSplit(
+ int startIndex,
+ Integer endIndex,
+ InputTypeStrategy inputTypeStrategy) {
+ this.startIndex = startIndex;
+ this.endIndex = endIndex;
+ this.inputTypeStrategy = inputTypeStrategy;
+ }
+ }
+
+ private static final class SplitCallContext implements CallContext {
+
+ private final ArgumentsSplit split;
+ private final CallContext originalCallContext;
+
+ private SplitCallContext(
+ ArgumentsSplit split,
+ CallContext originalCallContext) {
+ this.split = split;
+ this.originalCallContext = originalCallContext;
+ }
+
+ @Override
+ public DataTypeFactory getDataTypeFactory() {
+ return originalCallContext.getDataTypeFactory();
+ }
+
+ @Override
+ public FunctionDefinition getFunctionDefinition() {
+ return originalCallContext.getFunctionDefinition();
+ }
+
+ @Override
+ public boolean isArgumentLiteral(int pos) {
+ return originalCallContext.isArgumentLiteral(pos + split.startIndex);
+ }
+
+ @Override
+ public boolean isArgumentNull(int pos) {
+ return originalCallContext.isArgumentLiteral(pos + split.startIndex);
+ }
+
+ @Override
+ public <T> Optional<T> getArgumentValue(int pos, Class<T> clazz) {
+ return originalCallContext.getArgumentValue(pos + split.startIndex, clazz);
+ }
+
+ @Override
+ public String getName() {
+ return originalCallContext.getName();
+ }
+
+ @Override
+ public List<DataType> getArgumentDataTypes() {
+ List<DataType> originalArgumentDataTypes = originalCallContext.getArgumentDataTypes();
+ return originalArgumentDataTypes.subList(
+ split.startIndex,
+ split.endIndex != null ? split.endIndex : originalArgumentDataTypes.size());
+ }
+
+ @Override
+ public Optional<DataType> getOutputDataType() {
+ return originalCallContext.getOutputDataType();
+ }
+ }
+
+ /**
+ * A Builder for {@link SubSequenceInputTypeStrategy}.
+ */
+ @PublicEvolving
Review comment:
currently, all built-in strategies are internal
##########
File path: flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/InputTypeStrategies.java
##########
@@ -61,6 +62,17 @@
*/
public static final WildcardInputTypeStrategy WILDCARD = new WildcardInputTypeStrategy();
+ /**
+ * An strategy that lets you apply other strategies for subsequences of
+ * the actual arguments.
+ *
+ * <p>The {@link SequenceInputTypeStrategy} should be preferred in most of the cases. Use this strategy
+ * only if you need to apply a common logic to a subsequence of the arguments.
+ */
+ public static SubSequenceStrategyBuilder startSequences() {
Review comment:
shall we call it `patternSequence`? Because it reminds me of a regex or something similar. Or `compositeSequence`? The name should indicate that this is not the default goto method for sequences.
##########
File path: flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/expressions/ExpressionBuilder.java
##########
@@ -90,9 +91,8 @@ public static UnresolvedCallExpression isNull(Expression input) {
return call(IS_NULL, input);
}
- public static UnresolvedCallExpression ifThenElse(Expression condition, Expression ifTrue,
- Expression ifFalse) {
- return call(IF, condition, ifTrue, ifFalse);
+ public static UnresolvedCallExpression ifThenElse(Expression condition, Expression ifTrue, Expression ifFalse) {
+ return call(IF, call(IS_TRUE, condition), ifTrue, ifFalse);
Review comment:
could you elaborate on this change? So if the condition is NULL we will always execute the false branch?
##########
File path: flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/SameArgumentsInputTypeStrategy.java
##########
@@ -63,13 +73,40 @@ public ArgumentCount getArgumentCount() {
.collect(Collectors.toList())
);
+ if (!commonType.isPresent()) {
+ throw callContext.newValidationError("Could not find a common type for arguments: %s", argumentDataTypes);
Review comment:
only throw if `throwOnFailure` is true
##########
File path: flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/SameArgumentsInputTypeStrategy.java
##########
@@ -23,37 +23,47 @@
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.InputTypeStrategy;
import org.apache.flink.table.types.inference.Signature;
+import org.apache.flink.table.types.logical.LegacyTypeInformationType;
import org.apache.flink.table.types.logical.LogicalType;
import org.apache.flink.table.types.logical.utils.LogicalTypeGeneralization;
import org.apache.flink.table.types.utils.TypeConversions;
+import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Optional;
import java.util.stream.Collectors;
+import java.util.stream.IntStream;
/**
- * {@link InputTypeStrategy} specific for {@link org.apache.flink.table.functions.BuiltInFunctionDefinitions#ARRAY}.
- *
- * <p>It expects at least one argument. All the arguments must have a common super type.
+ * An {@link InputTypeStrategy} that expects that all arguments have a common type.
*/
@Internal
-public class ArrayInputTypeStrategy implements InputTypeStrategy {
+public class SameArgumentsInputTypeStrategy implements InputTypeStrategy {
+ private static final Signature.Argument COMMON_ARGUMENT = Signature.Argument.of("<COMMON>");
+ private final ArgumentCount argumentCount;
+
+ public SameArgumentsInputTypeStrategy(ArgumentCount argumentCount) {
+ this.argumentCount = argumentCount;
+ }
+
@Override
public ArgumentCount getArgumentCount() {
- return ConstantArgumentCount.from(1);
+ return argumentCount;
}
@Override
public Optional<List<DataType>> inferInputTypes(
- CallContext callContext,
- boolean throwOnFailure) {
+ CallContext callContext,
+ boolean throwOnFailure) {
List<DataType> argumentDataTypes = callContext.getArgumentDataTypes();
- if (argumentDataTypes.size() == 0) {
- return Optional.empty();
+
+ if (callContext.getArgumentDataTypes()
+ .stream()
+ .anyMatch(dataType -> dataType.getLogicalType() instanceof LegacyTypeInformationType)) {
Review comment:
convert to logical types at the beginning of this method to reuse them here and below
##########
File path: flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/SubSequenceInputTypeStrategy.java
##########
@@ -0,0 +1,260 @@
+/*
+ * 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.annotation.PublicEvolving;
+import org.apache.flink.table.catalog.DataTypeFactory;
+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 javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+/**
+ * An {@link InputTypeStrategy} that lets you apply other strategies for subsequences of
+ * the actual arguments.
+ *
+ * <p>The {@link SequenceInputTypeStrategy} should be preferred in most of the cases. Use this strategy
+ * only if you need to apply a common logic to a subsequence of the arguments.
+ */
+@Internal
+public final class SubSequenceInputTypeStrategy implements InputTypeStrategy {
+
+ private final List<ArgumentsSplit> argumentsSplits;
+ private final ArgumentCount argumentCount;
+
+ private SubSequenceInputTypeStrategy(List<ArgumentsSplit> argumentsSplits, ArgumentCount argumentCount) {
+ this.argumentsSplits = argumentsSplits;
Review comment:
nit: add some `Precondition.checkNotNull`
##########
File path: flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/SameArgumentsInputTypeStrategy.java
##########
@@ -63,13 +73,40 @@ public ArgumentCount getArgumentCount() {
.collect(Collectors.toList())
);
+ if (!commonType.isPresent()) {
+ throw callContext.newValidationError("Could not find a common type for arguments: %s", argumentDataTypes);
+ }
+
return commonType.map(type -> Collections.nCopies(
argumentDataTypes.size(),
TypeConversions.fromLogicalToDataType(type)));
}
@Override
public List<Signature> getExpectedSignatures(FunctionDefinition definition) {
- return Collections.singletonList(Signature.of(Signature.Argument.of("*")));
+ Optional<Integer> minCount = argumentCount.getMinCount();
+ Optional<Integer> maxCount = argumentCount.getMaxCount();
+
+ int numberOfMandatoryArguments = 0;
+ if (minCount.isPresent()) {
+ numberOfMandatoryArguments = minCount.get();
+ }
+
+ if (maxCount.isPresent()) {
+ List<Signature> signatures = new ArrayList<>();
+ IntStream.range(numberOfMandatoryArguments, maxCount.get() + 1)
Review comment:
nit: use mapToObj and collect?
##########
File path: flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/SubSequenceInputTypeStrategy.java
##########
@@ -0,0 +1,260 @@
+/*
+ * 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.annotation.PublicEvolving;
+import org.apache.flink.table.catalog.DataTypeFactory;
+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 javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+/**
+ * An {@link InputTypeStrategy} that lets you apply other strategies for subsequences of
+ * the actual arguments.
+ *
+ * <p>The {@link SequenceInputTypeStrategy} should be preferred in most of the cases. Use this strategy
+ * only if you need to apply a common logic to a subsequence of the arguments.
+ */
+@Internal
+public final class SubSequenceInputTypeStrategy implements InputTypeStrategy {
+
+ private final List<ArgumentsSplit> argumentsSplits;
+ private final ArgumentCount argumentCount;
+
+ private SubSequenceInputTypeStrategy(List<ArgumentsSplit> argumentsSplits, ArgumentCount argumentCount) {
+ this.argumentsSplits = argumentsSplits;
+ this.argumentCount = argumentCount;
+ }
+
+ @Override
+ public ArgumentCount getArgumentCount() {
+ return argumentCount;
+ }
+
+ @Override
+ public Optional<List<DataType>> inferInputTypes(
+ CallContext callContext,
+ boolean throwOnFailure) {
+
+ List<DataType> result = new ArrayList<>();
+
+ for (ArgumentsSplit argumentsSplit : argumentsSplits) {
+ Optional<List<DataType>> splitDataTypes = argumentsSplit.inputTypeStrategy
+ .inferInputTypes(new SplitCallContext(argumentsSplit, callContext), throwOnFailure);
+
+ if (splitDataTypes.isPresent()) {
+ result.addAll(splitDataTypes.get());
+ } else {
+ if (throwOnFailure) {
+ throw callContext.newValidationError(
+ "Could not infer arguments in range: [%d, %d].",
+ argumentsSplit.startIndex,
+ argumentsSplit.endIndex);
+ }
+ return Optional.empty();
+ }
+ }
+
+ return Optional.of(result);
+ }
+
+ @Override
+ public List<Signature> getExpectedSignatures(FunctionDefinition definition) {
+ return argumentsSplits.stream()
+ .map(computeSubSignatures(definition))
+ .reduce(this::crossJoin)
+ .get()
+ .stream().map(Signature::of)
+ .collect(Collectors.toList());
+ }
+
+ private Function<ArgumentsSplit, List<List<Signature.Argument>>> computeSubSignatures(
+ FunctionDefinition definition) {
+ return argumentsSplit -> argumentsSplit.inputTypeStrategy.getExpectedSignatures(definition)
+ .stream()
+ .map(Signature::getArguments)
+ .collect(Collectors.toList());
+ }
+
+ private List<List<Signature.Argument>> crossJoin(
+ List<List<Signature.Argument>> signatureStarts,
+ List<List<Signature.Argument>> signatureEnds) {
+ return signatureStarts.stream()
+ .flatMap(x ->
+ signatureEnds.stream().map(y -> {
+ ArrayList<Signature.Argument> joined = new ArrayList<>(x);
+ joined.addAll(y);
+ return joined;
+ }))
+ .collect(Collectors.toList());
+ }
+
+ private static final class ArgumentsSplit {
+ private final int startIndex;
+ private final @Nullable Integer endIndex;
+
+ private final InputTypeStrategy inputTypeStrategy;
+
+ public ArgumentsSplit(
+ int startIndex,
+ Integer endIndex,
+ InputTypeStrategy inputTypeStrategy) {
+ this.startIndex = startIndex;
+ this.endIndex = endIndex;
+ this.inputTypeStrategy = inputTypeStrategy;
+ }
+ }
+
+ private static final class SplitCallContext implements CallContext {
+
+ private final ArgumentsSplit split;
+ private final CallContext originalCallContext;
+
+ private SplitCallContext(
+ ArgumentsSplit split,
+ CallContext originalCallContext) {
+ this.split = split;
+ this.originalCallContext = originalCallContext;
+ }
+
+ @Override
+ public DataTypeFactory getDataTypeFactory() {
+ return originalCallContext.getDataTypeFactory();
+ }
+
+ @Override
+ public FunctionDefinition getFunctionDefinition() {
+ return originalCallContext.getFunctionDefinition();
+ }
+
+ @Override
+ public boolean isArgumentLiteral(int pos) {
+ return originalCallContext.isArgumentLiteral(pos + split.startIndex);
+ }
+
+ @Override
+ public boolean isArgumentNull(int pos) {
+ return originalCallContext.isArgumentLiteral(pos + split.startIndex);
+ }
+
+ @Override
+ public <T> Optional<T> getArgumentValue(int pos, Class<T> clazz) {
+ return originalCallContext.getArgumentValue(pos + split.startIndex, clazz);
+ }
+
+ @Override
+ public String getName() {
+ return originalCallContext.getName();
+ }
+
+ @Override
+ public List<DataType> getArgumentDataTypes() {
+ List<DataType> originalArgumentDataTypes = originalCallContext.getArgumentDataTypes();
+ return originalArgumentDataTypes.subList(
+ split.startIndex,
+ split.endIndex != null ? split.endIndex : originalArgumentDataTypes.size());
+ }
+
+ @Override
+ public Optional<DataType> getOutputDataType() {
+ return originalCallContext.getOutputDataType();
+ }
+ }
+
+ /**
+ * A Builder for {@link SubSequenceInputTypeStrategy}.
+ */
+ @PublicEvolving
+ public static class SubSequenceStrategyBuilder {
+ private final List<ArgumentsSplit> argumentsSplits = new ArrayList<>();
+ private int currentPos = 0;
+
+ /**
+ * Defines that we expect a single argument at the next position.
+ */
+ public SubSequenceStrategyBuilder argument(ArgumentTypeStrategy argumentTypeStrategy) {
+ SequenceInputTypeStrategy singleArgumentStrategy = new SequenceInputTypeStrategy(
+ Collections.singletonList(argumentTypeStrategy), null);
+ argumentsSplits.add(new ArgumentsSplit(currentPos, currentPos + 1, singleArgumentStrategy));
+ currentPos += 1;
+ return this;
+ }
+
+ /**
+ * Defines a common {@link InputTypeStrategy} for the next arguments. Given
+ * input strategy must expect a constant number of arguments. That means that both
+ * the minimum and maximum number of arguments must be defined and equal to each other.
+ *
+ * <p>If you need a varying logic use {@link #finishWithvarying(InputTypeStrategy)}.
+ */
+ public SubSequenceStrategyBuilder subSequence(InputTypeStrategy inputTypeStrategy) {
+ Optional<Integer> maxCount = inputTypeStrategy.getArgumentCount().getMaxCount();
+ Optional<Integer> minCount = inputTypeStrategy.getArgumentCount().getMinCount();
+ if (!maxCount.isPresent() || !minCount.isPresent() || !maxCount.get().equals(minCount.get())) {
+ throw new IllegalArgumentException("Both the minimum and maximum number of expected arguments must" +
+ " be defined and equal to each other.");
+ }
+ argumentsSplits.add(new ArgumentsSplit(currentPos, currentPos + maxCount.get(), inputTypeStrategy));
+ currentPos += maxCount.get();
+ return this;
+ }
+
+ /**
+ * Defines a common {@link InputTypeStrategy} for the next arguments. Given
+ * input strategy must expect a varying number of arguments. That means that the
+ * maximum number of arguments must not be defined.
+ */
+ public InputTypeStrategy finishWithvarying(InputTypeStrategy inputTypeStrategy) {
+ ArgumentCount strategyArgumentCount = inputTypeStrategy.getArgumentCount();
+ Optional<Integer> maxCount = strategyArgumentCount.getMaxCount();
+ if (maxCount.isPresent()) {
+ throw new IllegalArgumentException("The maximum number of arguments must not be defined.");
+ }
+ argumentsSplits.add(new ArgumentsSplit(currentPos, null, inputTypeStrategy));
+ int minCount = currentPos;
+ Optional<Integer> strategyMinCount = strategyArgumentCount.getMinCount();
+ if (strategyMinCount.isPresent()) {
Review comment:
same comment as bevor, why don't we use the fluent `.orElse` instead of `.isPresent()`.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org